"""Reproduce the Harris report using only aggregates from an official HCAD ZIP.

This standalone tool never opens an application database. Raw account identifiers
are used only to detect duplicates; no property-level records leave the process.
"""

from __future__ import annotations

import argparse
import csv
import hashlib
import io
import json
import re
import statistics
import zipfile
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path

YEAR = 2026
MIN_GROUP = 100
SOURCE_URL = "https://download.hcad.org/data/CAMA/2026/Real_acct_owner.zip"
SLUG = "2026-market-value-down-appraised-value-up"
MONEY_FIELDS = ("tot_mkt_val", "prior_tot_mkt_val", "tot_appr_val", "prior_tot_appr_val")
REQUIRED = {"acct", "yr", "state_class", "site_addr_3", "Neighborhood_Code", "value_status", *MONEY_FIELDS}


def positive_dollars(value: str) -> int | None:
    value = value.strip()
    if not re.fullmatch(r"[0-9]+(?:\.0+)?", value):
        return None
    number = int(value.split(".")[0])
    return number if number > 0 else None


def normalize_zip(value: str) -> str | None:
    value = value.strip()
    return value[:5] if re.fullmatch(r"[0-9]{5}(?:-[0-9]{4})?", value) else None


def direction(current: int, prior: int) -> str:
    return "down" if current < prior else "up" if current > prior else "unchanged"


def percent(numerator: int, denominator: int) -> float:
    return round(100 * numerator / denominator, 4) if denominator else 0.0


def analyze_rows(rows, area_names: dict[str, str], min_group: int = MIN_GROUP) -> dict:
    counters = Counter({key: 0 for key in (
        "source_rows", "malformed_rows", "other_year_or_class", "target_rows",
        "invalid_account", "missing_invalid_or_nonpositive_values", "appraised_above_market",
        "matched", "missing_zip", "missing_area", "affected", "material_affected", "market_down",
    )})
    matrix = {market: {appraised: 0 for appraised in ("down", "unchanged", "up")}
              for market in ("down", "unchanged", "up")}
    statuses = Counter()
    zip_groups: dict[str, dict] = {}
    area_groups: dict[str, dict] = {}
    seen: set[str] = set()
    deltas: dict[str, list[float]] = {key: [] for key in ("market_pct", "appraised_pct", "market_dollars", "appraised_dollars")}

    def add_group(groups, code, name, market_down, affected, material, zip_code):
        group = groups.setdefault(code, {"code": code, "name": name, "matched": 0,
            "market_down": 0, "affected": 0, "material_affected": 0, "zips": set()})
        group["matched"] += 1
        group["market_down"] += market_down
        group["affected"] += affected
        group["material_affected"] += material
        if zip_code:
            group["zips"].add(zip_code)

    for row in rows:
        counters["source_rows"] += 1
        if row is None:
            counters["malformed_rows"] += 1
            continue
        if row["yr"].strip() != str(YEAR) or row["state_class"].strip() != "A1":
            counters["other_year_or_class"] += 1
            continue
        counters["target_rows"] += 1
        account = row["acct"].strip()
        if not re.fullmatch(r"[0-9]{13}", account):
            counters["invalid_account"] += 1
            continue
        if account in seen:
            raise ValueError("Duplicate target account in source; publication aborted")
        seen.add(account)
        amounts = [positive_dollars(row[key]) for key in MONEY_FIELDS]
        if any(value is None for value in amounts):
            counters["missing_invalid_or_nonpositive_values"] += 1
            continue
        market, prior_market, appraised, prior_appraised = amounts
        if appraised > market or prior_appraised > prior_market:
            counters["appraised_above_market"] += 1
            continue
        counters["matched"] += 1
        statuses[row["value_status"].strip() or "Unspecified"] += 1
        market_direction = direction(market, prior_market)
        appraised_direction = direction(appraised, prior_appraised)
        matrix[market_direction][appraised_direction] += 1
        market_down = market < prior_market
        affected = market_down and appraised > prior_appraised
        material = affected and (prior_market - market) * 100 >= prior_market and (appraised - prior_appraised) * 100 >= prior_appraised
        counters["market_down"] += market_down
        counters["affected"] += affected
        counters["material_affected"] += material
        if affected:
            deltas["market_pct"].append(100 * (market - prior_market) / prior_market)
            deltas["appraised_pct"].append(100 * (appraised - prior_appraised) / prior_appraised)
            deltas["market_dollars"].append(market - prior_market)
            deltas["appraised_dollars"].append(appraised - prior_appraised)
        zip_code = normalize_zip(row["site_addr_3"])
        area_code = row["Neighborhood_Code"].strip()
        if zip_code:
            add_group(zip_groups, zip_code, zip_code, market_down, affected, material, zip_code)
        else:
            counters["missing_zip"] += 1
        if area_code in area_names:
            add_group(area_groups, area_code, area_names[area_code], market_down, affected, material, zip_code)
        else:
            counters["missing_area"] += 1

    if not counters["matched"]:
        raise ValueError("No usable 2026 A1 records; publication aborted")
    if counters["malformed_rows"] > max(10, counters["source_rows"] * 0.001):
        raise ValueError("Too many malformed source rows; inspect source format before publication")
    assert counters["source_rows"] == counters["malformed_rows"] + counters["other_year_or_class"] + counters["target_rows"]
    assert counters["target_rows"] == sum(counters[key] for key in ("matched", "invalid_account", "missing_invalid_or_nonpositive_values", "appraised_above_market"))
    assert sum(sum(row.values()) for row in matrix.values()) == counters["matched"]
    assert sum(g["matched"] for g in zip_groups.values()) + counters["missing_zip"] == counters["matched"]
    assert sum(g["matched"] for g in area_groups.values()) + counters["missing_area"] == counters["matched"]

    def publish_groups(groups):
        published = []
        omitted_groups = omitted_records = 0
        for group in groups.values():
            if group["matched"] < min_group:
                omitted_groups += 1
                omitted_records += group["matched"]
                continue
            group["zips"] = sorted(group["zips"])
            published.append(group)
        published.sort(key=lambda group: (-group["affected"], group["code"]))
        return published, {"groups": omitted_groups, "matched_records": omitted_records}

    zips, omitted_zips = publish_groups(zip_groups)
    areas, omitted_areas = publish_groups(area_groups)
    return {
        "year": YEAR, "prior_year": YEAR - 1, "state_class": "A1", "minimum_group": min_group,
        "counts": dict(counters), "matrix": matrix, "value_status_counts": dict(sorted(statuses.items())),
        "affected_share_all": percent(counters["affected"], counters["matched"]),
        "affected_share_market_down": percent(counters["affected"], counters["market_down"]),
        "affected_medians": {key: round(statistics.median(values), 4) if values else None for key, values in deltas.items()},
        "zips": zips, "areas": areas, "omitted_small_groups": {"zips": omitted_zips, "areas": omitted_areas},
    }


