"""
SQLite storage layer for aerodrome meteorological analytics.

Tables:
    observations  — Hourly analytic time-series (10 float variables per timestamp).
    outlier_log   — Audit trail of values treated by physical-limit enforcement.
    aerodromes    — Catalog of aerodromes available on the ICEA/DECEA website.

Example:
    >>> from db import get_connection, ensure_schema, upsert_analytics, query_analytics
    >>> conn = get_connection(Path("met.db"))
    >>> ensure_schema(conn)
    >>> upsert_analytics(conn, "SBGR", anl_df)
    >>> df = query_analytics(conn, "SBGR", "2024-01-01", "2024-12-31")
"""

import sqlite3
from datetime import date, datetime, timezone
from pathlib import Path
from typing import Optional

import pandas as pd

# ── Analytic columns ──────────────────────────────────────────────────────────
ANL_COLS: list[str] = [
    "T", "Td", "UR", "QNH", "WS", "WG", "WD", "VIS", "TETO", "PREC"
]

# ── Physical limits per variable ──────────────────────────────────────────────
PHYSICAL_LIMITS: dict[str, tuple[float, float]] = {
    "T":    (-25.0,   55.0),   # °C  — Brazilian climate extremes
    "Td":   (-30.0,   40.0),   # °C  — dew point
    "UR":   (  0.0,  100.0),   # %
    "QNH":  (940.0, 1060.0),   # hPa — absolute world records
    "WS":   (  0.0,  200.0),   # kt
    "WG":   (  0.0,  250.0),   # kt  — Cat-5 hurricane gust ≈ 175 kt
    "WD":   (  0.0,  360.0),   # °
    "VIS":  (  0.0, 9999.0),   # dam
    "TETO": (  0.0, 9999.0),   # dam
    "PREC": (  0.0,  500.0),   # mm  — world record hourly ≈ 300 mm/h
}

# ── SQL schema ────────────────────────────────────────────────────────────────
_SCHEMA = """
CREATE TABLE IF NOT EXISTS observations (
    aerodrome TEXT NOT NULL,
    dt        TEXT NOT NULL,
    T    REAL, Td   REAL, UR   REAL, QNH  REAL,
    WS   REAL, WG   REAL, WD   REAL,
    VIS  REAL, TETO REAL, PREC REAL,
    PRIMARY KEY (aerodrome, dt)
);
CREATE INDEX IF NOT EXISTS idx_aero_dt ON observations(aerodrome, dt);

CREATE TABLE IF NOT EXISTS outlier_log (
    id         INTEGER PRIMARY KEY AUTOINCREMENT,
    aerodrome  TEXT NOT NULL,
    dt         TEXT NOT NULL,
    variable   TEXT NOT NULL,
    orig_value REAL NOT NULL,
    treatment  TEXT NOT NULL DEFAULT 'SET_NULL',
    reason     TEXT,
    applied_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_outlier_aero ON outlier_log(aerodrome, variable);

CREATE TABLE IF NOT EXISTS aerodromes (
    icao       TEXT PRIMARY KEY,
    nome       TEXT,
    uf         TEXT,
    updated_at TEXT
);
"""


# ── Connection ────────────────────────────────────────────────────────────────
def get_connection(db_path: Path) -> sqlite3.Connection:
    """Opens (or creates) a SQLite database with WAL mode enabled.

    Args:
        db_path: Filesystem path to the ``.db`` file.  Parent directories are
            created automatically if they do not exist.

    Returns:
        An open :class:`sqlite3.Connection` configured with WAL journal mode,
        NORMAL synchronous writes, and a 32 MB page cache.
    """
    db_path.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(str(db_path), check_same_thread=False)
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA synchronous=NORMAL")
    conn.execute("PRAGMA cache_size=-32000")
    return conn


def ensure_schema(conn: sqlite3.Connection) -> None:
    """Creates all tables and indices if they do not already exist.

    Args:
        conn: Open SQLite connection.
    """
    conn.executescript(_SCHEMA)
    conn.commit()


# ── Physical-limit clipping ───────────────────────────────────────────────────
def _clip_to_limits(df: pd.DataFrame) -> pd.DataFrame:
    """Sets values outside physical limits to NaN (no audit log entry).

    Args:
        df: DataFrame whose columns may include any subset of ANL_COLS.

    Returns:
        Copy of *df* with out-of-range values replaced by ``float("nan")``.
    """
    df = df.copy()
    for col, (lo, hi) in PHYSICAL_LIMITS.items():
        if col in df.columns:
            mask = df[col].notna() & ((df[col] < lo) | (df[col] > hi))
            df.loc[mask, col] = float("nan")
    return df


