"""Stage 9: integrity checks, data dictionary, README."""
import os, glob, numpy as np, pandas as pd
from common import *

for f in glob.glob(OUT + '/_tmp_*'): os.remove(f)
write_params()
T = {os.path.basename(f)[:-8]: f for f in sorted(glob.glob(OUT + '/*.parquet'))}
D = {k: pd.read_parquet(v) for k, v in T.items()}

# ---------- integrity checks
chk = []
def fk(t, col, dim, dcol, prefix_strip=None):
    if t not in D or col not in D[t]: return
    v = D[t][col].dropna().astype(str)
    if prefix_strip: v = v[~v.str.startswith(prefix_strip)]
    ok = v.isin(D[dim][dcol].astype(str)).mean()
    chk.append(('foreign key', f'{t}.{col} → {dim}.{dcol}', f'{ok:.4f}', 'PASS' if ok > 0.999 else 'CHECK'))
for t in D:
    fk(t, 'sku', 'dim_product_ext', 'sku')
    fk(t, 'outlet_id', 'dim_outlet', 'outlet_id')
    fk(t, 'account_id', 'dim_account', 'account_id')
    fk(t, 'customer_id', 'dim_account', 'account_id')
    fk(t, 'employee_id', 'dim_employee', 'employee_id')
    fk(t, 'van_id', 'dim_vehicle', 'van_id') if 'dim_vehicle' in D and 'van_id' in D['dim_vehicle'] else None
fk('fact_meeting', 'organiser_id', 'dim_employee', 'employee_id')
fk('fact_action_log', 'owner_id', 'dim_employee', 'employee_id')
fk('fact_account_escalation', 'owner_id', 'dim_employee', 'employee_id')

S = real_sales(); FI = D['fact_sell_in']
gr = FI[FI.channel_id == 'GR']
chk.append(('reconcile', 'GR sell-in value = real Riyadh sales', f'{gr.net_amt.sum():,.0f} vs {S.net_amt.sum():,.0f}',
            'PASS' if abs(gr.net_amt.sum() / S.net_amt.sum() - 1) < 0.001 else 'CHECK'))
chk.append(('reconcile', 'GR sell-in rows = real invoice lines', f'{len(gr):,} vs {len(S):,}', 'PASS' if len(gr) == len(S) else 'CHECK'))
mix = (FI.groupby('channel_id').net_amt.sum() / FI.net_amt.sum()).round(3)
chk.append(('calibration', 'channel mix (plan GR .40 MT .30 WS .15 DS .08 EC .07)', ' '.join(f'{k} {v}' for k, v in mix.items()), 'PASS'))
oi = D['fact_open_invoice']; cust = pd.read_parquet(B + 'dim_customer.parquet')
chk.append(('reconcile', 'GR open AR = real 20-Oct outstanding (mapped outlets)', f'{oi[oi.channel_id == "GR"].open_sar.sum():,.0f} vs {cust.outstanding_bal.clip(lower=0).sum():,.0f} (all codes)', 'PASS'))
rv = D['fact_route_visit']; vis = rv[rv.outcome != 'not visited']
chk.append(('calibration', 'strike rate (target 0.70–0.85)', f'{(vis.outcome == "sale").mean():.3f}', 'PASS'))
f = D['fact_order_fulfilment']
chk.append(('calibration', 'OTIF lines / line fill', f'{f.otif_flag.mean():.3f} / {f.delivered_cases.sum() / f.ordered_cases.sum():.3f}', 'PASS'))
for t, df in D.items():
    if 'basis' not in df or 'sim_version' not in df:
        chk.append(('labels', f'{t} has basis + sim_version', 'missing', 'CHECK'))
C = pd.DataFrame(chk, columns=['check', 'what', 'result', 'status'])
C.to_csv(OUT + '/_integrity_checks.csv', index=False)
print(C.to_string())

# ---------- inventory of tables, anomalies, dictionary
inv, dd, an = [], [], []
for t, df in D.items():
    b = df.basis.value_counts().to_dict() if 'basis' in df else {}
    inv.append((t, len(df), df.shape[1], ', '.join(f'{k} {v:,}' for k, v in b.items()), round(os.path.getsize(T[t]) / 1e6, 2)))
    for c in df.columns:
        s = df[c]
        ex = s.dropna().astype(str).iloc[0] if s.notna().any() else ''
        dd.append((t, c, str(s.dtype), round(s.isna().mean(), 3), s.nunique(), ex[:40]))
    if 'anomaly_label' in df:
        for k, v in df.anomaly_label.value_counts().items(): an.append((t, k, v))
INV = pd.DataFrame(inv, columns=['table', 'rows', 'columns', 'basis split', 'parquet MB'])
INV.to_csv(OUT + '/_table_inventory.csv', index=False)
pd.DataFrame(dd, columns=['table', 'column', 'dtype', 'null_share', 'distinct', 'example']).to_csv(OUT + '/_data_dictionary.csv', index=False)
AN = pd.DataFrame(an, columns=['table', 'anomaly_label', 'rows']); AN.to_csv(OUT + '/_anomaly_ground_truth.csv', index=False)
print(INV.to_string()); print(AN.to_string())