"""F1-conf-04: reconcile property-sales export to provider control totals.
Run from workspace root: ~/wp29/venv/bin/python out/analysis.py"""
import csv, json, re, datetime
import pandas as pd

DATA, OUT = "data", "out"
meta = json.load(open(f"{DATA}/metadata.json"))
with open(f"{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"]
n_lines = len(lines)

ID_RE = re.compile(r"S\d{5}")
MUNIS = set(meta["n_by_municipality"])

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

# Step 1: structurally readable (5 fields, valid sale_id)
s1 = [r for r in lines if len(r) == 5 and ID_RE.fullmatch(r[0])]
# Step 2: all fields parse unambiguously
PRICE_RE = re.compile(r"\d{1,3}(,\d{3})+|\d+")
def fields_ok(r):
    return (r[1] in MUNIS and valid_date(r[2]) and PRICE_RE.fullmatch(r[3]) is not None
            and re.fullmatch(r"\d+(\.\d+)?", r[4]) is not None and float(r[4]) > 0)
s2 = [r for r in s1 if fields_ok(r)]
# Step 3: deduplicate sale_id (keep none if conflicting duplicates)
ids = pd.Series([r[0] for r in s2])
dup = set(ids[ids.duplicated(keep=False)])
s3 = [r for r in s2 if r[0] not in dup]

n_thousands_sep = sum(1 for r in s3 if "," in r[3])
df = pd.DataFrame(s3, columns=header)
df["price_cad"] = df["price_cad"].str.replace(",", "", regex=False).astype("int64")
df["area_m2"] = df["area_m2"].astype(float)
df = df.sort_values("sale_id").reset_index(drop=True)
df.to_csv(f"{OUT}/analytic.csv", index=False)

steps = [
    ("1", "keep lines with exactly 5 CSV fields and sale_id matching S+5 digits", n_lines, len(s1),
     "garbled/corrupted lines cannot be read unambiguously; not reconstructed"),
    ("2", "keep rows whose municipality, ISO date, price and area all parse", len(s1), len(s2),
     "fields must be readable unambiguously"),
    ("3", "drop rows with duplicated sale_id", len(s2), len(s3), "sale_id must be unique"),
]
pd.DataFrame([dict(step=a, rule=b, n_before=c, n_after=d, n_removed=c - d, rationale=e)
              for a, b, c, d, e in steps]).to_csv(f"{OUT}/exclusions.csv", index=False)

n_ctrl, sum_ctrl = int(meta["n_records"]), int(meta["sum_price_cad"])
n_valid = len(df); sum_valid = int(df["price_cad"].sum())
n_unread = n_ctrl - n_valid
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"]}

present = set(df["sale_id"])
missing = [f"S{i:05d}" for i in range(1, n_ctrl + 1) if f"S{i:05d}" not in present]
issues = []
if n_lines - len(s1):
    issues.append({"issue": f"{n_lines - len(s1)} data lines in sales.csv are garbled (random characters, wrong field count, malformed sale_id); "
                            f"unreadable records per municipality implied by control totals: {muni_gap}",
                   "handling": "excluded; not reconstructed or imputed; their implied price sum reported as sum_price_unreadable_implied"})
if n_thousands_sep:
    issues.append({"issue": f"{n_thousands_sep} readable records have price_cad written as a quoted string with comma thousands separators (e.g. \"280,000\")",
                   "handling": "parsed unambiguously by removing the thousands separator (CSV quoting makes the field boundary unambiguous); retained"})
if len(s1) - len(s2):
    issues.append({"issue": f"{len(s1)-len(s2)} structurally readable rows had unparseable fields", "handling": "excluded"})
if len(s2) - len(s3):
    issues.append({"issue": f"{len(s2)-len(s3)} rows with duplicated sale_id", "handling": "excluded"})

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()), 2),
    "median_price_per_m2_valid": round(float((df["price_cad"] / df["area_m2"]).median()), 4),
    "feasibility": "complete" if n_unread == 0 else "partial",
    "feasibility_reason": None if n_unread == 0 else
        (f"{n_unread} of the {n_ctrl} records in the provider's control count could not be analysed: "
         f"their lines in sales.csv are garbled (corrupted characters / wrong field count / malformed sale_id) and cannot be read "
         f"unambiguously; they were not reconstructed. Their implied price sum is {sum_ctrl - sum_valid} CAD."),
    "data_issues": issues,
    "n_input_rows": {"sales.csv": n_lines},
    "random_seed": None,
    "n_by_municipality_valid": {k: int(v) for k, v in sorted(by_muni.items())},
    "n_by_municipality_unreadable_implied": muni_gap,
    "missing_sale_ids_range": [missing[0], missing[-1]] if missing else None,
    "n_missing_sale_ids": len(missing),
    "mean_price_unreadable_implied": round((sum_ctrl - sum_valid) / n_unread, 2) if n_unread else None,
    "mean_price_control_implied": round(sum_ctrl / n_ctrl, 2),
}
json.dump(res, open(f"{OUT}/results.json", "w"), indent=2)
print(json.dumps(res, indent=2))