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

from __future__ import annotations

import json
import re
import sqlite3
import subprocess
import threading
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)

from web2_youtube_live_store import (
    extract_first_url,
    extract_source_room_id,
    extract_youtube_key_suffix,
)

_STREAM_TITLE_TS_RE = re.compile(r"_(?P<ts>\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2})$")
_LOG_TS_RE = re.compile(r"^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}):")
_FAIL_ADDR_RE = re.compile(
    r"获取失败的地址是:\('原画',\s*'(https?://[^']+)',\s*'主播:\s*([^']+)'\)"
)
_ANCHOR_LABEL_RE = re.compile(r"主播[:：]\s*([^\s'\"()]+)")
_JUNK_END_REASONS = frozenset({"not_found", "idle", "pm2_online", "stopped", "errored", "error"})
_ACTIVE_BY_PROCESS: dict[str, int] = {}

from web2_douyin_source_display import (
    _entry_for_title,
    _url_entries_from_lines,
    extract_douyin_source_meta,
)


def _normalize_process(process: str) -> str:
    return (process or "").strip() or "youtube"


def _push_date_from_ts(ts: float) -> str:
    return datetime.fromtimestamp(ts).strftime("%Y-%m-%d")


def _parse_stream_title(stream_title: str) -> tuple[str, str]:
    title = (stream_title or "").strip()
    match = _STREAM_TITLE_TS_RE.search(title)
    if not match:
        return title, ""
    anchor = title[: match.start()].rstrip("_")
    stamp = match.group("ts")
    push_date = stamp.split("_", 1)[0] if "_" in stamp else ""
    return anchor, push_date


def _timestamp_from_stream_title(stream_title: str) -> float | None:
    match = _STREAM_TITLE_TS_RE.search(stream_title or "")
    if not match:
        return None
    try:
        return datetime.strptime(match.group("ts"), "%Y-%m-%d_%H-%M-%S").timestamp()
    except ValueError:
        return None


def _parse_log_line_ts(line: str) -> float | None:
    match = _LOG_TS_RE.match(line.strip())
    if not match:
        return None
    try:
        return datetime.strptime(match.group(1), "%Y-%m-%dT%H:%M:%S").timestamp()
    except ValueError:
        return None


def _parse_local_wall_time(line: str, log_ts: float | None) -> float | None:
    if log_ts is None:
        return None
    local_hms = ""
    for part in reversed(line.split("|")):
        token = part.strip()
        if re.fullmatch(r"\d{2}:\d{2}:\d{2}", token):
            local_hms = token
            break
    if not local_hms:
        return log_ts
    try:
        base = datetime.fromtimestamp(log_ts)
        hour, minute, second = (int(x) for x in local_hms.split(":"))
        return base.replace(hour=hour, minute=minute, second=second, microsecond=0).timestamp()
    except (ValueError, OSError):
        return log_ts


def _clean_douyin_url(url: str) -> str:
    return extract_first_url(url) or ""


def _build_anchor_url_index(url_lines: str) -> dict[str, tuple[str, str]]:
    index: dict[str, tuple[str, str]] = {}
    for entry in _url_entries_from_lines(url_lines):
        title = (entry.get("source_title") or "").strip()
        url = _clean_douyin_url(entry.get("source_url") or "")
        room_id = entry.get("room_id") or ""
        if title and url:
            index[title] = (url, room_id)
    return index


def _lookup_room_id(anchor_name: str, url_lines: str) -> tuple[str, str]:
    anchor = (anchor_name or "").strip()
    for line in (url_lines or "").splitlines():
        stripped = line.strip()
        if not stripped or stripped.startswith("#"):
            continue
        if anchor and anchor not in stripped:
            continue
        url = extract_first_url(stripped)
        room_id = extract_source_room_id(stripped, url)
        if url:
            return url, room_id
    meta = extract_douyin_source_meta(url_lines, anchor)
    return str(meta.get("source_url") or ""), str(meta.get("source_room_id") or "")