# ── Upsert analytics ──────────────────────────────────────────────────────────
def upsert_analytics(
    conn: sqlite3.Connection,
    aerodrome: str,
    anl_df: pd.DataFrame,
) -> int:
    """Inserts or replaces rows in the observations table.

    Physical limits are applied automatically before insertion; out-of-range
    values are silently set to NULL.

    Args:
        conn: Open SQLite connection with schema already applied.
        aerodrome: ICAO code of the aerodrome (e.g. ``"SBGR"``).
        anl_df: DataFrame with a ``_dt`` column (datetime-like) and any subset
            of :data:`ANL_COLS` as float columns.

    Returns:
        Number of rows processed (inserted or replaced).
    """
    if anl_df.empty:
        return 0

    anl_df = _clip_to_limits(anl_df)

    cols_present  = [c for c in ANL_COLS if c in anl_df.columns]
    placeholders  = ", ".join(["?"] * (2 + len(cols_present)))
    col_names     = ", ".join(["aerodrome", "dt"] + cols_present)
    sql = (
        f"INSERT OR REPLACE INTO observations ({col_names}) VALUES ({placeholders})"
    )

    def _dt_str(v: object) -> str:
        if isinstance(v, (datetime, pd.Timestamp)):
            return v.strftime("%Y-%m-%d %H:%M:%S")
        return str(v)[:19]

    rows = []
    for _, row in anl_df.iterrows():
        vals: list = [aerodrome, _dt_str(row["_dt"])]
        for c in cols_present:
            v = row[c]
            vals.append(None if pd.isna(v) else float(v))
        rows.append(tuple(vals))

    with conn:
        conn.executemany(sql, rows)
    return len(rows)


# ── Physical-limit repair on existing rows ────────────────────────────────────
def apply_physical_limits(
    conn: sqlite3.Connection,
    aerodrome: Optional[str] = None,
) -> int:
    """Audits and corrects out-of-range values already stored in the database.

    Every corrected value is logged in ``outlier_log`` before being set to
    NULL.  Pass ``aerodrome=None`` to process all aerodromes.

    Args:
        conn: Open SQLite connection with schema already applied.
        aerodrome: ICAO code to restrict processing, or ``None`` for all.

    Returns:
        Total number of outlier records created and corrected.
    """
    now   = datetime.now(timezone.utc).isoformat()
    total = 0
    aero_filter        = "AND aerodrome=?" if aerodrome else ""
    aero_params_suffix = (aerodrome,) if aerodrome else ()

    for col, (lo, hi) in PHYSICAL_LIMITS.items():
        rows = conn.execute(
            f"SELECT aerodrome, dt, {col} FROM observations "
            f"WHERE {col} IS NOT NULL AND {col} < ? {aero_filter}",
            (lo,) + aero_params_suffix,
        ).fetchall()
        if rows:
            conn.executemany(
                "INSERT INTO outlier_log"
                "(aerodrome,dt,variable,orig_value,treatment,reason,applied_at)"
                " VALUES(?,?,?,?,?,?,?)",
                [(r[0], r[1], col, r[2], "SET_NULL", f"below_limit:{lo}", now)
                 for r in rows],
            )
            total += len(rows)

        rows = conn.execute(
            f"SELECT aerodrome, dt, {col} FROM observations "
            f"WHERE {col} IS NOT NULL AND {col} > ? {aero_filter}",
            (hi,) + aero_params_suffix,
        ).fetchall()
        if rows:
            conn.executemany(
                "INSERT INTO outlier_log"
                "(aerodrome,dt,variable,orig_value,treatment,reason,applied_at)"
                " VALUES(?,?,?,?,?,?,?)",
                [(r[0], r[1], col, r[2], "SET_NULL", f"above_limit:{hi}", now)
                 for r in rows],
            )
            total += len(rows)

    where_clause = f"WHERE aerodrome='{aerodrome}'" if aerodrome else ""
    conn.execute(f"""
        UPDATE observations SET
            T    = CASE WHEN T    IS NOT NULL AND (T    < -25  OR T    > 55  ) THEN NULL ELSE T    END,
            Td   = CASE WHEN Td   IS NOT NULL AND (Td   < -30  OR Td   > 40  ) THEN NULL ELSE Td   END,
            UR   = CASE WHEN UR   IS NOT NULL AND (UR   < 0    OR UR   > 100 ) THEN NULL ELSE UR   END,
            QNH  = CASE WHEN QNH  IS NOT NULL AND (QNH  < 940  OR QNH  > 1060) THEN NULL ELSE QNH  END,
            WS   = CASE WHEN WS   IS NOT NULL AND (WS   < 0    OR WS   > 200 ) THEN NULL ELSE WS   END,
            WG   = CASE WHEN WG   IS NOT NULL AND (WG   < 0    OR WG   > 250 ) THEN NULL ELSE WG   END,
            WD   = CASE WHEN WD   IS NOT NULL AND (WD   < 0    OR WD   > 360 ) THEN NULL ELSE WD   END,
            VIS  = CASE WHEN VIS  IS NOT NULL AND (VIS  < 0    OR VIS  > 9999) THEN NULL ELSE VIS  END,
            TETO = CASE WHEN TETO IS NOT NULL AND (TETO < 0    OR TETO > 9999) THEN NULL ELSE TETO END,
            PREC = CASE WHEN PREC IS NOT NULL AND (PREC < 0    OR PREC > 500 ) THEN NULL ELSE PREC END
        {where_clause}
    """)
    conn.commit()
    return total


