# =============================================================================
# QWHPI — Quebec Weekly Housing Price Index
# Author  : Simon-Pierre Boucher
# Contact : contact@spboucher.ai
# File    : api/app/services/data.py
# Purpose : Data access layer — canonical parquet lake (source of truth),
#           in-memory cached, vintage-keyed for ETag support.
# =============================================================================
"""Data access for the API.

Reads the processed Parquet lake (never raw CSVs). Frames are cached in
memory and refreshed when the canonical file's mtime changes (the weekly
scheduler swaps files atomically). ``DATA_DIR`` can be overridden via the
``QWHPI_DATA_DIR`` environment variable (used by docker-compose).
"""

from __future__ import annotations

import os
import threading
from pathlib import Path

import polars as pl

_REPO_ROOT = Path(__file__).resolve().parents[3]
DATA_DIR = Path(os.environ.get("QWHPI_DATA_DIR", _REPO_ROOT / "data" / "processed"))

CANONICAL = DATA_DIR / "qhpi_monthly.parquet"
COVERAGE = DATA_DIR / "coverage_matrix_monthly.parquet"
LIQUIDITY = DATA_DIR / "monthly_liquidity.parquet"
ASSESSMENT = DATA_DIR / "assessment_gap_monthly.parquet"
MUNICIPALITIES = DATA_DIR / "municipalities.parquet"
FIRST_RELEASE = DATA_DIR / "first_release_monthly.parquet"

_lock = threading.Lock()
_cache: dict[str, tuple[float, pl.DataFrame]] = {}


def _load(path: Path) -> pl.DataFrame:
    mtime = path.stat().st_mtime
    with _lock:
        hit = _cache.get(str(path))
        if hit and hit[0] == mtime:
            return hit[1]
        df = pl.read_parquet(path)
        _cache[str(path)] = (mtime, df)
        return df


def canonical() -> pl.DataFrame:
    return _load(CANONICAL)


def coverage() -> pl.DataFrame:
    return _load(COVERAGE)


def liquidity() -> pl.DataFrame:
    return _load(LIQUIDITY)


def assessment_gap() -> pl.DataFrame:
    return _load(ASSESSMENT)


def municipalities() -> pl.DataFrame:
    return _load(MUNICIPALITIES)


def first_release() -> pl.DataFrame | None:
    return _load(FIRST_RELEASE) if FIRST_RELEASE.exists() else None


def data_vintage() -> str:
    return str(canonical()["data_vintage"][0])


def model_version() -> str:
    return str(canonical()["model_version"][0])


def series(geography: str, property_type: str,
           period_from: str | None = None, period_to: str | None = None) -> pl.DataFrame:
    df = canonical().filter(
        (pl.col("geography_id") == geography)
        & (pl.col("property_type") == property_type)
    )
    if period_from:
        df = df.filter(pl.col("period") >= period_from)
    if period_to and period_to != "latest":
        df = df.filter(pl.col("period") <= period_to)
    return df.sort("period")


def geography_ids() -> list[str]:
    return canonical()["geography_id"].unique().sort().to_list()