"""Phase 2.4 — Audit of the rule-based category labels on a stratified random sample.

Step 1 (sample)   : 25 listings per category (20 categories) from the assortment sample, seed 20261101
                    -> data/processed/category_audit_sample.csv   (NEVER overwritten once annotations exist)
Step 2 (annotate) : two independent LLM annotators (Anthropic claude-opus-5-5, OpenAI gpt-5.5) label each listing with one of
                    the 20 canonical keys using the platform's own taxonomy definitions; every call is logged with usage.
Step 3 (report)   : agreement matrices, per-category precision of the rule label against the LLM consensus,
                    list of disagreements for operator adjudication -> experiments/category_audit.json

Usage: python -m src.validation.category_audit sample | annotate | report
"""
from __future__ import annotations
import json, os, sys, time, concurrent.futures as cf
import pandas as pd
from dotenv import load_dotenv
from src.common import ROOT, P, SEED, get_logger, ledger, register_output, write_json, BASE_PROC, BASE_EXP

log = get_logger("validation.category_audit")
load_dotenv(ROOT / ".env")
SAMPLE = BASE_PROC / "category_audit_sample.csv"
TAXO = {
 "erable": "Érable & miel: sirop, beurre, flocons d'érable; miel, pollen, hydromel non alcoolisé",
 "epicerie": "Épicerie fine: confitures, sauces, condiments, épices, vinaigres, huiles, farines, granolas, collations, marinades, produits d'épicerie en général (hors catégories plus spécifiques)",
 "cafe_the": "Café & thé: café torréfié, thés, tisanes, matcha, infusions",
 "chocolat": "Chocolat & confiseries: chocolat, caramels, guimauves, bonbons",
 "boulangerie": "Boulangerie & pâtisserie: pains, biscuits, pâtisseries",
 "viande_poisson": "Viandes & poissons: boucherie, charcuterie, jerky, gibier, fumoir, poissons, fruits de mer",
 "fromage_laitier": "Fromages & produits laitiers: fromages, yogourts, crème glacée",
 "boissons": "Boissons non alcoolisées: kombucha, limonade, jus, sodas, eau",
 "alcool": "Bières, vins, cidres, spiritueux, hydromel alcoolisé",
 "sante_beaute": "Beauté & soins: savons, cosmétiques, soins du corps/barbe/cheveux, parfums, suppléments et produits de santé",
 "maison": "Maison & déco: chandelles, céramique, ébénisterie, textiles de maison, literie, vaisselle, décoration, meubles",
 "mode": "Mode & accessoires: vêtements, tuques, lainages, maroquinerie, chaussures, sacs, accessoires de mode, lunettes",
 "bijoux": "Bijoux: joaillerie, bijoux faits main, montres",
 "art": "Art & artisanat: estampes, affiches, illustrations, photographie, papeterie, verre soufflé, sculpture, macramé, livres et objets culturels, fournitures d'artisanat",
 "enfants": "Enfants & bébés: jouets, doudous, vêtements enfants, puériculture",
 "animaux": "Animaux: gâteries, colliers, laisses, lits, jouets pour animaux, aliments pour animaux",
 "plein_air": "Plein air & sport: camping, raquettes, chasse/pêche, vélo, équipement sportif",
 "jardin": "Jardin & agriculture: semences, plantes, serres, compost, engrais, outils de jardin",
 "industriel": "Manufacturier & spécialisé: outillage, équipements, quincaillerie, composants, pièces, véhicules et remorques, matériaux de construction, fournitures professionnelles",
 "autre": "Autre: ne correspond clairement à aucune catégorie ci-dessus",
}
SYSTEM = ("Tu es un annotateur de catalogue. On te donne une fiche produit d'une boutique en ligne québécoise. Choisis la catégorie canonique la plus "
          "appropriée parmi la taxonomie fournie (une seule clé). Réponds UNIQUEMENT en JSON: {\"category\": \"<clé>\", \"confidence\": <0-1>}.\n\nTaxonomie:\n" +
          "\n".join(f"- {k}: {v}" for k, v in TAXO.items()))


def fiche(r) -> str:
    return (f"Titre: {r.title}\nType (marchand): {r.product_type or ''}\nCatégories marchand: {r.merchant_categories or ''}\n"
            f"Tags: {r.tags_list}\nMarque/vendor: {r.vendor or ''}\nPrix: {r.price} {r.currency}\nDescription (début): {str(r.description_head or '')[:300]}")


def sample():
    if SAMPLE.exists():
        log.error("%s exists — refusing to overwrite an annotated sample", SAMPLE); return
    li = pd.read_parquet(P["interim"] / "listings_clean.parquet")
    a = li[li.f_analysis_assort]
    s = a.groupby("category", group_keys=False).apply(lambda g: g.sample(min(25, len(g)), random_state=SEED))
    cols = ["uid", "store_id", "category", "title", "product_type", "merchant_categories", "tags_list", "vendor", "price", "currency", "description_head"]
    s[cols].to_csv(SAMPLE, index=False)
    log.info("sample %s rows -> %s", len(s), SAMPLE)


