"""SQLite persistence: quote history, report idempotency, delivery receipts, raw collector payloads.

Single-threaded use only (one connection, WAL mode). Dates are stored as ISO ``YYYY-MM-DD`` text so
lexicographic ordering equals chronological ordering.
"""

from __future__ import annotations

import logging
import math
import sqlite3
from dataclasses import dataclass
from datetime import UTC, date, datetime
from pathlib import Path
from typing import Self

from us2kr.models import BriefingResult, Quote
from us2kr.symbols import POINT_CHANGE_GROUPS, display_name, group_of

log = logging.getLogger(__name__)

# Symbols for which history-based notes are generated (order = order of notes in the briefing).
HISTORY_SYMBOLS: tuple[str, ...] = ("^GSPC", "^IXIC", "^SOX", "^VIX", "KRW=X", "^TNX", "NVDA", "EWY")
HISTORY_SESSIONS = 10  # look back at most this many stored sessions
MIN_STREAK_DAYS = 3
MIN_SESSIONS_FOR_AVG = 5  # need at least this many prior sessions to judge "unusual" moves
BIG_MOVE_RATIO = 2.0
VIX_LEVELS: tuple[float, ...] = (20.0, 30.0)
KRW_STEP = 50.0
# Streak contiguity: two stored rows count as consecutive sessions only when the calendar gap is at most
# a weekend plus one holiday (Fri -> Tue) AND the later row's prev_close matches the earlier row's close
# (yfinance revisions can shift a close slightly, hence the relative tolerance).
MAX_SESSION_GAP_DAYS = 4
PREV_CLOSE_TOLERANCE = 0.002

_SCHEMA = """
CREATE TABLE IF NOT EXISTS quotes (
    session_date TEXT NOT NULL,
    symbol       TEXT NOT NULL,
    close        REAL,
    prev_close   REAL,
    change_pct   REAL,
    change_abs   REAL,
    day_high     REAL,
    day_low      REAL,
    volume       REAL,
    PRIMARY KEY (session_date, symbol)
);
CREATE TABLE IF NOT EXISTS reports (
    id           INTEGER PRIMARY KEY,
    krx_date     TEXT NOT NULL,
    kind         TEXT NOT NULL,
    session_date TEXT,
    model_used   TEXT,
    degraded     INTEGER NOT NULL DEFAULT 0,
    markdown     TEXT,
    result_json  TEXT,
    created_at   TEXT,
    UNIQUE (krx_date, kind)
);
CREATE TABLE IF NOT EXISTS deliveries (
    id        INTEGER PRIMARY KEY,
    report_id INTEGER NOT NULL,
    channel   TEXT NOT NULL,
    receipt   TEXT,
    sent_at   TEXT
);
CREATE TABLE IF NOT EXISTS raw (
    session_date TEXT NOT NULL,
    name         TEXT NOT NULL,
    payload_json TEXT,
    created_at   TEXT,
    PRIMARY KEY (session_date, name)
);
"""


@dataclass(frozen=True)
class _Row:
    """One stored daily quote (subset of columns needed for history notes)."""

    session_date: str
    close: float | None
    prev_close: float | None
    change_pct: float | None
    change_abs: float | None


def _now_iso() -> str:
    return datetime.now(UTC).isoformat(timespec="seconds")