def _resolve_anchor_url(
    anchor: str,
    url_index: dict[str, tuple[str, str]],
    url_lines: str,
    log_url_index: dict[str, tuple[str, str]] | None = None,
) -> tuple[str, str]:
    name = (anchor or "").strip()
    if not name:
        return "", ""
    if log_url_index and name in log_url_index:
        return log_url_index[name]
    entry = _entry_for_title(_url_entries_from_lines(url_lines), name)
    if entry:
        return _clean_douyin_url(entry["source_url"]), entry["room_id"]
    if name in url_index:
        url, room_id = url_index[name]
        return _clean_douyin_url(url), room_id
    meta = extract_douyin_source_meta(url_lines, name)
    url = _clean_douyin_url(str(meta.get("source_url") or ""))
    room_id = str(meta.get("source_room_id") or "")
    if url:
        return url, room_id
    return _lookup_room_id(name, url_lines)


def _session_has_signal(row: sqlite3.Row | dict[str, Any]) -> bool:
    data = dict(row) if not isinstance(row, dict) else row
    anchor = str(data.get("anchor_name") or "").strip()
    source_url = str(data.get("source_url") or "").strip()
    stream_title = str(data.get("stream_title") or "").strip()
    room_id = str(data.get("source_room_id") or "").strip()
    return bool(anchor or source_url or stream_title or room_id)


def _is_junk_session_row(row: sqlite3.Row | dict[str, Any]) -> bool:
    if _session_has_signal(row):
        return False
    data = dict(row) if not isinstance(row, dict) else row
    end_reason = str(data.get("end_reason") or "").strip().lower()
    if end_reason in _JUNK_END_REASONS:
        return True
    meta_raw = data.get("metadata_json") or data.get("metadata") or {}
    if isinstance(meta_raw, str):
        try:
            meta = json.loads(meta_raw) if meta_raw else {}
        except json.JSONDecodeError:
            meta = {}
    else:
        meta = meta_raw if isinstance(meta_raw, dict) else {}
    if str(meta.get("trigger") or "") == "pm2_online":
        return True
    return not end_reason and str(data.get("status") or "") == "active"


def purge_junk_relay_sessions(*, process: str = "", db_path: Path) -> int:
    proc = _normalize_process(process) if process else ""
    conn = _connect(db_path)
    removed = 0
    try:
        rows = conn.execute(f"SELECT * FROM {_TABLE}").fetchall()
        for row in rows:
            if proc and str(row["process"] or "") != proc:
                continue
            if _is_junk_session_row(row):
                conn.execute(f"DELETE FROM {_TABLE} WHERE id = ?", (int(row["id"]),))
                removed += 1
        conn.commit()
    finally:
        conn.close()
    return removed


