#!/usr/bin/env python3 """Nettoyage et enrichissement d'un catalogue produit. Usage : python3 enrichir.py catalogue-source.csv Produit trois fichiers : catalogue-enrichi.csv le catalogue normalisé, prêt à importer a-decider.csv les lignes qu'une machine ne doit PAS trancher seule rapport-enrichissement.md taux de remplissage avant/après et journal des corrections Principe directeur : **le script ne devine jamais.** Il normalise ce qui est déterministe (unités, formats, casse, doublons, clés de contrôle) et il isole ce qui demande une décision humaine dans `a-decider.csv`. Un pipeline qui invente une valeur manquante transforme un catalogue incomplet en catalogue faux, ce qui est strictement pire. """ import csv, os, re, sys, unicodedata from collections import Counter, defaultdict from decimal import Decimal, InvalidOperation HERE = os.path.dirname(os.path.abspath(__file__)) # ---------------------------------------------------------------- référentiels MARQUES = {"kerhan": "Kerhan", "nordvik": "Nordvik", "aravis": "Aravis"} TAXONOMIE = { "randonnee": ("Randonnée", "Sport et loisirs > Plein air > Randonnée"), "cuisine-outdoor": ("Cuisine outdoor", "Sport et loisirs > Plein air > Camping > Réchauds"), "eclairage": ("Éclairage", "Quincaillerie > Outils > Lampes et frontales"), "couchage": ("Couchage", "Sport et loisirs > Plein air > Camping > Couchage"), "hydratation": ("Hydratation", "Sport et loisirs > Plein air > Hydratation"), } ALIAS_CATEGORIE = { "randonnée": "randonnee", "randonnee": "randonnee", "rando / trek": "randonnee", "randonnée > sacs": "randonnee", "trek": "randonnee", "cuisine outdoor": "cuisine-outdoor", "cuisine": "cuisine-outdoor", "réchauds": "cuisine-outdoor", "cooking": "cuisine-outdoor", "cuisine / réchauds": "cuisine-outdoor", "éclairage": "eclairage", "eclairage": "eclairage", "lampes": "eclairage", "lighting": "eclairage", "eclairage > frontales": "eclairage", "couchage": "couchage", "sommeil": "couchage", "sleeping": "couchage", "bivouac / couchage": "couchage", "hydratation": "hydratation", "gourdes": "hydratation", "drink": "hydratation", "boissons": "hydratation", } COULEURS = {"noir": "Noir", "bleu marine": "Bleu marine", "vert lichen": "Vert lichen", "terracotta": "Terracotta"} MATIERES = { "nylon ripstop 210d": "Nylon ripstop 210D", "nylon 210d": "Nylon ripstop 210D", "polyester 600d": "Polyester 600D", "aluminium 7075": "Aluminium 7075", "acier inoxydable": "Acier inoxydable", "inox": "Acier inoxydable", "duvet 700 cuin": "Duvet 700 CUIN", } VIDES = {"", "-", "n/a", "n/c", "nc", "null", "none", "à compléter", "a completer", " "} # matière plausible par famille — sert à détecter les erreurs de saisie fournisseur MATIERE_PLAUSIBLE = { "randonnee": {"Nylon ripstop 210D", "Polyester 600D", "Aluminium 7075"}, "cuisine-outdoor": {"Acier inoxydable", "Aluminium 7075"}, "eclairage": {"Aluminium 7075", "Polyester 600D"}, "couchage": {"Duvet 700 CUIN", "Nylon ripstop 210D", "Polyester 600D", "Aluminium 7075"}, "hydratation": {"Acier inoxydable", "Aluminium 7075"}, } # attribut métier porté par le libellé, par famille de produit ATTRIBUT = { "sac a dos": ("volume_l", r"(\d+)\s*L\b", "L"), "rechaud": ("poids_annonce_g", r"(\d+)\s*g\b", "g"), "lampe frontale": ("flux_lm", r"(\d+)\s*lm\b", "lm"), "duvet": ("temperature_c", r"(-?\d+)\s*°C", "°C"), "matelas gonflable": ("epaisseur_cm", r"(\d+)\s*cm\b", "cm"), "batons de marche": ("longueur_cm", r"(\d+)\s*cm\b", "cm"), "gourde isotherme": ("contenance_ml", r"(\d+)\s*ml\b", "ml"), "tente": ("places", r"(\d+)\s*pl\b", "pl"), } # ---------------------------------------------------------------- utilitaires def sans_accent(s): return "".join(c for c in unicodedata.normalize("NFD", s) if unicodedata.category(c) != "Mn") def nettoyer_texte(s): s = (s or "").replace("\xa0", " ") s = re.sub(r"<[^>]+>", " ", s) s = s.replace(" ", " ").replace("&", "&") return re.sub(r"\s+", " ", s).strip() def est_vide(s): return nettoyer_texte(s).lower() in VIDES def titre_propre(s): s = nettoyer_texte(s) if s and s == s.upper() and len(s) > 3: # libellé tout en capitales s = s.capitalize() # « 30L » -> « 30 L », « 5 °C » -> « 5 °C » s = re.sub(r"(\d)\s*(L|g|lm|cm|ml|pl)\b", r"\1 \2", s) s = re.sub(r"(-?\d)\s*°\s*C\b", r"\1 °C", s) return re.sub(r"\s+", " ", s).strip() def parser_prix(s): s = nettoyer_texte(s).upper().replace("EUR", "").replace("€", "").strip() s = s.replace(" ", "").replace(",", ".") if not s: return None try: v = Decimal(s) except InvalidOperation: return None return v.quantize(Decimal("0.01")) if v > 0 else None def parser_poids_g(s): s = nettoyer_texte(s).lower().replace(",", ".") if not s: return None m = re.match(r"^([\d.]+)\s*(kg|g|grammes|gramme)?$", s) if not m: return None try: v = float(m.group(1)) except ValueError: return None return int(round(v * 1000)) if m.group(2) == "kg" else int(round(v)) def gtin13_valide(g): g = nettoyer_texte(g) if not re.fullmatch(r"\d{13}", g) or set(g) == {"0"}: return False corps, cle = g[:12], int(g[12]) calc = (10 - sum((3 if i % 2 else 1) * int(d) for i, d in enumerate(corps)) % 10) % 10 return calc == cle def slug(s): s = sans_accent(nettoyer_texte(s)).lower() s = re.sub(r"[^a-z0-9]+", "-", s) return re.sub(r"-+", "-", s).strip("-") def famille(libelle): base = sans_accent(nettoyer_texte(libelle)).lower() for cle in sorted(ATTRIBUT, key=len, reverse=True): if base.startswith(cle): return cle return None # ---------------------------------------------------------------- traitement COLS_SORTIE = ["reference", "libelle", "slug", "marque", "categorie", "categorie_chemin", "attribut_nom", "attribut_valeur", "prix_ttc_eur", "poids_g", "couleur", "matiere", "gtin13", "gtin_statut", "stock", "complet"] def traiter(source): with open(source, encoding="utf-8", newline="") as f: brut = list(csv.DictReader(f)) corrections = Counter() a_decider = [] vus = {} sortie = [] for ligne in brut: ref = nettoyer_texte(ligne["reference"]).upper() lib = titre_propre(ligne["libelle"]) if not lib: corrections["libellé vide — ligne écartée"] += 1 a_decider.append({"reference": ref, "motif": "libellé vide", "valeur_source": repr(ligne["libelle"]), "action_attendue": "fournir le libellé ou supprimer la référence"}) continue if ligne["libelle"] != lib: corrections["libellé nettoyé (HTML, espaces, casse, unités recollées)"] += 1 # marque mk = nettoyer_texte(ligne["marque"]).lower() marque = MARQUES.get(mk) if marque is None: marque = nettoyer_texte(ligne["marque"]).title() corrections["marque inconnue du référentiel"] += 1 elif nettoyer_texte(ligne["marque"]) != marque: corrections["casse de marque normalisée"] += 1 # catégorie cat_src = nettoyer_texte(ligne["categorie"]).lower() cle = ALIAS_CATEGORIE.get(cat_src) if cle is None: a_decider.append({"reference": ref, "motif": "catégorie hors référentiel", "valeur_source": ligne["categorie"], "action_attendue": "rattacher à une catégorie de la taxonomie"}) cat_lib, cat_chemin = nettoyer_texte(ligne["categorie"]), "" else: cat_lib, cat_chemin = TAXONOMIE[cle] if nettoyer_texte(ligne["categorie"]) != cat_lib: corrections["catégorie libre rattachée à la taxonomie"] += 1 # prix prix = parser_prix(ligne["prix_ttc"]) if prix is None: a_decider.append({"reference": ref, "motif": "prix illisible ou nul", "valeur_source": ligne["prix_ttc"], "action_attendue": "saisir le prix TTC"}) elif nettoyer_texte(ligne["prix_ttc"]) != f"{prix}": corrections["prix converti en décimal (formats €, EUR, virgule, espaces)"] += 1 # poids poids = parser_poids_g(ligne["poids"]) if poids is None and not est_vide(ligne["poids"]): a_decider.append({"reference": ref, "motif": "poids illisible", "valeur_source": ligne["poids"], "action_attendue": "saisir le poids en grammes"}) elif poids is not None and nettoyer_texte(ligne["poids"]) != str(poids): corrections["poids converti en grammes (kg, g, « grammes »)"] += 1 # couleur / matière c = nettoyer_texte(ligne["couleur"]).lower() couleur = COULEURS.get(c, "" if c in VIDES else nettoyer_texte(ligne["couleur"]).capitalize()) if couleur and nettoyer_texte(ligne["couleur"]) != couleur: corrections["couleur normalisée"] += 1 m = nettoyer_texte(ligne["matiere"]).lower() matiere = MATIERES.get(m, "" if m in VIDES else nettoyer_texte(ligne["matiere"])) if matiere and nettoyer_texte(ligne["matiere"]) != matiere: corrections["matière normalisée"] += 1 if matiere and cle and matiere not in MATIERE_PLAUSIBLE[cle]: a_decider.append({"reference": ref, "motif": "matière incohérente avec la catégorie", "valeur_source": f'{matiere} sur « {cat_lib} »', "action_attendue": "confirmer ou corriger la matière"}) # GTIN g = nettoyer_texte(ligne["gtin"]) if not g: gtin, statut = "", "absent" elif gtin13_valide(g): gtin, statut = g, "valide" else: gtin, statut = "", "invalide" a_decider.append({"reference": ref, "motif": "GTIN invalide (clé de contrôle ou longueur)", "valeur_source": g, "action_attendue": "fournir un GTIN-13 valide ou laisser vide"}) corrections["GTIN invalide écarté"] += 1 # stock s = nettoyer_texte(ligne["stock"]) if s.lstrip("-").isdigit(): stock = max(0, int(s)) if int(s) < 0: corrections["stock négatif ramené à 0"] += 1 else: stock = 0 if s: corrections["stock non numérique ramené à 0"] += 1 # --- enrichissement : extraction de l'attribut métier depuis le libellé fam = famille(lib) attr_nom = attr_val = "" if fam: nom, motif, unite = ATTRIBUT[fam] mm = re.search(motif, lib, re.IGNORECASE) if mm: attr_nom, attr_val = nom, mm.group(1) corrections[f"attribut « {nom} » extrait du libellé"] += 1 champs = dict(reference=ref, libelle=lib, slug=slug(f"{marque} {lib}"), marque=marque, categorie=cat_lib, categorie_chemin=cat_chemin, attribut_nom=attr_nom, attribut_valeur=attr_val, prix_ttc_eur=str(prix) if prix is not None else "", poids_g=str(poids) if poids is not None else "", couleur=couleur, matiere=matiere, gtin13=gtin, gtin_statut=statut, stock=str(stock)) obligatoires = ["libelle", "marque", "categorie", "prix_ttc_eur", "poids_g", "couleur", "matiere", "attribut_valeur"] champs["complet"] = "oui" if all(champs[k] for k in obligatoires) else "non" # --- déduplication sur la référence : on garde la ligne la plus renseignée if ref in vus: corrections["doublon de référence fusionné"] += 1 garde = vus[ref] fusion = dict(garde) for k in COLS_SORTIE: if not fusion.get(k) and champs.get(k): fusion[k] = champs[k] if garde["prix_ttc_eur"] != champs["prix_ttc_eur"] and champs["prix_ttc_eur"]: a_decider.append({"reference": ref, "motif": "doublon avec deux prix différents", "valeur_source": f'{garde["prix_ttc_eur"]} vs {champs["prix_ttc_eur"]}', "action_attendue": "choisir le prix de référence"}) fusion["complet"] = "oui" if all(fusion[k] for k in obligatoires) else "non" vus[ref] = fusion else: vus[ref] = champs sortie.append(ref) lignes = [vus[r] for r in sortie] return brut, lignes, corrections, a_decider def taux_remplissage(rows, colonnes, lecteur): return {c: sum(1 for r in rows if lecteur(r, c)) for c in colonnes} def main(): source = sys.argv[1] if len(sys.argv) > 1 else os.path.join(HERE, "catalogue-source.csv") brut, lignes, corrections, a_decider = traiter(source) with open(os.path.join(HERE, "catalogue-enrichi.csv"), "w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=COLS_SORTIE); w.writeheader(); w.writerows(lignes) with open(os.path.join(HERE, "a-decider.csv"), "w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=["reference", "motif", "valeur_source", "action_attendue"]) w.writeheader(); w.writerows(a_decider) # ---- rapport paires = [("libellé", "libelle", "libelle"), ("marque", "marque", "marque"), ("catégorie", "categorie", "categorie"), ("prix", "prix_ttc", "prix_ttc_eur"), ("poids", "poids", "poids_g"), ("couleur", "couleur", "couleur"), ("matière", "matiere", "matiere"), ("GTIN valide", "gtin", "gtin13")] nb_b, nb_a = len(brut), len(lignes) def rempli_source(r, c): return not est_vide(r.get(c, "")) and (gtin13_valide(r[c]) if c == "gtin" else True) lignes_rapport = [ "# Rapport d'enrichissement de catalogue", "", f"**Source :** `{os.path.basename(source)}` — {nb_b} lignes, {len(brut[0])} colonnes", f"**Sortie :** `catalogue-enrichi.csv` — {nb_a} références uniques, {len(COLS_SORTIE)} colonnes", f"**À trancher par un humain :** `a-decider.csv` — {len(a_decider)} lignes", "", "---", "", "## 1. Taux de remplissage, avant et après", "", "| Champ | Avant | Après | Écart |", "|---|---:|---:|---:|", ] for libelle, cs, cd in paires: av = sum(1 for r in brut if rempli_source(r, cs)) ap = sum(1 for r in lignes if r.get(cd)) pav, pap = 100 * av / nb_b, 100 * ap / nb_a lignes_rapport.append(f"| {libelle} | {pav:.1f} % | {pap:.1f} % | {pap - pav:+.1f} pt |") attr = sum(1 for r in lignes if r["attribut_valeur"]) chemin = sum(1 for r in lignes if r["categorie_chemin"]) complets = sum(1 for r in lignes if r["complet"] == "oui") lignes_rapport += [ f"| attribut métier structuré | 0,0 % | {100*attr/nb_a:.1f} % | +{100*attr/nb_a:.1f} pt |", f"| chemin de taxonomie | 0,0 % | {100*chemin/nb_a:.1f} % | +{100*chemin/nb_a:.1f} pt |", f"| slug d'URL | 0,0 % | 100,0 % | +100,0 pt |", "", f"**Références complètes sur tous les champs obligatoires : {complets}/{nb_a} " f"({100*complets/nb_a:.1f} %).**", "", "Le champ « attribut métier » est le vrai apport : le volume d'un sac, la température de", "confort d'un duvet ou le flux d'une frontale étaient noyés dans le libellé, en texte libre.", "Ils sont désormais des colonnes exploitables — donc filtrables en boutique, et lisibles", "par un flux Google Shopping.", "", "---", "", "## 2. Corrections appliquées", "", "| Correction | Occurrences |", "|---|---:|", ] for k, v in corrections.most_common(): lignes_rapport.append(f"| {k} | {v} |") lignes_rapport += ["", f"**Total : {sum(corrections.values())} corrections déterministes.**", ""] motifs = Counter(d["motif"] for d in a_decider) lignes_rapport += ["---", "", "## 3. Ce que la machine n'a pas tranché", "", f"{len(a_decider)} lignes sont remontées dans `a-decider.csv`.", "", "| Motif | Lignes |", "|---|---:|"] for k, v in motifs.most_common(): lignes_rapport.append(f"| {k} | {v} |") lignes_rapport += [ "", "Aucune de ces valeurs n'a été devinée. Un pipeline qui invente un prix manquant ou", "rattache une catégorie inconnue « au plus proche » ne produit pas un catalogue complet,", "il produit un catalogue faux — et l'erreur devient invisible. Le temps humain se", "concentre ici, et nulle part ailleurs.", "", "---", "", "## 4. Contrôles exécutés", "", "| Contrôle | Règle |", "|---|---|", "| Doublon de référence | fusion des deux lignes, la valeur renseignée l'emporte sur le vide ; prix divergents remontés |", "| GTIN-13 | longueur 13 + clé de contrôle modulo 10 recalculée ; les codes tout à zéro sont rejetés |", "| Prix | `12,90 €`, `12.90EUR`, ` 12.90 ` → `12.90` ; zéro et négatif rejetés |", "| Poids | `780g`, `0,78 kg`, `780 grammes` → `780` |", "| Libellé | balises HTML retirées, espaces insécables et multiples réduits, capitales corrigées, unités recollées |", "| Catégorie | rattachement à une taxonomie fermée + chemin Google Shopping ; hors référentiel = remonté |", "| Matière | cohérence avec la catégorie (un duvet en acier inoxydable est une erreur de saisie) |", "| Stock | valeurs négatives et non numériques ramenées à 0 |", "| Slug | translittération sans accent, minuscules, tirets |", "", "---", "", "## 5. Reproductibilité", "", "```bash", "python3 generer-source.py # régénère la source, graine fixe", "python3 enrichir.py # régénère les trois fichiers de sortie", "```", "", "Le traitement est déterministe : deux exécutions sur la même source donnent le même", "octet. C'est ce qui rend le rapport opposable.", ] with open(os.path.join(HERE, "rapport-enrichissement.md"), "w", encoding="utf-8") as f: f.write("\n".join(lignes_rapport) + "\n") print(f"{nb_b} lignes source -> {nb_a} références uniques") print(f"{sum(corrections.values())} corrections, {len(a_decider)} lignes à décider") print(f"complétude : {complets}/{nb_a} ({100*complets/nb_a:.1f} %)") if __name__ == "__main__": main()