class Store:
    """Thin sqlite3 wrapper. Creates parent directories and tables on construction."""

    def __init__(self, db_path: Path) -> None:
        self.db_path = Path(db_path)
        self.db_path.parent.mkdir(parents=True, exist_ok=True)
        self._conn = sqlite3.connect(str(self.db_path), timeout=30, isolation_level=None)
        self._conn.row_factory = sqlite3.Row
        self._conn.execute("PRAGMA journal_mode=WAL")
        self._conn.execute("PRAGMA synchronous=NORMAL")
        self._conn.executescript(_SCHEMA)
        log.debug("store opened at %s", self.db_path)

    # ------------------------------------------------------------------ quotes
    def save_quotes(self, session_date: date, quotes: list[Quote]) -> None:
        """Upsert the day's quotes keyed by (session_date, symbol)."""
        rows = [
            (
                session_date.isoformat(),
                q.symbol,
                q.close,
                q.prev_close,
                q.change_pct,
                q.change_abs,
                q.day_high,
                q.day_low,
                q.volume,
            )
            for q in quotes
        ]
        with self._conn:
            self._conn.executemany(
                """
                INSERT INTO quotes (session_date, symbol, close, prev_close, change_pct, change_abs,
                                    day_high, day_low, volume)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(session_date, symbol) DO UPDATE SET
                    close = excluded.close,
                    prev_close = excluded.prev_close,
                    change_pct = excluded.change_pct,
                    change_abs = excluded.change_abs,
                    day_high = excluded.day_high,
                    day_low = excluded.day_low,
                    volume = excluded.volume
                """,
                rows,
            )
        log.info("saved %d quotes for %s", len(rows), session_date)

    def load_quotes(self, session_date: date) -> list[sqlite3.Row]:
        """Return stored quote rows for one session (mainly for tests / debugging)."""
        cur = self._conn.execute(
            "SELECT * FROM quotes WHERE session_date = ? ORDER BY symbol", (session_date.isoformat(),)
        )
        return list(cur.fetchall())

    def load_quotes_between(self, start: date, end: date) -> list[sqlite3.Row]:
        """Quote rows for every stored session in [start, end], ordered by (symbol, session_date)."""
        cur = self._conn.execute(
            """
            SELECT * FROM quotes WHERE session_date >= ? AND session_date <= ?
            ORDER BY symbol, session_date
            """,
            (start.isoformat(), end.isoformat()),
        )
        return list(cur.fetchall())

    def _history(self, symbol: str, before: date, n: int) -> list[_Row]:
        """Up to ``n`` stored rows for ``symbol`` strictly before ``before``, oldest first."""
        cur = self._conn.execute(
            """
            SELECT session_date, close, prev_close, change_pct, change_abs
            FROM quotes WHERE symbol = ? AND session_date < ?
            ORDER BY session_date DESC LIMIT ?
            """,
            (symbol, before.isoformat(), n),
        )
        rows = [
            _Row(r["session_date"], r["close"], r["prev_close"], r["change_pct"], r["change_abs"])
            for r in cur
        ]
        rows.reverse()
        return rows

    def history_notes(self, session_date: date, quotes: list[Quote]) -> list[str]:
        """Korean one-liners derived from stored history + today's quotes.

        Covers: N-day streaks (N>=3), today's move vs the 10-session mean absolute move,
        VIX crossing 20/30 and USD/KRW crossing a multiple of 50 won. Returns [] without history.
        """
        today = {q.symbol: q for q in quotes}
        notes: list[str] = []
        for sym in HISTORY_SYMBOLS:
            q = today.get(sym)
            if q is None or q.close is None:
                continue
            hist = self._history(sym, session_date, HISTORY_SESSIONS)
            if not hist:
                continue
            name = display_name(sym)
            points = (q.group or group_of(sym)) in POINT_CHANGE_GROUPS
            today_delta = _delta(q.change_pct, q.change_abs, q.close, q.prev_close, points)
            deltas = [_delta(r.change_pct, r.change_abs, r.close, r.prev_close, points) for r in hist]

            # Streaks may only count sessions forming an unbroken chain ending today; a hole in the table
            # (daemon down past the catch-up window, prices collector failed) must not read as "N일 연속".
            chain = _contiguous_tail(hist, session_date, q.prev_close)
            chain_deltas = [_delta(r.change_pct, r.change_abs, r.close, r.prev_close, points) for r in chain]
            note = _streak_note(name, chain_deltas + [today_delta])
            if note:
                notes.append(note)
            note = _big_move_note(name, deltas, today_delta, points)  # a sample mean: holes are harmless
            if note:
                notes.append(note)

            # Level crossings must agree with the rendered table, which is built from q.prev_close (the
            # collector's directly preceding bar). Stored history is only a fallback: it may be a still-live
            # snapshot for 24h instruments (KRW=X) or 2+ sessions old after a missed run.
            prev_close = q.prev_close if q.prev_close is not None else hist[-1].close
            if prev_close is None:
                continue
            if sym == "^VIX":
                notes.extend(_level_cross_notes(name, prev_close, q.close, list(VIX_LEVELS), 1))
            elif sym == "KRW=X":
                notes.extend(
                    _level_cross_notes(name, prev_close, q.close, _krw_levels(prev_close, q.close), 0, "원")
                )
        return notes

    # ----------------------------------------------------------------- reports
    def has_report(self, krx_date: date, kind: str = "briefing") -> bool:
        cur = self._conn.execute(
            "SELECT 1 FROM reports WHERE krx_date = ? AND kind = ? LIMIT 1", (krx_date.isoformat(), kind)
        )
        return cur.fetchone() is not None

    def save_report(self, result: BriefingResult, kind: str = "briefing") -> int:
        """Upsert the briefing on (krx_date, kind); return the report id."""
        krx = result.payload.krx_date.isoformat()
        with self._conn:
            self._conn.execute(
                """
                INSERT INTO reports (krx_date, kind, session_date, model_used, degraded, markdown,
                                     result_json, created_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(krx_date, kind) DO UPDATE SET
                    session_date = excluded.session_date,
                    model_used = excluded.model_used,
                    degraded = excluded.degraded,
                    markdown = excluded.markdown,
                    result_json = excluded.result_json,
                    created_at = excluded.created_at
                """,
                (
                    krx,
                    kind,
                    result.payload.session_date.isoformat(),
                    result.model_used,
                    int(result.degraded),
                    result.markdown,
                    result.model_dump_json(),
                    _now_iso(),
                ),
            )
            row = self._conn.execute(
                "SELECT id FROM reports WHERE krx_date = ? AND kind = ?", (krx, kind)
            ).fetchone()
        report_id = int(row["id"])
        log.info("saved report id=%d krx_date=%s kind=%s degraded=%s", report_id, krx, kind, result.degraded)
        return report_id

    def reports_between(self, start: date, end: date, kind: str = "briefing") -> list[sqlite3.Row]:
        """(krx_date, session_date, result_json) rows with start <= krx_date <= end, oldest first."""
        cur = self._conn.execute(
            """
            SELECT krx_date, session_date, result_json FROM reports
            WHERE kind = ? AND krx_date >= ? AND krx_date <= ? ORDER BY krx_date
            """,
            (kind, start.isoformat(), end.isoformat()),
        )
        return list(cur.fetchall())

    def mark_sent(self, report_id: int, channel: str, receipt: str) -> None:
        with self._conn:
            self._conn.execute(
                "INSERT INTO deliveries (report_id, channel, receipt, sent_at) VALUES (?, ?, ?, ?)",
                (report_id, channel, receipt, _now_iso()),
            )

    def deliveries_for(self, report_id: int) -> list[sqlite3.Row]:
        cur = self._conn.execute("SELECT * FROM deliveries WHERE report_id = ? ORDER BY id", (report_id,))
        return list(cur.fetchall())

    # --------------------------------------------------------------------- raw
    def save_raw(self, session_date: date, name: str, payload_json: str) -> None:
        with self._conn:
            self._conn.execute(
                """
                INSERT INTO raw (session_date, name, payload_json, created_at) VALUES (?, ?, ?, ?)
                ON CONFLICT(session_date, name) DO UPDATE SET
                    payload_json = excluded.payload_json, created_at = excluded.created_at
                """,
                (session_date.isoformat(), name, payload_json, _now_iso()),
            )

    def load_raw(self, session_date: date, name: str) -> str | None:
        row = self._conn.execute(
            "SELECT payload_json FROM raw WHERE session_date = ? AND name = ?",
            (session_date.isoformat(), name),
        ).fetchone()
        return None if row is None else row["payload_json"]

    # ------------------------------------------------------------------- misc
    def close(self) -> None:
        try:
            self._conn.close()
        except sqlite3.Error:  # pragma: no cover - defensive
            log.debug("store close failed", exc_info=True)

    def __enter__(self) -> Self:
        return self

    def __exit__(self, *exc: object) -> None:
        self.close()


