"""YouTube relay session history (simplified from sh2 youtube_relay_log.py)."""

from __future__ import annotations

import json
import re
import sqlite3
import time
import uuid
from datetime import datetime
from pathlib import Path
from typing import Any

_TABLE = "youtube_relay_sessions"
_RECORDING_LINE_RE = re.compile(
    r"正在录制\s*\d+\s*(?:个|路)\s*[:：]\s*([^[\n]+?)\[原画\s*([0-9:]+)\]",
    re.IGNORECASE,
)
_PUSH_START_RE = re.compile(r"开始推流到 YouTube:\s*(.+?)\s*$")
_INCOMING_URL_RE = re.compile(r"传入地址:\s*(https?://\S+)", re.IGNORECASE)
_ROOM_RE = re.compile(r"live\.douyin\.com/([a-zA-Z0-9_-]+)", re.IGNORECASE)

_ACTIVE_BY_PROCESS: dict[str, int] = {}


def relay_db_path(base_dir: Path) -> Path:
    path = base_dir / "runtime" / "youtube_relay" / "sessions.sqlite"
    path.parent.mkdir(parents=True, exist_ok=True)
    return path


def _connect(db_path: Path) -> sqlite3.Connection:
    conn = sqlite3.connect(str(db_path), check_same_thread=False)
    conn.row_factory = sqlite3.Row
    conn.execute(
        f"""
        CREATE TABLE IF NOT EXISTS {_TABLE} (
          id INTEGER PRIMARY KEY AUTOINCREMENT,
          session_uuid TEXT NOT NULL UNIQUE,
          process TEXT NOT NULL DEFAULT '',
          channel_name TEXT NOT NULL DEFAULT '',
          anchor_name TEXT NOT NULL DEFAULT '',
          source_url TEXT NOT NULL DEFAULT '',
          source_room_id TEXT NOT NULL DEFAULT '',
          stream_title TEXT NOT NULL DEFAULT '',
          youtube_key_suffix TEXT NOT NULL DEFAULT '',
          started_at REAL NOT NULL,
          ended_at REAL,
          duration_seconds REAL NOT NULL DEFAULT 0,
          push_date TEXT NOT NULL DEFAULT '',
          status TEXT NOT NULL DEFAULT 'active',
          end_reason TEXT NOT NULL DEFAULT '',
          metadata_json TEXT NOT NULL DEFAULT '{{}}',
          created_at REAL NOT NULL,
          updated_at REAL NOT NULL
        )
        """
    )
    conn.execute(
        f"CREATE INDEX IF NOT EXISTS idx_relay_sess_proc ON {_TABLE}(process, started_at DESC)"
    )
    conn.commit()
    return conn


def _row_to_session(row: sqlite3.Row) -> dict[str, Any]:
    data = dict(row)
    meta_raw = data.pop("metadata_json", "{}")
    try:
        data["metadata"] = json.loads(meta_raw) if meta_raw else {}
    except json.JSONDecodeError:
        data["metadata"] = {}
    if data.get("ended_at"):
        data["duration_seconds"] = int(float(data.get("duration_seconds") or 0))
    elif float(data.get("duration_seconds") or 0) > 0:
        data["duration_seconds"] = int(float(data["duration_seconds"]))
    else:
        data["duration_seconds"] = int(max(0, time.time() - float(data.get("started_at") or time.time())))
    return data


def _parse_duration_token(token: str) -> int:
    parts = (token or "").strip().split(":")
    try:
        if len(parts) == 3:
            return int(parts[0]) * 3600 + int(parts[1]) * 60 + int(float(parts[2]))
        if len(parts) == 2:
            return int(parts[0]) * 60 + int(float(parts[1]))
    except (TypeError, ValueError):
        pass
    return 0


def _room_from_text(text: str) -> str:
    m = _ROOM_RE.search(text or "")
    return m.group(1) if m else ""


def _lookup_url(anchor: str, url_lines: str) -> tuple[str, str]:
    anchor = (anchor or "").strip()
    for line in (url_lines or "").splitlines():
        t = line.strip()
        if not t or t.startswith("#"):
            continue
        if anchor and anchor in t:
            return t.split("#")[0].strip(), _room_from_text(t)
    for line in (url_lines or "").splitlines():
        t = line.strip()
        if t and not t.startswith("#") and "douyin" in t.lower():
            return t.split("#")[0].strip(), _room_from_text(t)
    return "", ""


