Produits complémentaires v2 (ancrage taxonomie famille) — Implementation Plan

# Produits complémentaires v2 (ancrage taxonomie famille) — Implementation Plan

> For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Refondre la brique « Compléments IA » rejetée en v1. On n'infère plus la catégorie d'un produit depuis son titre SAP abrégé : on la lit dans ecom_products (postgres :5433). L'IA raisonne sur la famille réelle (206 familles, taxonomie warehouse) et propose, pour chaque ancre, des familles complémentaires classées A (consommable/nécessaire) ou B (complément d'usage), puis on déplie en SKU concrets actifs triés par ventes. Livrable : XLSX 4 onglets (README · A · B · Mix) au format large du fichier baseline, validé sur les 20 premières ancres (CAFETERIA : cafés + biscuits).

Architecture: Une couche pure et testable (core/complement_families.py) fait tout le calcul non-IA : échantillonnage de profil famille, tri/filtre des SKU candidats, classement A/B des arêtes, expansion famille→SKU avec display_sequence, assemblage du format large. La couche IA isolée et injectable (core/ai_typing.py, déjà existante) est étendue de deux fonctions — profile_family (libellé + résumé d'une famille) et propose_family_complements (graphe famille→familles A/B) — sur le pattern llm_call injectable déjà en place (get_llm_backend, cache, modèle Sonnet/Haiku). Un script d'orchestration (scripts/refresh_product_complements_v2.py) câble le tout : lecture des 20 ancres → enrichissement postgres → profils familles (IA, caché) → graphe A/B (IA) → ventes (timescale) → 3 vues → XLSX + JSON.

Tech Stack: Python 3.12 (venv\Scripts\python.exe, c'est celui qui a psycopg2/openpyxl ; .venv est en 3.14 sans psycopg2), pytest, backend LLM via core.ai_typing.get_llm_backend (CLI claude -p abonnement par défaut, SDK seulement si --allow-api-billing), openpyxl, psycopg2 (via database.crm_db / database.timeseries_db).

Référence spec : docs/superpowers/specs/2026-06-08-product-complements-v2-design.md

---

Faits warehouse vérifiés (introspection 2026-06-08, ne pas re-deviner)

• ecom_products (postgres :5433) possède bien toutes les colonnes de la spec :

• section_taxonomy fournit world + label_fr/label_en par section_code (pas de libellé de famille → à dériver par IA).
• Famille = family_code seul, globalement unique (FR : 206 familles, et count(distinct (section_code,family_code)) = 206 aussi). 21 sections, 840 (sec,fam,cat).
• status_code est numérique (98 ≈ 282k actifs, ' ' 160k, 94/99/95 mineurs) — PAS le vocabulaire Continued/New/Delisted (celui-ci n'existe que dans le fichier baseline). Le filtre « actif » fiable sur ecom_products est donc not_salable_flag IS NOT TRUE AND not_visible_flag IS NOT TRUE (444 404 produits FR passent ; 4 320 sont not_salable et/ou not_visible). On dérive un libellé d'affichage simple (Active/Inactive) — on ne fabrique pas de faux Continued.
• 20 premières ancres du fichier imports/lucas/20260608_094900_Produits_comple_mentaires_20260526_1.xlsx : 100 % jointes à ecom_products, réparties sur 2 familles — 001002 (cafés, 8..20) et 001001 (biscuits/snacks, 1..8), world CAFETERIA.
• World CAFETERIA = 7 familles seulement (001001 snacks, 001002 cafés/boissons chaudes/capsules, 001003 boissons froides/jus/eau, 001004 électro hygiène, 001005 gobelets/serviettes/service, 001006 bacs de recyclage, 001000 item spécial à ignorer). ⇒ univers candidat même-world petit et pertinent (gobelets 001005 = complément A/B évident du café).
• Ventes : ecom_order_lines est sur timescale :5434 (URI TIMESCALE_URI, défaut postgresql://leadcontagion:lc_ts_2026@localhost:5434/lc_timeseries), clé product_reference, quantity/sales_amount, order_date, FR du 2025-04-30 au 2026-06-04 (16,5 M lignes). Cross-DB interdit : tirer les ventes sur timescale, enrichir sur postgres, joindre en Python.
• product_complements (postgres :5433) = baseline en lecture seule (10 665 paires · 5 458 ancres), colonnes anchor_reference, complement_reference, complement_status, display_sequence, .... Ne pas la modifier.

---

File Structure

| Fichier | Responsabilité | |---|---| | core/complement_families.py (créer) | Fonctions pures : rank_family_skus, build_family_profile_sample, order_complement_families, assemble_anchor_blocks, build_wide_rows. Aucun I/O réseau/DB. | | core/ai_typing.py (modifier) | +profile_family, +propose_family_complements (réutilisent _norm_type, _parse_json_obj, _parse_json_obj_list, get_llm_backend). | | scripts/refresh_product_complements_v2.py (créer) | Orchestration offline : 20 ancres → enrichissement postgres → profils familles (IA, caché) → graphe A/B (IA) → ventes timescale → 3 vues → XLSX (README+A+B+Mix+Baseline) + JSON. CLI --n-anchors 20 --model sonnet --llm-backend cli|api --max-complements 20. | | tests/test_complement_families.py (créer) | TDD des 5 fonctions pures (valeurs exactes). | | tests/test_ai_typing.py (modifier) | +2 tests des nouvelles fonctions IA, llm_call stubbé (zéro réseau). |

Conventions verrouillées (utilisées par toutes les tasks) :

• Une famille = son family_code (chaîne, ex. "001002"). C'est l'identifiant.
• Une arête famille (classée par l'IA) = dict {"comp_family": str, "klass": "A"|"B", "relation": str, "force": float, "justification": str}. force ∈ [0,1].
• Un SKU candidat = dict {"reference": str(18 digits), "description": str, "sales": int|float, "not_salable": bool, "not_visible": bool, "status_code": str}.
• Un bloc complément (sortie) = dict {"reference", "description", "status", "display_sequence", "klass", "comp_family"}.
• family_skus: dict[str, list[dict]] = {family_code: [SKU candidat déjà trié]}.
• Le format large = [anchor_ref, anchor_desc] + 20 blocs de 4 [code complément, description, statut, display_sequence] (cellules vides si < 20).

---

Task 1: rank_family_skus — SKU actifs d'une famille triés par ventes

Files: Create core/complement_families.py · Test tests/test_complement_families.py

• [ ] Step 1: Write the failing test

`python # tests/test_complement_families.py from core.complement_families import rank_family_skus

def _sku(ref, sales, salable=False, visible=False, status="98", desc=None): # salable/visible here mean the flag values (not_salable / not_visible) return {"reference": ref, "description": desc or f"P{ref}", "sales": sales, "not_salable": salable, "not_visible": visible, "status_code": status}

def test_rank_family_skus_active_only_and_sales_order(): cands = [ _sku("a", 100), _sku("b", 250), _sku("c", 999, salable=True), # not_salable -> dropped _sku("d", 999, visible=True), # not_visible -> dropped _sku("e", 10), ] out = rank_family_skus(cands, top_n=2) assert [s["reference"] for s in out] == ["b", "a"] # active, top-2 by sales # status label derived, sales preserved assert out[0]["status"] == "Active" # ties broken by reference asc for determinism tied = rank_family_skus([_sku("z", 5), _sku("a", 5), _sku("m", 5)], top_n=3) assert [s["reference"] for s in tied] == ["a", "m", "z"] # active_only=False keeps inactive but still ranks by sales keep = rank_family_skus(cands, top_n=5, active_only=False) assert [s["reference"] for s in keep] == ["c", "d", "b", "a", "e"] assert keep[0]["status"] == "Inactive" `

• [ ] Step 2: Run test to verify it fails — venv\Scripts\python.exe -m pytest tests/test_complement_families.py::test_rank_family_skus_active_only_and_sales_order -v → FAIL (ImportError).

• [ ] Step 3: Write minimal implementation

`python # core/complement_families.py """Produits complémentaires v2 — fonctions pures (ancrage taxonomie famille).

Aucun I/O réseau ni DB. Méthode : docs/superpowers/specs/2026-06-08-product-complements-v2-design.md """ from __future__ import annotations

def _is_active(sku: dict) -> bool: """Actif = ni not_salable ni not_visible (les codes status_code numériques ne portent pas le vocabulaire Continued/New/Delisted — cf. plan §Faits).""" return not sku.get("not_salable") and not sku.get("not_visible")

def rank_family_skus(candidates, top_n: int, active_only: bool = True) -> list: """SKU d'une famille triés par ventes décroissantes, top_n retenus.

active_only filtre sur not_salable/not_visible. Tri stable : ventes desc puis reference asc (déterministe). Ajoute un libellé status Active/Inactive lisible pour l'affichage.""" pool = [s for s in candidates if (not active_only or _is_active(s))] pool.sort(key=lambda s: (-(s.get("sales") or 0), s["reference"])) out = [] for s in pool[:top_n]: out.append({s, "status": "Active" if _is_active(s) else "Inactive"}) return out `

• [ ] Step 4: Run test → PASS.
• [ ] Step 5: Commit — git add core/complement_families.py tests/test_complement_families.py && git commit -m "feat(complements-v2): rank_family_skus (active filter + sales order)"

---

Task 2: build_family_profile_sample — échantillon déterministe pour profilage IA

Files: Modify core/complement_families.py · Test tests/test_complement_families.py

Le profil famille (Phase 2 spec) a besoin d'un échantillon de ~12 produits texte. Fonction pure et déterministe (tri par reference) pour que le cache IA soit stable.

• [ ] Step 1: Write the failing test

`python # tests/test_complement_families.py (append) from core.complement_families import build_family_profile_sample

def test_build_family_profile_sample_deterministic_and_capped(): prods = [ {"reference": "002", "web_title": "Café moulu Miko", "product_description": "PAQ 250G CAFE"}, {"reference": "001", "web_title": "Café grains Lavazza", "product_description": "1KG GRAINS"}, {"reference": "003", "web_title": "", "product_description": "CAPS TASSIMO"}, ] out = build_family_profile_sample(prods, sample_size=2) # sorted by reference asc, capped at sample_size, web_title preferred over desc assert out == ["Café grains Lavazza — 1KG GRAINS", "Café moulu Miko — PAQ 250G CAFE"] # falls back to product_description when web_title is empty out3 = build_family_profile_sample(prods, sample_size=3) assert out3[2] == "CAPS TASSIMO" `

• [ ] Step 2: Run test → FAIL (ImportError).

• [ ] Step 3: Write minimal implementation

`python # core/complement_families.py (append) def build_family_profile_sample(products, sample_size: int = 12) -> list: """Échantillon texte déterministe d'une famille pour le profilage IA.

Trie par reference asc, prend sample_size produits, formate 'web_title — product_description' (web_title prioritaire ; si vide, on ne garde que la description). Pur, sans I/O.""" ordered = sorted(products, key=lambda p: p.get("reference", "")) lines = [] for p in ordered[:sample_size]: title = (p.get("web_title") or "").strip() desc = (p.get("product_description") or "").strip() if title and desc: lines.append(f"{title} — {desc}") else: lines.append(title or desc) return lines `

• [ ] Step 4: Run test → PASS.
• [ ] Step 5: Commit — git commit -am "feat(complements-v2): build_family_profile_sample (deterministic, capped)"

---

Task 3: order_complement_families — filtre + ordre A-avant-B des arêtes

Files: Modify core/complement_families.py · Test tests/test_complement_families.py

• [ ] Step 1: Write the failing test

`python # tests/test_complement_families.py (append) from core.complement_families import order_complement_families

def _edge(fam, klass, force): return {"comp_family": fam, "klass": klass, "force": force, "relation": "x", "justification": "j"}

def test_order_complement_families_klass_then_force(): edges = [_edge("001005", "A", 0.6), _edge("001001", "B", 0.7), _edge("001003", "A", 0.9)] # default: A block first (force desc), then B block (force desc) allo = order_complement_families(edges) assert [e["comp_family"] for e in allo] == ["001003", "001005", "001001"] # klass filter assert [e["comp_family"] for e in order_complement_families(edges, klass="A")] \ == ["001003", "001005"] assert [e["comp_family"] for e in order_complement_families(edges, klass="B")] \ == ["001001"] `

• [ ] Step 2: Run test → FAIL.

• [ ] Step 3: Write minimal implementation

`python # core/complement_families.py (append) _KLASS_RANK = {"A": 0, "B": 1}

def order_complement_families(edges, klass: str | None = None) -> list: """Ordonne les arêtes famille d'une ancre : A avant B, puis force desc.

klass (si fourni) restreint à 'A' ou 'B'. Pur.""" sel = [e for e in edges if klass is None or e.get("klass") == klass] sel.sort(key=lambda e: (_KLASS_RANK.get(e.get("klass"), 9), -(e.get("force") or 0.0))) return sel `

• [ ] Step 4: Run test → PASS.
• [ ] Step 5: Commit — git commit -am "feat(complements-v2): order_complement_families (A-before-B, force desc)"

---

Task 4: assemble_anchor_blocks — expansion famille→SKU + display_sequence (cap 20)

Files: Modify core/complement_families.py · Test tests/test_complement_families.py

• [ ] Step 1: Write the failing test

`python # tests/test_complement_families.py (append) from core.complement_families import assemble_anchor_blocks

def test_assemble_anchor_blocks_sequence_dedup_cap(): ordered = [ {"comp_family": "001005", "klass": "A", "force": 0.9}, {"comp_family": "001003", "klass": "B", "force": 0.7}, ] family_skus = { "001005": [{"reference": "g1", "description": "Gobelet", "status": "Active"}, {"reference": "g2", "description": "Serviette", "status": "Active"}], "001003": [{"reference": "g1", "description": "Gobelet", "status": "Active"}, {"reference": "j1", "description": "Jus", "status": "Active"}], } blocks = assemble_anchor_blocks(ordered, family_skus, max_total=20) # walk families in order, assign 1.. ; g1 dedup across families assert [(b["reference"], b["display_sequence"], b["klass"]) for b in blocks] == [ ("g1", 1, "A"), ("g2", 2, "A"), ("j1", 3, "B")] assert blocks[0]["comp_family"] == "001005" # cap respected capped = assemble_anchor_blocks(ordered, family_skus, max_total=2) assert len(capped) == 2 and [b["reference"] for b in capped] == ["g1", "g2"] `

• [ ] Step 2: Run test → FAIL.

• [ ] Step 3: Write minimal implementation

`python # core/complement_families.py (append) def assemble_anchor_blocks(ordered_families, family_skus: dict, max_total: int = 20) -> list: """Déplie les familles ordonnées d'une ancre en blocs SKU concrets.

Parcourt ordered_families (sortie de order_complement_families), tire les SKU déjà classés de family_skus[comp_family], déduplique les reference déjà retenues, attribue display_sequence 1.., plafonne à max_total. Pur.""" blocks = [] seen = set() for edge in ordered_families: fam = edge["comp_family"] for sku in family_skus.get(fam, []): ref = sku["reference"] if ref in seen: continue seen.add(ref) blocks.append({ "reference": ref, "description": sku.get("description") or "", "status": sku.get("status") or "", "display_sequence": len(blocks) + 1, "klass": edge.get("klass"), "comp_family": fam, }) if len(blocks) >= max_total: return blocks return blocks `

• [ ] Step 4: Run test → PASS.
• [ ] Step 5: Commit — git commit -am "feat(complements-v2): assemble_anchor_blocks (family->SKU, seq, dedup, cap)"

---

Task 5: build_wide_rows — matrice format large (= shape du fichier baseline)

Files: Modify core/complement_families.py · Test tests/test_complement_families.py

• [ ] Step 1: Write the failing test

`python # tests/test_complement_families.py (append) from core.complement_families import build_wide_rows, WIDE_HEADER

def test_build_wide_rows_layout(): anchors = [{"reference": "0000000000000471006", "description": "Biscuits"}] blocks_by_anchor = {"0000000000000471006": [ {"reference": "c1", "description": "Café", "status": "Active", "display_sequence": 1}, ]} rows = build_wide_rows(anchors, blocks_by_anchor, n_blocks=2) # col0 anchor ref, col1 anchor desc, then 2 blocks of 4 cells (2nd empty) assert rows[0] == ["0000000000000471006", "Biscuits", "c1", "Café", "Active", 1, "", "", "", ""] # header matches: 2 fixed cols + n_blocks*4 hdr = WIDE_HEADER(2) assert hdr[:2] == ["Anchor SAP", "Anchor Description"] assert hdr[2:6] == ["Consumable Web 1", "Local SAP Description 1", "Local Status Current Year 1", "Display Sequence 1"] assert len(hdr) == 2 + 2 * 4 # anchor with no complements -> just the 2 fixed cells + empty blocks empty = build_wide_rows([{"reference": "x", "description": "d"}], {}, n_blocks=1) assert empty[0] == ["x", "d", "", "", "", ""] `

• [ ] Step 2: Run test → FAIL.

• [ ] Step 3: Write minimal implementation

`python # core/complement_families.py (append) def WIDE_HEADER(n_blocks: int) -> list: """En-tête du format large : 2 colonnes ancre + n_blocks×4.""" head = ["Anchor SAP", "Anchor Description"] for i in range(1, n_blocks + 1): head += [f"Consumable Web {i}", f"Local SAP Description {i}", f"Local Status Current Year {i}", f"Display Sequence {i}"] return head

def build_wide_rows(anchors, blocks_by_anchor: dict, n_blocks: int = 20) -> list: """Matrice format large : 1 ligne / ancre, jusqu'à n_blocks blocs de 4 [code complément, description, statut, display_sequence], cellules vides au-delà des compléments disponibles. Pur.""" rows = [] for a in anchors: ref = a["reference"] row = [ref, a.get("description") or ""] blocks = blocks_by_anchor.get(ref, [])[:n_blocks] for b in blocks: row += [b["reference"], b.get("description") or "", b.get("status") or "", b.get("display_sequence")] # pad to n_blocks*4 row += [""] ((n_blocks - len(blocks)) 4) rows.append(row) return rows `

• [ ] Step 4: Run full pure suite — venv\Scripts\python.exe -m pytest tests/test_complement_families.py -v → PASS (5 tests).
• [ ] Step 5: Commit — git commit -am "feat(complements-v2): build_wide_rows + WIDE_HEADER (baseline layout)"

---

Task 6: ai_typing.profile_family — libellé + résumé d'une famille (IA, stubbé)

Files: Modify core/ai_typing.py · Test tests/test_ai_typing.py

Réutilise llm_call injectable + _parse_json_obj. En test on injecte un faux ; en prod l'orchestrateur passe get_llm_backend(...).

• [ ] Step 1: Write the failing test

`python # tests/test_ai_typing.py (append) import json from core.ai_typing import profile_family

def test_profile_family_parses_label_and_summary(): captured = {}

def fake_llm(prompt: str) -> str: captured["prompt"] = prompt return json.dumps({"label": " Cafés ", "summary": "Cafés moulus et en grains pour la pause."})

out = profile_family("001002", ["Café grains Lavazza — 1KG", "Café moulu Miko — 250G"], llm_call=fake_llm) assert out == {"family_code": "001002", "label": "Cafés", "summary": "Cafés moulus et en grains pour la pause."} # sample lines were passed into the prompt assert "Lavazza" in captured["prompt"] # malformed JSON -> graceful empty-ish profile (never raises) bad = profile_family("009", ["x"], llm_call=lambda p: "not json") assert bad["family_code"] == "009" and bad["label"] == "" `

• [ ] Step 2: Run test → FAIL (ImportError).

• [ ] Step 3: Write minimal implementation

`python # core/ai_typing.py (append) FAMILY_PROFILE_PROMPT = """Tu décris une FAMILLE de produits B2B (catalogue \ Lyreco) à partir d'un échantillon de produits réels. Donne un libellé court et \ une phrase de contenu — base-toi UNIQUEMENT sur l'échantillon, n'invente rien.

Échantillon de la famille : {sample}

Réponds UNIQUEMENT par un objet JSON \ {{"label": "<2-4 mots>", "summary": ""}}, \ sans texte autour."""

def profile_family(family_code: str, sample_lines, llm_call) -> dict: """Profil IA d'une famille : {family_code, label, summary}.

sample_lines = sortie de complement_families.build_family_profile_sample. Tolérant : une réponse inexploitable renvoie label/summary vides plutôt que de lever (l'orchestrateur garde la famille avec un profil pauvre).""" prompt = FAMILY_PROFILE_PROMPT.format( sample="\n".join(f"- {l}" for l in sample_lines)) try: parsed = _parse_json_obj(llm_call(prompt)) except Exception: parsed = {} return { "family_code": family_code, "label": (parsed.get("label") or "").strip(), "summary": (parsed.get("summary") or "").strip(), } `

• [ ] Step 4: Run test → PASS.
• [ ] Step 5: Commit — git add core/ai_typing.py tests/test_ai_typing.py && git commit -m "feat(complements-v2): ai_typing.profile_family (label+summary)"

---

Task 7: ai_typing.propose_family_complements — graphe famille→familles A/B (IA, stubbé)

Files: Modify core/ai_typing.py · Test tests/test_ai_typing.py

Pour une famille ancre (profil) + une liste de familles candidates (profils), l'IA renvoie les familles complémentaires classées A/B avec force+justification, choisies UNIQUEMENT dans les candidates. On filtre tout comp_family hors-liste et l'auto-référence.

• [ ] Step 1: Write the failing test

`python # tests/test_ai_typing.py (append) import json from core.ai_typing import propose_family_complements

def test_propose_family_complements_filters_and_normalises(): anchor = {"family_code": "001002", "label": "Cafés", "summary": "Cafés."} candidates = [ {"family_code": "001005", "label": "Gobelets", "summary": "Gobelets, serviettes."}, {"family_code": "001001", "label": "Biscuits", "summary": "Biscuits."}, ]

def fake_llm(prompt: str) -> str: return json.dumps([ {"comp_family": "001005", "klass": "a", "relation": "accessoire-de", "force": 0.9, "justification": "le café se sert dans des gobelets"}, {"comp_family": "001001", "klass": "B", "relation": "complément-usage", "force": 0.6, "justification": "biscuits avec le café"}, {"comp_family": "001002", "klass": "A", "force": 1.0, "justification": "auto-référence -> ignorée"}, {"comp_family": "999999", "klass": "A", "force": 1.0, "justification": "hors candidats -> ignorée"}, ])

edges = propose_family_complements(anchor, candidates, llm_call=fake_llm) by_fam = {e["comp_family"]: e for e in edges} assert set(by_fam) == {"001005", "001001"} # self + unknown dropped assert by_fam["001005"]["klass"] == "A" # upper-cased assert by_fam["001005"]["force"] == 0.9 assert by_fam["001001"]["klass"] == "B" # bad klass defaults to "B" (conservative: not a hard A claim) bad = propose_family_complements( anchor, candidates, llm_call=lambda p: json.dumps([{"comp_family": "001005", "klass": "Z", "force": 0.5}])) assert bad[0]["klass"] == "B" `

• [ ] Step 2: Run test → FAIL.

• [ ] Step 3: Write minimal implementation

`python # core/ai_typing.py (append) FAMILY_COMPLEMENT_PROMPT = """Tu raisonnes sur la COMPLÉMENTARITÉ entre FAMILLES \ de produits B2B. NE te base PAS sur des ventes : seulement sur l'usage réel.

FAMILLE ANCRE : {anchor_label} ({anchor_code}) — {anchor_summary}

Parmi les familles candidates ci-dessous UNIQUEMENT, lesquelles sont \ complémentaires de l'ancre, et de quelle nature ?

• klass "A" = CONSOMMABLE / nécessaire pour utiliser/consommer l'ancre \

• klass "B" = complément d'usage, acheté dans le même contexte sans être \

Familles candidates : {candidates}

Réponds UNIQUEMENT par un tableau JSON \ [{{"comp_family": "", "klass": "A|B", \ "relation": "", \ "force": <0..1>, "justification": ""}}], vide [] si aucune. \ N'invente aucun code de famille."""

def propose_family_complements(anchor_profile: dict, candidate_profiles, llm_call) -> list: """Graphe famille→familles A/B pour une ancre.

Filtre les familles hors candidate_profiles et l'auto-référence. klass normalisée en 'A'/'B' (toute valeur inattendue → 'B', conservateur : on ne revendique pas une relation A douteuse). Renvoie list[{comp_family, klass, relation, force, justification}].""" cand_codes = {c["family_code"] for c in candidate_profiles} cand_lines = "\n".join( f"- {c['family_code']}: {c.get('label','')} — {c.get('summary','')}" for c in candidate_profiles) prompt = FAMILY_COMPLEMENT_PROMPT.format( anchor_code=anchor_profile["family_code"], anchor_label=anchor_profile.get("label", ""), anchor_summary=anchor_profile.get("summary", ""), candidates=cand_lines, ) try: items = _parse_json_obj_list(llm_call(prompt)) except Exception: return [] out = [] for it in items: fam = str(it.get("comp_family", "")).strip() if not fam or fam == anchor_profile["family_code"] or fam not in cand_codes: continue klass = str(it.get("klass", "")).strip().upper() if klass not in ("A", "B"): klass = "B" out.append({ "comp_family": fam, "klass": klass, "relation": it.get("relation") or "autre", "force": float(it.get("force", 0.0) or 0.0), "justification": it.get("justification") or "", }) return out `

• [ ] Step 4: Run full IA suite — venv\Scripts\python.exe -m pytest tests/test_ai_typing.py -v → PASS (anciens + 2 nouveaux).
• [ ] Step 5: Commit — git commit -am "feat(complements-v2): ai_typing.propose_family_complements (A/B graph)"

---

Task 8: scripts/refresh_product_complements_v2.py — orchestration + XLSX/JSON

Files: Create scripts/refresh_product_complements_v2.py

Intégration (DB + LLM réels) : pas de test unitaire — vérifiée par un run réel sur les 20 ancres. Toute la logique calculatoire est déjà couverte (Tasks 1-7).

• [ ] Step 1: Write the script

`python # scripts/refresh_product_complements_v2.py """Refresh 'Produits complémentaires v2' (FR) — ancrage taxonomie famille.

20 ancres du fichier source -> enrichissement ecom_products (famille/world/marque) -> profils familles (IA, caché) sur l'univers same-world -> graphe famille->familles A/B (IA) -> ventes (timescale) -> 3 vues (A / B / Mix) au format large -> XLSX (README + A + B + Mix + Baseline) + JSON. Méthode : docs/superpowers/specs/2026-06-08-product-complements-v2-design.md

Run : venv\\Scripts\\python.exe -m scripts.refresh_product_complements_v2 --n-anchors 20 venv\\Scripts\\python.exe -m scripts.refresh_product_complements_v2 --model sonnet --llm-backend cli """ from __future__ import annotations

import argparse import json import sys from collections import Counter, defaultdict from datetime import datetime, timezone from pathlib import Path

_REPO = Path(__file__).resolve().parent.parent if str(_REPO) not in sys.path: sys.path.insert(0, str(_REPO))

import openpyxl from database.crm_db import CRMDatabase from database.timeseries_db import TimeseriesDatabase from core.complement_families import ( rank_family_skus, build_family_profile_sample, order_complement_families, assemble_anchor_blocks, build_wide_rows, WIDE_HEADER, ) from core.ai_typing import profile_family, propose_family_complements, get_llm_backend

COUNTRY = "FR" EXPORT_DIR = _REPO / "exports" SOURCE_XLSX = _REPO / "imports" / "lucas" / "20260608_094900_Produits_comple_mentaires_20260526_1.xlsx" PROFILE_CACHE = EXPORT_DIR / "complements_v2_family_profiles.json" SALES_DAYS = 365

def step(msg: str) -> None: print(f" [{datetime.now().strftime('%H:%M:%S')}] {msg}", flush=True)

def pad(code: str) -> str: return str(code).strip().zfill(18)

def read_anchors(n: int) -> list: """n premières ancres du fichier source (ordre du fichier).""" wb = openpyxl.load_workbook(SOURCE_XLSX, read_only=True) ws = wb["Sheet1"] anchors = [] for r in ws.iter_rows(min_row=2, values_only=True): if r[0] is None or str(r[0]).strip() == "": continue anchors.append({"reference": pad(r[0]), "raw": str(r[0]).strip(), "description": (r[1] or "")}) if len(anchors) >= n: break return anchors

def compute(n_anchors: int, max_complements: int, llm_call) -> dict: crm = CRMDatabase() ts = TimeseriesDatabase()

step(f"Phase 1/6 — lecture {n_anchors} ancres + enrichissement ecom_products…") anchors = read_anchors(n_anchors) refs = [a["reference"] for a in anchors] meta = {r["product_reference"]: r for r in crm.query(""" SELECT p.product_reference, p.family_code, p.brand, p.manufacturer, p.product_description, t.world FROM ecom_products p LEFT JOIN section_taxonomy t ON p.section_code = t.section_code AND p.source_country = t.source_country WHERE p.source_country = %s AND p.product_reference = ANY(%s) """, (COUNTRY, refs))} for a in anchors: m = meta.get(a["reference"], {}) a["family_code"] = m.get("family_code") a["world"] = m.get("world") anchor_families = sorted({a["family_code"] for a in anchors if a["family_code"]}) worlds = sorted({a["world"] for a in anchors if a["world"]}) step(f" → {len(anchors)} ancres · familles {anchor_families} · worlds {worlds}")

step("Phase 2/6 — univers candidat (familles same-world) + profils IA (caché)…") cand_rows = crm.query(""" SELECT p.family_code, p.product_reference, p.web_title, p.product_description, p.not_salable_flag, p.not_visible_flag, p.status_code FROM ecom_products p JOIN section_taxonomy t ON p.section_code = t.section_code AND p.source_country = t.source_country WHERE p.source_country = %s AND t.world = ANY(%s) """, (COUNTRY, worlds)) fam_products: dict[str, list] = defaultdict(list) for r in cand_rows: fam_products[r["family_code"]].append(r) candidate_families = sorted(f for f in fam_products if f and f not in ("001000",)) step(f" → {len(candidate_families)} familles candidates · {len(cand_rows)} produits")

cache = json.loads(PROFILE_CACHE.read_text(encoding="utf-8")) if PROFILE_CACHE.exists() else {} profiles: dict[str, dict] = {} for fam in sorted(set(anchor_families) | set(candidate_families)): if fam in cache: profiles[fam] = cache[fam] continue sample = build_family_profile_sample(fam_products.get(fam, []), sample_size=12) profiles[fam] = profile_family(fam, sample, llm_call) cache[fam] = profiles[fam] PROFILE_CACHE.write_text(json.dumps(cache, ensure_ascii=False, indent=2), encoding="utf-8") step(f" → {len(profiles)} profils familles")

step("Phase 3/6 — graphe famille→familles A/B (IA)…") edges_by_family: dict[str, list] = {} for fam in anchor_families: cand_profiles = [profiles[f] for f in candidate_families if f != fam] edges_by_family[fam] = propose_family_complements( profiles[fam], cand_profiles, llm_call) n_edges = sum(len(v) for v in edges_by_family.values()) step(f" → {n_edges} arêtes famille (A/B)")

step(f"Phase 4/6 — ventes FR (timescale, {SALES_DAYS}j) + tri SKU par famille…") comp_families = sorted({e["comp_family"] for v in edges_by_family.values() for e in v}) comp_refs = [r["product_reference"] for r in cand_rows if r["family_code"] in comp_families] sales: Counter = Counter() for start in range(0, len(comp_refs), 20000): for r in ts.query(""" SELECT product_reference, SUM(quantity)::bigint AS q FROM ecom_order_lines WHERE source_country = %s AND order_date >= CURRENT_DATE - (%s || ' days')::INTERVAL AND product_reference = ANY(%s) GROUP BY product_reference """, (COUNTRY, SALES_DAYS, comp_refs[start:start + 20000])): sales[r["product_reference"]] = int(r["q"] or 0) family_skus: dict[str, list] = {} for fam in comp_families: cands = [{"reference": r["product_reference"], "description": r["product_description"] or r["web_title"] or "", "sales": sales.get(r["product_reference"], 0), "not_salable": r["not_salable_flag"], "not_visible": r["not_visible_flag"], "status_code": r["status_code"]} for r in cand_rows if r["family_code"] == fam] family_skus[fam] = rank_family_skus(cands, top_n=max_complements) step(f" → {len(comp_families)} familles complément dépliées en SKU")

step("Phase 5/6 — assemblage 3 vues (A / B / Mix)…") views = {"A": {}, "B": {}, "mix": {}} for a in anchors: fam = a["family_code"] edges = edges_by_family.get(fam, []) views["A"][a["reference"]] = assemble_anchor_blocks( order_complement_families(edges, klass="A"), family_skus, max_complements) views["B"][a["reference"]] = assemble_anchor_blocks( order_complement_families(edges, klass="B"), family_skus, max_complements) views["mix"][a["reference"]] = assemble_anchor_blocks( order_complement_families(edges), family_skus, max_complements)

step("Phase 6/6 — baseline (site actuel) pour comparaison…") baseline = crm.query(""" SELECT anchor_reference, complement_reference, complement_description, complement_status, display_sequence FROM product_complements WHERE source_country = %s AND anchor_reference = ANY(%s) ORDER BY anchor_reference, display_sequence """, (COUNTRY, refs))

return { "computed_at": datetime.now(timezone.utc).isoformat(), "country": COUNTRY, "n_anchors": len(anchors), "max_complements": max_complements, "anchor_families": anchor_families, "worlds": worlds, "candidate_families": candidate_families, "n_edges": n_edges, "anchors": anchors, "profiles": profiles, "edges_by_family": edges_by_family, "views": views, "baseline": baseline, }

def write_xlsx(out_path: Path, data: dict, n_blocks: int) -> None: from openpyxl import Workbook from openpyxl.styles import Alignment, Font, PatternFill from openpyxl.utils import get_column_letter

wb = Workbook() head_fill = PatternFill("solid", fgColor="2A2A38") head_font = Font(bold=True, color="FFFFFF")

def sheet(name): ws = wb.active if wb.sheetnames == ["Sheet"] else wb.create_sheet(name) ws.title = name return ws

def styled_header(ws, headers): ws.append(headers) for c in ws[1]: c.fill = head_fill; c.font = head_font c.alignment = Alignment(vertical="center", wrap_text=True) ws.freeze_panes = "A2"

# README ws = sheet("README") for line in [ ["Produits complémentaires v2 — FR — ancrage taxonomie famille"], [f"Généré : {data['computed_at']} · {data['n_anchors']} ancres · " f"worlds {', '.join(data['worlds'])}"], [], ["Méthode : la catégorie n'est plus devinée depuis le titre SAP mais LUE " "dans ecom_products (famille warehouse)."], ["L'IA propose des FAMILLES complémentaires classées A (consommable/" "nécessaire) ou B (complément d'usage),"], ["puis on déplie en SKU actifs (ni not_salable ni not_visible) triés par " "ventes FR (365j)."], [], ["Onglets : A — consommables · B — compléments larges · Mix (A+B) · " "Baseline (site actuel)."], ["Format large : col A = ancre SAP, col B = description, puis 20 blocs " "[code complément · description · statut · display_sequence]."], [f"Familles candidates (same-world) : {', '.join(data['candidate_families'])}"], [f"Arêtes famille A/B proposées : {data['n_edges']}"], ]: ws.append(line) ws.column_dimensions["A"].width = 110

anchors = data["anchors"] header = WIDE_HEADER(n_blocks) for key, title in [("A", "A — consommables"), ("B", "B — compléments larges"), ("mix", "Mix (A+B)")]: ws = sheet(title) styled_header(ws, header) for row in build_wide_rows(anchors, data["views"][key], n_blocks=n_blocks): ws.append(row) ws.column_dimensions["A"].width = 20 ws.column_dimensions["B"].width = 38

# Baseline (long format, read-only comparison) ws = sheet("Baseline (site actuel)") styled_header(ws, ["Anchor SAP", "Complement SAP", "Complement Description", "Status", "Display Sequence"]) for b in data["baseline"]: ws.append([b["anchor_reference"], b["complement_reference"], b["complement_description"], b["complement_status"], b["display_sequence"]])

# Graphe familles (audit) ws = sheet("Graphe familles") styled_header(ws, ["anchor_family", "comp_family", "comp_label", "klass", "relation", "force", "justification"]) prof = data["profiles"] for fam, edges in data["edges_by_family"].items(): for e in sorted(edges, key=lambda x: (x["klass"], -x["force"])): ws.append([fam, e["comp_family"], (prof.get(e["comp_family"]) or {}).get("label", ""), e["klass"], e["relation"], round(e["force"], 2), (e["justification"] or "")[:160]]) for col, w in [("A", 14), ("B", 14), ("C", 22), ("G", 70)]: ws.column_dimensions[col].width = w

wb.save(out_path)

def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--n-anchors", type=int, default=20, dest="n_anchors") ap.add_argument("--max-complements", type=int, default=20, dest="max_complements") ap.add_argument("--model", default="sonnet") ap.add_argument("--llm-backend", default="cli", choices=["cli", "api"], dest="llm_backend") ap.add_argument("--allow-api-billing", action="store_true", dest="allow_api_billing") args = ap.parse_args()

EXPORT_DIR.mkdir(parents=True, exist_ok=True) llm_call = get_llm_backend(prefer=args.llm_backend, model=args.model, allow_api_billing=args.allow_api_billing)

data = compute(args.n_anchors, args.max_complements, llm_call)

today = datetime.now(timezone.utc).strftime("%Y-%m-%d") out_xlsx = EXPORT_DIR / f"product_complements_v2_{COUNTRY}_{today}.xlsx" out_json = out_xlsx.with_suffix(".json") step(f"écriture {out_json.name}…") out_json.write_text(json.dumps(data, indent=2, default=str, ensure_ascii=False), encoding="utf-8") step(f"écriture {out_xlsx.name}…") write_xlsx(out_xlsx, data, n_blocks=args.max_complements)

print("\n" + "=" * 70) print("PRODUITS COMPLÉMENTAIRES v2 — FR — SNAPSHOT") print("=" * 70) print(f" Ancres: {data['n_anchors']} · Familles ancre: {data['anchor_families']}") print(f" Familles candidates: {len(data['candidate_families'])} · " f"Arêtes A/B: {data['n_edges']}") print(f" exports/{out_xlsx.name}") return 0

if __name__ == "__main__": sys.exit(main()) `

• [ ] Step 2: Smoke-run on the 20 anchors

Run: venv\Scripts\python.exe -m scripts.refresh_product_complements_v2 --n-anchors 20 --model sonnet --llm-backend cli Expected: termine sans exception ; écrit exports/product_complements_v2_FR_.xlsx + .json ; récap avec Familles ancre: ['001001', '001002'], arêtes A/B non nulles. (Nécessite claude sur le PATH pour la voie abonnement + DBs Docker up. La phase profils est cachée → un 2e run est quasi instantané.)

• [ ] Step 3: Eyeball the output

Ouvrir l'XLSX : (a) onglet A — consommables : pour les cafés (001002), les blocs doivent pointer des gobelets/serviettes (001005) actifs ; (b) onglet B : biscuits/boissons dans le même contexte ; (c) Graphe familles : les comp_family ∈ candidates same-world, klass A/B cohérent, justification lisible ; (d) Baseline : les compléments actuels du site pour comparer. Noter qu'A peut être clairsemé pour les ancres consommables (attendu, cf. spec §6).

• [ ] Step 4: Run the full pure + IA suite once more

Run: venv\Scripts\python.exe -m pytest tests/test_complement_families.py tests/test_ai_typing.py -v Expected: PASS (5 pures + tests IA).

• [ ] Step 5: Commit — git add scripts/refresh_product_complements_v2.py && git commit -m "feat(complements-v2): refresh script (anchors->families A/B->SKU->xlsx/json)"

---

Task 9 (OPTIONNELLE — après relecture Pierre/Marie des 20 lignes): extension + persistance

À ne faire qu'après que Pierre/Marie aient relu les 20 lignes (garde-fou spec §2 « relu par Pierre/Marie sur les 20 lignes, puis étendues »). Deux extensions possibles, non requises pour la validation :

• --n-anchors au-delà de 20 (étendre l'univers candidat au-delà du same-world si A trop clairsemé pour les durables) ;
• persistance postgres (product_complements_v2 : ancre, complément, klass, force, display_sequence, computed_at) pour brancher un panel. Vérifier l'API réelle de CRMDatabase (présence d'executemany) avant d'écrire l'upsert — ne pas présumer.

---

Self-Review

• Spec coverage : §4 Phase 1 (anchors+enrich)→Task 8 P1 ; §4 Phase 2 (profils famille)→Tasks 2+6, orchestré P2 ; §4 Phase 3 (graphe A/B)→Task 7, orchestré P3 ; resserrage A durable→hook noté (anchors CAFETERIA = consommables, A sémantique, conforme §4) ; §4 Phase 4 (expansion famille→SKU, actifs, tri ventes)→Tasks 1+4, orchestré P4 ; §4 Phase 5 (3 vues + display_sequence)→Tasks 3+4+5, orchestré P5 ; §4 Phase 6 / §Sorties (README+A+B+Mix+Baseline, format large)→Task 8 write_xlsx ; §5 architecture (core pur + ai_typing réutilisé + script)→respectée ; §6 garde-fous (statut actif dur, A clairsemé attendu, baseline read-only)→Tasks 1 + 8 ; §7 hors-périmètre (multi-saut, tout catalogue, écriture base, GB)→Task 9 optionnelle / non planifié.
• Faits warehouse : tous vérifiés par introspection (colonnes, 206 familles, status_code numérique → filtre via flags, 20 ancres 100 % jointes en 2 familles CAFETERIA, ventes sur timescale). Aucun chemin supposé.
• Type consistency : family_code = chaîne partout ; arête famille {comp_family,klass,force,relation,justification} produite par Task 7, consommée par Tasks 3/4/8 ; SKU candidat {reference,description,sales,not_salable,not_visible,status_code} produit en P4, consommé par Task 1 → bloc {reference,description,status,display_sequence,klass,comp_family} consommé par Tasks 4/5 ; family_skus: {family_code: [SKU triés]} cohérent Task 1↔4↔8 ; signature llm_call(prompt)->str identique Tasks 6/7 et orchestrateur (get_llm_backend). build_wide_rows consomme exactement views[key] (blocks_by_anchor) + anchors.
• Placeholders : aucun TODO/TBD ; tout le code des steps est complet. La note « vérifier l'API CRMDatabase » est dans la Task optionnelle 9.