def _close_session_row(
    conn: sqlite3.Connection,
    row_id: int,
    *,
    ended_at: float | None = None,
    reason: str = "",
    duration_seconds: int | None = None,
) -> None:
    row = conn.execute(f"SELECT * FROM {_TABLE} WHERE id = ?", (row_id,)).fetchone()
    if not row or row["ended_at"]:
        return
    end = float(ended_at or time.time())
    duration = duration_seconds
    if duration is None:
        duration = int(max(0, end - float(row["started_at"])))
    conn.execute(
        f"""
        UPDATE {_TABLE}
        SET ended_at = ?, duration_seconds = ?, status = 'ended', end_reason = ?, updated_at = ?
        WHERE id = ?
        """,
        (end, int(duration), (reason or "")[:500], end, row_id),
    )
    process = str(row["process"] or "")
    if _ACTIVE_BY_PROCESS.get(process) == row_id:
        _ACTIVE_BY_PROCESS.pop(process, None)


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 _covers_dir(base_dir: Path) -> Path:
    path = base_dir / "runtime" / "youtube_relay" / "covers"
    path.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 '',
          cover_path 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)"
    )
    try:
        cols = {row[1] for row in conn.execute(f"PRAGMA table_info({_TABLE})").fetchall()}
        if "cover_path" not in cols:
            conn.execute(f"ALTER TABLE {_TABLE} ADD COLUMN cover_path TEXT NOT NULL DEFAULT ''")
    except sqlite3.OperationalError:
        pass
    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())))
    data["cover_url"] = (
        f"/youtube/relay/sessions/{data['id']}/cover" if data.get("cover_path") else ""
    )
    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 = "",
    youtube_key_suffix: str = "",
    metadata: dict[str, Any] | None = None,
    capture_cover_url: str = "",
    db_path: Path,
    base_dir: Path | None = None,
) -> dict[str, Any]:
    proc = _normalize_process(process)
    now = time.time()
    anchor = (anchor_name or "").strip()
    if not source_url:
        source_url = extract_first_url(stream_title)
    if not source_room_id:
        source_room_id = extract_source_room_id(source_url, anchor_name, stream_title)
    conn = _connect(db_path)
    try:
        open_rows = conn.execute(
            f"SELECT id, anchor_name FROM {_TABLE} WHERE process = ? AND ended_at IS NULL",
            (proc,),
        ).fetchall()
        for row in open_rows:
            if str(row["anchor_name"] or "").strip() == anchor and anchor:
                session = _row_to_session(
                    conn.execute(f"SELECT * FROM {_TABLE} WHERE id = ?", (int(row["id"]),)).fetchone()
                )
                conn.commit()
                return session
            _close_session_row(conn, int(row["id"]), ended_at=now, reason="source_switch")
        session_uuid = uuid.uuid4().hex
        push_date = _push_date_from_ts(now)
        _, title_date = _parse_stream_title(stream_title)
        if title_date:
            push_date = title_date
        title_started = _timestamp_from_stream_title(stream_title)
        if title_started:
            now = title_started
        conn.execute(
            f"""
            INSERT INTO {_TABLE} (
              session_uuid, process, channel_name, anchor_name, source_url, source_room_id,
              stream_title, youtube_key_suffix, started_at, duration_seconds, push_date,
              cover_path, status, metadata_json, created_at, updated_at
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, '', 'active', ?, ?, ?)
            """,
            (
                session_uuid,
                proc,
                channel_name or "",
                anchor,
                source_url or "",
                source_room_id or "",
                stream_title or "",
                youtube_key_suffix or "",
                now,
                push_date,
                json.dumps(metadata or {}, ensure_ascii=False),
                now,
                now,
            ),
        )
        session_id = int(conn.execute("SELECT last_insert_rowid()").fetchone()[0])
        _ACTIVE_BY_PROCESS[proc] = session_id
        conn.commit()
        session = _row_to_session(
            conn.execute(f"SELECT * FROM {_TABLE} WHERE id = ?", (session_id,)).fetchone()
        )
    finally:
        conn.close()
    cover_url = capture_cover_url or source_url
    if cover_url and session.get("id") and base_dir:
        threading.Thread(
            target=_capture_cover_safe,
            args=(int(session["id"]), cover_url, db_path, base_dir),
            daemon=True,
        ).start()
    return session


def end_relay_session(
    process: str,
    *,
    anchor_name: str = "",
    reason: str = "stopped",
    duration_seconds: int | None = None,
    db_path: Path,
) -> dict[str, Any] | None:
    proc = _normalize_process(process)
    conn = _connect(db_path)
    try:
        if anchor_name.strip():
            row = conn.execute(
                f"""
                SELECT id FROM {_TABLE}
                WHERE process = ? AND ended_at IS NULL AND anchor_name = ?
                ORDER BY started_at DESC LIMIT 1
                """,
                (proc, anchor_name.strip()),
            ).fetchone()
        else:
            row = conn.execute(
                f"""
                SELECT id FROM {_TABLE}
                WHERE process = ? AND ended_at IS NULL
                ORDER BY started_at DESC LIMIT 1
                """,
                (proc,),
            ).fetchone()
        if not row:
            conn.commit()
            return None
        _close_session_row(
            conn,
            int(row["id"]),
            reason=reason or "ended",
            duration_seconds=duration_seconds,
        )
        conn.commit()
        closed = conn.execute(f"SELECT * FROM {_TABLE} WHERE id = ?", (int(row["id"]),)).fetchone()
        return _row_to_session(closed) if closed else None
    finally:
        conn.close()