# ----------------------------------------------------------------------- helpers


def _delta(
    change_pct: float | None,
    change_abs: float | None,
    close: float | None,
    prev_close: float | None,
    points: bool,
) -> float | None:
    """Daily move in the unit relevant for the symbol: yield points for rates, percent otherwise."""
    if points:
        if change_abs is not None:
            return change_abs
        if close is not None and prev_close is not None:
            return close - prev_close
        return None
    if change_pct is not None:
        return change_pct
    if close is not None and prev_close:
        return (close / prev_close - 1.0) * 100.0
    return None


def _is_predecessor(older: _Row, newer_date: str, newer_prev_close: float | None) -> bool:
    """True when ``older`` looks like the trading session immediately before ``newer_date``."""
    gap = (date.fromisoformat(newer_date) - date.fromisoformat(older.session_date)).days
    if gap <= 0 or gap > MAX_SESSION_GAP_DAYS:
        return False
    if older.close and newer_prev_close is not None:
        return abs(newer_prev_close / older.close - 1.0) <= PREV_CLOSE_TOLERANCE
    return True  # no prices to cross-check; the calendar gap is plausible


def _contiguous_tail(hist: list[_Row], session_date: date, today_prev_close: float | None) -> list[_Row]:
    """Longest suffix of ``hist`` (oldest first) forming an unbroken session chain ending at today."""
    tail: list[_Row] = []
    newer_date, newer_prev = session_date.isoformat(), today_prev_close
    for r in reversed(hist):
        if not _is_predecessor(r, newer_date, newer_prev):
            break
        tail.append(r)
        newer_date, newer_prev = r.session_date, r.prev_close
    tail.reverse()
    return tail


