272 lines
10 KiB
Python
272 lines
10 KiB
Python
"""YouTube relay quality scoring (from sh2 youtube_quality_guard.py, client-simplified)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
import time
|
||
from typing import Any
|
||
|
||
from web2_douyin_source_display import extract_douyin_source_meta
|
||
|
||
_FFMPEG_KV_RE = re.compile(
|
||
r"(?P<key>frame|fps|bitrate|speed|drop|dup)\s*=\s*(?P<value>[^\s]+)", re.IGNORECASE
|
||
)
|
||
_TIME_RE = re.compile(r"time\s*=\s*(?P<value>\d{2}:\d{2}:\d{2}(?:\.\d+)?)", re.IGNORECASE)
|
||
_URL_RE = re.compile(r"https?://[^\s\"'<>]+", re.IGNORECASE)
|
||
_ROOM_RE = re.compile(r"live\.douyin\.com/([a-zA-Z0-9_-]+)", re.IGNORECASE)
|
||
|
||
|
||
def _as_float(value: Any) -> float | None:
|
||
text = str(value or "").strip().lower().replace("x", "").replace("kbits/s", "").replace("kb/s", "")
|
||
text = text.replace(",", "")
|
||
match = re.search(r"-?\d+(?:\.\d+)?", text)
|
||
if not match:
|
||
return None
|
||
try:
|
||
return float(match.group(0))
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def _as_int(value: Any) -> int | None:
|
||
parsed = _as_float(value)
|
||
return None if parsed is None else max(0, int(parsed))
|
||
|
||
|
||
_IDLE_MARKERS = (
|
||
"等待直播",
|
||
"waiting",
|
||
"获取失败",
|
||
"douyin web api failed",
|
||
"无录制任务",
|
||
"网址内容获取失败",
|
||
"没有正在监测的直播",
|
||
)
|
||
|
||
|
||
def _line_implies_relay_idle(line: str) -> bool:
|
||
lower = line.lower()
|
||
if "frame=" in lower and "speed=" in lower:
|
||
return False
|
||
return any(marker in lower for marker in _IDLE_MARKERS) or "error" in lower or "failed" in lower
|
||
|
||
|
||
def _latest_ffmpeg_stats_line(text: str, *, lookback: int = 12) -> str:
|
||
lines = [line.strip() for line in str(text or "").splitlines() if line.strip()]
|
||
if not lines:
|
||
return ""
|
||
tail = lines[-lookback:]
|
||
stats_idx = -1
|
||
stats_line = ""
|
||
for idx in range(len(tail) - 1, -1, -1):
|
||
line = tail[idx]
|
||
lower = line.lower()
|
||
if "frame=" in lower and "speed=" in lower:
|
||
stats_idx = idx
|
||
stats_line = line
|
||
break
|
||
if stats_idx < 0:
|
||
return ""
|
||
if any(_line_implies_relay_idle(line) for line in tail[stats_idx + 1 :]):
|
||
return ""
|
||
return stats_line
|
||
|
||
|
||
def parse_ffmpeg_metrics(*texts: str) -> dict[str, Any]:
|
||
raw = "\n".join(str(item or "") for item in texts)
|
||
tail = raw[-16000:]
|
||
metrics: dict[str, Any] = {}
|
||
stats_line = _latest_ffmpeg_stats_line(tail)
|
||
if stats_line:
|
||
for match in _FFMPEG_KV_RE.finditer(stats_line):
|
||
key = match.group("key").lower()
|
||
value = match.group("value")
|
||
if key in {"frame", "drop", "dup"}:
|
||
metrics[key] = _as_int(value)
|
||
elif key in {"fps", "speed", "bitrate"}:
|
||
metrics[key] = _as_float(value)
|
||
time_match = _TIME_RE.search(stats_line)
|
||
if time_match:
|
||
metrics["encoded_time"] = time_match.group("value")
|
||
lower = tail.lower()
|
||
metrics["error_count"] = sum(
|
||
lower.count(token)
|
||
for token in (
|
||
"error",
|
||
"failed",
|
||
"connection refused",
|
||
"timed out",
|
||
"broken pipe",
|
||
"invalid data",
|
||
)
|
||
)
|
||
metrics["reconnect_count"] = sum(lower.count(token) for token in ("reconnect", "retry", "connection reset"))
|
||
return metrics
|
||
|
||
|
||
def _count_urls(text: str) -> int:
|
||
return len(_URL_RE.findall(str(text or "")))
|
||
|
||
|
||
def _ini_has_key(text: str) -> bool:
|
||
return bool(re.search(r"^\s*key\s*=\s*\S+", str(text or ""), re.IGNORECASE | re.MULTILINE))
|
||
|
||
|
||
def _ini_has_rtmps(text: str) -> bool:
|
||
lower = str(text or "").lower()
|
||
return any(token in lower for token in ("rtmps = 是", "rtmps=true", "rtmps = true", "rtmps=1"))
|
||
|
||
|
||
def _source_hint(url_lines: str, recent_log: str) -> dict[str, str]:
|
||
combined = f"{url_lines}\n{recent_log}"
|
||
urls = _URL_RE.findall(combined)
|
||
source_url = urls[0] if urls else ""
|
||
room = _room_from_text(source_url) if source_url else ""
|
||
return {"source_url": source_url, "source_room_id": room}
|
||
|
||
|
||
def _room_from_text(text: str) -> str:
|
||
m = _ROOM_RE.search(text or "")
|
||
return m.group(1) if m else ""
|
||
|
||
|
||
def _lane_score(row: dict[str, Any], metrics: dict[str, Any], cfg: dict[str, str]) -> tuple[int, list[dict[str, str]]]:
|
||
score = 100
|
||
issues: list[dict[str, str]] = []
|
||
|
||
if not row.get("ffmpeg_active") and not row.get("is_pushing"):
|
||
score -= 35
|
||
issues.append({"level": "critical", "text": "FFmpeg 未检测到推流进程"})
|
||
if not row.get("is_pushing"):
|
||
score -= 25
|
||
issues.append({"level": "critical", "text": "当前没有稳定推流输出"})
|
||
|
||
speed = metrics.get("speed") or row.get("ffmpeg_speed")
|
||
if isinstance(speed, (int, float)) and speed < 0.98:
|
||
score -= 18
|
||
issues.append({"level": "warning", "text": f"编码速度 {speed:.2f}x，低于实时要求"})
|
||
if row.get("ffmpeg_speed_low"):
|
||
score -= 10
|
||
issues.append({"level": "warning", "text": "近期推流速度持续偏低"})
|
||
if row.get("ffmpeg_progress_stalled"):
|
||
score -= 15
|
||
issues.append({"level": "warning", "text": "FFmpeg 进度停滞"})
|
||
|
||
biz = str(row.get("business_status") or "")
|
||
if biz == "youtube_key_rejected":
|
||
score -= 40
|
||
issues.append({"level": "critical", "text": "YouTube 串流密钥被拒绝"})
|
||
elif biz == "misconfigured":
|
||
score -= 30
|
||
issues.append({"level": "critical", "text": "串流密钥或配置不完整"})
|
||
|
||
if int(metrics.get("error_count") or 0) > 0:
|
||
score -= min(30, int(metrics.get("error_count") or 0) * 6)
|
||
issues.append({"level": "warning", "text": "日志出现 FFmpeg/网络错误"})
|
||
|
||
url_count = _count_urls(cfg.get("url_lines", ""))
|
||
has_key = _ini_has_key(cfg.get("youtube_ini_text", ""))
|
||
if not has_key:
|
||
score = max(0, score - 30)
|
||
issues.append({"level": "critical", "text": "YouTube 推流 key 未配置"})
|
||
if url_count == 0:
|
||
score = max(0, score - 15)
|
||
issues.append({"level": "warning", "text": "没有配置转播源 URL"})
|
||
if has_key and not _ini_has_rtmps(cfg.get("youtube_ini_text", "")):
|
||
score = max(0, score - 8)
|
||
issues.append({"level": "warning", "text": "建议使用 RTMPS 推流"})
|
||
|
||
watch = str(cfg.get("watch_url") or "").strip()
|
||
if not watch:
|
||
score = max(0, score - 6)
|
||
issues.append({"level": "info", "text": "未配置 YouTube 目标监听地址（弹幕/观众数）"})
|
||
|
||
if not issues and row.get("is_pushing"):
|
||
issues.append({"level": "ok", "text": "推流链路正常"})
|
||
return max(0, min(100, score)), issues[:6]
|
||
|
||
|
||
def build_youtube_quality_guard_report(
|
||
rows: list[dict[str, Any]],
|
||
*,
|
||
live_summary: dict[str, Any] | None = None,
|
||
config_by_process: dict[str, dict[str, str]] | None = None,
|
||
) -> dict[str, Any]:
|
||
live_summary = live_summary or {}
|
||
config_by_process = config_by_process or {}
|
||
lanes: list[dict[str, Any]] = []
|
||
recommendations: list[dict[str, str]] = []
|
||
|
||
for row in rows:
|
||
process = str(row.get("process") or row.get("pm2") or "")
|
||
pushing = bool(row.get("is_pushing"))
|
||
metrics = (
|
||
parse_ffmpeg_metrics(str(row.get("recent_log") or ""), str(row.get("recent_error") or ""))
|
||
if pushing
|
||
else {}
|
||
)
|
||
cfg = config_by_process.get(process) or {}
|
||
source_meta = extract_douyin_source_meta(cfg.get("url_lines", ""), str(row.get("recent_log") or ""))
|
||
source_title = str(source_meta.get("source_title") or row.get("source_title") or "").strip()
|
||
source_room_id = str(source_meta.get("source_room_id") or row.get("source_room_id") or "").strip()
|
||
source_url = str(source_meta.get("source_url") or row.get("source_url") or "").strip()
|
||
source_display = str(
|
||
source_meta.get("source_display")
|
||
or row.get("source_display")
|
||
or source_title
|
||
or (f"抖音直播间 {source_room_id}" if source_room_id else "")
|
||
).strip()
|
||
score, issues = _lane_score(row, metrics, cfg)
|
||
rtmps = _ini_has_rtmps(cfg.get("youtube_ini_text", ""))
|
||
lanes.append(
|
||
{
|
||
"process": process,
|
||
"label": row.get("label") or process,
|
||
"source_title": source_title,
|
||
"source_room_id": source_room_id,
|
||
"source_url": source_url,
|
||
"source_display": source_display,
|
||
"status": row.get("business_status") or row.get("process_status") or "",
|
||
"is_pushing": pushing,
|
||
"ffmpeg_active": bool(row.get("ffmpeg_active") or row.get("ffmpeg_pids")),
|
||
"score": score,
|
||
"metrics": metrics,
|
||
"rtmps": rtmps,
|
||
"has_stream_key": _ini_has_key(cfg.get("youtube_ini_text", "")),
|
||
"issues": issues,
|
||
}
|
||
)
|
||
|
||
pushing_count = sum(1 for lane in lanes if lane["is_pushing"])
|
||
active_events = int(
|
||
((live_summary.get("stats") or {}).get("active_events") or len(live_summary.get("active") or [])) or 0
|
||
)
|
||
sample_count = int(((live_summary.get("stats") or {}).get("sample_count") or 0) or 0)
|
||
viewer_peak = int(((live_summary.get("stats") or {}).get("viewer_peak") or 0) or 0)
|
||
score = int(sum(lane["score"] for lane in lanes) / len(lanes)) if lanes else 0
|
||
|
||
if pushing_count == 0:
|
||
recommendations.append({"level": "critical", "text": "没有正在推流的线路"})
|
||
if active_events == 0 and pushing_count > 0:
|
||
recommendations.append({"level": "warning", "text": "推流已存在但事件中心未形成活动事件"})
|
||
if sample_count == 0 and pushing_count > 0:
|
||
recommendations.append({"level": "info", "text": "尚无 Studio/浏览器采样，可安装 ingest 扩展增强观测"})
|
||
elif all(lane["score"] >= 80 for lane in lanes if lane["is_pushing"]):
|
||
recommendations.append({"level": "ok", "text": "推流质量良好，保持监控即可"})
|
||
if any(lane["status"] == "relay_slow" for lane in lanes):
|
||
recommendations.append({"level": "warning", "text": "存在推流速度偏低的线路，检查 CPU/GPU 与网络"})
|
||
|
||
return {
|
||
"generated_at": time.time(),
|
||
"score": max(0, min(100, score)),
|
||
"summary": {
|
||
"lanes": len(lanes),
|
||
"pushing_lanes": pushing_count,
|
||
"active_events": active_events,
|
||
"sample_count": sample_count,
|
||
"viewer_peak": viewer_peak,
|
||
},
|
||
"lanes": lanes,
|
||
"recommendations": recommendations[:8],
|
||
}
|