def find_member(archive: zipfile.ZipFile, basename: str) -> str:
    matches = [name for name in archive.namelist() if name.replace("\\", "/").split("/")[-1].lower() == basename]
    if len(matches) != 1:
        raise ValueError(f"Expected exactly one {basename} archive member")
    return matches[0]


def analyze_archive(path: Path) -> dict:
    with zipfile.ZipFile(path) as archive:
        member = find_member(archive, "real_acct.txt")
        names_member = find_member(archive, "real_neighborhood_code.txt")
        area_names = {}
        with archive.open(names_member) as source:
            reader = csv.DictReader(io.TextIOWrapper(source, encoding="latin-1", newline=""), delimiter="\t")
            if not {"cd", "dscr"}.issubset(reader.fieldnames or []):
                raise ValueError("Missing appraisal-area schema")
            for row in reader:
                code, name = row["cd"].strip(), " ".join(row["dscr"].split())
                if code in area_names and area_names[code] != name:
                    raise ValueError("Conflicting appraisal-area names")
                if code and name:
                    area_names[code] = name
        with archive.open(member) as source:
            reader = csv.reader(io.TextIOWrapper(source, encoding="latin-1", newline=""), delimiter="\t")
            columns = next(reader)
            if len(columns) != len(set(columns)) or not REQUIRED.issubset(columns):
                raise ValueError("Missing or duplicate required property columns")
            rows = (dict(zip(columns, values)) if len(values) == len(columns) else None for values in reader)
            report = analyze_rows(rows, area_names)
        report["source"] = {
            "url": SOURCE_URL,
            "member": member,
            "snapshot_date": "-".join(f"{part:02d}" for part in archive.getinfo(member).date_time[:3]),
            "archive_bytes": path.stat().st_size,
        }
    with path.open("rb") as source:
        report["source"]["sha256"] = hashlib.file_digest(source, "sha256").hexdigest()
    return report


def csv_cell(value):
    if isinstance(value, str) and value.startswith(("=", "+", "-", "@", "\t", "\r")):
        return "'" + value
    return value


def write_outputs(report: dict, root: Path) -> None:
    data_dir = root / "frontend/src/data/research"
    public_dir = root / "frontend/public/research"
    data_dir.mkdir(parents=True, exist_ok=True)
    public_dir.mkdir(parents=True, exist_ok=True)
    content = json.dumps(report, ensure_ascii=False, indent=2) + "\n"
    (data_dir / f"{SLUG}.json").write_text(content, encoding="utf-8")
    (public_dir / f"{SLUG}.json").write_text(content, encoding="utf-8")
    (public_dir / f"{SLUG}-analysis.py").write_text(Path(__file__).read_text(encoding="utf-8"), encoding="utf-8")
    for kind in ("zips", "areas"):
        with (public_dir / f"{SLUG}-{kind}.csv").open("w", newline="", encoding="utf-8-sig") as output:
            writer = csv.writer(output)
            writer.writerow(["code", "name", "matched_homes", "market_value_fell", "market_fell_appraised_rose", "share_of_matched_pct", "both_changes_at_least_1pct", "zip_codes"])
            for group in report[kind]:
                writer.writerow([csv_cell(value) for value in [group["code"], group["name"], group["matched"], group["market_down"], group["affected"], percent(group["affected"], group["matched"]), group["material_affected"], " ".join(group["zips"]) ]])


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("archive", type=Path)
    parser.add_argument("--source-last-modified", required=True, help="Last-Modified header from the downloaded archive response")
    parser.add_argument("--root", type=Path, default=Path(__file__).resolve().parents[1])
    args = parser.parse_args()
    report = analyze_archive(args.archive)
    report["version"] = 1
    report["generated_at"] = datetime.now(timezone.utc).isoformat(timespec="seconds")
    report["source"]["last_modified"] = args.source_last_modified
    write_outputs(report, args.root)
    print(json.dumps({key: value for key, value in report.items() if key not in ("zips", "areas")}, indent=2))
    print(f"Published aggregate coverage: {len(report['zips'])} ZIP codes, {len(report['areas'])} appraisal areas")


if __name__ == "__main__":
    main()