def _anthropic(text, model="claude-opus-5-5"):
    import anthropic
    c = anthropic.Anthropic()
    t0 = time.time()
    m = c.messages.create(model=model, max_tokens=600, thinking={"type": "adaptive"}, output_config={"effort": "low"}, system=SYSTEM, messages=[{"role": "user", "content": text}])
    out = next((b.text for b in m.content if getattr(b, "type", "") == "text"), "").strip()
    return out, {"in": m.usage.input_tokens, "out": m.usage.output_tokens, "s": round(time.time() - t0, 2)}


def _openai(text, model="gpt-5.5"):
    from openai import OpenAI
    c = OpenAI()
    t0 = time.time()
    m = c.chat.completions.create(model=model, messages=[{"role": "system", "content": SYSTEM}, {"role": "user", "content": text}],
                                  max_completion_tokens=400, response_format={"type": "json_object"})
    out = (m.choices[0].message.content or "").strip()
    return out, {"in": m.usage.prompt_tokens, "out": m.usage.completion_tokens, "s": round(time.time() - t0, 2)}


def parse(out):
    try:
        j = json.loads(out[out.find("{"): out.rfind("}") + 1])
        k = str(j.get("category", "")).strip().lower()
        return (k if k in TAXO else "INVALID"), float(j.get("confidence", float("nan")))
    except Exception:
        return "PARSE_ERROR", float("nan")


def annotate():
    df = pd.read_csv(SAMPLE)
    for col, fn in (("llm_a", _anthropic), ("llm_b", _openai)):
        if col in df.columns and df[col].notna().all() and not df[col].isin(["PARSE_ERROR", "INVALID"]).any():
            log.info("%s already annotated", col); continue
        if col not in df.columns:
            df[col] = None; df[col + "_conf"] = None
        df.loc[df[col].isin(["PARSE_ERROR", "INVALID"]), col] = None
        todo = df.index[df[col].isna()].tolist()
        usage = {"in": 0, "out": 0, "calls": 0, "errors": 0}
        def work(i):
            r = df.loc[i]
            for attempt in range(6):
                try:
                    out, u = fn(fiche(r)); return i, out, u
                except Exception as e:  # noqa: BLE001
                    time.sleep(min(60, 3 * 2 ** attempt)); err = str(e)[:120]
            return i, f"ERROR {err}", {"in": 0, "out": 0, "s": 0}
        with cf.ThreadPoolExecutor(6) as ex:
            for i, out, u in ex.map(work, todo):
                k, conf = parse(out); df.loc[i, col] = k; df.loc[i, col + "_conf"] = conf
                usage["in"] += u["in"]; usage["out"] += u["out"]; usage["calls"] += 1; usage["errors"] += out.startswith("ERROR")
        df.to_csv(SAMPLE, index=False)
        ledger("category_audit.annotate", annotator=col, model=("claude-opus-5-5" if col == "llm_a" else "gpt-5.5"), **usage)
        log.info("%s done: %s", col, usage)


def report():
    df = pd.read_csv(SAMPLE)
    out = {"n": len(df)}
    for a, b in (("category", "llm_a"), ("category", "llm_b"), ("llm_a", "llm_b")):
        out[f"agree_{a}_{b}"] = round(float((df[a] == df[b]).mean()), 4)
    cons = df.llm_a.where(df.llm_a == df.llm_b)
    out["consensus_share"] = round(float(cons.notna().mean()), 4)
    sub = df[cons.notna()]
    out["rule_precision_vs_consensus"] = round(float((sub.category == sub.llm_a).mean()), 4)
    out["per_category_precision_vs_consensus"] = {k: {"n_consensus": int((sub.category == k).sum()), "precision": round(float((sub[sub.category == k].llm_a == k).mean()), 3) if (sub.category == k).any() else None} for k in TAXO}
    conf = pd.crosstab(sub.category, sub.llm_a)
    out["confusion_rule_vs_consensus"] = {r: {c: int(v) for c, v in row.items() if v} for r, row in conf.to_dict("index").items()}
    dis = df[(df.category != df.llm_a) | (df.category != df.llm_b)]
    dis[["uid", "store_id", "category", "llm_a", "llm_b", "title", "product_type"]].to_csv(BASE_PROC / "category_audit_disagreements.csv", index=False)
    out["n_disagreements"] = len(dis)
    write_json(out, BASE_EXP / "category_audit.json", "category_audit.report")
    log.info(json.dumps({k: v for k, v in out.items() if "per_" not in k and "confusion" not in k}))


if __name__ == "__main__":
    {"sample": sample, "annotate": annotate, "report": report}[sys.argv[1]]()