def _streak_note(name: str, deltas: list[float | None]) -> str | None:
    """'{name} N일 연속 상승/하락' when the last N>=3 deltas share a sign (last element = today)."""
    if not deltas or deltas[-1] is None or deltas[-1] == 0:
        return None
    up = deltas[-1] > 0
    n = 0
    for d in reversed(deltas):
        if d is None or d == 0 or (d > 0) != up:
            break
        n += 1
    if n < MIN_STREAK_DAYS:
        return None
    return f"{name} {n}일 연속 {'상승' if up else '하락'}"


def _big_move_note(
    name: str, hist_deltas: list[float | None], today: float | None, points: bool
) -> str | None:
    """Flag today's move when it exceeds BIG_MOVE_RATIO x the mean absolute move of stored sessions."""
    moves = [abs(d) for d in hist_deltas if d is not None]
    if today is None or len(moves) < MIN_SESSIONS_FOR_AVG:
        return None
    mean = sum(moves) / len(moves)
    if mean <= 0 or abs(today) <= BIG_MOVE_RATIO * mean:
        return None
    ratio = abs(today) / mean
    if points:
        return (
            f"{name} 오늘 {today * 100:+.1f}bp — 최근 {len(moves)}일 평균 변동폭({mean * 100:.1f}bp)의 "
            f"{ratio:.1f}배, 평소보다 큰 변동"
        )
    return f"{name} 오늘 {today:+.2f}% — 최근 {len(moves)}일 평균 변동폭({mean:.2f}%)의 {ratio:.1f}배, 평소보다 큰 변동"


def _krw_levels(prev: float, close: float) -> list[float]:
    """Candidate multiples of KRW_STEP spanning [prev, close]; crossing is decided by the caller."""
    lo, hi = sorted((prev, close))
    first = math.floor(lo / KRW_STEP) * KRW_STEP
    last = math.ceil(hi / KRW_STEP) * KRW_STEP
    return [first + i * KRW_STEP for i in range(round((last - first) / KRW_STEP) + 1)]


def _fmt_pair(a: float, b: float, decimals: int) -> tuple[str, str]:
    """Format two values, adding precision (up to 4 dp) until the strings differ.

    Otherwise a 1399.99 -> 1400.0 break would read "1,400 → 1,400", contradicting itself.
    """
    sa, sb = f"{a:,.{decimals}f}", f"{b:,.{decimals}f}"
    for d in range(decimals + 1, 5):
        if sa != sb:
            break
        sa, sb = f"{a:,.{d}f}", f"{b:,.{d}f}"
    return sa, sb


def _level_cross_notes(
    name: str, prev: float, close: float, levels: list[float], decimals: int, unit: str = ""
) -> list[str]:
    """Note for a level strictly crossed between prev and close.

    Up: prev < L < close; down: prev > L > close.

    Touching a level exactly (landing on it or leaving from it) is not a break.
    """
    notes: list[str] = []
    crossed = [lvl for lvl in levels if (prev < lvl < close) or (prev > lvl > close)]
    if not crossed:
        return notes
    # report only the level nearest to today's close (avoid spam if several were crossed)
    lvl = min(crossed, key=lambda x: abs(x - close))
    direction = "상향 돌파" if close > prev else "하향 이탈"
    lvl_txt = f"{lvl:,.0f}{unit}"
    prev_txt, close_txt = _fmt_pair(prev, close, decimals)
    notes.append(f"{name} {lvl_txt} {direction} ({prev_txt} → {close_txt})")
    return notes