# ── Outlier audit queries ─────────────────────────────────────────────────────
def get_outlier_summary(
    conn: sqlite3.Connection,
    aerodrome: Optional[str] = None,
) -> pd.DataFrame:
    """Returns per-variable outlier counts and statistics from the audit log.

    Args:
        conn: Open SQLite connection.
        aerodrome: Filter to a single ICAO code, or ``None`` for all.

    Returns:
        DataFrame with columns ``variable``, ``aerodrome``, ``n_outliers``,
        ``min_orig``, ``max_orig``, ``reason``, ``last_applied``.
    """
    aero_filter = "WHERE aerodrome=?" if aerodrome else ""
    params      = (aerodrome,) if aerodrome else ()
    return pd.read_sql_query(
        f"""
        SELECT
            variable,
            aerodrome,
            COUNT(*)              AS n_outliers,
            MIN(orig_value)       AS min_orig,
            MAX(orig_value)       AS max_orig,
            reason,
            MAX(applied_at)       AS last_applied
        FROM outlier_log
        {aero_filter}
        GROUP BY variable, aerodrome, reason
        ORDER BY variable, aerodrome
        """,
        conn, params=params,
    )


def get_outlier_detail(
    conn: sqlite3.Connection,
    aerodrome: Optional[str] = None,
    variable:  Optional[str] = None,
    limit: int = 500,
) -> pd.DataFrame:
    """Returns row-level outlier records from the audit log.

    Args:
        conn: Open SQLite connection.
        aerodrome: Filter by ICAO code, or ``None`` for all.
        variable: Filter by variable name (e.g. ``"T"``), or ``None`` for all.
        limit: Maximum number of rows to return.

    Returns:
        DataFrame with columns ``aerodrome``, ``dt``, ``variable``,
        ``orig_value``, ``reason``, ``applied_at``.
    """
    conditions: list[str] = []
    params: list[object]  = []
    if aerodrome:
        conditions.append("aerodrome=?")
        params.append(aerodrome)
    if variable:
        conditions.append("variable=?")
        params.append(variable)
    where = ("WHERE " + " AND ".join(conditions)) if conditions else ""
    return pd.read_sql_query(
        f"SELECT aerodrome, dt, variable, orig_value, reason, applied_at "
        f"FROM outlier_log {where} ORDER BY applied_at DESC, aerodrome, dt "
        f"LIMIT {limit}",
        conn, params=params,
    )


def has_outlier_log(conn: sqlite3.Connection) -> bool:
    """Returns True if at least one outlier record exists in the audit log.

    Args:
        conn: Open SQLite connection.
    """
    n = conn.execute("SELECT COUNT(*) FROM outlier_log").fetchone()[0]
    return n > 0


# ── Aerodrome catalog ─────────────────────────────────────────────────────────
def upsert_aerodromes(
    conn: sqlite3.Connection,
    aerodromes: list[dict],
) -> int:
    """Inserts or updates aerodrome catalog entries.

    Args:
        conn: Open SQLite connection with schema already applied.
        aerodromes: List of dicts, each with keys ``icao``, ``nome``, ``uf``.

    Returns:
        Number of records processed.
    """
    if not aerodromes:
        return 0
    now = datetime.now(timezone.utc).isoformat()
    with conn:
        conn.executemany(
            "INSERT OR REPLACE INTO aerodromes(icao, nome, uf, updated_at)"
            " VALUES(?, ?, ?, ?)",
            [(a["icao"], a.get("nome", ""), a.get("uf", ""), now)
             for a in aerodromes if a.get("icao")],
        )
    return len(aerodromes)


def list_all_aerodromes(conn: sqlite3.Connection) -> list[dict]:
    """Returns all entries from the aerodrome catalog table.

    Args:
        conn: Open SQLite connection.

    Returns:
        List of ``{"icao": str, "nome": str, "uf": str}`` dicts ordered by
        ICAO code.  Returns an empty list if the catalog is empty.
    """
    rows = conn.execute(
        "SELECT icao, nome, uf FROM aerodromes ORDER BY icao"
    ).fetchall()
    return [{"icao": r[0], "nome": r[1] or "", "uf": r[2] or ""} for r in rows]