def start_relay_session(
    *,
    process: str,
    anchor_name: str = "",
    source_url: str = "",
    source_room_id: str = "",
    channel_name: str = "",
    stream_title: str = "",
    metadata: dict[str, Any] | None = None,
    db_path: Path,
) -> dict[str, Any]:
    proc = (process or "youtube").strip()
    now = time.time()
    session_uuid = uuid.uuid4().hex
    push_date = datetime.now().strftime("%Y-%m-%d")
    conn = _connect(db_path)
    try:
        cur = conn.execute(
            f"""
            INSERT INTO {_TABLE} (
              session_uuid, process, channel_name, anchor_name, source_url, source_room_id,
              stream_title, started_at, push_date, status, metadata_json, created_at, updated_at
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'active', ?, ?, ?)
            """,
            (
                session_uuid,
                proc,
                channel_name,
                anchor_name,
                source_url,
                source_room_id or _room_from_text(source_url),
                stream_title,
                now,
                push_date,
                json.dumps(metadata or {}, ensure_ascii=False),
                now,
                now,
            ),
        )
        conn.commit()
        sid = int(cur.lastrowid)
        _ACTIVE_BY_PROCESS[proc] = sid
        row = conn.execute(f"SELECT * FROM {_TABLE} WHERE id = ?", (sid,)).fetchone()
        return _row_to_session(row) if row else {}
    finally:
        conn.close()


def end_relay_session(
    process: str,
    *,
    reason: str = "stopped",
    duration_seconds: int | None = None,
    db_path: Path,
) -> dict[str, Any] | None:
    proc = (process or "youtube").strip()
    conn = _connect(db_path)
    try:
        sid = _ACTIVE_BY_PROCESS.get(proc)
        if not sid:
            row = conn.execute(
                f"SELECT id FROM {_TABLE} WHERE process = ? AND ended_at IS NULL ORDER BY started_at DESC LIMIT 1",
                (proc,),
            ).fetchone()
            sid = int(row["id"]) if row else None
        if not sid:
            return None
        now = time.time()
        dur = duration_seconds
        if dur is None:
            row = conn.execute(f"SELECT started_at, duration_seconds FROM {_TABLE} WHERE id = ?", (sid,)).fetchone()
            if row:
                stored = float(row["duration_seconds"] or 0)
                dur = int(stored) if stored > 0 else int(max(0, now - float(row["started_at"])))
            else:
                dur = 0
        conn.execute(
            f"""
            UPDATE {_TABLE}
            SET ended_at = ?, duration_seconds = ?, status = 'ended', end_reason = ?, updated_at = ?
            WHERE id = ?
            """,
            (now, dur, reason, now, sid),
        )
        conn.commit()
        _ACTIVE_BY_PROCESS.pop(proc, None)
        row = conn.execute(f"SELECT * FROM {_TABLE} WHERE id = ?", (sid,)).fetchone()
        return _row_to_session(row) if row else None
    finally:
        conn.close()


def sync_relay_from_recent_log(
    process: str,
    recent_log: str,
    *,
    channel_name: str = "",
    url_lines: str = "",
    db_path: Path,
) -> dict[str, Any] | None:
    proc = (process or "youtube").strip()
    anchor = ""
    duration_token = ""
    for line in reversed((recent_log or "").splitlines()):
        match = _RECORDING_LINE_RE.search(line)
        if match:
            anchor = match.group(1).strip(" :?")
            duration_token = match.group(2).strip()
            break
    if not anchor:
        if "无录制任务" in (recent_log or "")[-2000:]:
            closed = end_relay_session(proc, reason="idle", db_path=db_path)
            return {"closed": closed} if closed else None
        return None

    source_url, room_id = _lookup_url(anchor, url_lines)
    row = None
    conn = _connect(db_path)
    try:
        row = conn.execute(
            f"""
            SELECT id, anchor_name FROM {_TABLE}
            WHERE process = ? AND ended_at IS NULL ORDER BY started_at DESC LIMIT 1
            """,
            (proc,),
        ).fetchone()
        if row and str(row["anchor_name"] or "").strip() == anchor:
            if duration_token:
                duration = _parse_duration_token(duration_token)
                conn.execute(
                    f"UPDATE {_TABLE} SET duration_seconds = ?, updated_at = ? WHERE id = ?",
                    (duration, time.time(), int(row["id"])),
                )
                conn.commit()
            return {"updated": int(row["id"])}
    finally:
        conn.close()

    if row:
        end_relay_session(proc, reason="source_switch", db_path=db_path)
    session = start_relay_session(
        process=proc,
        anchor_name=anchor,
        source_url=source_url,
        source_room_id=room_id,
        channel_name=channel_name,
        metadata={"sync": "recent_log"},
        db_path=db_path,
    )
    if duration_token and session.get("id"):
        duration = _parse_duration_token(duration_token)
        conn = _connect(db_path)
        try:
            conn.execute(
                f"UPDATE {_TABLE} SET duration_seconds = ?, updated_at = ? WHERE id = ?",
                (duration, time.time(), int(session["id"])),
            )
            conn.commit()
        finally:
            conn.close()
    return session


