# -*- coding: utf-8 -*-
"""Integrity checks across API -> ODS -> DWD."""
from __future__ import annotations

from dataclasses import dataclass
from datetime import date, datetime, time, timedelta
from pathlib import Path
from typing import Any, Dict, Iterable, List, Tuple
from zoneinfo import ZoneInfo

import json

from config.settings import AppConfig
from database.connection import DatabaseConnection
from tasks.dwd_load_task import DwdLoadTask
from scripts.check_ods_gaps import run_gap_check

AMOUNT_KEYWORDS = ("amount", "money", "fee", "balance")


@dataclass(frozen=True)
class IntegrityWindow:
    start: datetime
    end: datetime
    label: str
    granularity: str


def _ensure_tz(dt: datetime, tz: ZoneInfo) -> datetime:
    if dt.tzinfo is None:
        return dt.replace(tzinfo=tz)
    return dt.astimezone(tz)


def _month_start(day: date) -> date:
    return date(day.year, day.month, 1)


def _next_month(day: date) -> date:
    if day.month == 12:
        return date(day.year + 1, 1, 1)
    return date(day.year, day.month + 1, 1)


def _date_to_start(dt: date, tz: ZoneInfo) -> datetime:
    return datetime.combine(dt, time.min).replace(tzinfo=tz)


def _date_to_end_exclusive(dt: date, tz: ZoneInfo) -> datetime:
    return datetime.combine(dt, time.min).replace(tzinfo=tz) + timedelta(days=1)


def build_history_windows(start_dt: datetime, end_dt: datetime, tz: ZoneInfo) -> List[IntegrityWindow]:
    """Build weekly windows for current month, monthly windows for earlier months."""
    start_dt = _ensure_tz(start_dt, tz)
    end_dt = _ensure_tz(end_dt, tz)
    if end_dt <= start_dt:
        return []

    start_date = start_dt.date()
    end_date = end_dt.date()
    current_month_start = _month_start(end_date)

    windows: List[IntegrityWindow] = []
    cur = start_date
    while cur <= end_date:
        month_start = _month_start(cur)
        month_end_exclusive = _next_month(cur)
        range_start = max(cur, month_start)
        range_end = min(end_date, month_end_exclusive - timedelta(days=1))

        if month_start == current_month_start:
            week_start = range_start
            while week_start <= range_end:
                week_end = min(week_start + timedelta(days=6), range_end)
                w_start_dt = _date_to_start(week_start, tz)
                w_end_dt = _date_to_end_exclusive(week_end, tz)
                if w_start_dt < end_dt and w_end_dt > start_dt:
                    windows.append(
                        IntegrityWindow(
                            start=max(w_start_dt, start_dt),
                            end=min(w_end_dt, end_dt),
                            label=f"week_{week_start.isoformat()}",
                            granularity="week",
                        )
                    )
                week_start = week_end + timedelta(days=1)
        else:
            m_start_dt = _date_to_start(range_start, tz)
            m_end_dt = _date_to_end_exclusive(range_end, tz)
            if m_start_dt < end_dt and m_end_dt > start_dt:
                windows.append(
                    IntegrityWindow(
                        start=max(m_start_dt, start_dt),
                        end=min(m_end_dt, end_dt),
                        label=f"month_{month_start.isoformat()}",
                        granularity="month",
                    )
                )
        cur = month_end_exclusive

    return windows


def _split_table(name: str, default_schema: str) -> Tuple[str, str]:
    if "." in name:
        schema, table = name.split(".", 1)
        return schema, table
    return default_schema, name


def _pick_time_column(dwd_cols: Iterable[str], ods_cols: Iterable[str]) -> str | None:
    lower_cols = {c.lower() for c in dwd_cols} & {c.lower() for c in ods_cols}
    for candidate in DwdLoadTask.FACT_ORDER_CANDIDATES:
        if candidate.lower() in lower_cols:
            return candidate.lower()
    return None


def _fetch_columns(cur, schema: str, table: str) -> Tuple[List[str], Dict[str, str]]:
    cur.execute(
        """
        SELECT column_name, data_type
        FROM information_schema.columns
        WHERE table_schema = %s AND table_name = %s
        ORDER BY ordinal_position
        """,
        (schema, table),
    )
    cols = []
    types: Dict[str, str] = {}
    for name, data_type in cur.fetchall():
        cols.append(name)
        types[name.lower()] = (data_type or "").lower()
    return cols, types