def close_active_relay_sessions(
    process: str,
    *,
    reason: str = "",
    ended_at: float | None = None,
    db_path: Path,
) -> list[dict[str, Any]]:
    proc = _normalize_process(process)
    conn = _connect(db_path)
    try:
        rows = conn.execute(
            f"SELECT id FROM {_TABLE} WHERE process = ? AND ended_at IS NULL ORDER BY started_at DESC",
            (proc,),
        ).fetchall()
        for row in rows:
            _close_session_row(conn, int(row["id"]), ended_at=ended_at, reason=reason)
        conn.commit()
        if not rows:
            return []
        ids = [int(row["id"]) for row in rows]
        return [
            _row_to_session(item)
            for item in conn.execute(
                f"SELECT * FROM {_TABLE} WHERE id IN ({','.join('?' * len(ids))})",
                ids,
            ).fetchall()
        ]
    finally:
        conn.close()


def on_relay_push_start(
    *,
    process: str,
    anchor_name: str,
    record_url: str | None,
    stream_title: str,
    real_url: str,
    channel_name: str = "",
    youtube_ini_text: str = "",
    db_path: Path,
    base_dir: Path | None = None,
) -> dict[str, Any]:
    source_url = (record_url or "").strip() or extract_first_url(record_url or "")
    if not source_url and real_url:
        source_url = real_url
    room_id = extract_source_room_id(record_url or "", real_url or "", anchor_name)
    return start_relay_session(
        process=process,
        anchor_name=anchor_name,
        source_url=source_url,
        source_room_id=room_id,
        stream_title=stream_title,
        channel_name=channel_name,
        youtube_key_suffix=extract_youtube_key_suffix(youtube_ini_text),
        metadata={"real_url": real_url, "record_url": record_url or ""},
        capture_cover_url=real_url or source_url,
        db_path=db_path,
        base_dir=base_dir,
    )


def on_relay_push_end(
    process: str,
    *,
    anchor_name: str,
    reason: str = "push_stopped",
    db_path: Path,
) -> dict[str, Any] | None:
    return end_relay_session(process, anchor_name=anchor_name, reason=reason, db_path=db_path)


def get_relay_session(session_id: int | str, *, db_path: Path) -> dict[str, Any] | None:
    conn = _connect(db_path)
    try:
        row = conn.execute(
            f"SELECT * FROM {_TABLE} WHERE id = ? OR session_uuid = ?",
            (session_id, str(session_id)),
        ).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,
    base_dir: Path | None = None,
) -> 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 = close_active_relay_sessions(proc, reason="idle", db_path=db_path)
            return {"closed": closed} if closed else None
        return None

    source_url, room_id = _resolve_anchor_url(anchor, _build_anchor_url_index(url_lines), url_lines)
    if not source_url:
        meta = extract_douyin_source_meta(url_lines, recent_log)
        source_url = str(meta.get("source_url") or "")
        room_id = str(meta.get("source_room_id") or room_id)
    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, anchor_name=str(row["anchor_name"] or ""), 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"},
        capture_cover_url=source_url,
        db_path=db_path,
        base_dir=base_dir or db_path.parent.parent.parent,
    )
    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 = "",
    date_from: str = "",
    date_to: 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 date_from:
        clauses.append("push_date >= ?")
        params.append(date_from.strip())
    if date_to:
        clauses.append("push_date <= ?")
        params.append(date_to.strip())
    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))
    purge_junk_relay_sessions(process=process, db_path=db_path)
    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()
        sessions = [_row_to_session(r) for r in rows if _session_has_signal(r)]
        total = conn.execute(
            f"SELECT COUNT(*) FROM {_TABLE} {where}",
            params,
        ).fetchone()[0]
        return {
            "sessions": sessions,
            "total": int(total),
            "limit": safe_limit,
            "offset": safe_offset,
        }
    finally:
        conn.close()


def clear_imported_relay_sessions(process: str, *, db_path: Path) -> int:
    proc = _normalize_process(process)
    conn = _connect(db_path)
    try:
        cur = conn.execute(
            f"DELETE FROM {_TABLE} WHERE process = ? AND end_reason = 'imported_from_log'",
            (proc,),
        )
        conn.commit()
        return int(cur.rowcount or 0)
    finally:
        conn.close()


