"""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, os
from datetime import date
import numpy as np
import pandas as pd

DATA, OUT = "data", "out"
os.makedirs(OUT, exist_ok=True)
meta = json.load(open(os.path.join(DATA, "metadata.json")))

with open(os.path.join(DATA, "sales.csv"), newline="", encoding="utf-8") as f:
    raw = f.read()
lines = raw.split("\n")
if lines and lines[-1] == "":
    lines = lines[:-1]
header, body = lines[0], lines[1:]
n_lines = len(body)
cols = ["sale_id", "municipality", "sale_date", "price_cad", "area_m2"]
assert next(csv.reader([header])) == cols

ID_RE = re.compile(r"^S\d{5}$")
MUNI_RE = re.compile(r"^[A-Za-z][A-Za-z\- ]*$")
INT_RE = re.compile(r"^\d+$")
THOUS_RE = re.compile(r"^\d{1,3}(,\d{3})+$")
FLOAT_RE = re.compile(r"^\d+(\.\d+)?$")

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

# Step 2: structural parse (exactly 5 CSV fields, RFC-4180 quoting)
step1 = []
for ln, line in enumerate(body, start=2):
    fields = next(csv.reader([line]))
    if len(fields) == 5:
        step1.append((ln, fields))
# Step 3: field validity (each field unambiguous)
step2, n_thousands = [], 0
for ln, (sid, muni, sd, pr, ar) in step1:
    if not (ID_RE.match(sid) and MUNI_RE.match(muni) and parse_date(sd) and FLOAT_RE.match(ar)):
        continue
    if INT_RE.match(pr):
        price = int(pr)
    elif THOUS_RE.match(pr):
        price = int(pr.replace(",", "")); n_thousands += 1
    else:
        continue
    area = float(ar)
    if price <= 0 or area <= 0:
        continue
    step2.append(dict(sale_id=sid, municipality=muni, sale_date=sd, price_cad=price, area_m2=area, _line=ln))
# Step 4: duplicate sale_id (would be ambiguous) -> drop all copies
df = pd.DataFrame(step2)
dup_mask = df.duplicated("sale_id", keep=False)
step3 = df[~dup_mask].copy()

ledger = [
    dict(step=1, rule="data lines read (excluding header)", n_before=n_lines, n_after=n_lines, n_removed=0,
         rationale="all physical data lines in data/sales.csv"),
    dict(step=2, rule="line parses (RFC-4180 CSV) into exactly 5 fields", n_before=n_lines, n_after=len(step1),
         n_removed=n_lines - len(step1), rationale="corrupted/garbled export lines cannot be read; not reconstructed"),
    dict(step=3, rule="every field valid: sale_id S+5 digits, municipality alphabetic, ISO date, price integer (optionally comma thousands-separated), area numeric, both > 0",
         n_before=len(step1), n_after=len(step2), n_removed=len(step1) - len(step2),
         rationale="fields must be readable unambiguously; not reconstructed"),
    dict(step=4, rule="sale_id unique (all copies of duplicated ids dropped)", n_before=len(step2), n_after=len(step3),
         n_removed=int(dup_mask.sum()), rationale="duplicated ids would make records ambiguous"),
]
pd.DataFrame(ledger).to_csv(os.path.join(OUT, "exclusions.csv"), index=False)

analytic = step3[cols].sort_values("sale_id").reset_index(drop=True)
analytic.to_csv(os.path.join(OUT, "analytic.csv"), index=False)

n_valid = int(len(analytic))
n_ctrl = int(meta["n_records"])
sum_ctrl = int(meta["sum_price_cad"])
sum_valid = int(analytic["price_cad"].sum())
n_unread = n_ctrl - n_valid
excluded_lines = sorted(set(range(2, n_lines + 2)) - set(step3["_line"]))
present = set(int(s[1:]) for s in analytic["sale_id"])
missing_ids = sorted(set(range(1, n_ctrl + 1)) - present)
by_muni = analytic["municipality"].value_counts().to_dict()
muni_gap = {k: int(v - by_muni.get(k, 0)) for k, v in meta["n_by_municipality"].items()}

def rng(xs):
    return f"{xs[0]}-{xs[-1]}" if xs and xs[-1] - xs[0] + 1 == len(xs) else str(xs)

n_bad_lines = n_lines - len(step3)
issues = []
if n_bad_lines:
    issues.append(dict(
        issue=(f"{n_bad_lines} data lines are corrupted/garbled (file lines {rng(excluded_lines)}): no valid sale_id or field structure. "
               f"Sale ids S{missing_ids[0]:05d}-S{missing_ids[-1]:05d} ({len(missing_ids)} ids) are absent from the readable records.")
              if missing_ids else f"{n_bad_lines} data lines could not be read",
        handling="Excluded from the analytic dataset; not reconstructed or imputed. Counted as unreadable records; their price sum is only implied by the control total."))
if n_thousands:
    issues.append(dict(issue=f"{n_thousands} records have price_cad exported as a quoted string with comma thousands separator (e.g. \"280,000\").",
                       handling="Kept: under standard CSV quoting the field is unambiguous; separator removed and value converted to integer CAD."))
if int(dup_mask.sum()):
    issues.append(dict(issue=f"{int(dup_mask.sum())} rows share duplicated sale_id", handling="All copies excluded."))
if n_unread:
    issues.append(dict(issue=f"Readable record counts by municipality fall short of the control n_by_municipality (control minus readable): {muni_gap}.",
                       handling="Reported only; the shortfall sums to the number of unreadable records. Municipality of unreadable records is implied by control totals, not observed."))

feas = "complete" if n_unread == 0 else "partial"
reason = None if feas == "complete" else (
    f"{n_unread} of the {n_ctrl} records in the provider's control count could not be analysed: "
    f"{n_bad_lines} export lines (file lines {rng(excluded_lines)}) are corrupted and cannot be read unambiguously "
    f"(sale ids S{missing_ids[0]:05d}-S{missing_ids[-1]:05d} missing); they were not reconstructed. "
    f"Their implied price sum is {sum_ctrl - sum_valid} CAD.")

results = dict(
    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(analytic["price_cad"].mean()), 6),
    median_price_per_m2_valid=round(float(np.median(analytic["price_cad"] / analytic["area_m2"])), 6),
    feasibility=feas,
    feasibility_reason=reason,
    data_issues=issues,
    n_input_rows={"sales.csv": n_lines},
    random_seed=None,
    # supplementary keys (not required by TASK.md)
    implied_mean_price_unreadable=round((sum_ctrl - sum_valid) / n_unread, 6) if n_unread else None,
    n_price_thousands_separator=n_thousands,
    n_valid_if_thousands_sep_excluded=n_valid - n_thousands,
    control_mean_price_all_records=round(sum_ctrl / n_ctrl, 6),
    n_by_municipality_valid={k: int(v) for k, v in sorted(by_muni.items())},
    n_by_municipality_gap=muni_gap,
)
with open(os.path.join(OUT, "results.json"), "w") as f:
    json.dump(results, f, indent=2)
print(json.dumps({k: results[k] for k in ["n_valid_records", "n_unreadable_records", "sum_price_valid",
                                          "sum_price_unreadable_implied", "mean_price_valid",
                                          "median_price_per_m2_valid", "feasibility"]}, indent=1))