import os import json import asyncio import sys import paramiko sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) from utils.utils import map_csv_file, not_import_client_csv from mapping.customers import ( customerContact, companyData, companyRepresentative, customerAdresses, companyGroup ) from general.customers.customer_class import ShopifyCustomer from api.customers import create_customer, update_customer, get_all_customers # Fonction utilitaire pour télécharger un fichier depuis le SFTP def download_sftp_file(sftp, remote_folder, file_name, local_folder) -> str: local_path = os.path.join(local_folder, file_name) remote_path = os.path.join(remote_folder, file_name) sftp.get(remote_path, local_path) return local_path # Build a ShopifyCustomer object from contact data def build_shopify_contact(contact: dict) -> ShopifyCustomer: shopify_contact = ShopifyCustomer( firstname=contact.get("contactFirstName", ""), lastname=contact.get("contactLastName", ""), email=contact.get("contactEmail", ""), phone=contact.get("contactPhone", ""), note="", tax_exempt=False ) if contact.get("contactId"): shopify_contact.add_tag(f"Identifiant Contact : {contact.get('contactId')}") if contact.get("companyId"): shopify_contact.add_tag(f"Identifiant Company : {contact.get('companyId')}") if contact.get("contactCivility"): shopify_contact.add_metafield("custom", "civilite", contact.get("contactCivility")) if contact.get("contactFirstName"): shopify_contact.add_metafield("custom", "first_name", contact.get("contactFirstName")) if contact.get("contactFunction"): shopify_contact.add_metafield("custom", "fonction", contact.get("contactFunction")) if contact.get("contactService"): shopify_contact.add_metafield("custom", "service", contact.get("contactService")) return shopify_contact # Return company info for a given contact def return_company_info_for_contact(contact: dict, contact_mapping, companies_csv: str) -> dict: companies = map_csv_file(companies_csv, companyData) company_id = contact.get("companyId") if not company_id: return None for comp in companies: if comp.get("companyId") and comp.get("companyId").strip() == company_id.strip(): return comp return None # Add company info and related tags/metafields to the Shopify customer def add_company_info_to_shopify_customer(shopify_contact: ShopifyCustomer, merged_contact: dict): if merged_contact.get("companyType"): tag = f"Type de l'entreprise: {merged_contact.get('companyType')}" if tag not in shopify_contact.tags: shopify_contact.add_tag(tag) if merged_contact.get("companyReference"): tag = f"Ref unique de l'entreprise: {merged_contact.get('companyReference')}" if tag not in shopify_contact.tags: shopify_contact.add_tag(tag) if merged_contact.get("companyCategoryCode"): tag = f"Code catégorie de l'entreprise: {merged_contact.get('companyCategoryCode')}" if tag not in shopify_contact.tags: shopify_contact.add_tag(tag) if merged_contact.get("companyName"): shopify_contact.add_addresses( company=merged_contact.get("companyName"), name=merged_contact.get("companyName"), firstname="", lastname="", address1=merged_contact.get("companyAdress", ""), address2=merged_contact.get("companyAdressComplement", ""), address3="", zip_code=merged_contact.get("companyZipCode", ""), city=merged_contact.get("companyCity", ""), country=merged_contact.get("companyCountry", ""), phone=merged_contact.get("companyPhone", ""), default=True ) if merged_contact.get("companyMobile"): tag = f"Deuxième téléphone: {merged_contact.get('companyMobile')}" if tag not in shopify_contact.tags: shopify_contact.add_tag(tag) if merged_contact.get("companyTvaNumber"): shopify_contact.add_metafield("custom", "tva", merged_contact.get("companyTvaNumber")) if merged_contact.get("companyPaymentMode"): shopify_contact.add_metafield("custom", "mode_de_paiement", merged_contact.get("companyPaymentMode")) if merged_contact.get("companyPaymentDelay"): shopify_contact.add_metafield("custom", "delai_de_paiement", merged_contact.get("companyPaymentDelay")) if merged_contact.get("companyGPV"): shopify_contact.add_metafield("custom", "gpv", merged_contact.get("companyGPV")) if merged_contact.get("companyGroupId"): shopify_contact.add_metafield("custom", "id_groupe_client", merged_contact.get("companyGroupId")) if merged_contact.get("companySiret"): shopify_contact.add_metafield("custom", "siret", merged_contact.get("companySiret")) if merged_contact.get("companyShippingMode"): shopify_contact.add_metafield("custom", "mode_expedition", merged_contact.get("companyShippingMode")) notes = "" if merged_contact.get("companyNotes"): notes += merged_contact.get("companyNotes") if merged_contact.get("companyPrintableNotes"): notes += " " + merged_contact.get("companyPrintableNotes") if notes and not shopify_contact.note: shopify_contact.note = notes.strip() if merged_contact.get("companyCurrency"): shopify_contact.add_metafield("custom", "currency", merged_contact.get("companyCurrency")) if merged_contact.get("companyState"): tag = f"Status: {merged_contact.get('companyState')}" if tag not in shopify_contact.tags: shopify_contact.add_tag(tag) # Add representative tag based on company info def add_representative_tag(shopify_contact: ShopifyCustomer, company_id: str, rep_csv: str): reps = map_csv_file(rep_csv, companyRepresentative) for rep in reps: rep_company_id = rep.get("companyId") if rep_company_id and company_id and rep_company_id.strip() == company_id.strip(): rep_name = rep.get("representativeLastName") if rep_name: tag = f"Responsable : {rep_name.strip()}" if tag not in shopify_contact.tags: shopify_contact.add_tag(tag) break # Add additional address (non default) if contact id matches address CSV def add_customer_addresses(shopify_contact: ShopifyCustomer, merged_contact: dict, addr_csv: str): addresses = map_csv_file(addr_csv, customerAdresses) contact_id = merged_contact.get("contactId", "").strip() for addr in addresses: addr_contact_id = addr.get("contactId", "").strip() if addr_contact_id == contact_id: shopify_contact.add_addresses( company=merged_contact.get("companyName", ""), name=addr.get("recipientName", ""), firstname="", lastname="", address1=addr.get("adress", ""), address2=addr.get("adressComplement", ""), address3="", zip_code=addr.get("zipCode", ""), city=addr.get("city", ""), country=addr.get("country", ""), phone="", default=False ) # Add group tag if a matching group is found in the CSV def add_group_tag(shopify_contact: ShopifyCustomer, group_id: str, group_csv: str): groups = map_csv_file(group_csv, companyGroup) for grp in groups: grp_group_id = grp.get("companyId") if grp_group_id and group_id and grp_group_id.strip() == group_id.strip(): group_ref = grp.get("companyReference") if group_ref and group_ref.strip() != "": tag = f"Identifiant du groupe : {grp_group_id.strip()}" if tag not in shopify_contact.tags: shopify_contact.add_tag(tag) break # Create a mapping of existing Shopify customers by company ID def get_customer_id_(all_customers, contact_id): for customer in all_customers: tags = customer.get("tags", "") if isinstance(tags, str): tags = [tag.strip() for tag in tags.split(",")] for tag in tags: if tag.startswith("Identifiant Contact :"): comp_id = tag.split(":", 1)[1].strip() if comp_id == contact_id: return customer.get("id") return None async def main(): # Détails de connexion SFTP host = "emde.autarcia.com" port = 22 username = "ftp-shoppingfeed" password = "5vm_*2I[f2yL2A6J!/dE" remote_folder = "/exports/CSV/exp_swing" # Répertoire local pour stocker les fichiers clients base_dir = os.path.dirname(os.path.abspath(__file__)) project_dir = os.path.abspath(os.path.join(base_dir, '..')) local_customers_folder = os.path.join(project_dir, 'files', 'customers') os.makedirs(local_customers_folder, exist_ok=True) # Établir la connexion SFTP et télécharger les fichiers CSV nécessaires transport = paramiko.Transport((host, port)) try: transport.connect(username=username, password=password) sftp = paramiko.SFTPClient.from_transport(transport) companies_csv = download_sftp_file(sftp, remote_folder, "SW_DATA_CUSTOMER_CLI_2680.CSV", local_customers_folder) rep_csv = download_sftp_file(sftp, remote_folder, "SW_DATA_CUSTOMER_REP_2679.CSV", local_customers_folder) addr_csv = download_sftp_file(sftp, remote_folder, "SW_DATA_CUSTOMERADDR_2677.CSV", local_customers_folder) group_csv = download_sftp_file(sftp, remote_folder, "SW_GT_CUSTOMERGROUP_2669.CSV", local_customers_folder) contacts_csv = download_sftp_file(sftp, remote_folder, "SW_DATA_CUSTOMERCONT_2678.CSV", local_customers_folder) finally: sftp.close() transport.close() token_index = 0 all_customers = await get_all_customers(token_index=token_index) contacts = map_csv_file(contacts_csv, customerContact) if not contacts: print("Aucun contact trouvé dans le fichier CSV.") return contacts_with_email = [contact for contact in contacts if contact.get("contactEmail", "").strip()] contacts_without_email = [contact for contact in contacts if not contact.get("contactEmail", "").strip()] print(f"Nombre de contacts avec email : {len(contacts_with_email)}") print(f"Nombre de contacts sans email (ignorés) : {len(contacts_without_email)}") if contacts_without_email: source_filename = os.path.basename(contacts_csv) not_import_client_csv(contacts_without_email, source_filename) email_contacts = {} contacts_with_email_count = 0 contacts_without_email_count = 0 for contact in contacts: email = contact.get("contactEmail") if email and email.strip(): contacts_with_email_count += 1 if email not in email_contacts: email_contacts[email] = [] email_contacts[email].append(contact) else: contacts_without_email_count += 1 unique_email_contacts = [] for email, contacts_group in email_contacts.items(): commercial_found = False for contact in contacts_group: if contact.get("contactType", "").lower() == "commercial": unique_email_contacts.append(contact) commercial_found = True break if not commercial_found and contacts_group: unique_email_contacts.append(contacts_group[0]) all_contacts_to_process = unique_email_contacts print(f"Nombre total de contacts à traiter : {len(all_contacts_to_process)}") commercial_contacts = [] other_contacts = [] for contact in all_contacts_to_process: if contact.get("contactType", "").lower() == "commercial": commercial_contacts.append(contact) else: other_contacts.append(contact) print(f"Nombre de contacts commerciaux : {len(commercial_contacts)}") print(f"Nombre d'autres contacts : {len(other_contacts)}") # Traitement des contacts commerciaux for contact in commercial_contacts: company_info = return_company_info_for_contact(contact, customerContact, companies_csv) merged_contact = contact.copy() if company_info: for key, value in company_info.items(): if key not in merged_contact or merged_contact[key] in [None, ""]: merged_contact[key] = value shopify_customer = build_shopify_contact(merged_contact) if company_info: add_company_info_to_shopify_customer(shopify_customer, merged_contact) add_representative_tag(shopify_customer, company_info.get("companyId"), rep_csv) add_group_tag(shopify_customer, merged_contact.get("companyGroupId"), group_csv) add_customer_addresses(shopify_customer, merged_contact, addr_csv) customer_data = shopify_customer.create_array() contact_id = merged_contact.get("contactId") gpv = merged_contact.get("companyGPV") if gpv: shopify_customer.add_tag(f"code_tarif_{gpv}") existing_customer_id = get_customer_id_(all_customers, contact_id) if existing_customer_id: customer_data["customer"]["id"] = existing_customer_id result = await update_customer(existing_customer_id, customer_data, token_index=token_index) if result: print(f"Commercial {contact['contactId']} mis à jour PRIORITAIREMENT") else: print(f"La mise à jour du commercial {contact['contactId']} a échoué.") else: result = await create_customer(customer_data, token_index=token_index) if result and "customer" in result: print(f"Commercial {contact['contactId']} créé PRIORITAIREMENT") else: print(f"La création du commercial {contact['contactId']} a échoué.") # Traitement des autres contacts for contact in other_contacts: company_info = return_company_info_for_contact(contact, customerContact, companies_csv) merged_contact = contact.copy() if company_info: for key, value in company_info.items(): if key not in merged_contact or merged_contact[key] in [None, ""]: merged_contact[key] = value shopify_customer = build_shopify_contact(merged_contact) if company_info: add_company_info_to_shopify_customer(shopify_customer, merged_contact) add_representative_tag(shopify_customer, company_info.get("companyId"), rep_csv) add_group_tag(shopify_customer, merged_contact.get("companyGroupId"), group_csv) add_customer_addresses(shopify_customer, merged_contact, addr_csv) customer_data = shopify_customer.create_array() contact_id = merged_contact.get("contactId") gpv = merged_contact.get("companyGPV") if gpv: shopify_customer.add_tag(f"code_tarif_{gpv}") existing_customer_id = get_customer_id_(all_customers, contact_id) if existing_customer_id: customer_data["customer"]["id"] = existing_customer_id result = await update_customer(existing_customer_id, customer_data, token_index=token_index) if result: print(f"Client {contact['contactId']} mis à jour") else: print(f"La mise à jour du client {contact['contactId']} a échoué.") else: result = await create_customer(customer_data, token_index=token_index) if result and "customer" in result: print(f"Client {contact['contactId']} créé") else: print(f"La création du client {contact['contactId']} a échoué.") print("Traitement terminé.") print("Tous les clients ont été traités avec succès.") if __name__ == "__main__": asyncio.run(main())