def _finalize_import_payload(payload: dict[str, Any]) -> dict[str, Any]:
    started_at = float(payload.get("started_at") or 0)
    duration = int(payload.get("duration_seconds") or 0)
    ended_at = payload.get("ended_at")
    if ended_at is None and duration > 0 and started_at > 0:
        ended_at = started_at + duration
    payload["ended_at"] = ended_at
    payload["duration_seconds"] = duration
    return payload


def _new_import_payload(
    *,
    process: str,
    anchor_name: str,
    stream_title: str = "",
    started_at: float,
    push_date: str = "",
    source_url: str = "",
    source_room_id: str = "",
    duration_seconds: int = 0,
    youtube_key_suffix: str = "",
) -> dict[str, Any]:
    anchor = (anchor_name or "").strip()
    title = (stream_title or "").strip()
    if not push_date:
        _, push_date = _parse_stream_title(title)
    if not push_date and started_at:
        push_date = _push_date_from_ts(started_at)
    return {
        "process": process,
        "anchor_name": anchor,
        "stream_title": title,
        "started_at": started_at,
        "push_date": push_date,
        "source_url": source_url,
        "source_room_id": source_room_id,
        "duration_seconds": duration_seconds,
        "youtube_key_suffix": youtube_key_suffix,
        "ended_at": None,
    }


def _persist_import_row(
    conn: sqlite3.Connection,
    payload: dict[str, Any],
    *,
    started_at: float,
    ended_at: float | None,
    duration_seconds: int,
) -> None:
    anchor = str(payload.get("anchor_name") or "").strip()
    stream_title = str(payload.get("stream_title") or "").strip()
    if not anchor and not stream_title:
        return
    if duration_seconds > 0 and started_at > 0:
        computed_end = started_at + duration_seconds
        if ended_at is None or ended_at < started_at:
            ended_at = computed_end
    dedupe_started = _timestamp_from_stream_title(stream_title) or started_at
    existing = conn.execute(
        f"""
        SELECT id FROM {_TABLE}
        WHERE process = ? AND anchor_name = ? AND stream_title = ? AND ABS(started_at - ?) < 180
        LIMIT 1
        """,
        (payload.get("process") or "youtube", anchor, stream_title, dedupe_started),
    ).fetchone()
    if not existing and stream_title:
        existing = conn.execute(
            f"""
            SELECT id FROM {_TABLE}
            WHERE process = ? AND anchor_name = ? AND stream_title = ?
            LIMIT 1
            """,
            (payload.get("process") or "youtube", anchor, stream_title),
        ).fetchone()
    if existing:
        return
    now = time.time()
    source_url = _clean_douyin_url(str(payload.get("source_url") or ""))
    source_room_id = str(
        payload.get("source_room_id") or extract_source_room_id(source_url, stream_title, anchor)
    )
    conn.execute(
        f"""
        INSERT INTO {_TABLE} (
          session_uuid, process, channel_name, anchor_name, source_url, source_room_id,
          stream_title, youtube_key_suffix, started_at, ended_at, duration_seconds, push_date,
          cover_path, status, end_reason, metadata_json, created_at, updated_at
        ) VALUES (?, ?, '', ?, ?, ?, ?, ?, ?, ?, ?, ?, '', 'ended', 'imported_from_log', ?, ?, ?)
        """,
        (
            uuid.uuid4().hex,
            payload.get("process") or "youtube",
            anchor,
            source_url,
            source_room_id,
            stream_title,
            payload.get("youtube_key_suffix") or "",
            started_at,
            ended_at,
            duration_seconds,
            payload.get("push_date") or _push_date_from_ts(started_at),
            json.dumps({"import": True}, ensure_ascii=False),
            now,
            now,
        ),
    )


