#!/usr/bin/env python3
"""Step 01 — Build the sampling frame: house listings (Immo-Ka copy) joined to the
assessment-roll unit (Vrai Prix copy) and the Vrai Prix production evaluation.

Inputs (read-only, never copied into the repository):
  IMMOKA_DB   = immoka.db   (listings, vp_eval)
  VRAIPRIX_DB = vraiprix.db (units, transactions, market_index)
Outputs:
  data/processed/frame.parquet          one row per eligible listing (all exclusion flags kept)
  results/tables/frame_exclusions.csv   logged exclusion cascade
  results/raw/frame_summary.json        counts by source/region/strata

Eligibility rules (R0–R8) are logged one by one; nothing is silently dropped.
"""
from __future__ import annotations

import json
import os
import re
import sqlite3
import sys
from pathlib import Path

import numpy as np
import pandas as pd
import yaml

ROOT = Path(__file__).resolve().parents[2]
CFG = yaml.safe_load(open(ROOT / "configs" / "data.yaml"))

IMMOKA_DB = os.environ.get("IMMOKA_DB", CFG["sources"]["immoka_db"])
VRAIPRIX_DB = os.environ.get("VRAIPRIX_DB", CFG["sources"]["vraiprix_db"])

HOUSE_TYPES = tuple(CFG["frame"]["house_property_types"])
MATCH_M = CFG["frame"]["max_match_m"]

# Québec administrative regions by municipality code prefix are not in the data; we use the
# listing's `region` when present and otherwise the roll municipality name through a coarse
# lookup of the largest CMAs (good enough for stratification; reported as such).
CMA_HINTS = {
    "montreal": ["Montréal", "Laval", "Longueuil", "Brossard", "Terrebonne", "Repentigny", "Blainville", "Mirabel",
                 "Saint-Jérôme", "Boucherville", "Saint-Eustache", "Mascouche", "Châteauguay", "Vaudreuil-Dorion",
                 "Sainte-Julie", "Saint-Constant", "Candiac", "La Prairie", "Chambly", "Saint-Bruno-de-Montarville",
                 "Varennes", "Sainte-Thérèse", "Boisbriand", "Rosemère", "Lorraine", "Bois-des-Filion", "Deux-Montagnes",
                 "Pointe-Claire", "Dollard-Des Ormeaux", "Kirkland", "Beaconsfield", "Dorval", "Mont-Royal", "Westmount",
                 "Côte-Saint-Luc", "Hampstead", "Montréal-Ouest", "Sainte-Anne-de-Bellevue", "Baie-D'Urfé", "Senneville",
                 "L'Île-Perrot", "Pincourt", "Saint-Lazare", "Hudson", "Les Cèdres", "Mercier", "Beauharnois",
                 "Saint-Amable", "Saint-Basile-le-Grand", "McMasterville", "Beloeil", "Mont-Saint-Hilaire", "Otterburn Park",
                 "Carignan", "Richelieu", "Saint-Mathias-sur-Richelieu", "Marieville", "L'Assomption", "Saint-Sulpice",
                 "Charlemagne", "Lavaltrie", "Saint-Lin--Laurentides", "Sainte-Anne-des-Plaines", "Saint-Colomban",
                 "Gore", "Oka", "Pointe-Calumet", "Saint-Joseph-du-Lac", "Sainte-Marthe-sur-le-Lac", "Saint-Placide",
                 "Delson", "Saint-Philippe", "Saint-Isidore", "Saint-Mathieu", "Léry", "Kahnawake", "Montréal-Est"],
    "quebec": ["Québec", "Lévis", "L'Ancienne-Lorette", "Saint-Augustin-de-Desmaures", "Boischatel", "Shannon",
               "Sainte-Brigitte-de-Laval", "Stoneham-et-Tewkesbury", "Lac-Beauport", "Fossambault-sur-le-Lac",
               "Sainte-Catherine-de-la-Jacques-Cartier", "Lac-Saint-Joseph", "Château-Richer", "L'Ange-Gardien",
               "Saint-Henri", "Saint-Lambert-de-Lauzon", "Beaumont", "Lac-Delage", "Saint-Gabriel-de-Valcartier",
               "Wendake", "Notre-Dame-des-Anges"],
    "gatineau": ["Gatineau", "Chelsea", "Cantley", "Val-des-Monts", "L'Ange-Gardien", "La Pêche", "Pontiac"],
    "sherbrooke": ["Sherbrooke", "Magog", "Ascot Corner", "Saint-Denis-de-Brompton", "Stoke", "Compton",
                   "Waterville", "North Hatley", "Hatley"],
    "trois-rivieres": ["Trois-Rivières", "Bécancour", "Saint-Maurice", "Champlain", "Notre-Dame-du-Mont-Carmel",
                       "Yamachiche", "Saint-Étienne-des-Grès"],
    "saguenay": ["Saguenay", "Saint-Honoré", "Saint-Fulgence", "Saint-David-de-Falardeau", "Larouche",
                 "Saint-Charles-de-Bourget"],
}


