import os import sys import json import asyncio import paramiko # Ajouter le dossier parent au PYTHONPATH # ─── Add the project root (one level up) *first* to sys.path ─── SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) PROJECT_ROOT = os.path.abspath(os.path.join(SCRIPT_DIR, '..')) sys.path.insert(0, PROJECT_ROOT) 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_without_variations_map, product_your_ref_map ) from mapping.collectionsERP import product_to_collections_map, collections_map from api.products import create_shopify_product, update_shopify_product, get_all_products, delete_shopify_product 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 # Limite de 8 appels simultanés (2 appels/s * 8 tokens = 16 appels/s max) 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')) # Assurer l'existence des dossiers locaux os.makedirs(mapping_dir, exist_ok=True) os.makedirs(collections_mapping_dir, exist_ok=True) # Détails de connexion 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 CSV sur le serveur SFTP 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" products_parents_filename = "SW_DATA_PRODUCT2_2696.CSV" product_without_variations_filename = "SW_DATA_PRODUCT3_2698.CSV" products_details_filename = "EXTRANET_CATALOGUE_3000.CSV" collections_filename = "DATA_RUBRIQ_ID_SHOP_2941.CSV" products_collections_join_filename = "DATA_RUBRIQUE_PRD_SH_2940.CSV" product_your_ref_filename = "SW_REFERENCING2_2661.CSV" # Chemins locaux vers les fichiers CSV (initialisés avec des valeurs par défaut) 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) products_parents_file = os.path.join(mapping_dir, products_parents_filename) product_without_variations_file = os.path.join(mapping_dir, product_without_variations_filename) products_details_file = os.path.join(mapping_dir, products_details_filename) product_your_ref_file = os.path.join(mapping_dir, product_your_ref_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, products_parents_file, product_without_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) products_parents_file = download_sftp_file(REMOTE_FOLDER_SWING, products_parents_filename, mapping_dir) product_without_variations_file = download_sftp_file(REMOTE_FOLDER_SWING, product_without_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) product_your_ref_file = download_sftp_file(REMOTE_FOLDER_SWING, product_your_ref_filename, mapping_dir) # ------------------------------------------------------------------------- # Le reste du code de traitement des produits reste inchangé # ------------------------------------------------------------------------- async def process_single_product(product, session, token_index, all_shopify_products, semaphore, quanti_typer_product, products_pricing, product_stocking, product_contains_variant, products_details, collections, product_collection_join,product_your_ref_file): async with semaphore: # Récupération des champs principaux product_reference = product.get('product_reference') product_id = product.get('product_id') product_end_of_life = product.get('product_end_of_life') title = product.get('product_designation') product_ean = product.get('product_ean') 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_details = extract_data_by_label('product_id', product_id, products_details) product_pricing_data = extract_datas_by_label('product_id', product_id, products_pricing) product_your_ref_data = extract_datas_by_label('product_id', product_id, product_your_ref_file) references = [] if product_your_ref_data: references = list(set( item['product_reference'] for item in product_your_ref_data if item.get('product_reference') )) product_stock_data = extract_data_by_label('product_id', product_id, product_stocking) quanti_typer_products = extract_data_by_label('product_id', product_id, quanti_typer_product) product_collection_join_data = extract_datas_by_label('product_id', product_id, product_collection_join) # Traitement du prix de base product_base_price = 0.0 if product_pricing_data: for product_pricing in product_pricing_data: if product_pricing.get('product_GPV') == "PPC": base_price_str = product_pricing.get('product_base_price') try: product_base_price = float(base_price_str.replace(',', '.')) if base_price_str else 0.0 except Exception: pass # Vérification du statut du produit vendor = None product_stock = product_stock_data.get('product_stock') if product_stock_data else 0 end_of_life_stock = product_stock_data.get('end_of_life') product_next_stock_date = product_stock_data.get('product_next_stock_date') if product_stock_data else "" image_keys = ['url1', 'url2', 'url3', 'url4'] description = None weight = None image_service = ImageService() vendor = None weight = None description = None material = None height = None width = None thickness = None if product_details: vendor = product_details.get('brand') weight = product_details.get('weight') description = product_details.get('detailed_description') material = product_details.get('mat_pat_value') height = product_details.get('hau_pat_value') width = product_details.get('lar_pat_value') thickness = product_details.get('epa_pat_value') # Télécharger les images et les ajouter via ImageService images = {key: download_image(product_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 product_to_push = Product(title, description, vendor, '') if quanti_typer_products: try: quanti_typer_products = int(quanti_typer_products.get('quantity_per_buy')) except Exception: quanti_typer_products = 0 if quanti_typer_products != 0: product_to_push.add_metafield("custom", "vendu_par", quanti_typer_products, 'number_integer') if product_next_stock_date: product_to_push.add_metafield("custom", "next_available", product_next_stock_date, 'single_line_text_field') if thickness: product_to_push.add_metafield("custom", "thickness", thickness, 'single_line_text_field') if product_id: product_to_push.add_metafield("custom", "product_id", product_id, 'single_line_text_field') if material: product_to_push.add_metafield("custom", "material", material, 'single_line_text_field') if height: product_to_push.add_metafield("custom", "height", height, 'single_line_text_field') if width: product_to_push.add_metafield("custom", "width", width, 'single_line_text_field') if product_stat_code2: product_to_push.add_metafield("custom", "product_stat_code2", product_stat_code2, 'single_line_text_field') if product_stat_code4: product_to_push.add_metafield("custom", "product_stat_code4", product_stat_code4, 'single_line_text_field') if product_end_of_life == 1: inventory_policy = "deny" else: inventory_policy = "continue" variant = { "option1": "Default", "sku": product_reference, "price": product_base_price, "inventory_management": "shopify", "inventory_policy": inventory_policy, "inventory_quantity": product_stock, "weight": weight, "barcode": product_ean, "metafields": [ { "namespace": "custom", "key": "variant_id", "value": product_id, "type": "single_line_text_field", } ] } product_to_push.add_variant(variant) tags = TagService() product_id_tag = 'Product_id: ' + product_id tags.add_tag(product_id_tag) for reference in references: tags.add_tag(reference) # 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, product_id_tag) if founded_product is None and product_base_price > 0.0 and status != '1': print(f"Création du produit avec le SKU {product_id}") # await create_shopify_product(session, product_data_formatted, token_index) else: if founded_product is None: return founded_product_id = founded_product.get('id') if product_base_price == 0.0 or status == '1': print(f"[DEBUG] Suppression ? base_price: {product_base_price}, status: {status}") responce = await delete_shopify_product(session, founded_product_id, token_index) print(f"[DEBUG] Résultat suppression : {responce}") # else: # print(f"Mise à jour du produit avec le SKU {product_id}") # await update_shopify_product(session, founded_product_id, product_data_formatted, token_index) async def process_products(): semaphore = asyncio.Semaphore(MAX_CONCURRENT_TASKS) all_shopify_products = await get_all_products() # Charger les fichiers CSV à partir des chemins locaux (téléchargés via SFTP) products_without_variations = map_csv_file(product_without_variations_file, product_without_variations_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) product_contains_variant = map_csv_file(product_contains_variant_file, product_contains_variant_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) product_your_ref_join = map_csv_file(product_your_ref_file, product_your_ref_map) import aiohttp async with aiohttp.ClientSession() as session: tasks = [] for idx, product in enumerate(products_without_variations): 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, product_contains_variant, products_details, collections, product_collection_join,product_your_ref_join )) await asyncio.gather(*tasks) async def main(): # download_all_files() await process_products() if __name__ == "__main__": asyncio.run(main())