import csv from proteus import config, Model # XML-RPC Configuration HTTPS = 'https://' SERVER_URL = 'itsa.open-squared.tech' DATABASE_NAME = 'tradon' USERNAME = 'admin' PASSWORD = 'dsproject' # CSV Configuration CSV_FILE_PATH = r'C:\Users\SylvainDUVERNAY\Open Squared\Production - Documents\TRADON Implementation\ITSA\Reference Data\Loaders\Parties.csv' # Default values DEFAULT_COUNTRY = 'US' # Default country code if not specified def connect_to_tryton(): """Establish connection to Tryton via XML-RPC""" print(f"Connecting to Tryton server: {SERVER_URL}") print(f"Database: {DATABASE_NAME}") print(f"Username: {USERNAME}") try: config.set_xmlrpc(f'{HTTPS}{USERNAME}:{PASSWORD}@{SERVER_URL}/{DATABASE_NAME}/') print("✓ Connected successfully!\n") return True except Exception as e: print(f"✗ Connection failed: {e}") print("\nTroubleshooting:") print(" - Verify the server URL is correct and accessible") print(" - Check that the Tryton server is running") print(" - Verify username and password are correct") print(" - Make sure you can access the server in a browser") return False def get_country(country_code): """Find country by code""" Country = Model.get('country.country') if not country_code: country_code = DEFAULT_COUNTRY countries = Country.find([('code', '=', country_code.upper())]) if countries: return countries[0] else: print(f" ⚠ Warning: Country '{country_code}' not found, using '{DEFAULT_COUNTRY}'") default_countries = Country.find([('code', '=', DEFAULT_COUNTRY)]) if default_countries: return default_countries[0] # Get first available country as last resort all_countries = Country.find([]) if all_countries: print(f" ⚠ Using first available country: {all_countries[0].name}") return all_countries[0] raise ValueError("No countries found in database!") def get_subdivision(country, subdivision_code): """Find country subdivision (state/province) by code""" if not subdivision_code: return None Subdivision = Model.get('country.subdivision') # Search for subdivision with matching code and country subdivisions = Subdivision.find([ ('code', '=', f"{country.code}-{subdivision_code}"), ('country', '=', country.id) ]) if subdivisions: return subdivisions[0] # Try without country prefix subdivisions = Subdivision.find([ ('code', 'ilike', f"%{subdivision_code}"), ('country', '=', country.id) ]) if subdivisions: return subdivisions[0] print(f" ⚠ Warning: Subdivision '{subdivision_code}' not found for country {country.code}") return None def check_party_exists_by_name(name): """Check if party with given name already exists""" Party = Model.get('party.party') parties = Party.find([('name', '=', name)]) return parties[0] if parties else None def create_party_with_addresses(row): """Create a new party with address(es) using proteus""" Party = Model.get('party.party') Address = Model.get('party.address') # Create party - let Tryton auto-generate the code party = Party() party.name = row['name'] if row.get('tax_identifier'): party.tax_identifier = row['tax_identifier'] if row.get('vat_code'): party.vat_code = row['vat_code'] # Save the party FIRST (without addresses) party.save() # Check if we have meaningful address data # Require at least street OR city to be present (not empty) has_street = bool(row.get('street')) has_city = bool(row.get('city')) has_postal_code = bool(row.get('postal_code')) has_country = bool(row.get('country_code')) # Create address only if we have at least street OR city if has_street or has_city: address = Address() # Link to the party we just created address.party = party if row.get('address_name'): address.name = row['address_name'] if has_street: address.street = row['street'] if has_city: address.city = row['city'] # Use postal_code instead of zip if has_postal_code: address.postal_code = row['postal_code'] # Get country if has_country: country_code = row['country_code'] country = get_country(country_code) else: country = get_country(DEFAULT_COUNTRY) address.country = country # Get subdivision (state/province) if provided if row.get('subdivision_code'): subdivision = get_subdivision(country, row['subdivision_code']) if subdivision: address.subdivision = subdivision # Save the address separately address.save() # Clean up any empty addresses that might have been auto-created # Reload party to get fresh data party = Party(party.id) # Find and delete empty addresses addresses_to_delete = [] for addr in party.addresses: # Consider an address empty if it has no street, city, or postal_code is_empty = ( (not addr.street or not addr.street.strip()) and (not addr.city or not addr.city.strip()) and (not addr.postal_code or not addr.postal_code.strip()) ) if is_empty: addresses_to_delete.append(addr) # Delete empty addresses if addresses_to_delete: Address.delete(addresses_to_delete) print(f" ℹ Cleaned up {len(addresses_to_delete)} empty address(es)") # Reload party one more time to return clean data party = Party(party.id) return party def import_parties(csv_file): """Import parties from CSV file""" imported_count = 0 skipped_count = 0 error_count = 0 errors = [] # Track names we've already processed in this run processed_names = set() print(f"{'='*70}") print(f"Importing parties from: {csv_file}") print(f"{'='*70}\n") try: # Open with utf-8-sig to handle BOM with open(csv_file, 'r', encoding='utf-8-sig') as file: reader = csv.DictReader(file) # Debug: Show detected columns print(f"Detected columns: {reader.fieldnames}\n") for row_num, row in enumerate(reader, start=2): try: # Clean up values name = row.get('name', '').strip() tax_identifier = row.get('tax_identifier', '').strip() vat_code = row.get('vat_code', '').strip() # Address fields address_name = row.get('address_name', '').strip() street = row.get('street', '').strip() city = row.get('city', '').strip() # Handle both 'zip' and 'postal_code' column names postal_code = row.get('postal_code', '').strip() or row.get('zip', '').strip() country_code = row.get('country_code', '').strip() subdivision_code = row.get('subdivision_code', '').strip() # Skip empty rows if not name: continue # Skip if postal_code is 'NULL' or '0' if postal_code and postal_code.upper() in ['NULL', '0']: postal_code = '' print(f"Processing Row {row_num}: {name}") # Check if we've already processed this name in this import run if name in processed_names: print(f" ⚠ Duplicate name in CSV: '{name}'") print(f" Skipping duplicate entry...\n") skipped_count += 1 continue # Check if party already exists in database existing_party = check_party_exists_by_name(name) if existing_party: print(f" ⚠ Party '{name}' already exists with code: {existing_party.code}") print(f" Skipping...\n") skipped_count += 1 processed_names.add(name) continue # Create the party with address row_data = { 'name': name, 'tax_identifier': tax_identifier, 'vat_code': vat_code, 'address_name': address_name, 'street': street, 'city': city, 'postal_code': postal_code, 'country_code': country_code, 'subdivision_code': subdivision_code } party = create_party_with_addresses(row_data) # Mark this name as processed processed_names.add(name) print(f" ✓ Created party") print(f" Party ID: {party.id}") print(f" Auto-generated Code: {party.code}") print(f" Name: {name}") if tax_identifier: print(f" Tax Identifier: {tax_identifier}") if vat_code: print(f" VAT Code: {vat_code}") if party.addresses: print(f" Addresses: {len(party.addresses)}") for addr in party.addresses: addr_street = (addr.street[:50] + '...') if addr.street and len(addr.street) > 50 else (addr.street or 'N/A') addr_city = addr.city if addr.city else 'N/A' addr_postal = addr.postal_code if addr.postal_code else 'N/A' print(f" - {addr_street}") print(f" {addr_city}, {addr_postal}") else: print(f" Addresses: 0 (no address data provided)") print() imported_count += 1 except Exception as e: error_msg = f"Row {row_num} - {name}: {str(e)}" errors.append(error_msg) error_count += 1 print(f"✗ Error on row {row_num}: {e}\n") import traceback traceback.print_exc() # Summary print(f"{'='*70}") print("IMPORT SUMMARY") print(f"{'='*70}") print(f"Successfully imported: {imported_count} parties") print(f"Skipped (already exist or duplicates): {skipped_count} parties") print(f"Errors: {error_count}") if errors: print(f"\nError details:") for error in errors: print(f" - {error}") print(f"\n{'='*70}") except FileNotFoundError: print(f"✗ Error: CSV file not found at {csv_file}") print(f"Please update CSV_FILE_PATH in the script with the correct path.") except Exception as e: print(f"✗ Fatal error: {e}") import traceback traceback.print_exc() def verify_import(): """Verify imported parties""" Party = Model.get('party.party') print(f"\n{'='*70}") print("VERIFICATION - Parties") print(f"{'='*70}\n") # Find all parties (or limit to recently created ones) parties = Party.find([], order=[('id', 'DESC')]) if parties: print(f"Found {len(parties)} parties (showing last 20):\n") print(f"{'Code':<15} {'Name':<40} {'Addresses':<10}") print("-" * 70) for party in parties[:20]: # Show last 20 created code = party.code or 'N/A' name = party.name[:39] if party.name else 'N/A' addr_count = len(party.addresses) if party.addresses else 0 print(f"{code:<15} {name:<40} {addr_count:<10}") else: print("No parties found") print() def list_available_countries(): """List all available countries""" print(f"\n{'='*70}") print("AVAILABLE COUNTRIES (first 20)") print(f"{'='*70}\n") Country = Model.get('country.country') countries = Country.find([]) if countries: print(f"Found {len(countries)} countries:\n") for country in countries[:20]: # Show first 20 print(f" - {country.code}: {country.name}") if len(countries) > 20: print(f" ... and {len(countries) - 20} more") else: print("No countries found") print() def main(): print("="*70) print("TRYTON PARTY IMPORT SCRIPT") print("Using Proteus with XML-RPC Connection") print("Party codes will be auto-generated by Tryton") print("="*70) print() # Connect to Tryton using XML-RPC if not connect_to_tryton(): return 1 # Optional: List available countries # Uncomment if you want to see what's available in your database # list_available_countries() # Import parties import_parties(CSV_FILE_PATH) # Verify import verify_import() return 0 if __name__ == '__main__': exit(main())