import os import sys import asyncio import paramiko import aiohttp import re # Ajouter le dossier parent au PYTHONPATH sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) from utils.utils import ( map_csv_file, find_product_by_id, extract_data_by_label, download_image, extract_datas_by_label ) from mapping.products import ( quantity_per_product_map, products_pricing_map, product_stock_map, product_contains_variant_map, products_details_map, products_parents_map, product_variant_details_map, ) from mapping.collectionsERP import product_to_collections_map, collections_map from api.products import create_shopify_product, update_shopify_product, get_all_products,add_linked_products_metafields from general.products.product import Product from general.products.product_generation_service import ProductGenerationService from general.products.image_service import ImageService from general.products.tag_service import TagService MAX_CONCURRENT_TASKS = 8 # Définir le chemin de base pour les fichiers de mapping base_dir = os.path.dirname(__file__) mapping_dir = os.path.abspath(os.path.join(base_dir, '..', 'files', 'products')) collections_mapping_dir = os.path.abspath(os.path.join(base_dir, '..', 'files', 'collections')) os.makedirs(mapping_dir, exist_ok=True) os.makedirs(collections_mapping_dir, exist_ok=True) # Limite de 8 appels simultanés (2 appels/s * 8 tokens = 16 appels/s max) # Détails SFTP SFTP_HOST = "emde.autarcia.com" SFTP_PORT = 22 SFTP_USERNAME = "ftp-shoppingfeed" SFTP_PASSWORD = "5vm_*2I[f2yL2A6J!/dE" REMOTE_FOLDER_SWING = "/exports/CSV/exp_swing" REMOTE_FOLDER_SHOPIFY = "/exports/CSV/shopify" def download_sftp_file(remote_folder: str, file_name: str, local_folder: str) -> str: local_path = os.path.join(local_folder, file_name) transport = paramiko.Transport((SFTP_HOST, SFTP_PORT)) try: transport.connect(username=SFTP_USERNAME, password=SFTP_PASSWORD) sftp = paramiko.SFTPClient.from_transport(transport) remote_path = os.path.join(remote_folder, file_name) sftp.get(remote_path, local_path) print(f"Téléchargement de {remote_path} réussi dans {local_path}") finally: sftp.close() transport.close() return local_path # Noms des fichiers à télécharger pour les mappings produits quanti_typer_product_filename = "SW_DATA_PCB_2667.CSV" products_pricing_filename = "SW_DATA_SELLSCONDITI_2671.CSV" product_stock_filename = "SW_PRODUCT_STOCK_2701.CSV" product_contains_variant_filename = "SW_DATA_PRODUCT1_2694.CSV" product_variants_details_filename = "SW_DATA_PDTVAL DECLI_2668.CSV" products_parents_filename = "SW_DATA_PRODUCT2_2696.CSV" product_with_variations_filename = "SW_DATA_PRODUCT1_2694.CSV" product_variations_filename = "SW_DATA_PRODUCT1_2694.CSV" products_details_filename = "EXTRANET_CATALOGUE_3000.CSV" # Noms des fichiers à télécharger pour les mappings collections collections_filename = "DATA_RUBRIQ_ID_SHOP_2941.CSV" products_collections_join_filename = "DATA_RUBRIQUE_PRD_SH_2940.CSV" # Variables globales qui contiendront les chemins locaux vers les fichiers téléchargés quanti_typer_product_file = os.path.join(mapping_dir, quanti_typer_product_filename) products_pricing_file = os.path.join(mapping_dir, products_pricing_filename) product_stock_file = os.path.join(mapping_dir, product_stock_filename) product_contains_variant_file = os.path.join(mapping_dir, product_contains_variant_filename) product_variants_details_file = os.path.join(mapping_dir, product_variants_details_filename) products_parents_file = os.path.join(mapping_dir, products_parents_filename) product_with_variations_file = os.path.join(mapping_dir, product_with_variations_filename) product_variations_file = os.path.join(mapping_dir, product_variations_filename) products_details_file = os.path.join(mapping_dir, products_details_filename) collections_file = os.path.join(collections_mapping_dir, collections_filename) products_collections_join_file = os.path.join(collections_mapping_dir, products_collections_join_filename) def download_all_files(): global quanti_typer_product_file, products_pricing_file, product_stock_file, product_contains_variant_file, product_variants_details_file, products_parents_file, product_with_variations_file, product_variations_file, products_details_file, collections_file, products_collections_join_file quanti_typer_product_file = download_sftp_file(REMOTE_FOLDER_SWING, quanti_typer_product_filename, mapping_dir) products_pricing_file = download_sftp_file(REMOTE_FOLDER_SWING, products_pricing_filename, mapping_dir) product_stock_file = download_sftp_file(REMOTE_FOLDER_SWING, product_stock_filename, mapping_dir) product_contains_variant_file = download_sftp_file(REMOTE_FOLDER_SWING, product_contains_variant_filename, mapping_dir) product_variants_details_file = download_sftp_file(REMOTE_FOLDER_SWING, product_variants_details_filename, mapping_dir) products_parents_file = download_sftp_file(REMOTE_FOLDER_SWING, products_parents_filename, mapping_dir) product_with_variations_file = download_sftp_file(REMOTE_FOLDER_SWING, product_with_variations_filename, mapping_dir) product_variations_file = download_sftp_file(REMOTE_FOLDER_SWING, product_variations_filename, mapping_dir) products_details_file = download_sftp_file(REMOTE_FOLDER_SHOPIFY, products_details_filename, mapping_dir) collections_file = download_sftp_file(REMOTE_FOLDER_SHOPIFY, collections_filename, collections_mapping_dir) products_collections_join_file = download_sftp_file(REMOTE_FOLDER_SHOPIFY, products_collections_join_filename, collections_mapping_dir) async def process_single_product(product, session, token_index, all_shopify_products, semaphore, quanti_typer_product, products_pricing, product_stocking, products_details, collections, product_collection_join, products_variants, products_variants_details): async with semaphore: product_reference = product.get('product_reference') product_id = product.get('product_id') title = product.get('product_designation') status = product.get('product_blocked') product_made_in = product.get('product_made_in') product_stat_code2 = product.get('product_stat_code2') product_stat_code4 = product.get('product_stat_code4') # Extraction des données complémentaires product_with_variant_details = extract_data_by_label('product_id', product_id, products_details) product_collection_join_data = extract_datas_by_label('product_id', product_id, product_collection_join) product_variants = extract_datas_by_label('product_root_id', product_id, products_variants) is_more_than_100 = False if not product_variants: return elif len(product_variants) <= 1: return elif len(product_variants) >= 100: is_more_than_100 = True if status == '1': return # Ignorer les produits bloqués vendor = None image_keys = ['url1', 'url2', 'url3', 'url4'] image_service = ImageService() vendor = None weight = None description = None if product_with_variant_details: vendor = product_with_variant_details.get('brand') description = product_with_variant_details.get('detailed_description') # Télécharger les images et les ajouter via ImageService images = {key: download_image(product_with_variant_details.get(key), '../images') for key in image_keys} for key in image_keys: if images.get(key): image_service.add_image( f'https://vps.theplaceb2b.com/sites/EMDE/images/{images[key]}', product_reference ) # Création de l'instance du produit à envoyer à Shopify if not is_more_than_100: product_to_push = Product(title, description, vendor, '') product_to_push.add_option('Taille') for product_variant in product_variants: variant_id = product_variant.get('product_id') variant_reference = product_variant.get('product_reference') variant_ean = product_variant.get('product_ean') variant_stock_data = extract_data_by_label('product_id', variant_id, product_stocking) variant_pricing_data = extract_datas_by_label('product_id', variant_id, products_pricing) quanti_typer_variant = extract_data_by_label('product_id', variant_id, quanti_typer_product) variant_details = extract_data_by_label('product_id', variant_id, products_variants_details) variant_3000_details = extract_data_by_label('product_id', variant_id, products_details) variant_stock = variant_stock_data.get('product_stock') if variant_stock_data else 0 variant_next_stock_date = variant_stock_data.get('product_next_stock_date') if variant_stock_data else "" key_color = variant_details.get('key_color') # Exemple : "166" ou "001" size = variant_details.get('size') color = variant_details.get('color') print(quanti_typer_variant) if quanti_typer_variant and quanti_typer_variant != '1': print(variant_id + "========" + quanti_typer_variant) try: quanti_typer_variant = int(quanti_typer_variant) except Exception: quanti_typer_variant = 0 if variant_pricing_data: for variant_pricing in variant_pricing_data: if variant_pricing.get('product_GPV') == "PPC": base_price_str = variant_pricing.get('product_base_price') try: variant_base_price = float(base_price_str.replace(',', '.')) if base_price_str else 0.0 except Exception: return else: return if color: if '-' in color: color = color.split('-')[-1].strip() elif color.startswith(key_color): color = color[len(key_color):].strip() else: color = color.strip() if variant_base_price == 0.0: continue if variant_3000_details: weight = variant_3000_details.get('kgs_pat_value') variant = { "option1": size, "sku": variant_reference, "price": variant_base_price, "inventory_management": "shopify", "inventory_policy": "continue", "inventory_quantity": variant_stock, "weight": weight, "barcode": variant_ean, } if color != "Sans couleur": product_to_push.add_option('Couleur') variant["option2"] = color product_to_push.add_variant(variant) # Récupérer l'indice de la dernière variante ajoutée variant_index = len(product_to_push.variants) - 1 # Ajouter un metafield à cette variante if variant_next_stock_date: product_to_push.add_variant_metafield(variant_index, "custom", "next_available", variant_next_stock_date, 'single_line_text_field') if quanti_typer_variant: try: quanti_typer_variant = int(quanti_typer_variant) except Exception: quanti_typer_variant = 0 product_to_push.add_variant_metafield(variant_index, "custom", "vendu_par", quanti_typer_variant, 'single_line_text_field') if variant_id: product_to_push.add_variant_metafield(variant_index, "custom", "variant_id", variant_id, 'single_line_text_field') tags = TagService() parent_id_tag = 'parent_id: ' + product_id tags.add_tag(parent_id_tag) tags.add_tag('customisable') product_to_push.add_metafield("custom", "product_id", product_id, 'single_line_text_field') # Traitement des collections if product_collection_join_data: for product_collection in product_collection_join_data: collection_id = product_collection.get('collection_id') collection_data = extract_data_by_label('collection_id', collection_id, collections) if collection_data: if isinstance(collection_data, list): collection_data = collection_data[0] collection_title = collection_data.get('collection_title') if collection_title: tags.add_tag('Collection: ' + collection_title) if product_made_in: tags.add_tag('CODE_REMISE: ' + product_made_in) # Génération des données formatées pour Shopify product_service = ProductGenerationService(product_to_push, image_service, tags) product_data_formatted = product_service.get_formatted_product_data() founded_product = find_product_by_id(all_shopify_products, parent_id_tag) if founded_product is None: print(f"Création du produit avec le SKU {product_reference}") await create_shopify_product(session, product_data_formatted, token_index) else: founded_product_id = founded_product.get("id") print(f"Mise à jour du produit avec le SKU {product_reference}") await update_shopify_product(session, founded_product_id, product_data_formatted, token_index) else: return variants_by_size = {} products_ids = [] for product_variant in product_variants: variant_id = product_variant.get('product_id') variant_details = extract_data_by_label('product_id', variant_id, products_variants_details) raw_size = variant_details.get('size', '') or '' size = re.sub(r'\s+', '', raw_size) # On s'assure qu'une taille est bien présente if not size: continue # Regrouper selon la taille if size not in variants_by_size: variants_by_size[size] = [] variants_by_size[size].append(product_variant) # Pour chaque taille (groupe de variantes), créer un produit distinct for size, variants in variants_by_size.items(): # Création du produit pour cette taille product_to_push = Product(title+'-'+size, description, vendor, '') # On ajoute l'option "Taille" pour indiquer qu'on a branché ce filtre product_to_push.add_option('Taille') # Stocker la taille dans un metafield pour le produit product_to_push.add_metafield("custom", "taille", size, 'single_line_text_field') # Préparer les tags et les liens avec le produit parent tags = TagService() parent_id_tag = 'parent_id: ' + product_id size_specific_tag = product_id + "_" + size size_specific_tag = 'splited_id: ' + size_specific_tag tags.add_tag(size_specific_tag) # On ajoute le tag commun pour pouvoir relier tous les produits issus du même produit initial tags.add_tag(parent_id_tag) tags.add_tag('SPLITED') tags.add_tag(f'size_{size}') # Stocker l'id du produit initial dans les metafields product_to_push.add_metafield("custom", "product_id", product_id, 'single_line_text_field') # Traiter chacune des variantes de ce groupe (pour la taille "size") for product_variant in variants: variant_id = product_variant.get('product_id') variant_reference = product_variant.get('product_reference') variant_ean = product_variant.get('product_ean') variant_stock_data = extract_data_by_label('product_id', variant_id, product_stocking) variant_pricing_data = extract_datas_by_label('product_id', variant_id, products_pricing) quanti_typer_variant = extract_data_by_label('product_id', variant_id, quanti_typer_product) variant_details = extract_data_by_label('product_id', variant_id, products_variants_details) variant_3000_details = extract_data_by_label('product_id', variant_id, products_details) variant_stock = variant_stock_data.get('product_stock') if variant_stock_data else 0 variant_next_stock_date = variant_stock_data.get('product_next_stock_date') if variant_stock_data else "" key_color = variant_details.get('key_color') # Par exemple "166" ou "001" color = variant_details.get('color') # Conversion de la quantité if quanti_typer_variant: try: quanti_typer_variant = int(quanti_typer_variant) except Exception: quanti_typer_variant = 0 # Récupération du prix if variant_pricing_data: variant_base_price = 0.0 for variant_pricing in variant_pricing_data: if variant_pricing.get('product_GPV') == "PPC": base_price_str = variant_pricing.get('product_base_price') try: variant_base_price = float(base_price_str.replace(',', '.')) if base_price_str else 0.0 except Exception: continue # Passe cette variante si le prix est invalide else: continue # On saute la variante si aucune donnée de prix n'est présente # Nettoyage de la couleur (si renseignée) if color: if '-' in color: color = color.split('-')[-1].strip() elif color.startswith(key_color): color = color[len(key_color):].strip() else: color = color.strip() if variant_base_price == 0.0: continue # On saute les variantes sans prix # Poids (si disponible) weight = 0.0 if variant_3000_details: weight = variant_3000_details.get('kgs_pat_value', 0.0) # Constitution du dictionnaire de la variante variant = { "sku": variant_reference, "price": variant_base_price, "inventory_management": "shopify", "inventory_policy": "continue", "inventory_quantity": variant_stock, "weight": weight, "barcode": variant_ean, } # Ajout de l'option "Couleur" si nécessaire if color and color != "Sans couleur": product_to_push.add_option('Couleur') variant["option2"] = color # Ajouter la variante au produit product_to_push.add_variant(variant) # Récupération de l'indice de la dernière variante ajoutée variant_index = len(product_to_push.variants) - 1 # Ajout des métachamps pour cette variante (date de réapprovisionnement, etc.) if variant_next_stock_date: product_to_push.add_variant_metafield(variant_index, "custom", "next_available", variant_next_stock_date, 'single_line_text_field') if quanti_typer_variant: try: quanti_typer_variant = int(quanti_typer_variant) except Exception: quanti_typer_variant = 0 product_to_push.add_variant_metafield(variant_index, "custom", "vendu_par", quanti_typer_variant, 'single_line_text_field') if variant_id: product_to_push.add_variant_metafield(variant_index, "custom", "variant_id", variant_id, 'single_line_text_field') # Traitement des collections et autres tags if product_collection_join_data: for product_collection in product_collection_join_data: collection_id = product_collection.get('collection_id') collection_data = extract_data_by_label('collection_id', collection_id, collections) if collection_data: if isinstance(collection_data, list): collection_data = collection_data[0] collection_title = collection_data.get('collection_title') if collection_title: tags.add_tag('Collection: ' + collection_title) if product_made_in: tags.add_tag('CODE_REMISE: ' + product_made_in) # Génération des données formattées pour Shopify product_service = ProductGenerationService(product_to_push, image_service, tags) product_data_formatted = product_service.get_formatted_product_data() # Pour lier ces produits entre eux, on peut par exemple composer un tag spécifique # Vérifier si le produit existe déjà sur Shopify (en fonction du tag ou d'un autre identifiant) founded_product = find_product_by_id(all_shopify_products, size_specific_tag) if founded_product is None: created_product = await create_shopify_product(session, product_data_formatted, token_index) shopify_id = created_product['product']['id'] if shopify_id: products_ids.append(shopify_id) else: founded_product_id = founded_product.get("id") print(f"Mise à jour du produit pour la taille {size}") updated_product = await update_shopify_product(session, founded_product_id, product_data_formatted, token_index) shopify_id = updated_product['product']['id'] products_ids.append(shopify_id) await add_linked_products_metafields(session,products_ids,token_index) async def process_products(): semaphore = asyncio.Semaphore(MAX_CONCURRENT_TASKS) all_shopify_products = await get_all_products() # Charger les fichiers CSV (les chemins ont été mis à jour après téléchargement SFTP) products_parents = map_csv_file(products_parents_file, products_parents_map) quanti_typer_product = map_csv_file(quanti_typer_product_file, quantity_per_product_map) products_pricing = map_csv_file(products_pricing_file, products_pricing_map) product_stocking = map_csv_file(product_stock_file, product_stock_map) products_variants = map_csv_file(product_contains_variant_file, product_contains_variant_map) products_variants_details = map_csv_file(product_variants_details_file, product_variant_details_map) products_details = map_csv_file(products_details_file, products_details_map) collections = map_csv_file(collections_file, collections_map) product_collection_join = map_csv_file(products_collections_join_file, product_to_collections_map) async with aiohttp.ClientSession() as session: tasks = [] for idx, product in enumerate(products_parents): token_index = idx % 8 # Distribution cyclique des tokens tasks.append(process_single_product(product, session, token_index, all_shopify_products, semaphore, quanti_typer_product, products_pricing, product_stocking, products_details, collections, product_collection_join, products_variants, products_variants_details)) await asyncio.gather(*tasks) async def main(): download_all_files() await process_products() if __name__ == "__main__": asyncio.run(main()) sys.exit(0)