1070 lines
36 KiB
Python
1070 lines
36 KiB
Python
"""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()
|