# Author: Simon-Pierre Boucher — contact@spboucher.ai
#
"""Helpers for the 2026-10-17 revision extensions (scripts/07_revision_extensions.py).

Language detection, unit-corrected lot size, floor area, forward sortation
areas (FSA), fixed-effects / clustered OLS and out-of-sample metrics.
"""

import json
import re
import sqlite3

import numpy as np
import pandas as pd
import statsmodels.api as sm

FR_STOPWORDS = set(
    "le la les des une un et est avec dans pour sur au aux du de à cette ce cet "
    "ses sa son vous nous très plus rez chaussée cuisine chambre salle bain terrain "
    "étage sous-sol près".split()
)
EN_STOPWORDS = set(
    "the and with for this that is are has have from your you in on at of to a an "
    "kitchen bedroom bathroom basement floor lot home house living room large new".split()
)

SQM_PER_UNIT = {"m2": 1.0, "sqft": 0.09290304, "ac": 4046.8564, "hec": 10000.0}


def detect_language(text):
    """Return 'fr', 'en' or 'mixed' from stop-word counts (ratio rule 1.5)."""
    words = re.findall(r"[a-zàâçéèêëîïôûùüÿœ'-]+", text.lower())
    n_fr = sum(w in FR_STOPWORDS for w in words)
    n_en = sum(w in EN_STOPWORDS for w in words)
    if n_fr > 1.5 * n_en:
        return "fr"
    if n_en > 1.5 * n_fr:
        return "en"
    return "mixed"


def _area_sqm(size_str):
    """Convert a Realtor.ca size string ('98900 m2', '5000 sqft', '2 ac') to m²."""
    if not size_str or not isinstance(size_str, str):
        return np.nan
    m = re.match(r"^\s*([\d.,]+)\s*([a-zA-Z0-9]*)", size_str)
    if not m:
        return np.nan
    try:
        value = float(m.group(1).replace(",", ""))
    except ValueError:
        return np.nan
    unit = m.group(2).lower()
    if unit not in SQM_PER_UNIT or value <= 0:
        return np.nan
    return value * SQM_PER_UNIT[unit]


def load_revision_fields(db_path, ids):
    """Postal code / FSA, municipality, lot area (m²), floor area (m²) for the sample."""
    conn = sqlite3.connect(db_path)
    rows = conn.execute(
        "SELECT id, data FROM properties WHERE category = 'house'"
    ).fetchall()
    conn.close()
    wanted = set(ids)
    recs = {}
    for pid, data_json in rows:
        if pid not in wanted:
            continue
        try:
            d = json.loads(data_json)
        except (json.JSONDecodeError, TypeError):
            continue
        postal = d.get("PostalCode") or ""
        addr = d.get("Property", {}).get("Address", {}).get("AddressText", "")
        city = addr.split("|")[-1].split(",")[0].strip() if "|" in addr else ""
        recs[pid] = {
            "postal_code": postal,
            "fsa": postal[:3].upper(),
            "municipality": city,
            "lot_sqm": _area_sqm(d.get("Land", {}).get("SizeTotal", "")),
            "floor_sqm": _area_sqm(d.get("Building", {}).get("SizeInterior", "")),
            "scraped_at": d.get("scrapedAt"),
        }
    out = pd.DataFrame([recs.get(i, {}) for i in ids])
    out.index = pd.Index(ids)
    return out


def ols(y, X, cov_type="HC3", groups=None):
    """OLS with HC3 or cluster-robust covariance."""
    if cov_type == "cluster":
        return sm.OLS(y, X).fit(cov_type="cluster", cov_kwds={"groups": groups})
    return sm.OLS(y, X).fit(cov_type=cov_type)


def zscore(df, columns):
    X = (df[columns] - df[columns].mean()) / df[columns].std(ddof=0)
    return X


def within(df, columns, group):
    """Demean columns within groups (fixed-effects transformation)."""
    return df[columns] - df.groupby(group)[columns].transform("mean")


def block_f_test(model, variables):
    """Wald F-test that all coefficients in `variables` are zero (robust cov)."""
    names = list(model.params.index)
    R = np.zeros((len(variables), len(names)))
    for i, v in enumerate(variables):
        R[i, names.index(v)] = 1.0
    res = model.f_test(R)
    return float(np.squeeze(res.fvalue)), float(res.pvalue), int(res.df_num)


def oos_metrics(y_true_log, y_pred_log):
    """Out-of-sample R² (log scale), RMSE (log), MdAPE (level)."""
    resid = y_true_log - y_pred_log
    ss_res = float(np.sum(resid ** 2))
    ss_tot = float(np.sum((y_true_log - y_true_log.mean()) ** 2))
    price = np.exp(y_true_log)
    mdape = float(np.median(np.abs(np.exp(y_pred_log) - price) / price)) * 100
    return {"r2_oos": 1 - ss_res / ss_tot,
            "rmse_log": float(np.sqrt(np.mean(resid ** 2))),
            "mdape_pct": mdape}