def _amount_columns(cols: List[str], types: Dict[str, str]) -> List[str]:
    numeric_types = {"numeric", "double precision", "integer", "bigint", "smallint", "real", "decimal"}
    out = []
    for col in cols:
        lc = col.lower()
        if types.get(lc) not in numeric_types:
            continue
        if any(key in lc for key in AMOUNT_KEYWORDS):
            out.append(lc)
    return out


def _count_table(cur, schema: str, table: str, time_col: str | None, window: IntegrityWindow | None) -> int:
    where = ""
    params: List[Any] = []
    if time_col and window:
        where = f'WHERE "{time_col}" >= %s AND "{time_col}" < %s'
        params = [window.start, window.end]
    sql = f'SELECT COUNT(1) FROM "{schema}"."{table}" {where}'
    cur.execute(sql, params)
    row = cur.fetchone()
    return int(row[0] if row else 0)


def _sum_column(cur, schema: str, table: str, col: str, time_col: str | None, window: IntegrityWindow | None) -> float:
    where = ""
    params: List[Any] = []
    if time_col and window:
        where = f'WHERE "{time_col}" >= %s AND "{time_col}" < %s'
        params = [window.start, window.end]
    sql = f'SELECT COALESCE(SUM("{col}"), 0) FROM "{schema}"."{table}" {where}'
    cur.execute(sql, params)
    row = cur.fetchone()
    return float(row[0] if row else 0)


def run_dwd_vs_ods_check(
    *,
    cfg: AppConfig,
    window: IntegrityWindow | None,
    include_dimensions: bool,
) -> Dict[str, Any]:
    dsn = cfg["db"]["dsn"]
    session = cfg["db"].get("session")
    db_conn = DatabaseConnection(dsn=dsn, session=session)
    try:
        with db_conn.conn.cursor() as cur:
            results: List[Dict[str, Any]] = []
            table_map = DwdLoadTask.TABLE_MAP
            for dwd_table, ods_table in table_map.items():
                if not include_dimensions and ".dim_" in dwd_table:
                    continue
                schema_dwd, name_dwd = _split_table(dwd_table, "billiards_dwd")
                schema_ods, name_ods = _split_table(ods_table, "billiards_ods")
                try:
                    dwd_cols, dwd_types = _fetch_columns(cur, schema_dwd, name_dwd)
                    ods_cols, ods_types = _fetch_columns(cur, schema_ods, name_ods)
                    time_col = _pick_time_column(dwd_cols, ods_cols)
                    count_dwd = _count_table(cur, schema_dwd, name_dwd, time_col, window)
                    count_ods = _count_table(cur, schema_ods, name_ods, time_col, window)

                    dwd_amount_cols = _amount_columns(dwd_cols, dwd_types)
                    ods_amount_cols = _amount_columns(ods_cols, ods_types)
                    common_amount_cols = sorted(set(dwd_amount_cols) & set(ods_amount_cols))
                    amounts: List[Dict[str, Any]] = []
                    for col in common_amount_cols:
                        dwd_sum = _sum_column(cur, schema_dwd, name_dwd, col, time_col, window)
                        ods_sum = _sum_column(cur, schema_ods, name_ods, col, time_col, window)
                        amounts.append(
                            {
                                "column": col,
                                "dwd_sum": dwd_sum,
                                "ods_sum": ods_sum,
                                "diff": dwd_sum - ods_sum,
                            }
                        )

                    results.append(
                        {
                            "dwd_table": dwd_table,
                            "ods_table": ods_table,
                            "windowed": bool(time_col and window),
                            "window_col": time_col,
                            "count": {"dwd": count_dwd, "ods": count_ods, "diff": count_dwd - count_ods},
                            "amounts": amounts,
                        }
                    )
                except Exception as exc:  # noqa: BLE001
                    results.append(
                        {
                            "dwd_table": dwd_table,
                            "ods_table": ods_table,
                            "windowed": bool(window),
                            "window_col": None,
                            "count": {"dwd": None, "ods": None, "diff": None},
                            "amounts": [],
                            "error": f"{type(exc).__name__}: {exc}",
                        }
                    )

        total_count_diff = sum(
            int(item.get("count", {}).get("diff") or 0)
            for item in results
            if isinstance(item.get("count", {}).get("diff"), (int, float))
        )
        return {
            "tables": results,
            "total_count_diff": total_count_diff,
        }
    finally:
        db_conn.close()


