d61526698d
将 README 改写为产品文档；移除微信/EasyTier/工作台等遗留模块；新增 youLIVE 目标端监控、摄像头推流、质量守卫与转播历史能力。 Co-authored-by: Cursor <cursoragent@cursor.com>
434 lines
14 KiB
Python
434 lines
14 KiB
Python
"""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)
|