def cma_of(muni: str) -> str:
    if not isinstance(muni, str):
        return "other"
    m = muni.strip()
    for cma, names in CMA_HINTS.items():
        if m in names:
            return cma
    return "other"


def parse_details(s: str | None) -> dict:
    if not s:
        return {}
    try:
        d = json.loads(s)
        return d if isinstance(d, dict) else {}
    except Exception:
        return {}


def parse_images(s: str | None) -> list[str]:
    if not s:
        return []
    try:
        v = json.loads(s)
    except Exception:
        return []
    if isinstance(v, list):
        out = []
        for it in v:
            if isinstance(it, str):
                out.append(it)
            elif isinstance(it, dict) and it.get("url"):
                out.append(it["url"])
        return [u for u in out if u.startswith("http")]
    return []


def money(v) -> float | None:
    if v is None:
        return None
    s = re.sub(r"[^\d]", "", str(v))
    return float(s) if s else None


# Broker-entered fields we treat as *text-claimed* (not verified) attributes.
DETAIL_FIELDS = {
    "claim_siding": ["Revêtement", "Revêtements", "Revêtement extérieur", "External facing"],
    "claim_roof": ["Revêtement de la toiture", "Toiture", "Roofing"],
    "claim_foundation": ["Fondation"],
    "claim_windows": ["Fenestration", "Fenêtres", "Windows"],
    "claim_basement": ["Sous-sol", "Basement"],
    "claim_garage": ["Garage"],
    "claim_parking": ["Stationnement (total)", "Stationnement", "Nombre de stationnement intérieur"],
    "claim_kitchen": ["Cuisine", "Kitchen"],
    "claim_year_built": ["Année de construction", "Year of construction"],
    "claim_assess_land": ["Évaluation municipale (terrain)", "Évaluation municipale du terrain"],
    "claim_assess_building": ["Évaluation municipale (bâtiment)", "Évaluation municipale du bâtiment"],
    "claim_assess_total": ["Évaluation municipale"],
    "claim_assess_year": ["Évaluation municipale (année)", "Année de l'évaluation municipale"],
    "claim_building_dim": ["Dimensions du bâtiment", "Batisse dim.", "Building dimensions"],
    "claim_style": ["Style"],
    "claim_heating_energy": ["Énergie pour le chauffage"],
    "claim_heating_mode": ["Mode de chauffage", "Système de chauffage"],
    "claim_pool": ["Piscine", "Pool"],
}


