869 lines
32 KiB
Python
869 lines
32 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import sqlite3
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from web2_douyin_source_display import extract_douyin_source_meta
|
|
|
|
_DB_NAME = "hub_state.sqlite"
|
|
|
|
|
|
def live_db_path(base_dir: Path) -> Path:
|
|
path = Path(base_dir) / "runtime" / "youtube_live" / _DB_NAME
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
return path
|
|
_DOUYIN_ROOM_RE = re.compile(r"(?:https?://)?(?:www\.)?live\.douyin\.com/([A-Za-z0-9_-]+)", re.IGNORECASE)
|
|
_ROOM_HINT_RE = re.compile(r"(?:room[_\s-]?id|roomId|webcast_id)\D{0,8}([A-Za-z0-9_-]{4,})", re.IGNORECASE)
|
|
_URL_RE = re.compile(r"https?://[^\s\"'<>]+", re.IGNORECASE)
|
|
_KEY_RE = re.compile(r"^\s*key\s*=\s*(.+?)\s*$", re.IGNORECASE | re.MULTILINE)
|
|
|
|
|
|
def _db_path(db_path: Path | None = None) -> Path:
|
|
if db_path:
|
|
return db_path
|
|
root = Path(__file__).resolve().parent
|
|
return live_db_path(root)
|
|
|
|
|
|
def _chmod_runtime_files(path: Path) -> None:
|
|
try:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
os.chmod(path.parent, 0o775)
|
|
except OSError:
|
|
pass
|
|
for item in (path, path.with_name(f"{path.name}-wal"), path.with_name(f"{path.name}-shm")):
|
|
try:
|
|
if item.exists():
|
|
os.chmod(item, 0o664)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def _connect(db_path: Path | None = None) -> sqlite3.Connection:
|
|
path = _db_path(db_path)
|
|
_chmod_runtime_files(path)
|
|
conn = sqlite3.connect(str(path), timeout=12)
|
|
conn.row_factory = sqlite3.Row
|
|
conn.execute("PRAGMA busy_timeout=12000")
|
|
conn.execute("PRAGMA foreign_keys=ON")
|
|
conn.execute("PRAGMA synchronous=NORMAL")
|
|
try:
|
|
conn.execute("PRAGMA journal_mode=WAL")
|
|
except sqlite3.OperationalError:
|
|
pass
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS youtube_live_events (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
event_uuid TEXT NOT NULL UNIQUE,
|
|
process TEXT NOT NULL DEFAULT '',
|
|
mode TEXT NOT NULL DEFAULT '',
|
|
channel_id TEXT NOT NULL DEFAULT '',
|
|
channel_name TEXT NOT NULL DEFAULT '',
|
|
source_platform TEXT NOT NULL DEFAULT 'douyin',
|
|
source_url TEXT NOT NULL DEFAULT '',
|
|
source_room_id TEXT NOT NULL DEFAULT '',
|
|
source_title TEXT NOT NULL DEFAULT '',
|
|
target_platform TEXT NOT NULL DEFAULT 'youtube',
|
|
youtube_key_suffix TEXT NOT NULL DEFAULT '',
|
|
status TEXT NOT NULL DEFAULT 'active',
|
|
started_at REAL NOT NULL,
|
|
ended_at REAL,
|
|
duration_seconds REAL NOT NULL DEFAULT 0,
|
|
last_seen_at REAL NOT NULL,
|
|
created_at REAL NOT NULL,
|
|
updated_at REAL NOT NULL,
|
|
quality_score INTEGER NOT NULL DEFAULT 0,
|
|
viewer_peak INTEGER NOT NULL DEFAULT 0,
|
|
last_viewer_count INTEGER,
|
|
last_live_status TEXT NOT NULL DEFAULT '',
|
|
last_health TEXT NOT NULL DEFAULT '',
|
|
last_popup_text TEXT NOT NULL DEFAULT '',
|
|
last_page_url TEXT NOT NULL DEFAULT '',
|
|
last_page_title TEXT NOT NULL DEFAULT '',
|
|
control_actions_json TEXT NOT NULL DEFAULT '[]',
|
|
tags_json TEXT NOT NULL DEFAULT '[]',
|
|
snapshot_json TEXT NOT NULL DEFAULT '{}'
|
|
)
|
|
"""
|
|
)
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS youtube_live_samples (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
event_id INTEGER NOT NULL,
|
|
ts REAL NOT NULL,
|
|
source TEXT NOT NULL DEFAULT '',
|
|
live_status TEXT NOT NULL DEFAULT '',
|
|
viewer_count INTEGER,
|
|
like_count INTEGER,
|
|
chat_rate REAL,
|
|
health TEXT NOT NULL DEFAULT '',
|
|
popup_text TEXT NOT NULL DEFAULT '',
|
|
page_url TEXT NOT NULL DEFAULT '',
|
|
page_title TEXT NOT NULL DEFAULT '',
|
|
payload_json TEXT NOT NULL DEFAULT '{}',
|
|
FOREIGN KEY(event_id) REFERENCES youtube_live_events(id) ON DELETE CASCADE
|
|
)
|
|
"""
|
|
)
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS youtube_live_alerts (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
event_id INTEGER NOT NULL,
|
|
ts REAL NOT NULL,
|
|
severity TEXT NOT NULL DEFAULT 'info',
|
|
title TEXT NOT NULL DEFAULT '',
|
|
message TEXT NOT NULL DEFAULT '',
|
|
source TEXT NOT NULL DEFAULT '',
|
|
fingerprint TEXT NOT NULL DEFAULT '',
|
|
payload_json TEXT NOT NULL DEFAULT '{}',
|
|
FOREIGN KEY(event_id) REFERENCES youtube_live_events(id) ON DELETE CASCADE
|
|
)
|
|
"""
|
|
)
|
|
conn.execute("CREATE INDEX IF NOT EXISTS idx_youtube_live_events_status ON youtube_live_events(status, updated_at DESC)")
|
|
conn.execute("CREATE INDEX IF NOT EXISTS idx_youtube_live_events_process ON youtube_live_events(process, updated_at DESC)")
|
|
conn.execute("CREATE INDEX IF NOT EXISTS idx_youtube_live_samples_event ON youtube_live_samples(event_id, ts DESC)")
|
|
conn.execute(
|
|
"CREATE UNIQUE INDEX IF NOT EXISTS idx_youtube_live_alert_fingerprint "
|
|
"ON youtube_live_alerts(event_id, fingerprint)"
|
|
)
|
|
conn.commit()
|
|
_chmod_runtime_files(path)
|
|
return conn
|
|
|
|
|
|
def _json_dumps(value: Any) -> str:
|
|
return json.dumps(value, ensure_ascii=False, separators=(",", ":"), default=str)
|
|
|
|
|
|
def _json_loads(raw: str | None, fallback: Any) -> Any:
|
|
if not raw:
|
|
return fallback
|
|
try:
|
|
return json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
return fallback
|
|
|
|
|
|
def _row_to_event(row: sqlite3.Row) -> dict[str, Any]:
|
|
data = dict(row)
|
|
data["control_actions"] = _json_loads(data.pop("control_actions_json", "[]"), [])
|
|
data["tags"] = _json_loads(data.pop("tags_json", "[]"), [])
|
|
data["snapshot"] = _json_loads(data.pop("snapshot_json", "{}"), {})
|
|
if data.get("ended_at"):
|
|
data["duration_seconds"] = int(float(data.get("duration_seconds") or 0))
|
|
else:
|
|
data["duration_seconds"] = int(max(0, time.time() - float(data.get("started_at") or time.time())))
|
|
return data
|
|
|
|
|
|
def _row_to_sample(row: sqlite3.Row) -> dict[str, Any]:
|
|
data = dict(row)
|
|
data["payload"] = _json_loads(data.pop("payload_json", "{}"), {})
|
|
return data
|
|
|
|
|
|
def _row_to_alert(row: sqlite3.Row) -> dict[str, Any]:
|
|
data = dict(row)
|
|
data["payload"] = _json_loads(data.pop("payload_json", "{}"), {})
|
|
return data
|
|
|
|
|
|
def extract_source_room_id(*values: str | None) -> str:
|
|
for value in values:
|
|
text = str(value or "")
|
|
match = _DOUYIN_ROOM_RE.search(text)
|
|
if match:
|
|
return match.group(1)
|
|
match = _ROOM_HINT_RE.search(text)
|
|
if match:
|
|
return match.group(1)
|
|
return ""
|
|
|
|
|
|
def extract_first_url(text: str | None) -> str:
|
|
for line in str(text or "").splitlines():
|
|
stripped = line.strip()
|
|
if not stripped or stripped.startswith("#"):
|
|
continue
|
|
match = _URL_RE.search(stripped)
|
|
if match:
|
|
return match.group(0).rstrip(";,")
|
|
match = _URL_RE.search(str(text or ""))
|
|
return match.group(0).rstrip(";,") if match else ""
|
|
|
|
|
|
def extract_youtube_key_suffix(text: str | None) -> str:
|
|
match = _KEY_RE.search(str(text or ""))
|
|
if not match:
|
|
return ""
|
|
key = match.group(1).strip()
|
|
if not key:
|
|
return ""
|
|
return key[-6:] if len(key) > 6 else key
|
|
|
|
|
|
def _normalize_process(value: str | None) -> str:
|
|
cleaned = re.sub(r"[^A-Za-z0-9_.:-]+", "_", str(value or "").strip())
|
|
return cleaned[:96] or "youtube"
|
|
|
|
|
|
def _status_from_payload(payload: dict[str, Any]) -> str:
|
|
raw = " ".join(
|
|
str(payload.get(key) or "")
|
|
for key in ("live_status", "status", "business_status", "process_status", "health", "page_title")
|
|
).lower()
|
|
if any(token in raw for token in ("error", "failed", "断开", "异常", "错误")):
|
|
return "error"
|
|
if any(token in raw for token in ("live", "streaming", "excellent", "healthy", "直播中", "已开始", "正在直播")):
|
|
return "active"
|
|
if any(token in raw for token in ("waiting", "idle", "offline", "stopped", "等待", "未开始", "停止")):
|
|
return "idle"
|
|
return "active"
|
|
|
|
|
|
def _quality_score(payload: dict[str, Any]) -> int:
|
|
score = 60
|
|
raw = " ".join(str(payload.get(key) or "") for key in ("live_status", "health", "business_status", "process_status")).lower()
|
|
if any(token in raw for token in ("excellent", "healthy", "streaming", "直播中", "良好")):
|
|
score += 25
|
|
if any(token in raw for token in ("warning", "unstable", "poor", "buffer", "警告", "不佳", "卡顿")):
|
|
score -= 20
|
|
if any(token in raw for token in ("error", "failed", "disconnect", "异常", "错误", "断开")):
|
|
score -= 35
|
|
viewers = _coerce_int(payload.get("viewer_count"))
|
|
if viewers and viewers > 0:
|
|
score += min(10, viewers // 10 + 1)
|
|
return max(0, min(100, score))
|
|
|
|
|
|
def _coerce_int(value: Any) -> int | None:
|
|
if value is None or value == "":
|
|
return None
|
|
if isinstance(value, bool):
|
|
return None
|
|
if isinstance(value, (int, float)):
|
|
return max(0, int(value))
|
|
text = str(value).strip().replace(",", "")
|
|
match = re.search(r"\d+", text)
|
|
if not match:
|
|
return None
|
|
return max(0, int(match.group(0)))
|
|
|
|
|
|
def _action_history(row: sqlite3.Row | None, action: str, ts: float, note: str = "") -> str:
|
|
current = _json_loads(row["control_actions_json"] if row else "[]", [])
|
|
if not isinstance(current, list):
|
|
current = []
|
|
current.append({"action": action, "ts": ts, "note": note})
|
|
return _json_dumps(current[-80:])
|
|
|
|
|
|
def _active_event_row(
|
|
conn: sqlite3.Connection,
|
|
*,
|
|
process: str,
|
|
source_room_id: str = "",
|
|
source_url: str = "",
|
|
) -> sqlite3.Row | None:
|
|
if source_room_id:
|
|
row = conn.execute(
|
|
"""
|
|
SELECT * FROM youtube_live_events
|
|
WHERE ended_at IS NULL AND status IN ('active', 'error') AND source_room_id = ?
|
|
ORDER BY updated_at DESC LIMIT 1
|
|
""",
|
|
(source_room_id,),
|
|
).fetchone()
|
|
if row:
|
|
return row
|
|
if source_url:
|
|
row = conn.execute(
|
|
"""
|
|
SELECT * FROM youtube_live_events
|
|
WHERE ended_at IS NULL AND status IN ('active', 'error') AND source_url = ?
|
|
ORDER BY updated_at DESC LIMIT 1
|
|
""",
|
|
(source_url,),
|
|
).fetchone()
|
|
if row:
|
|
return row
|
|
return conn.execute(
|
|
"""
|
|
SELECT * FROM youtube_live_events
|
|
WHERE ended_at IS NULL AND status IN ('active', 'error') AND process = ?
|
|
ORDER BY updated_at DESC LIMIT 1
|
|
""",
|
|
(process,),
|
|
).fetchone()
|
|
|
|
|
|
def _upsert_event(
|
|
conn: sqlite3.Connection,
|
|
*,
|
|
process: str,
|
|
mode: str = "",
|
|
channel_id: str = "",
|
|
channel_name: str = "",
|
|
source_url: str = "",
|
|
source_room_id: str = "",
|
|
source_title: str = "",
|
|
youtube_key_suffix: str = "",
|
|
payload: dict[str, Any] | None = None,
|
|
action: str = "observe",
|
|
note: str = "",
|
|
) -> sqlite3.Row:
|
|
now = time.time()
|
|
process = _normalize_process(process)
|
|
payload = payload or {}
|
|
source_room_id = source_room_id or extract_source_room_id(source_url, source_title, payload.get("page_title"), payload.get("text"))
|
|
source_url = source_url or extract_first_url(str(payload.get("url") or ""))
|
|
live_status = str(payload.get("live_status") or payload.get("business_status") or payload.get("status") or "")
|
|
health = str(payload.get("health") or "")
|
|
page_url = str(payload.get("url") or payload.get("page_url") or "")
|
|
page_title = str(payload.get("page_title") or payload.get("title") or "")
|
|
popup_text = _primary_popup(payload)
|
|
viewer_count = _coerce_int(payload.get("viewer_count"))
|
|
score = _quality_score(payload)
|
|
status = _status_from_payload(payload)
|
|
row = _active_event_row(conn, process=process, source_room_id=source_room_id, source_url=source_url)
|
|
snapshot = {
|
|
"last_payload": payload,
|
|
"source_room_id": source_room_id,
|
|
"source_url": source_url,
|
|
"source_title": source_title,
|
|
"updated_at": now,
|
|
}
|
|
if row:
|
|
conn.execute(
|
|
"""
|
|
UPDATE youtube_live_events
|
|
SET mode = COALESCE(NULLIF(?, ''), mode),
|
|
channel_id = COALESCE(NULLIF(?, ''), channel_id),
|
|
channel_name = COALESCE(NULLIF(?, ''), channel_name),
|
|
source_url = COALESCE(NULLIF(?, ''), source_url),
|
|
source_room_id = COALESCE(NULLIF(?, ''), source_room_id),
|
|
source_title = COALESCE(NULLIF(?, ''), source_title),
|
|
youtube_key_suffix = COALESCE(NULLIF(?, ''), youtube_key_suffix),
|
|
status = CASE WHEN ? = 'idle' THEN status ELSE ? END,
|
|
last_seen_at = ?,
|
|
updated_at = ?,
|
|
quality_score = ?,
|
|
viewer_peak = CASE
|
|
WHEN ? IS NOT NULL AND ? > viewer_peak THEN ?
|
|
ELSE viewer_peak
|
|
END,
|
|
last_viewer_count = COALESCE(?, last_viewer_count),
|
|
last_live_status = COALESCE(NULLIF(?, ''), last_live_status),
|
|
last_health = COALESCE(NULLIF(?, ''), last_health),
|
|
last_popup_text = COALESCE(NULLIF(?, ''), last_popup_text),
|
|
last_page_url = COALESCE(NULLIF(?, ''), last_page_url),
|
|
last_page_title = COALESCE(NULLIF(?, ''), last_page_title),
|
|
control_actions_json = ?,
|
|
snapshot_json = ?
|
|
WHERE id = ?
|
|
""",
|
|
(
|
|
mode,
|
|
channel_id,
|
|
channel_name,
|
|
source_url,
|
|
source_room_id,
|
|
source_title,
|
|
youtube_key_suffix,
|
|
status,
|
|
status,
|
|
now,
|
|
now,
|
|
score,
|
|
viewer_count,
|
|
viewer_count,
|
|
viewer_count,
|
|
viewer_count,
|
|
live_status,
|
|
health,
|
|
popup_text,
|
|
page_url,
|
|
page_title,
|
|
_action_history(row, action, now, note),
|
|
_json_dumps(snapshot),
|
|
row["id"],
|
|
),
|
|
)
|
|
event_id = int(row["id"])
|
|
else:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO youtube_live_events(
|
|
event_uuid, process, mode, channel_id, channel_name, source_url, source_room_id, source_title,
|
|
youtube_key_suffix, status, started_at, last_seen_at, created_at, updated_at, quality_score,
|
|
viewer_peak, last_viewer_count, last_live_status, last_health, last_popup_text,
|
|
last_page_url, last_page_title, control_actions_json, snapshot_json
|
|
)
|
|
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
|
|
""",
|
|
(
|
|
str(uuid.uuid4()),
|
|
process,
|
|
mode,
|
|
channel_id,
|
|
channel_name,
|
|
source_url,
|
|
source_room_id,
|
|
source_title,
|
|
youtube_key_suffix,
|
|
"active" if status == "idle" else status,
|
|
now,
|
|
now,
|
|
now,
|
|
now,
|
|
score,
|
|
viewer_count or 0,
|
|
viewer_count,
|
|
live_status,
|
|
health,
|
|
popup_text,
|
|
page_url,
|
|
page_title,
|
|
_action_history(None, action, now, note),
|
|
_json_dumps(snapshot),
|
|
),
|
|
)
|
|
event_id = int(conn.execute("SELECT last_insert_rowid()").fetchone()[0])
|
|
row = conn.execute("SELECT * FROM youtube_live_events WHERE id = ?", (event_id,)).fetchone()
|
|
assert row is not None
|
|
return row
|
|
|
|
|
|
def _primary_popup(payload: dict[str, Any]) -> str:
|
|
raw = payload.get("popups") or payload.get("dialogs") or payload.get("alerts") or []
|
|
if isinstance(raw, str):
|
|
return raw.strip()[:500]
|
|
if isinstance(raw, list):
|
|
texts = [str(item).strip() for item in raw if str(item).strip()]
|
|
return " | ".join(texts)[:500]
|
|
return ""
|
|
|
|
|
|
def _payload_indicates_process_inactive(payload: dict[str, Any]) -> bool:
|
|
business_status = str(payload.get("business_status") or "").strip().lower()
|
|
if bool(payload.get("is_pushing")) or business_status in {"streaming", "relay_slow", "source_stalled"}:
|
|
return False
|
|
|
|
process_status = str(payload.get("process_status") or "").strip().lower()
|
|
raw = " ".join(
|
|
str(payload.get(key) or "")
|
|
for key in ("raw_status", "business_note", "recent_log", "recent_error", "live_status", "status")
|
|
).lower()
|
|
if any(token in raw for token in ("starting", "launching", "restarting", "\u542f\u52a8\u4e2d", "\u91cd\u542f\u4e2d")):
|
|
return False
|
|
|
|
stopped_process = {
|
|
"configured",
|
|
"disabled",
|
|
"idle",
|
|
"missing",
|
|
"not_running",
|
|
"ready",
|
|
"stopped",
|
|
"unknown",
|
|
}
|
|
active_business = {
|
|
"relay_starting",
|
|
"source_live",
|
|
"monitoring",
|
|
}
|
|
if business_status in active_business and process_status in {"online", "running", "launching"}:
|
|
return False
|
|
|
|
stopped_business = {
|
|
"configured",
|
|
"idle",
|
|
"inactive",
|
|
"misconfigured",
|
|
"not_running",
|
|
"ready",
|
|
"stale",
|
|
"stopped",
|
|
"unknown",
|
|
"waiting_source",
|
|
"youtube_key_rejected",
|
|
}
|
|
if process_status in stopped_process:
|
|
return True
|
|
if business_status in stopped_business and payload.get("is_pushing") is False:
|
|
return True
|
|
if payload.get("is_pushing") is False and any(
|
|
token in raw
|
|
for token in (
|
|
"not running",
|
|
"not started",
|
|
"configured but",
|
|
"process is configured",
|
|
"\u672a\u542f\u52a8",
|
|
"\u672a\u8fd0\u884c",
|
|
"\u5f53\u524d\u672a\u542f\u52a8",
|
|
)
|
|
):
|
|
return True
|
|
return False
|
|
|
|
|
|
def _insert_sample(conn: sqlite3.Connection, event_id: int, payload: dict[str, Any], source: str) -> None:
|
|
popup_text = _primary_popup(payload)
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO youtube_live_samples(
|
|
event_id, ts, source, live_status, viewer_count, like_count, chat_rate, health,
|
|
popup_text, page_url, page_title, payload_json
|
|
)
|
|
VALUES(?,?,?,?,?,?,?,?,?,?,?,?)
|
|
""",
|
|
(
|
|
event_id,
|
|
float(payload.get("captured_at") or time.time()),
|
|
source,
|
|
str(payload.get("live_status") or payload.get("business_status") or payload.get("status") or ""),
|
|
_coerce_int(payload.get("viewer_count")),
|
|
_coerce_int(payload.get("like_count")),
|
|
float(payload.get("chat_rate") or 0) if payload.get("chat_rate") not in (None, "") else None,
|
|
str(payload.get("health") or ""),
|
|
popup_text,
|
|
str(payload.get("url") or payload.get("page_url") or ""),
|
|
str(payload.get("page_title") or payload.get("title") or ""),
|
|
_json_dumps(payload),
|
|
),
|
|
)
|
|
|
|
|
|
def _insert_alerts(conn: sqlite3.Connection, event_id: int, payload: dict[str, Any], source: str) -> None:
|
|
popup_text = _primary_popup(payload)
|
|
messages: list[tuple[str, str, str]] = []
|
|
if popup_text:
|
|
messages.append(("warning", "Browser popup", popup_text))
|
|
health = str(payload.get("health") or "")
|
|
if health and any(token in health.lower() for token in ("error", "warning", "poor", "unstable", "异常", "错误", "警告", "不佳")):
|
|
messages.append(("warning", "Stream health", health))
|
|
live_status = str(payload.get("live_status") or "")
|
|
if live_status and any(token in live_status.lower() for token in ("error", "failed", "异常", "错误")):
|
|
messages.append(("critical", "Live status", live_status))
|
|
for severity, title, message in messages:
|
|
fingerprint = hashlib.sha1(f"{event_id}|{title}|{message[:240]}".encode("utf-8", "ignore")).hexdigest()
|
|
conn.execute(
|
|
"""
|
|
INSERT OR IGNORE INTO youtube_live_alerts(event_id, ts, severity, title, message, source, fingerprint, payload_json)
|
|
VALUES(?,?,?,?,?,?,?,?)
|
|
""",
|
|
(event_id, time.time(), severity, title, message[:1000], source, fingerprint, _json_dumps(payload)),
|
|
)
|
|
|
|
|
|
def record_control_action(
|
|
process: str,
|
|
action: str,
|
|
*,
|
|
script: str = "",
|
|
url_lines: str = "",
|
|
youtube_ini_text: str = "",
|
|
channel_id: str = "",
|
|
channel_name: str = "",
|
|
mode: str = "",
|
|
note: str = "",
|
|
db_path: Path | None = None,
|
|
) -> dict[str, Any]:
|
|
source_url = extract_first_url(url_lines)
|
|
source_room_id = extract_source_room_id(url_lines)
|
|
payload = {
|
|
"source": "control",
|
|
"status": "active",
|
|
"process_status": "running",
|
|
"script": script,
|
|
"action": action,
|
|
}
|
|
conn = _connect(db_path)
|
|
try:
|
|
if action in {"stop", "finish", "end"}:
|
|
return finish_active_events(process, reason=note or action, db_path=db_path)
|
|
if action == "restart":
|
|
finish_active_events(process, reason="restart", db_path=db_path)
|
|
row = _upsert_event(
|
|
conn,
|
|
process=process,
|
|
mode=mode,
|
|
channel_id=channel_id,
|
|
channel_name=channel_name,
|
|
source_url=source_url,
|
|
source_room_id=source_room_id,
|
|
youtube_key_suffix=extract_youtube_key_suffix(youtube_ini_text),
|
|
payload=payload,
|
|
action=action,
|
|
note=note,
|
|
)
|
|
_insert_sample(conn, int(row["id"]), payload, "control")
|
|
conn.commit()
|
|
return {"event": _row_to_event(row)}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def record_process_observation(
|
|
process: str,
|
|
payload: dict[str, Any],
|
|
*,
|
|
url_lines: str = "",
|
|
db_path: Path | None = None,
|
|
) -> dict[str, Any]:
|
|
if _payload_indicates_process_inactive(payload):
|
|
return finish_active_events(process, reason=str(payload.get("business_note") or payload.get("raw_status") or "process inactive"), db_path=db_path)
|
|
|
|
source_meta = extract_douyin_source_meta(
|
|
url_lines,
|
|
"\n".join(str(payload.get(key) or "") for key in ("recent_log", "recent_error")),
|
|
)
|
|
source_url = str(source_meta.get("source_url") or extract_first_url(url_lines))
|
|
source_room_id = str(source_meta.get("source_room_id") or extract_source_room_id(url_lines, payload.get("recent_log"), payload.get("recent_error")))
|
|
conn = _connect(db_path)
|
|
try:
|
|
row = _upsert_event(
|
|
conn,
|
|
process=process,
|
|
source_url=source_url,
|
|
source_room_id=source_room_id,
|
|
payload=payload,
|
|
action="process_observe",
|
|
)
|
|
event_id = int(row["id"])
|
|
last = conn.execute("SELECT ts FROM youtube_live_samples WHERE event_id = ? ORDER BY ts DESC LIMIT 1", (event_id,)).fetchone()
|
|
if not last or time.time() - float(last["ts"]) >= 5:
|
|
_insert_sample(conn, event_id, payload, "process")
|
|
conn.commit()
|
|
return {"event": _row_to_event(conn.execute("SELECT * FROM youtube_live_events WHERE id = ?", (event_id,)).fetchone())}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def ingest_browser_telemetry(payload: dict[str, Any], *, db_path: Path | None = None) -> dict[str, Any]:
|
|
process = _normalize_process(str(payload.get("process") or payload.get("lane") or "chrome_extension"))
|
|
source_url = str(payload.get("source_url") or "") or extract_first_url(
|
|
" ".join(str(payload.get(key) or "") for key in ("url", "page_title", "text", "source_text"))
|
|
)
|
|
source_room_id = str(payload.get("source_room_id") or "") or extract_source_room_id(
|
|
source_url,
|
|
payload.get("url"),
|
|
payload.get("page_title"),
|
|
payload.get("text"),
|
|
payload.get("source_text"),
|
|
)
|
|
conn = _connect(db_path)
|
|
try:
|
|
row = _upsert_event(
|
|
conn,
|
|
process=process,
|
|
mode=str(payload.get("mode") or "browser"),
|
|
channel_id=str(payload.get("channel_id") or ""),
|
|
channel_name=str(payload.get("channel_name") or ""),
|
|
source_url=source_url,
|
|
source_room_id=source_room_id,
|
|
source_title=str(payload.get("source_title") or payload.get("stream_title") or ""),
|
|
payload=payload,
|
|
action="browser_telemetry",
|
|
note=str(payload.get("url") or ""),
|
|
)
|
|
event_id = int(row["id"])
|
|
_insert_sample(conn, event_id, payload, str(payload.get("source") or "chrome_extension"))
|
|
_insert_alerts(conn, event_id, payload, str(payload.get("source") or "chrome_extension"))
|
|
prune_old_samples(conn)
|
|
conn.commit()
|
|
return {
|
|
"message": "telemetry stored",
|
|
"event": _row_to_event(conn.execute("SELECT * FROM youtube_live_events WHERE id = ?", (event_id,)).fetchone()),
|
|
}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def finish_active_events(process: str | None = None, *, reason: str = "manual", db_path: Path | None = None) -> dict[str, Any]:
|
|
now = time.time()
|
|
conn = _connect(db_path)
|
|
try:
|
|
if process:
|
|
rows = conn.execute(
|
|
"SELECT * FROM youtube_live_events WHERE ended_at IS NULL AND process = ?",
|
|
(_normalize_process(process),),
|
|
).fetchall()
|
|
else:
|
|
rows = conn.execute("SELECT * FROM youtube_live_events WHERE ended_at IS NULL").fetchall()
|
|
closed: list[dict[str, Any]] = []
|
|
for row in rows:
|
|
duration = max(0, now - float(row["started_at"]))
|
|
conn.execute(
|
|
"""
|
|
UPDATE youtube_live_events
|
|
SET ended_at = ?, duration_seconds = ?, status = 'ended', updated_at = ?,
|
|
control_actions_json = ?
|
|
WHERE id = ?
|
|
""",
|
|
(now, int(duration), now, _action_history(row, "finish", now, reason), int(row["id"])),
|
|
)
|
|
closed.append(_row_to_event(conn.execute("SELECT * FROM youtube_live_events WHERE id = ?", (int(row["id"]),)).fetchone()))
|
|
conn.commit()
|
|
return {"message": "events closed", "count": len(closed), "events": closed}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def list_live_events(
|
|
*,
|
|
status: str = "",
|
|
process: str = "",
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
db_path: Path | None = None,
|
|
) -> dict[str, Any]:
|
|
clauses: list[str] = []
|
|
params: list[Any] = []
|
|
if status:
|
|
if status == "active":
|
|
clauses.append("ended_at IS NULL")
|
|
else:
|
|
clauses.append("status = ?")
|
|
params.append(status)
|
|
if process:
|
|
clauses.append("process = ?")
|
|
params.append(_normalize_process(process))
|
|
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 youtube_live_events {where} ORDER BY updated_at DESC LIMIT ? OFFSET ?",
|
|
(*params, safe_limit, safe_offset),
|
|
).fetchall()
|
|
total = conn.execute(f"SELECT COUNT(*) FROM youtube_live_events {where}", params).fetchone()[0]
|
|
return {"events": [_row_to_event(row) for row in rows], "total": int(total), "limit": safe_limit, "offset": safe_offset}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_live_event(event_id: int | str, *, db_path: Path | None = None) -> dict[str, Any]:
|
|
conn = _connect(db_path)
|
|
try:
|
|
row = conn.execute("SELECT * FROM youtube_live_events WHERE id = ? OR event_uuid = ?", (event_id, str(event_id))).fetchone()
|
|
if not row:
|
|
return {}
|
|
samples = conn.execute(
|
|
"SELECT * FROM youtube_live_samples WHERE event_id = ? ORDER BY ts DESC LIMIT 240",
|
|
(int(row["id"]),),
|
|
).fetchall()
|
|
alerts = conn.execute(
|
|
"SELECT * FROM youtube_live_alerts WHERE event_id = ? ORDER BY ts DESC LIMIT 120",
|
|
(int(row["id"]),),
|
|
).fetchall()
|
|
event = _row_to_event(row)
|
|
event["samples"] = [_row_to_sample(item) for item in samples]
|
|
event["alerts"] = [_row_to_alert(item) for item in alerts]
|
|
event["sample_count"] = int(conn.execute("SELECT COUNT(*) FROM youtube_live_samples WHERE event_id = ?", (int(row["id"]),)).fetchone()[0])
|
|
return event
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def youtube_live_summary(*, db_path: Path | None = None) -> dict[str, Any]:
|
|
conn = _connect(db_path)
|
|
try:
|
|
active_rows = conn.execute(
|
|
"SELECT * FROM youtube_live_events WHERE ended_at IS NULL ORDER BY updated_at DESC LIMIT 12"
|
|
).fetchall()
|
|
latest_rows = conn.execute("SELECT * FROM youtube_live_events ORDER BY updated_at DESC LIMIT 16").fetchall()
|
|
alert_rows = conn.execute("SELECT * FROM youtube_live_alerts ORDER BY ts DESC LIMIT 20").fetchall()
|
|
total = int(conn.execute("SELECT COUNT(*) FROM youtube_live_events").fetchone()[0])
|
|
ended = int(conn.execute("SELECT COUNT(*) FROM youtube_live_events WHERE ended_at IS NOT NULL").fetchone()[0])
|
|
sample_count = int(conn.execute("SELECT COUNT(*) FROM youtube_live_samples").fetchone()[0])
|
|
total_duration = int(conn.execute("SELECT COALESCE(SUM(duration_seconds),0) FROM youtube_live_events").fetchone()[0] or 0)
|
|
active = [_row_to_event(row) for row in active_rows]
|
|
latest = [_row_to_event(row) for row in latest_rows]
|
|
return {
|
|
"active": active,
|
|
"latest": latest,
|
|
"alerts": [_row_to_alert(row) for row in alert_rows],
|
|
"stats": {
|
|
"total_events": total,
|
|
"ended_events": ended,
|
|
"active_events": len(active),
|
|
"sample_count": sample_count,
|
|
"total_duration_seconds": total_duration,
|
|
"viewer_peak": max([int(item.get("viewer_peak") or 0) for item in latest] or [0]),
|
|
},
|
|
"capabilities": [
|
|
"event_db",
|
|
"event_detail",
|
|
"source_room_trace",
|
|
"duration_trace",
|
|
"browser_extension_telemetry",
|
|
"viewer_count_capture",
|
|
"live_status_capture",
|
|
"popup_alert_capture",
|
|
"health_score",
|
|
"process_control_audit",
|
|
"sample_timeline",
|
|
"multi_lane_history",
|
|
"tampermonkey_compat",
|
|
],
|
|
}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def close_live_event(event_id: int | str, *, reason: str = "manual", db_path: Path | None = None) -> dict[str, Any]:
|
|
event = get_live_event(event_id, db_path=db_path)
|
|
if not event:
|
|
return {}
|
|
if event.get("ended_at"):
|
|
return {"event": event, "message": "already closed"}
|
|
now = time.time()
|
|
duration = max(0, now - float(event["started_at"]))
|
|
conn = _connect(db_path)
|
|
try:
|
|
row = conn.execute("SELECT * FROM youtube_live_events WHERE id = ?", (int(event["id"]),)).fetchone()
|
|
conn.execute(
|
|
"""
|
|
UPDATE youtube_live_events
|
|
SET ended_at = ?, duration_seconds = ?, status = 'ended', updated_at = ?,
|
|
control_actions_json = ?
|
|
WHERE id = ?
|
|
""",
|
|
(now, int(duration), now, _action_history(row, "close", now, reason), int(event["id"])),
|
|
)
|
|
conn.commit()
|
|
return {"message": "event closed", "event": get_live_event(event_id, db_path=db_path)}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def prune_old_samples(conn: sqlite3.Connection, *, keep_days: int = 14, keep_per_event: int = 2000) -> None:
|
|
cutoff = time.time() - max(1, keep_days) * 86400
|
|
conn.execute("DELETE FROM youtube_live_samples WHERE ts < ?", (cutoff,))
|
|
event_ids = [int(row[0]) for row in conn.execute("SELECT id FROM youtube_live_events ORDER BY updated_at DESC LIMIT 200").fetchall()]
|
|
for event_id in event_ids:
|
|
conn.execute(
|
|
"""
|
|
DELETE FROM youtube_live_samples
|
|
WHERE event_id = ?
|
|
AND id NOT IN (
|
|
SELECT id FROM youtube_live_samples WHERE event_id = ? ORDER BY ts DESC LIMIT ?
|
|
)
|
|
""",
|
|
(event_id, event_id, keep_per_event),
|
|
)
|