"""Phase 2.6 — Validation of the cross-merchant matching tiers on a labelled pair sample.

sample   : 60 random pairs per tier (A, B, C, D), seed 20261101 -> data/processed/match_validation_sample.csv (never overwritten)
annotate : two independent LLM annotators decide whether the two listings are the SAME product (same brand, same item,
           same format) sold by two different merchants: {"same": "yes"|"no"|"unsure", "reason": "..."}; usage logged
report   : precision per tier (share 'yes' among consensus labels), disagreement list for operator adjudication
Usage: python -m src.validation.match_validation sample | annotate | report
"""
from __future__ import annotations
import json, 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, write_json, BASE_PROC, BASE_EXP

log = get_logger("validation.match")
load_dotenv(ROOT / ".env")
SAMPLE = BASE_PROC / "match_validation_sample.csv"
SYSTEM = ("Tu es un expert en appariement de catalogues. On te montre deux fiches produit provenant de deux boutiques en ligne québécoises "
          "différentes. Décide si elles décrivent le MÊME produit (même marque/fabricant, même article, même format/contenance ; des variantes de "
          "couleur ou de taille du même article comptent comme le même produit si le prix affiché est le prix de base). Deux produits artisanaux "
          "similaires faits par des fabricants différents ne sont PAS le même produit. Réponds UNIQUEMENT en JSON: "
          "{\"same\": \"yes\"|\"no\"|\"unsure\", \"reason\": \"<10 mots max>\"}")


def card(r, s):
    return (f"Boutique: {r[f'store_{s}']}\nTitre: {r[f'title_{s}']}\nMarque/vendor: {r[f'vendor_{s}'] or ''}\nSKU: {r[f'sku_{s}'] or ''}\n"
            f"Catégorie marchand: {r[f'mcat_{s}'] or ''}\nPrix: {r[f'price_{s}']} CAD\nVariantes: {r[f'nv_{s}']}\nDescription: {str(r[f'desc_{s}'] or '')[:250]}")


def sample():
    if SAMPLE.exists():
        log.error("%s exists — refusing to overwrite", SAMPLE); return
    pr = pd.read_parquet(P["interim"] / "match_pairs.parquet")
    li = pd.read_parquet(P["interim"] / "listings_clean.parquet").set_index("uid")
    s = pr.groupby("tier", group_keys=False).apply(lambda g: g.sample(min(60, len(g)), random_state=SEED))
    for side in ("a", "b"):
        u = s[f"uid_{side}"]
        s[f"title_{side}"] = u.map(li.title); s[f"vendor_{side}"] = u.map(li.vendor); s[f"sku_{side}"] = u.map(li.sku)
        s[f"mcat_{side}"] = u.map(li.merchant_categories); s[f"desc_{side}"] = u.map(li.description_head)
    cols = ["uid_a", "uid_b", "tier", "score", "comparable", "store_a", "store_b", "price_a", "price_b", "nv_a", "nv_b"] + \
           [f"{k}_{s}" for s in ("a", "b") for k in ("title", "vendor", "sku", "mcat", "desc")]
    s[cols].to_csv(SAMPLE, index=False); log.info("sample %s pairs", len(s))


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


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


def parse(out):
    try:
        j = json.loads(out[out.find("{"): out.rfind("}") + 1]); v = str(j.get("same", "")).lower()
        return v if v in ("yes", "no", "unsure") else "INVALID"
    except Exception:
        return "PARSE_ERROR"


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():
            continue
        if col not in df.columns: df[col] = 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]; text = "FICHE 1\n" + card(r, "a") + "\n\nFICHE 2\n" + card(r, "b")
            for k in range(6):
                try: return i, *fn(text)
                except Exception as e: time.sleep(min(60, 3 * 2 ** k)); err = str(e)[:100]
            return i, f"ERROR {err}", {"in": 0, "out": 0}
        with cf.ThreadPoolExecutor(6) as ex:
            for i, out, u in ex.map(work, todo):
                df.loc[i, col] = parse(out); usage["in"] += u["in"]; usage["out"] += u["out"]; usage["calls"] += 1; usage["errors"] += out.startswith("ERROR")
        df.to_csv(SAMPLE, index=False); ledger("match_validation.annotate", annotator=col, **usage); log.info("%s %s", col, usage)


def report():
    df = pd.read_csv(SAMPLE)
    cons = df.llm_a.where(df.llm_a == df.llm_b)
    out = {"n": len(df), "agree_llm": round(float((df.llm_a == df.llm_b).mean()), 4), "consensus_share": round(float(cons.notna().mean()), 4)}
    out["precision_by_tier_consensus"] = {t: {"n": int((cons.notna() & (df.tier == t)).sum()), "yes": round(float((cons[df.tier == t] == "yes").mean()), 3),
                                              "no": round(float((cons[df.tier == t] == "no").mean()), 3)} for t in sorted(df.tier.unique())}
    out["precision_by_tier_comparable_only"] = {t: round(float((cons[(df.tier == t) & df.comparable] == "yes").mean()), 3) for t in sorted(df.tier.unique())}
    if "operator" in df.columns:
        out["precision_by_tier_operator"] = {t: round(float((df.loc[df.tier == t, "operator"] == "yes").mean()), 3) for t in sorted(df.tier.unique())}
    dis = df[(df.llm_a != df.llm_b) | (df.llm_a == "unsure")]
    dis.to_csv(BASE_PROC / "match_validation_disagreements.csv", index=False); out["n_disagreements"] = len(dis)
    write_json(out, BASE_EXP / "match_validation.json", "match_validation.report"); log.info(json.dumps(out))


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