def import_relay_sessions_from_log(
    log_path: Path,
    *,
    process: str = "youtube",
    url_lines: str = "",
    youtube_key_suffix: str = "",
    db_path: Path,
    replace_imported: bool = False,
) -> dict[str, Any]:
    path = Path(log_path)
    if not path.is_file():
        return {"imported": 0, "closed": 0, "error": "日志文件不存在"}
    text = path.read_text(encoding="utf-8", errors="replace")
    proc = _normalize_process(process)
    removed = clear_imported_relay_sessions(proc, db_path=db_path) if replace_imported else 0
    purge_junk_relay_sessions(process=proc, db_path=db_path)
    url_index = _build_anchor_url_index(url_lines)
    log_url_index: dict[str, tuple[str, str]] = {}
    for raw_line in text.splitlines():
        fail_match = _FAIL_ADDR_RE.search(raw_line)
        if fail_match:
            url = _clean_douyin_url(fail_match.group(1))
            anchor = fail_match.group(2).strip()
            if url and anchor:
                log_url_index[anchor] = (url, extract_source_room_id(url))
        incoming = _INCOMING_URL_RE.search(raw_line)
        if incoming:
            url = _clean_douyin_url(incoming.group(1))
            if url:
                log_url_index["__last_incoming__"] = (url, extract_source_room_id(url))
    imported = 0
    closed = 0
    active: dict[str, Any] | None = None
    sessions: list[dict[str, Any]] = []
    conn = _connect(db_path)
    try:
        for raw_line in text.splitlines():
            line = raw_line.strip()
            if not line:
                continue
            log_ts = _parse_log_line_ts(line)
            event_ts = _parse_local_wall_time(line, log_ts) or log_ts

            incoming = _INCOMING_URL_RE.search(line)
            if incoming:
                url = _clean_douyin_url(incoming.group(1))
                room_id = extract_source_room_id(url)
                log_url_index["__last_incoming__"] = (url, room_id)
                label = _ANCHOR_LABEL_RE.search(line)
                if label:
                    log_url_index[label.group(1).strip()] = (url, room_id)
                elif active and not active.get("source_url"):
                    active["source_url"] = url
                    active["source_room_id"] = room_id
                continue

            fail_match = _FAIL_ADDR_RE.search(line)
            if fail_match:
                url = _clean_douyin_url(fail_match.group(1))
                anchor = fail_match.group(2).strip()
                if url and anchor:
                    log_url_index[anchor] = (url, extract_source_room_id(url))
                continue

            push_match = _PUSH_START_RE.search(line)
            if push_match:
                stream_title = push_match.group(1).strip()
                anchor, push_date = _parse_stream_title(stream_title)
                started_at = _timestamp_from_stream_title(stream_title) or event_ts or time.time()
                if active:
                    sessions.append(_finalize_import_payload(active))
                    closed += 1
                source_url, source_room_id = _resolve_anchor_url(anchor, url_index, url_lines, log_url_index)
                active = _new_import_payload(
                    process=proc,
                    anchor_name=anchor,
                    stream_title=stream_title,
                    started_at=started_at,
                    push_date=push_date,
                    source_url=source_url,
                    source_room_id=source_room_id,
                    youtube_key_suffix=youtube_key_suffix,
                )
                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())
                if active and str(active.get("anchor_name") or "") == anchor:
                    active["duration_seconds"] = max(int(active.get("duration_seconds") or 0), duration)
                    if not active.get("source_url"):
                        source_url, source_room_id = _resolve_anchor_url(anchor, url_index, url_lines, log_url_index)
                        active["source_url"] = source_url
                        active["source_room_id"] = source_room_id
                elif active and str(active.get("anchor_name") or "") != anchor:
                    sessions.append(_finalize_import_payload(active))
                    closed += 1
                    started_at = (event_ts or time.time()) - duration if duration else (event_ts or time.time())
                    source_url, source_room_id = _resolve_anchor_url(anchor, url_index, url_lines, log_url_index)
                    active = _new_import_payload(
                        process=proc,
                        anchor_name=anchor,
                        started_at=started_at,
                        source_url=source_url,
                        source_room_id=source_room_id,
                        duration_seconds=duration,
                        youtube_key_suffix=youtube_key_suffix,
                    )
                else:
                    started_at = (event_ts or time.time()) - duration if duration else (event_ts or time.time())
                    source_url, source_room_id = _resolve_anchor_url(anchor, url_index, url_lines, log_url_index)
                    active = _new_import_payload(
                        process=proc,
                        anchor_name=anchor,
                        started_at=started_at,
                        source_url=source_url,
                        source_room_id=source_room_id,
                        duration_seconds=duration,
                        youtube_key_suffix=youtube_key_suffix,
                    )
                continue

            if "推流已彻底停止" in line and active:
                if event_ts and int(active.get("duration_seconds") or 0) > 0:
                    active["ended_at"] = float(active["started_at"]) + int(active["duration_seconds"])
                elif event_ts:
                    active["ended_at"] = event_ts
                sessions.append(_finalize_import_payload(active))
                active = None
                closed += 1

        if active:
            sessions.append(_finalize_import_payload(active))

        for payload in sessions:
            if not str(payload.get("anchor_name") or "").strip() and not str(payload.get("stream_title") or "").strip():
                continue
            if not payload.get("source_url"):
                source_url, source_room_id = _resolve_anchor_url(
                    str(payload.get("anchor_name") or ""),
                    url_index,
                    url_lines,
                    log_url_index,
                )
                payload["source_url"] = source_url
                payload["source_room_id"] = source_room_id or payload.get("source_room_id") or ""
            _persist_import_row(
                conn,
                payload,
                started_at=float(payload["started_at"]),
                ended_at=payload.get("ended_at"),
                duration_seconds=int(payload.get("duration_seconds") or 0),
            )
            imported += 1
        if youtube_key_suffix:
            conn.execute(
                f"""
                UPDATE {_TABLE}
                SET youtube_key_suffix = ?, updated_at = ?
                WHERE process = ? AND COALESCE(youtube_key_suffix, '') = ''
                """,
                (youtube_key_suffix, time.time(), proc),
            )
        conn.commit()
    finally:
        conn.close()
    return {
        "imported": imported,
        "closed": closed,
        "removed": removed,
        "process": proc,
        "log_path": str(path),
    }


