"""F1-conf-04: reconcile property-sales export to provider control totals.
Run from workspace root: ~/wp29/venv/bin/python out/analysis.py
Reads only data/; writes out/results.json, out/analytic.csv, out/exclusions.csv. No randomness."""
import csv, json, re, hashlib
from datetime import date
from pathlib import Path
import numpy as np
import pandas as pd

DATA, OUT = Path("data"), Path("out")
OUT.mkdir(exist_ok=True)
meta = json.loads((DATA / "metadata.json").read_text())
MUNIS = set(meta["n_by_municipality"])

with open(DATA / "sales.csv", newline="", encoding="utf-8") as f:
    rows = list(csv.reader(f))
header, lines = rows[0], rows[1:]
assert header == ["sale_id", "municipality", "sale_date", "price_cad", "area_m2"], header
n_lines = len(lines)

def to_num(s, thousands=False):
    s = s.strip()
    pat = r"\d{1,3}(,\d{3})+|\d+" if thousands else r"\d+(\.\d+)?"
    if not re.fullmatch(pat, s):
        return None
    return float(s.replace(",", ""))

def iso(s):
    if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", s):
        return None
    try:
        return date.fromisoformat(s)
    except ValueError:
        return None

ledger, cur = [], lines
def step(name, rule, keep, rationale):
    global cur
    before = len(cur)
    cur = [r for r in cur if keep(r)]
    ledger.append(dict(step=name, rule=rule, n_before=before, n_after=len(cur),
                       n_removed=before - len(cur), rationale=rationale))

step("1_field_count", "exactly 5 CSV fields (RFC-4180 quoting respected)", lambda r: len(r) == 5,
     "line cannot be split into the 5 declared columns; content unreadable, not reconstructed")
step("2_sale_id", "sale_id matches ^S\\d{5}$", lambda r: re.fullmatch(r"S\d{5}", r[0]) is not None,
     "identifier not in documented format; record cannot be identified unambiguously")
step("3_municipality", "municipality is one of the 5 control-total municipalities", lambda r: r[1] in MUNIS,
     "unknown municipality string; not guessed")
step("4_sale_date", "sale_date is a valid ISO YYYY-MM-DD date", lambda r: iso(r[2]) is not None,
     "date unreadable")
step("5_price", "price_cad is a positive integer (plain digits, or a quoted value with standard 3-digit thousands separators)",
     lambda r: (to_num(r[3], True) or 0) > 0, "price unreadable or non-positive")
step("6_area", "area_m2 is a positive decimal number", lambda r: (to_num(r[4]) or 0) > 0,
     "area unreadable or non-positive")
ids = [r[0] for r in cur]
dup = {i for i in ids if ids.count(i) > 1}
step("7_unique_id", "sale_id not duplicated among remaining rows", lambda r: r[0] not in dup,
     "duplicate identifiers would be ambiguous")

df = pd.DataFrame(cur, columns=header)
n_thousands = int(df["price_cad"].str.contains(",").sum())
df["price_cad"] = df["price_cad"].map(lambda s: int(to_num(s, True)))
df["area_m2"] = df["area_m2"].astype(float)
df = df.sort_values("sale_id").reset_index(drop=True)
df.to_csv(OUT / "analytic.csv", index=False)
pd.DataFrame(ledger).to_csv(OUT / "exclusions.csv", index=False)

n_valid = len(df)
n_ctrl = int(meta["n_records"])
sum_ctrl = int(meta["sum_price_cad"])
sum_valid = int(df["price_cad"].sum())
n_unread = n_ctrl - n_valid
missing_ids = sorted(set(f"S{i:05d}" for i in range(1, n_ctrl + 1)) - set(df["sale_id"]))
by_muni = df["municipality"].value_counts().to_dict()
muni_gap = {m: int(meta["n_by_municipality"][m] - by_muni.get(m, 0)) for m in meta["n_by_municipality"]}
removed = {e["step"]: e["n_removed"] for e in ledger}

issues = []
if removed["1_field_count"] or removed["2_sale_id"]:
    issues.append({"issue": f"{removed['1_field_count'] + removed['2_sale_id']} data lines are garbled (random characters, wrong number of fields, malformed sale_id); "
                   f"sale_ids {missing_ids[0]}..{missing_ids[-1]} ({len(missing_ids)} ids) are absent from the readable rows" if missing_ids else "garbled lines",
                   "handling": "excluded from the analytic dataset (not reconstructed or imputed); counted as unreadable; their implied price total is derived only as control sum minus valid sum"})
if n_thousands:
    issues.append({"issue": f"{n_thousands} rows have price_cad written as a quoted string with thousands separators (e.g. \"280,000\")",
                   "handling": "the CSV quoting makes the value unambiguous; separators removed and value parsed as integer CAD; rows retained"})
for k in ["3_municipality", "4_sale_date", "5_price", "6_area", "7_unique_id"]:
    if removed[k]:
        issues.append({"issue": f"{removed[k]} rows failed rule {k}", "handling": "excluded"})
if n_unread:
    issues.append({"issue": f"record count ({n_valid}) and price sum ({sum_valid}) of readable rows fall short of control totals ({n_ctrl}, {sum_ctrl}); per-municipality shortfall {muni_gap}",
                   "handling": "reported as reconciliation gap; feasibility set to partial; summary statistics computed on readable records only"})

feas = "complete" if n_unread == 0 else "partial"
reason = None if feas == "complete" else (
    f"{n_unread} of the {n_ctrl} control records could not be analysed: {n_lines} data lines were read, of which "
    f"{n_lines - n_valid} are corrupted/garbled export lines (wrong field count or malformed fields; ids {missing_ids[0]}-{missing_ids[-1]}) "
    f"whose content cannot be read unambiguously; they were not reconstructed. Their implied price total is {sum_ctrl - sum_valid} CAD.")

res = {
    "task_id": "F1-conf-04",
    "n_records_control": n_ctrl,
    "n_valid_records": n_valid,
    "n_unreadable_records": n_unread,
    "n_analysed": n_valid,
    "sum_price_control": sum_ctrl,
    "sum_price_valid": sum_valid,
    "sum_price_unreadable_implied": sum_ctrl - sum_valid,
    "mean_price_valid": round(float(df["price_cad"].mean()), 4),
    "median_price_per_m2_valid": round(float(np.median(df["price_cad"] / df["area_m2"])), 4),
    "feasibility": feas,
    "feasibility_reason": reason,
    "data_issues": issues,
    "n_input_rows": {"sales.csv": n_lines},
    "random_seed": None,
    # supplementary keys
    "n_by_municipality_valid": {m: int(by_muni.get(m, 0)) for m in meta["n_by_municipality"]},
    "n_by_municipality_shortfall": muni_gap,
    "n_price_thousands_separator_parsed": n_thousands,
    "missing_sale_id_range": [missing_ids[0], missing_ids[-1]] if missing_ids else None,
    "sale_date_range_valid": [df["sale_date"].min(), df["sale_date"].max()],
    "mean_price_unreadable_implied": round((sum_ctrl - sum_valid) / n_unread, 4) if n_unread else None,
    "input_sha256": {p.name: hashlib.sha256(p.read_bytes()).hexdigest() for p in sorted(DATA.iterdir())},
}
(OUT / "results.json").write_text(json.dumps(res, indent=2, ensure_ascii=False) + "\n")
print(json.dumps({k: res[k] for k in ["n_valid_records", "n_unreadable_records", "sum_price_valid",
                                       "mean_price_valid", "median_price_per_m2_valid", "feasibility"]}))