def _default_report_path(prefix: str) -> Path:
    root = Path(__file__).resolve().parents[1]
    stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    return root / "reports" / f"{prefix}_{stamp}.json"


def run_integrity_window(
    *,
    cfg: AppConfig,
    window: IntegrityWindow,
    include_dimensions: bool,
    task_codes: str,
    logger,
    write_report: bool,
    report_path: Path | None = None,
    window_split_unit: str | None = None,
    window_compensation_hours: int | None = None,
) -> Dict[str, Any]:
    total_seconds = max(0, int((window.end - window.start).total_seconds()))
    if total_seconds >= 86400:
        window_days = max(1, total_seconds // 86400)
        window_hours = 0
    else:
        window_days = 0
        window_hours = max(1, total_seconds // 3600 or 1)

    ods_payload = run_gap_check(
        cfg=cfg,
        start=window.start,
        end=window.end,
        window_days=window_days,
        window_hours=window_hours,
        page_size=int(cfg.get("api.page_size") or 200),
        chunk_size=500,
        sample_limit=50,
        sleep_per_window=0,
        sleep_per_page=0,
        task_codes=task_codes,
        from_cutoff=False,
        cutoff_overlap_hours=24,
        allow_small_window=True,
        logger=logger,
        window_split_unit=window_split_unit,
        window_compensation_hours=window_compensation_hours,
    )

    dwd_payload = run_dwd_vs_ods_check(
        cfg=cfg,
        window=window,
        include_dimensions=include_dimensions,
    )

    report = {
        "mode": "window",
        "window": {
            "start": window.start.isoformat(),
            "end": window.end.isoformat(),
            "label": window.label,
            "granularity": window.granularity,
        },
        "api_to_ods": ods_payload,
        "ods_to_dwd": dwd_payload,
        "generated_at": datetime.now(ZoneInfo(cfg.get("app.timezone", "Asia/Taipei"))).isoformat(),
    }

    if write_report:
        path = report_path or _default_report_path("data_integrity_window")
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
        report["report_path"] = str(path)

    return report


def run_integrity_history(
    *,
    cfg: AppConfig,
    start_dt: datetime,
    end_dt: datetime,
    include_dimensions: bool,
    task_codes: str,
    logger,
    write_report: bool,
    report_path: Path | None = None,
) -> Dict[str, Any]:
    tz = ZoneInfo(cfg.get("app.timezone", "Asia/Taipei"))
    windows = build_history_windows(start_dt, end_dt, tz)
    results: List[Dict[str, Any]] = []
    total_missing = 0
    total_errors = 0

    for window in windows:
        logger.info("校验窗口 起始=%s 结束=%s", window.start, window.end)
        payload = run_integrity_window(
            cfg=cfg,
            window=window,
            include_dimensions=include_dimensions,
            task_codes=task_codes,
            logger=logger,
            write_report=False,
        )
        results.append(payload)
        total_missing += int(payload.get("api_to_ods", {}).get("total_missing") or 0)
        total_errors += int(payload.get("api_to_ods", {}).get("total_errors") or 0)

    report = {
        "mode": "history",
        "start": _ensure_tz(start_dt, tz).isoformat(),
        "end": _ensure_tz(end_dt, tz).isoformat(),
        "windows": results,
        "total_missing": total_missing,
        "total_errors": total_errors,
        "generated_at": datetime.now(tz).isoformat(),
    }

    if write_report:
        path = report_path or _default_report_path("data_integrity_history")
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
        report["report_path"] = str(path)

    return report


def compute_last_etl_end(cfg: AppConfig) -> datetime | None:
    dsn = cfg["db"]["dsn"]
    session = cfg["db"].get("session")
    db_conn = DatabaseConnection(dsn=dsn, session=session)
    try:
        rows = db_conn.query(
            "SELECT MAX(window_end) AS mx FROM etl_admin.etl_run WHERE store_id = %s",
            (cfg.get("app.store_id"),),
        )
        mx = rows[0]["mx"] if rows else None
        if isinstance(mx, datetime):
            tz = ZoneInfo(cfg.get("app.timezone", "Asia/Taipei"))
            return _ensure_tz(mx, tz)
    finally:
        db_conn.close()
    return None