def main() -> None:
    con = sqlite3.connect(IMMOKA_DB)
    con.execute(f"ATTACH DATABASE '{VRAIPRIX_DB}' AS vp")
    q = f"""
    SELECT l.uid, l.source, l.url, l.address, l.city, l.region, l.property_type, l.price, l.bedrooms, l.bathrooms,
           l.area_sqft, l.lot_sqft, l.year_built AS listing_year_built, l.description, l.features, l.details, l.images,
           l.lat AS listing_lat, l.lng AS listing_lng, l.first_seen, l.last_seen, l.quality_score,
           e.unit_id, e.match_m, e.est AS vp_est, e.low AS vp_low, e.high AS vp_high, e.confidence_pct AS vp_conf,
           e.model_est AS vp_model_est, e.comps_est AS vp_comps_est, e.cost_est AS vp_cost_est, e.role_est AS vp_role_est,
           e.ens_est AS vp_ens_est, e.type_prop AS vp_type_prop, e.municipalite AS vp_municipalite,
           u.id_provinc, u.municipalite AS roll_municipalite, u.code_mun, u.lat AS roll_lat, u.lng AS roll_lng, u.cubf,
           u.type_prop AS roll_type_prop, u.annee_construction AS roll_year_built, u.aire_etages_m2 AS roll_floor_area_m2,
           u.superficie_terrain_m2 AS roll_lot_m2, u.nb_etages AS roll_storeys, u.nb_logements AS roll_dwellings,
           u.valeur_terrain AS roll_land_value, u.valeur_batiment AS roll_building_value, u.valeur_role AS roll_total_value,
           u.dat_cond_marche AS roll_market_date, u.est_2026 AS roll_model_est, u.p10 AS roll_model_p10, u.p90 AS roll_model_p90
    FROM listings l
    JOIN vp_eval e ON e.uid = l.uid
    LEFT JOIN vp.units u ON u.id_provinc = e.unit_id
    WHERE l.property_type IN ({",".join("?" * len(HOUSE_TYPES))})
    """
    df = pd.read_sql_query(q, con, params=HOUSE_TYPES)
    con.close()
    n0 = len(df)

    casc = []

    def rule(name: str, keep: pd.Series, desc: str):
        nonlocal df
        removed = int((~keep).sum())
        casc.append({"rule": name, "description": desc, "removed": removed, "remaining": int(keep.sum())})
        df = df[keep].copy()

    rule("R0", pd.Series(True, index=df.index), f"House listings (property_type in {HOUSE_TYPES}) evaluated by Vrai Prix: {n0}")
    rule("R1", df["vp_type_prop"].eq("unifamilial"), "Vrai Prix classifies the matched roll unit as single-family (unifamilial)")
    rule("R2", df["match_m"].le(MATCH_M), f"Listing geocode within {MATCH_M} m of the roll unit")
    rule("R3", df["roll_type_prop"].eq("unifamilial") & df["roll_dwellings"].fillna(1).le(1), "Roll unit is a single dwelling")
    rule("R4", df["roll_year_built"].gt(1800) & df["roll_floor_area_m2"].gt(20) & df["roll_land_value"].gt(0) & df["roll_building_value"].gt(0),
         "Complete roll fields: year built, floor area, land and building values")
    yb_ok = df["listing_year_built"].isna() | (df["listing_year_built"] - df["roll_year_built"]).abs().le(5)
    rule("R5", yb_ok, "Listing year built (if any) within ±5 years of the roll year built (match sanity)")
    df["n_images"] = df["images"].map(lambda s: len(parse_images(s)))
    rule("R6", df["n_images"].ge(CFG["frame"]["min_images"]), f"At least {CFG['frame']['min_images']} image URLs")
    rule("R7", df["price"].between(CFG["frame"]["price_min"], CFG["frame"]["price_max"]),
         f"Asking price within [{CFG['frame']['price_min']:,}, {CFG['frame']['price_max']:,}]")
    rule("R8", ~df["source"].isin(CFG["frame"]["excluded_sources"]),
         f"Sources with no structured building fields and few photos excluded: {CFG['frame']['excluded_sources']}")
    import urllib.parse
    bad_hosts = set(CFG["frame"]["robots_disallowed_hosts"])
    df["image_urls_tmp"] = df["images"].map(parse_images)
    df["share_disallowed"] = df["image_urls_tmp"].map(lambda l: np.mean([urllib.parse.urlparse(u).netloc in bad_hosts for u in l]) if l else 1.0)
    excluded_robots = df[df["share_disallowed"] > 0.5].copy()
    rule("R10", df["share_disallowed"].le(0.5), "Image host permits automated retrieval (robots.txt): hosts with 'Disallow: /' excluded")
    df = df.drop(columns=["image_urls_tmp", "share_disallowed"])
    # one listing per roll unit (duplicate listings of the same house across brokers): keep the one with most images
    df = df.sort_values(["id_provinc", "n_images"], ascending=[True, False])
    dup = df.duplicated("id_provinc", keep="first")
    rule("R9", ~dup, "One listing per roll unit (duplicates across sources: keep the listing with most images)")

    # derived fields
    det = df["details"].map(parse_details)
    for col, keys in DETAIL_FIELDS.items():
        df[col] = det.map(lambda d, ks=keys: next((d[k] for k in ks if k in d and d[k] not in (None, "", "Non spécifié")), None)).astype(object)
    for c in ["claim_assess_land", "claim_assess_building", "claim_assess_total"]:
        df[c] = df[c].map(money)
    for col in DETAIL_FIELDS:
        if col not in ("claim_assess_land", "claim_assess_building", "claim_assess_total", "claim_year_built"):
            df[col] = df[col].map(lambda v: None if (v is None or (not isinstance(v, (list, dict)) and pd.isna(v))) else str(v))
    df["claim_year_built"] = pd.to_numeric(df["claim_year_built"].map(lambda v: re.sub(r"[^\d]", "", str(v))[:4] if v else None), errors="coerce")
    df["image_urls"] = df["images"].map(parse_images)
    df["cma"] = df["roll_municipalite"].map(cma_of)
    df["age_2026"] = 2026 - df["roll_year_built"]
    df["age_band"] = pd.cut(df["age_2026"], [-1, 10, 25, 50, 75, 300], labels=["0-10", "11-25", "26-50", "51-75", "76+"])
    df["price_tier"] = pd.qcut(df["price"], 4, labels=["Q1", "Q2", "Q3", "Q4"])
    df["roll_building_share"] = df["roll_building_value"] / df["roll_total_value"]
    df["desc_len"] = df["description"].fillna("").str.len()

    out = ROOT / "data" / "processed" / "frame.parquet"
    df.drop(columns=["images"]).to_parquet(out, index=False)
    pd.DataFrame(casc).to_csv(ROOT / "results" / "tables" / "frame_exclusions.csv", index=False)
    # balance check: listings excluded for robots reasons vs retained, on roll characteristics
    cols = ["price", "roll_year_built", "roll_floor_area_m2", "roll_land_value", "roll_building_value", "n_images"]
    bal = pd.DataFrame({"excluded_robots_median": excluded_robots[cols].median(), "retained_median": df[cols].median(),
                        "excluded_n": len(excluded_robots), "retained_n": len(df)})
    bal.to_csv(ROOT / "results" / "tables" / "robots_exclusion_balance.csv")
    summary = {
        "n_frame": int(len(df)),
        "by_source": df["source"].value_counts().head(25).to_dict(),
        "by_cma": df["cma"].value_counts().to_dict(),
        "by_age_band": df["age_band"].value_counts().sort_index().to_dict(),
        "by_price_tier": df["price_tier"].value_counts().sort_index().to_dict(),
        "price_quartiles": [float(x) for x in df["price"].quantile([0, .25, .5, .75, 1]).values],
        "images_median": float(df["n_images"].median()),
        "claims_coverage": {c: float(df[c].notna().mean()) for c in DETAIL_FIELDS},
        "roll_market_date": df["roll_market_date"].value_counts().to_dict(),
        "vp_cost_est_coverage": float(df["vp_cost_est"].notna().mean()),
    }
    json.dump(summary, open(ROOT / "results" / "raw" / "frame_summary.json", "w"), indent=2, ensure_ascii=False, default=str)
    print(pd.DataFrame(casc).to_string(index=False))
    print(json.dumps({k: v for k, v in summary.items() if k != "by_source"}, indent=1, ensure_ascii=False, default=str))


if __name__ == "__main__":
    sys.exit(main())