def track_process_status_transition(
    process: str,
    old_status: str,
    new_status: str,
    *,
    db_path: Path,
) -> None:
    """d2ypp2 不在 PM2 状态切换时写会话；保留空实现避免脏数据。"""
    return


def _capture_cover_safe(session_id: int, stream_url: str, db_path: Path, base_dir: Path) -> None:
    try:
        capture_relay_cover(session_id, stream_url, db_path=db_path, base_dir=base_dir)
    except Exception:
        pass


def capture_relay_cover(
    session_id: int,
    stream_url: str,
    *,
    db_path: Path,
    base_dir: Path,
) -> str:
    url = (stream_url or "").strip()
    if not url or not url.startswith("http"):
        return ""
    covers = _covers_dir(base_dir)
    out_path = covers / f"{session_id}.jpg"
    headers = (
        "User-Agent: Mozilla/5.0 (iPhone; CPU iPhone OS 18_1 like Mac OS X) AppleWebKit/605.1.15\r\n"
        "Referer: https://live.douyin.com/\r\n"
    )
    cmd = [
        "ffmpeg",
        "-hide_banner",
        "-loglevel",
        "error",
        "-y",
        "-headers",
        headers,
        "-rw_timeout",
        "15000000",
        "-i",
        url,
        "-frames:v",
        "1",
        "-q:v",
        "3",
        str(out_path),
    ]
    proc = subprocess.run(cmd, capture_output=True, text=True, timeout=25)
    if proc.returncode != 0 or not out_path.is_file() or out_path.stat().st_size < 128:
        return ""
    conn = _connect(db_path)
    try:
        conn.execute(
            f"UPDATE {_TABLE} SET cover_path = ?, updated_at = ? WHERE id = ?",
            (str(out_path), time.time(), session_id),
        )
        conn.commit()
    finally:
        conn.close()
    return str(out_path)


def get_relay_cover_path(session_id: int | str, *, db_path: Path) -> Path | None:
    conn = _connect(db_path)
    try:
        row = conn.execute(
            f"SELECT cover_path FROM {_TABLE} WHERE id = ? OR session_uuid = ?",
            (session_id, str(session_id)),
        ).fetchone()
        if not row:
            return None
        path = Path(str(row["cover_path"] or ""))
        return path if path.is_file() else None
    finally:
        conn.close()