import sys import os import csv import json import asyncio import aiohttp from datetime import datetime # Ajout du chemin pour les imports personnalisés sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../..'))) from Harken.General.Product.product import Product from Harken.General.Product.image_service import ImageService from Harken.General.Product.tag_service import TagService from Harken.General.Product.product_generation_service import ProductGenerationService from Harken.API.tokens import access_tokens from Harken.API.product_create import create_shopify_product from Harken.API.product_update import update_shopify_product from Harken.API.dlete_product import delete_shopify_product from Harken.API.get_all_products import get_all_products from Harken.files.CSV_Converter.txt_to_csv import convert_txt_to_csv from Harken.General.Utils.product_utils import get_image_link, get_related_products, get_product_price from Harken.FTP.Utils import download_huggy_tarifs_base_txt, download_and_merge_huggii_products_files_txt # Configuration shop_name = 'https://harkenb2b.myshopify.com/' file_path_csv = '../files/Products/ProductsOutput.csv' file_path_json = '../Products/products.json' prices_file_path_csv = '../files/Tarifs/huggii_tarifs_base.csv' semaphore = asyncio.Semaphore(9) def get_product_by_sku(data, sku): matching_products = [] for product in data.get('products', []): for variant in product.get('variants', []): if variant.get('sku') == sku: matching_products.append(product) break # On passe au produit suivant après avoir trouvé un variant correspondant if not matching_products: return None # Trier les produits par 'updated_at' du plus récent au plus ancien matching_products.sort(key=lambda p: datetime.fromisoformat(p.get('updated_at', '1970-01-01T00:00:00Z')), reverse=True) # Garder uniquement le produit le plus récent et supprimer les autres de data['products'] latest_product = matching_products[0] # for product in matching_products: if product: if product.get('updated_at') == latest_product.get('updated_at'): data['products'] = [p for p in data['products'] if p not in matching_products or p == latest_product] return latest_product def read_csv_file(file_path): mapping = { "Code article": "item_code", "Code Famille": "family_code", "Désignation principale": "main_designation", "Actif": "status", "Catégorie": "category", "Sous-catégories": "subcategories", "Classe": "class", "Nature": "nature", "Code barre": "bar_code", "Code UE article": "Code_UE", "Poids": "weight", "Sous-Famille": "sub_family", "Longueur": "length", "Gamme": "range", "Marque": "brand", "Taille": "size", "Type 1": "type_1", "Type 2": "type_2", "Type 3": "type_3", "Type 4": "type_4", "Variation Code": "variation_code", "Variation Désignatio": "variation_name", "Produit Parent": "parent", "Variation Label": "variation_label", "NON PACKAGE": "no_package", "Visible sur B2B": "B2bVisible", } results = [] with open(file_path, 'r', encoding='ISO-8859-1') as csv_file: reader = csv.DictReader(csv_file, delimiter=';') headers = reader.fieldnames print("En-têtes CSV lus :", headers) print() for row in reader: english_row = {} for fr_key, en_key in mapping.items(): value = row.get(fr_key, "") if value is None: value = "" english_row[en_key] = value.strip() results.append(english_row) return results def instantiate_csv_file(prices_file_path_csv): try: with open(prices_file_path_csv, mode='r', encoding='ISO-8859-1') as csvfile: csv_reader = csv.reader(csvfile, delimiter=';') rows = list(csv_reader) return rows except Exception as e: print(f"Erreur lors de la lecture du fichier CSV : {e}") return None async def process_product(session, csv_product, json_data, token_index, prices, semaphore): for product in json_data.get('products', []): for variant in product.get('variants', []): sku = variant.get('sku') access_token = access_tokens[token_index % len(access_tokens)] if csv_product["status"] == "O" and csv_product["nature"] == "PV 2025 OK": try: sku = csv_product['item_code'] product_price = get_product_price(prices,sku) product_price = product_price.replace(',', '.') print(f"{sku} : {product_price}") product_price = float(product_price) product_price = round(product_price, 2) # image_link = get_image_link(sku) # image_link = 'https://api.huggii.com/Harken/images/'+sku # image_service = ImageService() # image_service.add_image(image_link, sku) founded_product = get_product_by_sku(json_data, sku) length = csv_product['length'].replace(',', '.') if csv_product['parent'] in variants_done: return if float(length) >= 2 and float(length) < 6: vendor_to_push = "2M+" elif float(length) >= 1.4 and float(length) < 6: vendor_to_push = "1.4M+" else: vendor_to_push = "" def longest_common_prefix(strings): if not strings: return "" prefix = strings[0] for s in strings[1:]: while not s.startswith(prefix): prefix = prefix[:-1] if prefix == "": return "" return prefix.strip() if csv_product["parent"] == "": title = f"{sku} - {csv_product['main_designation']}" else: sku = csv_product["parent"] variants_from_erp = get_related_products(csv_product['item_code']) designations = [] for variant in variants_from_erp: designation = variant['main_designation'] if "taille" in designation: designation = designation.split("taille")[0].strip() designations.append(designation) common_title = longest_common_prefix(designations) title = f"{sku} - {common_title}" product = Product( title=title, description=" ", vendor=vendor_to_push, product_type=csv_product['category'], ) weight = csv_product['weight'].replace(",", ".") if csv_product["parent"] == "": variant = { "option1": "Default", "sku": sku, "price": product_price, "inventory_management": "shopify", "inventory_policy": "continue", "weight": weight, "barcode": csv_product['bar_code'], } product.add_variant(variant) else: variants_done.append(csv_product['parent']) product.add_option(csv_product['variation_label']) for variant_to_check in variants_from_erp: print(variant_to_check) variat_price = get_product_price(prices,variant_to_check['item_code']) variat_price = variat_price.replace(',', '.') variat_price = float(variat_price) variat_price = round(variat_price, 2) weight = variant_to_check.get('weight', 0).replace(",", ".") variant_to_add = { "option1": variant_to_check["variation_name"], "sku": variant_to_check["item_code"], "price": variat_price, "inventory_management": "shopify", "inventory_policy": "continue", "weight": float(weight), # Par défaut, 0 si 'weight' est manquant "barcode": variant_to_check.get('bar_code', ""), } product.add_variant(variant_to_add) tag_service = TagService() prefix_fields = { 'Sous-catégorie: ': 'subcategories', 'Catégorie: ': 'category', 'Code Famille: ': 'family_code', 'Sous-Famille: ': 'sub_family', 'Gamme: ': 'range', 'Marque: ': 'brand', 'Classe: ': 'class', 'Nature: ': 'nature', 'NO_PACKAGE: ': 'no_package', } for prefix, field in prefix_fields.items(): if csv_product.get(field): tag_service.add_tag(prefix + csv_product[field]) tag_service.add_tag(f'REF: {sku}' ) simple_fields = ['type_1', 'type_2', 'type_3', 'type_4'] for field in simple_fields: if csv_product.get(field): tag_service.add_tag(csv_product[field]) if csv_product["B2bVisible"].upper() != "Y": tag_service.add_tag("B2B_hidden") formatted_tags = tag_service.get_tags_as_string() product.add_metafield("custom", "family_code", csv_product['family_code'], 'single_line_text_field') product.add_metafield("custom", "range", csv_product['range'], 'single_line_text_field') product.add_metafield("custom", "size", csv_product['size'], 'single_line_text_field') product.add_metafield("custom", "type_1", csv_product['type_1'], 'single_line_text_field') product.add_metafield("custom", "type_2", csv_product['type_2'], 'single_line_text_field') product.add_metafield("custom", "type_3", csv_product['type_3'], 'single_line_text_field') product.add_metafield("custom", "type_4", csv_product['type_4'], 'single_line_text_field') product.add_metafield("custom", "length", csv_product['length'], 'single_line_text_field') product_service = ProductGenerationService(product, tag_service) product_data = product_service.get_formatted_product_data() async with semaphore: if founded_product is None: print(f"Création du produit avec le SKU {sku}") await create_shopify_product(session, product_data, access_token) else: print(f"Mise à jour du produit avec le SKU {sku}") product_id = founded_product.get('id') await update_shopify_product(session, product_id, product_data, access_token) except Exception as e: print(f"Exception dans process_product pour le SKU {csv_product['item_code']}: {e}") elif csv_product["status"] == "O" and csv_product["nature"] != "PV 2025 OK": sku = csv_product['item_code'].lstrip('0') founded_product = get_product_by_sku(json_data, sku) if founded_product is not None: product_id = founded_product.get('id') if product_id: await delete_shopify_product(session, product_id, access_token) async def process_csv(file_path_csv, file_path_json, prices, semaphore): try: # Lecture des données CSV csv_data = read_csv_file(file_path_csv) # Lecture des données JSON existantes with open(file_path_json, 'r', encoding='ISO-8859-1') as json_file: json_data = json.load(json_file) async with aiohttp.ClientSession() as session: tasks = [] for index, csv_product in enumerate(csv_data): task = asyncio.create_task(process_product(session, csv_product, json_data, index, prices, semaphore)) tasks.append(task) if len(tasks) >= 15: results = await asyncio.gather(*tasks, return_exceptions=True) tasks = [] for result in results: if isinstance(result, Exception): print(f"Exception levée dans une tâche : {result}") if tasks: print(f"Traitement du dernier lot de {len(tasks)} tâches") results = await asyncio.gather(*tasks, return_exceptions=True) for result in results: if isinstance(result, Exception): print(f"Exception levée dans une tâche : {result}") except Exception as e: print(f"Erreur dans process_csv : {e}") if __name__ == "__main__": variants_done = [] # Variable globale utilisée dans process_product async def main(): # Créer le sémaphore dans la boucle d'événements active semaphore = asyncio.Semaphore(9) # Opérations synchrones préliminaires download_huggy_tarifs_base_txt() download_and_merge_huggii_products_files_txt() convert_txt_to_csv('../files/Tarifs/huggii_tarifs_base.txt', prices_file_path_csv) convert_txt_to_csv('../files/Products/merged_huggii_products.txt', file_path_csv) prices = instantiate_csv_file(prices_file_path_csv) print('Récupération des produits ...') # Appeler get_all_products dans la même boucle if not await get_all_products(access_tokens, shop_name, '2024-10'): sys.exit(0) # Traiter le CSV dans la même boucle await process_csv(file_path_csv, file_path_json, prices, semaphore) asyncio.run(main()) sys.exit(0)