import asyncio import aiohttp import csv import json import os import signal import sys from typing import Set # ➜ Ajout du chemin projet pour les imports internes auth_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")) if auth_root not in sys.path: sys.path.append(auth_root) from Harken.API.tokens import access_tokens # Liste de tokens d'API from Harken.API.get_all_products import get_all_products # Fonction utilitaire # --- Constantes de configuration --- SHOP_NAME: str = "https://harkenb2b.myshopify.com" API_VERSION: str = "2024-10" CSV_FILE: str = "sku_image_mapping.csv" IMAGES_BASE_URL: str = "https://api.huggii.com/Harken/images/" RATE_LIMIT: int = 2 # Requêtes simultanées par token TOKENS = access_tokens # Liste de tokens (rotation simple) STOP: bool = False # Flag d’interruption Ctrl‑C # --------------------------------------------------------------------------- # Utilitaires CSV / Shopify # --------------------------------------------------------------------------- def load_sku_image_map(csv_path: str) -> dict[str, str]: """Retourne un dict {SKU_normalisé: image_name} (NP supprimé).""" mapping: dict[str, str] = {} with open(csv_path, newline="", encoding="utf-8") as csvfile: reader = csv.DictReader(csvfile) for row in reader: sku = row["sku"].strip().upper() if sku.endswith("NP"): sku = sku[:-2] # retire le suffixe NP mapping[sku] = row["image_name"].strip() return mapping def variant_has_image(product: dict, variant_id: int) -> bool: """True si le variant possède déjà une image (peu importe laquelle).""" for image in product.get("images", []): if variant_id in image.get("variant_ids", []): return True return False # --------------------------------------------------------------------------- # Upload d’une image # --------------------------------------------------------------------------- async def upload_image( session: aiohttp.ClientSession, access_token: str, product_id: int, image_url: str, variant_id: int, semaphore: asyncio.Semaphore, ) -> None: """Envoie la requête d’upload et log le résultat.""" async with semaphore: if STOP: return try: url = f"{SHOP_NAME}/admin/api/{API_VERSION}/products/{product_id}/images.json" headers = { "X-Shopify-Access-Token": access_token, "Content-Type": "application/json", } payload = {"image": {"src": image_url, "variant_ids": [variant_id]}} async with session.post(url, headers=headers, json=payload) as response: if response.status in (200, 201): print(f"✅ Image ajoutée ➜ variant {variant_id}") else: error_text = await response.text() print( f"❌ Erreur {response.status} variant {variant_id} | " f"{error_text[:120]}…" ) except Exception as exc: # pylint: disable=broad-except print(f"❌ Exception variant {variant_id} : {exc}") await asyncio.sleep(0.5) # --------------------------------------------------------------------------- # Gestion SIGINT (Ctrl‑C) # --------------------------------------------------------------------------- def handle_sigint(_sig, _frame): # type: ignore[override] global STOP # noqa: PLW0603 print("\n🛑 Interruption reçue, arrêt après les tâches en cours…") STOP = True signal.signal(signal.SIGINT, handle_sigint) # --------------------------------------------------------------------------- # Main async # --------------------------------------------------------------------------- async def main() -> None: # noqa: C901 # --- Chargement CSV & Produits Shopify --- sku_image_map: dict[str, str] = load_sku_image_map(CSV_FILE) print(f"🖼 {len(sku_image_map)} images répertoriées dans le CSV") print("⏳ Récupération de la liste des produits Shopify…") products_json: str = await get_all_products(TOKENS, SHOP_NAME, API_VERSION) products_data: dict = json.loads(products_json) products: list[dict] = products_data.get("products", []) print(f"📦 {len(products)} produits chargés depuis Shopify") # --- Compteurs de diagnostic --- matched_sku: int = 0 # variants dont le SKU figure dans le CSV already_has_image: int = 0 # mais possède déjà une image to_upload: int = 0 # variants auxquels on va associer une nouvelle image found_skus: Set[str] = set() # pour identifier les SKU du CSV sans variant async with aiohttp.ClientSession() as session: semaphores = [asyncio.Semaphore(RATE_LIMIT) for _ in TOKENS] token_count = len(TOKENS) token_index = 0 tasks: list[asyncio.Task] = [] for product in products: product_id = product["id"] for variant in product.get("variants", []): if STOP: break raw_sku: str = variant.get("sku", "") sku: str = raw_sku.strip().upper() if sku.endswith("NP"): sku = sku[:-2] variant_id: int = variant["id"] if sku in sku_image_map: matched_sku += 1 found_skus.add(sku) if variant_has_image(product, variant_id): already_has_image += 1 continue # Pas besoin d’uploader # ➜ On doit uploader l’image to_upload += 1 image_url: str = IMAGES_BASE_URL + sku_image_map[sku] # Rotation simple des tokens pour contourner le rate‑limit Shopify access_token: str = TOKENS[token_index % token_count] semaphore: asyncio.Semaphore = semaphores[token_index % token_count] token_index += 1 task = asyncio.create_task( upload_image( session, access_token, product_id, image_url, variant_id, semaphore, ) ) tasks.append(task) # --- Lancement des uploads --- if tasks: print(f"🚀 Lancement de {len(tasks)} uploads d’images…") await asyncio.gather(*tasks, return_exceptions=True) else: print("ℹ️ Aucune image à uploader : vérifie les statistiques ci‑dessous.") # ------------------------------------------------------------------- # Statistiques finales # ------------------------------------------------------------------- print("\n📊 Résumé :") print(f" • Variants avec un SKU présent dans le CSV : {matched_sku}") print(f" • déjà illustrés : {already_has_image}") print(f" • à illustrer (uploads lancés) : {to_upload}") missing_skus = set(sku_image_map) - found_skus print( f" • SKU dans le CSV sans variant Shopify : {len(missing_skus)}" ) if missing_skus: # Affiche un échantillon pour inspection rapide sample = ", ".join(list(missing_skus)[:20]) print(f" ↳ Exemples : {sample}{'…' if len(missing_skus) > 20 else ''}") # --------------------------------------------------------------------------- # Entrée script # --------------------------------------------------------------------------- if __name__ == "__main__": asyncio.run(main())