"""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 pandas as pd

DATA, OUT = "data", "out"
meta = json.load(open(os.path.join(DATA, "metadata.json")))
munis = set(meta["n_by_municipality"])

with open(os.path.join(DATA, "sales.csv"), newline="") as f:
    lines = f.read().splitlines()
header, data_lines = lines[0], lines[1:]
assert header == "sale_id,municipality,sale_date,price_cad,area_m2"
n_lines = len(data_lines)
rows = list(csv.reader(data_lines))  # RFC-4180 parsing (handles quoted fields)

ID_RE = re.compile(r"^S\d{5}$")
PRICE_PLAIN = re.compile(r"^\d+$")
PRICE_THOUS = re.compile(r"^\d{1,3}(,\d{3})+$")
AREA_RE = re.compile(r"^\d+(\.\d+)?$")

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

steps = []
def step(name, rule, keep, cur, rationale):
    new = [r for r in cur if keep(r)]
    steps.append(dict(step=name, rule=rule, n_before=len(cur), n_after=len(new),
                      n_removed=len(cur) - len(new), rationale=rationale))
    return new

cur = rows
cur = step("1_field_count", "row has exactly 5 CSV fields", lambda r: len(r) == 5, cur,
           "corrupted/garbled export lines cannot be mapped to columns; not reconstructed")
cur = step("2_sale_id", "sale_id matches ^S\\d{5}$", lambda r: bool(ID_RE.match(r[0])), cur,
           "identifier must be readable")
cur = step("3_municipality", "municipality is one of the provider's municipalities", lambda r: r[1] in munis, cur,
           "municipality must be readable")
cur = step("4_sale_date", "sale_date is a valid ISO date YYYY-MM-DD", lambda r: valid_date(r[2]), cur,
           "date must be readable")
cur = step("5_price", "price_cad is an integer, optionally with comma thousands separators inside a quoted field",
           lambda r: bool(PRICE_PLAIN.match(r[3]) or PRICE_THOUS.match(r[3])), cur,
           "price must be unambiguous; quoted '280,000' is unambiguous under CSV quoting")
cur = step("6_area", "area_m2 is a positive decimal number",
           lambda r: bool(AREA_RE.match(r[4])) and float(r[4]) > 0, cur, "area must be readable and positive")
seen = set()
def first(r):
    if r[0] in seen: return False
    seen.add(r[0]); return True
cur = step("7_duplicate_id", "keep first occurrence of each sale_id", first, cur, "sale_id must be unique")

df = pd.DataFrame(cur, columns=["sale_id", "municipality", "sale_date", "price_cad", "area_m2"])
n_thous = int(df.price_cad.str.contains(",").sum())
df["price_cad"] = df.price_cad.str.replace(",", "", regex=False).astype("int64")
df["area_m2"] = df.area_m2.astype(float)
df.to_csv(os.path.join(OUT, "analytic.csv"), index=False)
pd.DataFrame(steps).to_csv(os.path.join(OUT, "exclusions.csv"), index=False)

n_valid = len(df)
n_control = int(meta["n_records"])
sum_control = int(meta["sum_price_cad"])
sum_valid = int(df.price_cad.sum())
n_unread = n_control - n_valid
n_badfields = steps[0]["n_removed"]
other_removed = sum(s["n_removed"] for s in steps[1:])
by_muni_valid = df.municipality.value_counts().to_dict()
by_muni_gap = {m: int(meta["n_by_municipality"][m] - by_muni_valid.get(m, 0)) for m in sorted(munis)}

issues = []
if n_badfields:
    bad_line_nos = [i + 2 for i, r in enumerate(rows) if len(r) != 5]
    issues.append({"issue": f"{n_badfields} data lines (file lines {min(bad_line_nos)}-{max(bad_line_nos)}) are garbled "
                            "(random characters, wrong number of fields) and cannot be read",
                   "handling": "excluded from the analytic dataset; not reconstructed or imputed; their count and implied "
                               "price sum are reported via reconciliation with the control totals"})
if other_removed:
    issues.append({"issue": f"{other_removed} rows with 5 fields failed field-level validity checks",
                   "handling": "excluded (see exclusions.csv)"})
if n_thous:
    issues.append({"issue": f"{n_thous} price_cad values are exported as quoted strings with comma thousands "
                            "separators (e.g. \"280,000\")",
                   "handling": "kept: the quoted field is unambiguous under CSV quoting; commas removed and parsed as integer CAD"})

feas = "complete" if n_unread == 0 and n_valid == n_control else "partial"
reason = None if feas == "complete" else (
    f"{n_unread} of the {n_control} records in the provider's control count could not be analysed: "
    f"{n_badfields} export lines are garbled (corrupted characters / wrong field count) and cannot be read "
    f"unambiguously; they account for an implied {sum_control - sum_valid} CAD of the control price sum "
    f"(municipality gaps vs control: {by_muni_gap}).")

res = {
    "task_id": "F1-conf-04",
    "n_records_control": n_control,
    "n_valid_records": n_valid,
    "n_unreadable_records": n_unread,
    "n_analysed": n_valid,
    "sum_price_control": sum_control,
    "sum_price_valid": sum_valid,
    "sum_price_unreadable_implied": sum_control - sum_valid,
    "mean_price_valid": round(float(df.price_cad.mean()), 4),
    "median_price_per_m2_valid": round(float((df.price_cad / df.area_m2).median()), 4),
    "feasibility": feas,
    "feasibility_reason": reason,
    "data_issues": issues,
    "n_input_rows": {"sales.csv": n_lines},
    "random_seed": None,
    "n_by_municipality_valid": {m: int(by_muni_valid.get(m, 0)) for m in sorted(munis)},
    "n_by_municipality_gap_vs_control": by_muni_gap,
    "mean_price_unreadable_implied": round((sum_control - sum_valid) / n_unread, 2) if n_unread else None,
    "sale_date_min_valid": df.sale_date.min(),
    "sale_date_max_valid": df.sale_date.max(),
}
json.dump(res, open(os.path.join(OUT, "results.json"), "w"), indent=2)
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"]}))