def list_aerodromes_with_data(conn: sqlite3.Connection) -> list[dict]:
    """Returns catalog entries for aerodromes that have rows in observations.

    Joins the ``observations`` table with the ``aerodromes`` catalog so the
    caller receives full display metadata (name, state) for each code that
    actually has data.

    Args:
        conn: Open SQLite connection with schema already applied.

    Returns:
        List of ``{"icao": str, "nome": str, "uf": str}`` dicts ordered by
        ICAO code.  Returns an empty list when no observations exist yet.
    """
    rows = conn.execute(
        """SELECT DISTINCT o.aerodrome,
                  COALESCE(a.nome, ''),
                  COALESCE(a.uf,   '')
           FROM   observations o
           LEFT JOIN aerodromes a ON a.icao = o.aerodrome
           ORDER  BY o.aerodrome"""
    ).fetchall()
    return [{"icao": r[0], "nome": r[1], "uf": r[2]} for r in rows]


# ── General queries ───────────────────────────────────────────────────────────
def query_analytics(
    conn: sqlite3.Connection,
    aerodrome: str,
    start_dt: Optional[str] = None,
    end_dt:   Optional[str] = None,
) -> pd.DataFrame:
    """Queries analytic observations for one aerodrome over an optional date range.

    Args:
        conn: Open SQLite connection.
        aerodrome: ICAO code to query.
        start_dt: Inclusive start date/datetime string (``"YYYY-MM-DD"`` or
            ``"YYYY-MM-DD HH:MM:SS"``).  No lower bound if ``None``.
        end_dt: Inclusive end date (``"YYYY-MM-DD"``).  When only a date is
            given, the comparison extends to ``23:59:59`` of that day.

    Returns:
        DataFrame with columns ``_dt`` (``datetime64``) plus the ANL_COLS
        floats.  Returns an empty DataFrame if no rows match.
    """
    sql    = (
        "SELECT dt, T, Td, UR, QNH, WS, WG, WD, VIS, TETO, PREC "
        "FROM observations WHERE aerodrome=?"
    )
    params: list[object] = [aerodrome]
    if start_dt:
        sql += " AND dt >= ?"
        params.append(start_dt)
    if end_dt:
        sql += " AND dt <= ?"
        params.append(end_dt + " 23:59:59" if len(end_dt) == 10 else end_dt)
    sql += " ORDER BY dt"
    df = pd.read_sql_query(sql, conn, params=params)
    if df.empty:
        return df
    df["_dt"] = pd.to_datetime(df["dt"], format="%Y-%m-%d %H:%M:%S", errors="coerce")
    return df.drop(columns=["dt"]).dropna(subset=["_dt"]).reset_index(drop=True)


def get_coverage(
    conn: sqlite3.Connection,
    aerodrome: str,
) -> Optional[tuple[date, date]]:
    """Returns the earliest and latest observation dates for one aerodrome.

    Args:
        conn: Open SQLite connection.
        aerodrome: ICAO code to inspect.

    Returns:
        ``(min_date, max_date)`` tuple, or ``None`` if no observations exist.
    """
    row = conn.execute(
        "SELECT MIN(dt), MAX(dt) FROM observations WHERE aerodrome=?",
        (aerodrome,),
    ).fetchone()
    if not row or row[0] is None:
        return None
    return date.fromisoformat(row[0][:10]), date.fromisoformat(row[1][:10])


def list_aerodromes(conn: sqlite3.Connection) -> list[str]:
    """Returns ICAO codes of all aerodromes that have at least one observation.

    Args:
        conn: Open SQLite connection.

    Returns:
        Sorted list of ICAO code strings.
    """
    rows = conn.execute(
        "SELECT DISTINCT aerodrome FROM observations ORDER BY aerodrome"
    ).fetchall()
    return [r[0] for r in rows]


def aerodrome_stats(conn: sqlite3.Connection, aerodrome: str) -> dict:
    """Returns basic row-count statistics for one aerodrome.

    Args:
        conn: Open SQLite connection.
        aerodrome: ICAO code to inspect.

    Returns:
        Dict with keys ``min_dt``, ``max_dt``, ``n_obs``, ``n_T``, ``n_VIS``.
        Returns an empty dict if no observations exist.
    """
    row = conn.execute(
        "SELECT MIN(dt), MAX(dt), COUNT(*), COUNT(T), COUNT(VIS) "
        "FROM observations WHERE aerodrome=?",
        (aerodrome,),
    ).fetchone()
    if not row or row[0] is None:
        return {}
    return {
        "min_dt": row[0], "max_dt": row[1],
        "n_obs":  row[2], "n_T":    row[3], "n_VIS": row[4],
    }