def list_relay_sessions(
    *,
    process: str = "",
    q: str = "",
    status: str = "",
    limit: int = 50,
    offset: int = 0,
    db_path: Path,
) -> dict[str, Any]:
    clauses: list[str] = []
    params: list[Any] = []
    if process:
        clauses.append("process = ?")
        params.append(process.strip())
    if status == "active":
        clauses.append("ended_at IS NULL")
    elif status == "ended":
        clauses.append("ended_at IS NOT NULL")
    if q.strip():
        like = f"%{q.strip()}%"
        clauses.append(
            "(anchor_name LIKE ? OR source_url LIKE ? OR source_room_id LIKE ? OR channel_name LIKE ?)"
        )
        params.extend([like, like, like, like])
    where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
    safe_limit = max(1, min(int(limit), 200))
    safe_offset = max(0, int(offset))
    conn = _connect(db_path)
    try:
        rows = conn.execute(
            f"SELECT * FROM {_TABLE} {where} ORDER BY started_at DESC LIMIT ? OFFSET ?",
            (*params, safe_limit, safe_offset),
        ).fetchall()
        total = conn.execute(f"SELECT COUNT(*) FROM {_TABLE} {where}", params).fetchone()[0]
        return {
            "sessions": [_row_to_session(r) for r in rows],
            "total": int(total),
            "limit": safe_limit,
            "offset": safe_offset,
        }
    finally:
        conn.close()


def import_relay_sessions_from_log(
    log_path: Path,
    *,
    process: str = "youtube",
    url_lines: str = "",
    db_path: Path,
) -> dict[str, Any]:
    if not log_path.is_file():
        return {"imported": 0, "error": "日志文件不存在"}
    text = log_path.read_text(encoding="utf-8", errors="replace")
    proc = (process or "youtube").strip()
    imported = 0
    active: dict[str, Any] | None = None
    conn = _connect(db_path)
    try:
        for raw_line in text.splitlines():
            line = raw_line.strip()
            if not line:
                continue
            push_match = _PUSH_START_RE.search(line)
            if push_match:
                if active:
                    _insert_imported(conn, active)
                    imported += 1
                title = push_match.group(1).strip()
                active = {
                    "process": proc,
                    "stream_title": title,
                    "anchor_name": title.split("_")[0][:64],
                    "started_at": time.time(),
                    "source_url": "",
                    "source_room_id": "",
                }
                continue
            rec_match = _RECORDING_LINE_RE.search(line)
            if rec_match:
                anchor = rec_match.group(1).strip(" :?")
                duration = _parse_duration_token(rec_match.group(2).strip())
                url, room = _lookup_url(anchor, url_lines)
                if active:
                    active["anchor_name"] = anchor
                    active["duration_seconds"] = duration
                    active["source_url"] = url
                    active["source_room_id"] = room
                else:
                    active = {
                        "process": proc,
                        "anchor_name": anchor,
                        "duration_seconds": duration,
                        "started_at": time.time() - duration,
                        "source_url": url,
                        "source_room_id": room,
                        "stream_title": "",
                    }
            incoming = _INCOMING_URL_RE.search(line)
            if incoming and active:
                url = incoming.group(1).strip()
                active["source_url"] = url
                active["source_room_id"] = _room_from_text(url)
        if active:
            _insert_imported(conn, active)
            imported += 1
        conn.commit()
    finally:
        conn.close()
    return {"imported": imported}


def _insert_imported(conn: sqlite3.Connection, payload: dict[str, Any]) -> None:
    now = time.time()
    started = float(payload.get("started_at") or now)
    conn.execute(
        f"""
        INSERT INTO {_TABLE} (
          session_uuid, process, anchor_name, source_url, source_room_id, stream_title,
          started_at, ended_at, duration_seconds, push_date, status, end_reason,
          metadata_json, created_at, updated_at
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'ended', 'imported_from_log', ?, ?, ?)
        """,
        (
            uuid.uuid4().hex,
            payload.get("process") or "youtube",
            payload.get("anchor_name") or "",
            payload.get("source_url") or "",
            payload.get("source_room_id") or "",
            payload.get("stream_title") or "",
            started,
            now,
            int(payload.get("duration_seconds") or 0),
            datetime.fromtimestamp(started).strftime("%Y-%m-%d"),
            json.dumps({"import": True}, ensure_ascii=False),
            now,
            now,
        ),
    )


def track_process_status_transition(
    process: str,
    old_status: str,
    new_status: str,
    *,
    db_path: Path,
) -> None:
    old_s = (old_status or "").strip().lower()
    new_s = (new_status or "").strip().lower()
    proc = (process or "").strip()
    if not proc:
        return
    if old_s != "online" and new_s == "online":
        start_relay_session(process=proc, metadata={"trigger": "pm2_online"}, db_path=db_path)
    elif old_s == "online" and new_s in {"stopped", "errored", "not_found"}:
        end_relay_session(proc, reason=new_s, db_path=db_path)