"""Are our NEVER-RETRIEVED true pairs concentrated in empty-address records?

LIFTOVER R4 reports, for bge-out: 49% of retrieval misses were empty-address candidates, and only
52% of those were ever retrieved [M]. And 344,883 pool records (3.34%) genuinely have an empty name
or address on the RAW TSVs.

That matters because our miss decomposition is:
    missed true pairs 2,974
      never retrieved 1,211  (40.7%)   <- the ONLY loss that is not ambiguity
      retrieved but rejected 1,763
and the never-retrieved group is the one remaining lever that a better scorer cannot touch. If it
is concentrated in empty-address records, a name-only index over just those 345K records is a
targeted, ~5-minute CPU fix rather than a full retrieval rebuild.

If it is NOT concentrated there, R4 is a bge-out-specific artefact and model-3 should skip it.
"""
import json
import os
import sys

import joblib
import numpy as np
import pandas as pd

sys.path.insert(0, "src")
import config as C
import decide as DC
from evaluate import macro_f05

RAW = os.environ.get("RAW_DIR", "/scratch/user4/dai/dataset_unstop/dataset/train")
b = joblib.load("work/results_A/model_bundle.joblib")
MM, THR = b["miss_mass"], b["param"]
CACHE = "work/val_scored_20000.parquet"
cand = pd.read_parquet(CACHE)
truths = {int(k): set(v) for k, v in json.load(open(CACHE + ".truth.json")).items()}
ids = sorted(truths)
shipped = DC.predict_sets(cand, "q", "c", "p", b["method"], THR, MM, True)

p = C.PATHS["train"]
s2 = pd.read_parquet(p["source2"], columns=["entity_id"])
s3 = pd.read_parquet(p["source3"], columns=["entity_id"])
n_s2 = len(s2)
pool_ids = np.concatenate([s2["entity_id"].values, s3["entity_id"].values])

raw = {}
for src in (2, 3):
    d = pd.read_csv(os.path.join(RAW, f"train_source{src}.tsv"), sep="\t", dtype=str,
                    keep_default_na=False, na_filter=False)
    for e, n, a in zip(d["entity_id"], d["business_name"], d["business_address"]):
        raw[e] = (n, a)

empty_addr = np.zeros(len(pool_ids), bool)
empty_name = np.zeros(len(pool_ids), bool)
for i, e in enumerate(pool_ids):
    n, a = raw[e]
    empty_addr[i] = not a.strip()
    empty_name[i] = not n.strip()
print(f"pool {len(pool_ids):,}")
print(f"  empty ADDRESS: {int(empty_addr.sum()):>9,}  ({empty_addr.mean():.3%})")
print(f"  empty NAME   : {int(empty_name.sum()):>9,}  ({empty_name.mean():.4%})")

cand_by_q = {int(q): set(g["c"].tolist()) for q, g in cand.groupby("q", sort=False)}

never, rejected, hit = [], [], []
for q in ids:
    t = truths.get(q, set())
    if not t:
        continue
    cs = cand_by_q.get(q, set())
    pred = set(shipped.get(q, []))
    never += list(t - cs)
    rejected += list((t & cs) - pred)
    hit += list(t & pred)
print(f"\ntrue pairs: retrieved+predicted {len(hit):,}  "
      f"retrieved+rejected {len(rejected):,}  never retrieved {len(never):,}")

print("\n" + "=" * 74)
print("EMPTY-ADDRESS RATE BY OUTCOME  (the test of LIFTOVER R4)")
print("=" * 74)
print(f"{'group':<26}{'n':>9}{'empty addr':>13}{'empty name':>13}")
for name, grp in (("predicted correctly", hit), ("retrieved but rejected", rejected),
                  ("NEVER RETRIEVED", never)):
    if not grp:
        continue
    a = np.array(grp)
    print(f"{name:<26}{len(a):>9,}{empty_addr[a].mean():>12.2%}{empty_name[a].mean():>13.3%}")
base = empty_addr.mean()
print(f"{'pool base rate':<26}{len(pool_ids):>9,}{base:>12.2%}")

a = np.array(never)
if len(a):
    lift = empty_addr[a].mean() / max(base, 1e-9)
    print(f"\n  never-retrieved empty-address enrichment vs pool: {lift:.2f}x")
    share = empty_addr[a].mean()
    print(f"  share of never-retrieved misses that are empty-address: {share:.2%}")
    print(f"  LIFTOVER reports 49% for bge-out")
    if share > 0.25:
        recover = len(a) * share
        print(f"\n  => a name-only index over the {int(empty_addr.sum()):,} empty-address pool")
        print(f"     records could reach up to {recover:,.0f} of our {len(a):,} never-retrieved")
        print(f"     pairs. At the measured ~0.0507 F0.5 per recovered TP and 20,000 entities,")
        print(f"     the CEILING is {recover*0.0507/len(ids):+.5f} - and only if every one is")
        print(f"     both retrieved AND then accepted by the scorer.")
    else:
        print("\n  => NOT concentrated in empty-address records. R4 is a bge-out artefact;")
        print("     model-3 should skip it.")

print("\n" + "=" * 74)
print("WHAT DO THE NEVER-RETRIEVED PAIRS LOOK LIKE? (10 samples)")
print("=" * 74)
s1raw = pd.read_csv(os.path.join(RAW, "train_source1.tsv"), sep="\t", dtype=str,
                    keep_default_na=False, na_filter=False)
s1p = pd.read_parquet(p["source1"], columns=["entity_id"])
s1_ids = s1p["entity_id"].values
s1map = dict(zip(s1raw["entity_id"], zip(s1raw["business_name"], s1raw["business_address"])))
shown = 0
for q in ids:
    t = truths.get(q, set())
    cs = cand_by_q.get(q, set())
    for c in (t - cs):
        qn, qa = s1map[s1_ids[q]]
        cn, ca = raw[pool_ids[c]]
        print(f"  S1: {qn[:46]!r:<50} | {qa[:40]!r}")
        print(f"  ->: {cn[:46]!r:<50} | {ca[:40]!r}")
        shown += 1
        break
    if shown >= 10:
        break