spb/fabri-ka

1#!/usr/bin/env python32"""Champ structuré `certifications` sur les fiches boutiques (vague 4).34Deux gisements :5  1. Les sources de découverte qui SONT des certifications/labels vérifiés6     (Aliments du Québec, Les Produits du Québec, Économusée, Arrêts7     gourmands) -> label dérivé de discovery_sources.8  2. Le répertoire public CARTV/SIPAB des entreprises certifiées bio9     (data/raw/cartv_bio.jsonl, moissonné par wave4_discovery.py) ->10     appariement par domaine (fort) puis par nom normalisé unique11     (prudent), avec certificateur + date.1213Écrit dans la DB (stores.certifications, JSON) ET dans data/stores.json14(champ `certifications`) — ⚠ sérialiser avec les autres écrivains du15registre. Rejouable (fusion sans doublon par `label`).1617Usage : .venv/bin/python scripts/backfill_certifications.py18"""19import json20import re21import sys22import unicodedata23from pathlib import Path2425ROOT = Path(__file__).resolve().parent.parent26sys.path.insert(0, str(ROOT))27sys.path.insert(0, str(ROOT / "scripts"))2829from fabrika import db            # noqa: E40230from aggregate import norm_domain  # noqa: E4023132SOURCE_LABELS = {33    "alimentsduquebec_ent": ("Aliments du Québec",34                             "Certification Aliments du Québec (adhérent vérifié)"),35    "lesproduitsduquebec_ent": ("Les Produits du Québec",36                                "Certification Les Produits du Québec"),37    "artisansaloeuvre": ("Économusée", "Accréditation réseau Économusée / "38                                       "Artisans à l'œuvre"),39    "arretsgourmands": ("Arrêt gourmand certifié",40                        "Arrêt gourmand certifié (Chaudière-Appalaches)"),41}4243STOP = {"inc", "enr", "ltee", "les", "la", "le", "de", "du", "des", "et"}444546def norm_name(name):47    s = unicodedata.normalize("NFKD", name or "").encode("ascii", "ignore").decode().lower()48    return " ".join(t for t in re.findall(r"[a-z0-9]+", s) if t not in STOP)495051def main():52    con = db.connect()53    stores = {r["id"]: dict(r) for r in con.execute(54        "SELECT id, name, discovery_sources, certifications FROM stores")}55    certs = {sid: {c["label"]: c for c in json.loads(s.get("certifications") or "[]")}56             for sid, s in stores.items()}5758    # 1) labels dérivés des sources de découverte certifiantes59    n_src = 060    for sid, s in stores.items():61        for src in json.loads(s.get("discovery_sources") or "[]"):62            if src in SOURCE_LABELS:63                label, detail = SOURCE_LABELS[src]64                if label not in certs[sid]:65                    certs[sid][label] = {"label": label, "detail": detail,66                                         "source": src}67                    n_src += 16869    # 2) répertoire CARTV bio70    cartv_path = ROOT / "data" / "raw" / "cartv_bio.jsonl"71    n_dom = n_name = 072    if cartv_path.exists():73        recs = [json.loads(l) for l in open(cartv_path)]74        by_dom = {}75        for r in recs:76            for w in r.get("websites", []):77                d = norm_domain(w)78                if d:79                    by_dom.setdefault(d, r)80        # index nom normalisé -> fiches CARTV (pour l'appariement prudent)81        by_name = {}82        for r in recs:83            by_name.setdefault(norm_name(r["title"]), []).append(r)84        name_to_sids = {}85        for sid, s in stores.items():86            name_to_sids.setdefault(norm_name(s["name"]), []).append(sid)8788        def add_bio(sid, r, how):89            label = "Biologique — CARTV"90            if label in certs[sid]:91                return 092            certs[sid][label] = {93                "label": label,94                "detail": "Appellation biologique du Québec (répertoire public "95                          "CARTV/SIPAB)",96                "certifier": r.get("certifier") or None,97                "since": r.get("cert_date") or None,98                "operations": r.get("operations") or [],99                "source": "cartv_bio", "match": how,100            }101            return 1102103        for sid in stores:104            if sid in by_dom:105                n_dom += add_bio(sid, by_dom[sid], "domaine")106        for nname, rs in by_name.items():107            if len(rs) != 1 or not nname or len(nname) < 6:108                continue109            sids = name_to_sids.get(nname, [])110            if len(sids) == 1 and "Biologique — CARTV" not in certs[sids[0]]:111                n_name += add_bio(sids[0], rs[0], "nom_unique")112113    # écriture DB114    n_stores = 0115    for sid, cmap in certs.items():116        if not cmap:117            continue118        con.execute("UPDATE stores SET certifications=? WHERE id=?",119                    (json.dumps(list(cmap.values()), ensure_ascii=False), sid))120        n_stores += 1121    con.commit(); con.close()122123    # écriture registre (champ informatif, la DB fait foi pour l'API)124    reg_path = ROOT / "data" / "stores.json"125    reg = json.load(open(reg_path))126    for s in reg["stores"]:127        cmap = certs.get(s["id"])128        if cmap:129            s["certifications"] = list(cmap.values())130    json.dump(reg, open(reg_path, "w"), ensure_ascii=False, indent=1)131132    print(f"certifications : {n_stores} boutiques porteuses "133          f"(+{n_src} labels de sources, +{n_dom} bio par domaine, "134          f"+{n_name} bio par nom unique)")135136137if __name__ == "__main__":138    main()139