#!/usr/bin/env python3
# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca
"""Phase 1 — profile the RAW layer: coverage, completeness, price tails, duplicates, roll match.

Writes results/profile/*.csv (machine-readable) used by docs/provenance_report.md.
"""
from __future__ import annotations

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
import polars as pl

from sharp_hpi import config as C

OUT = C.RESULTS / "profile"
OUT.mkdir(parents=True, exist_ok=True)


def main() -> None:
    tx = pl.read_parquet(C.RAW_TX)
    um = pl.read_parquet(C.RAW_UNITS_MATCHED)
    mun = pl.read_csv(C.MAMH_MUN_CSV, infer_schema_length=0, encoding="utf8-lossy").select(
        pl.col("mcode").alias("code_mun"), pl.col("munnom"), pl.col("regadm"), pl.col("mrc"), pl.col("mpopul").cast(pl.Int64, strict=False)
    )
    df = tx.join(um, on="id_provinc", how="left", suffix="_u").join(mun, on="code_mun", how="left")
    df = df.with_columns(
        pl.col("date").dt.strftime("%Y-%m").alias("month"),
        pl.col("date").dt.year().alias("year"),
        (pl.col("id_provinc").is_not_null() & (pl.col("id_provinc") != "")).alias("has_idprov"),
        pl.col("code_mun").is_not_null().alias("roll_matched"),
    )

    # 1. monthly coverage by provider type
    (
        df.group_by(["month", "property_type"]).agg(
            pl.len().alias("n"), pl.col("amount").median().alias("median_price"), pl.col("amount").mean().alias("mean_price"),
            pl.col("roll_matched").mean().alias("share_roll_matched"),
        ).sort(["month", "property_type"]).write_csv(OUT / "coverage_month_type.csv")
    )
    # 2. yearly completeness per column (transaction + roll fields)
    cols = ["amount", "lat", "lng", "property_type", "year_built", "floor_area", "building_type", "land_area", "valeur_role",
            "id_provinc", "cubf", "type_prop", "annee_construction", "aire_etages_m2", "superficie_terrain_m2", "nb_etages",
            "nb_logements", "lien_physique", "genre_construction", "unite_voisinage", "matricule", "valeur_terrain", "valeur_batiment"]
    comp = df.group_by("year").agg([pl.len().alias("n")] + [
        (pl.col(c).is_not_null() & (pl.col(c).cast(pl.Utf8) != "")).mean().alias(c) for c in cols
    ]).sort("year")
    comp.write_csv(OUT / "completeness_by_year.csv")
    # 2b. monthly completeness of the key hedonic variables (time-correlated missingness audit)
    df.group_by("month").agg(
        pl.len().alias("n"),
        pl.col("floor_area").is_not_null().mean().alias("floor_area"),
        pl.col("year_built").is_not_null().mean().alias("year_built"),
        (pl.col("building_type").is_not_null() & (pl.col("building_type") != "")).mean().alias("building_type"),
        pl.col("roll_matched").mean().alias("roll_matched"),
        (pl.col("unite_voisinage").is_not_null() & (pl.col("unite_voisinage") != "")).mean().alias("unite_voisinage"),
    ).sort("month").write_csv(OUT / "completeness_by_month.csv")
    # 3. price distribution per year × type
    df.group_by(["year", "property_type"]).agg(
        pl.len().alias("n"), pl.col("amount").min().alias("min"),
        *[pl.col("amount").quantile(q).alias(f"q{int(q*100):02d}") for q in (0.01, 0.05, 0.25, 0.5, 0.75, 0.95, 0.99)],
        pl.col("amount").max().alias("max"),
    ).sort(["year", "property_type"]).write_csv(OUT / "price_quantiles_year_type.csv")
    # 4. coverage by administrative region × year (roll municipality → MAMH region)
    df.group_by(["regadm", "year"]).agg(pl.len().alias("n")).sort(["regadm", "year"]).write_csv(OUT / "coverage_region_year.csv")
    # 5. municipality counts (how many municipalities per transaction count band)
    mcount = df.filter(pl.col("roll_matched")).group_by(["code_mun", "munnom", "regadm", "mpopul"]).agg(pl.len().alias("n_tx")).sort("n_tx", descending=True)
    mcount.write_csv(OUT / "municipality_counts.csv")
    bands = [0, 10, 25, 50, 100, 250, 500, 1000, 5000, 10**9]
    mcount.with_columns(pl.col("n_tx").cut(bands[1:-1], labels=[f"{bands[i]}-{bands[i+1]-1}" if bands[i+1] < 10**9 else f"{bands[i]}+" for i in range(len(bands)-1)]).alias("band")) \
        .group_by("band").agg(pl.len().alias("municipalities"), pl.col("n_tx").sum().alias("transactions")).sort("band").write_csv(OUT / "municipality_size_bands.csv")
    # 6. duplicates and multi-parcel candidates
    dup_exact = df.filter(pl.col("has_idprov")).group_by(["id_provinc", "date", "amount"]).agg(pl.len().alias("c")).filter(pl.col("c") > 1)
    same_parcel_date = df.group_by(["id_provinc", "date"]).agg(pl.len().alias("c")).filter((pl.col("c") > 1) & (pl.col("id_provinc") != ""))
    grp = df.with_columns(
        pl.col("street").str.replace(r"^\s*\d+[A-Za-z]?(?:-\d+)?[,\s]+", "").str.to_lowercase().alias("street_name"),
        (pl.col("lat") * 100).round(0).alias("lat_c"), (pl.col("lng") * 100).round(0).alias("lng_c"),
    )
    mp_street = grp.group_by(["date", "amount", "city", "street_name"]).agg(pl.len().alias("c")).filter(pl.col("c") > 1)
    mp_cell = grp.group_by(["date", "amount", "lat_c", "lng_c"]).agg(pl.len().alias("c")).filter(pl.col("c") > 1)
    turnover = df.filter(pl.col("has_idprov")).group_by("id_provinc").agg(pl.len().alias("n_sales"))
    pl.DataFrame({
        "metric": ["exact_duplicate_groups", "exact_duplicate_rows", "same_parcel_same_date_groups", "multiparcel_street_groups", "multiparcel_street_rows",
                   "multiparcel_cell_groups", "multiparcel_cell_rows", "parcels_with_gt12_sales", "rows_in_parcels_gt12_sales"],
        "value": [dup_exact.height, int(dup_exact["c"].sum()), same_parcel_date.height, mp_street.height, int(mp_street["c"].sum()),
                  mp_cell.height, int(mp_cell["c"].sum()), turnover.filter(pl.col("n_sales") > 12).height,
                  int(turnover.filter(pl.col("n_sales") > 12)["n_sales"].sum())],
    }).write_csv(OUT / "duplicates_multiparcel.csv")
    # 7. new construction and roll-vintage
    df.with_columns((pl.col("year") <= pl.col("annee_construction")).alias("sold_in_or_before_build_year")).group_by(["year", "property_type"]).agg(
        pl.len().alias("n"), pl.col("sold_in_or_before_build_year").mean().alias("share_new_build"),
    ).sort(["year", "property_type"]).write_csv(OUT / "new_construction_share.csv")
    df.group_by("dat_cond_marche").agg(pl.len().alias("n")).sort("n", descending=True).write_csv(OUT / "roll_market_condition_dates.csv")
    # 8. assessment ratio distribution (diagnostic only)
    r = df.with_columns((pl.col("amount") / pl.col("valeur_role")).alias("ratio")).filter(pl.col("valeur_role") > 0)
    r.group_by("year").agg(
        pl.len().alias("n"), *[pl.col("ratio").quantile(q).alias(f"q{int(q*100):02d}") for q in (0.01, 0.05, 0.25, 0.5, 0.75, 0.95, 0.99)],
        (pl.col("ratio") < 0.2).mean().alias("share_lt_0.2"), (pl.col("ratio") > 5).mean().alias("share_gt_5"),
    ).sort("year").write_csv(OUT / "assessment_ratio_year.csv")
    # 9. provider type × roll classification cross-tab
    df.with_columns(
        pl.when(pl.col("nb_logements").is_null()).then(pl.lit("NA")).when(pl.col("nb_logements") >= 6).then(pl.lit("6+")).otherwise(pl.col("nb_logements").cast(pl.Utf8)).alias("nl"),
    ).group_by(["property_type", "cubf", "nl", "lien_physique"]).agg(pl.len().alias("n")).sort("n", descending=True).write_csv(OUT / "type_crosstab.csv")
    # 10. daily volume (ingestion-gap audit): zero-transaction weekdays, by month
    daily = df.group_by("date").agg(pl.len().alias("n")).sort("date")
    full = pl.DataFrame({"date": pl.date_range(daily["date"].min(), daily["date"].max(), "1d", eager=True)}).join(daily, on="date", how="left").fill_null(0)
    full = full.with_columns(pl.col("date").dt.weekday().alias("wd"), pl.col("date").dt.strftime("%Y-%m").alias("month"))
    full.write_csv(OUT / "daily_counts.csv")
    full.filter(pl.col("wd") <= 5).group_by("month").agg(pl.len().alias("weekdays"), (pl.col("n") == 0).sum().alias("zero_weekdays"), pl.col("n").sum().alias("n")).sort("month").write_csv(OUT / "zero_weekdays_by_month.csv")
    print("profile written:", sorted(p.name for p in OUT.glob("*.csv")))


if __name__ == "__main__":
    main()