1771 lines
65 KiB
Python
1771 lines
65 KiB
Python
from fastapi import FastAPI, Query, Request
|
||
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse
|
||
from pathlib import Path
|
||
import asyncio
|
||
import html
|
||
import os
|
||
import time
|
||
from contextlib import asynccontextmanager
|
||
from typing import Optional
|
||
|
||
import psutil
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
|
||
from web2_supervisor import (
|
||
BuiltinSupervisor,
|
||
force_kill_ffmpeg_cross_platform,
|
||
get_process_info_pm2,
|
||
run_pm2_shell,
|
||
strip_ansi_codes,
|
||
use_builtin,
|
||
)
|
||
from web2_network_routes import router as network_router
|
||
from web2_proxy_config import read_proxy_config, write_proxy_config
|
||
from web2_system_metrics import snapshot as system_metrics_snapshot
|
||
from web2_youtube_oauth import (
|
||
fetch_stream_key_and_write_ini,
|
||
oauth_authorization_url,
|
||
oauth_exchange_code,
|
||
oauth_status as youtube_oauth_status_snapshot,
|
||
)
|
||
from web2_client_overview import build_client_overview
|
||
from web2_process_runtime import process_family, summarize_process_activity
|
||
from web2_douyin_source_display import extract_douyin_source_meta
|
||
from web2_douyin_preview import fetch_douyin_anchor_name
|
||
from web2_config_governance import (
|
||
list_config_backups,
|
||
restore_config_backup,
|
||
write_config_text,
|
||
)
|
||
from web2_config_insights import analyze_config_content, optimize_config_content, read_config_file
|
||
from web2_youtube_ingest_auth import ingest_config_payload, verify_ingest_payload
|
||
from web2_youtube_pro_runtime import sync_all_youtube_pro_runtime_files, validate_youtube_runtime_files
|
||
from web2_relay_pro import channel_id_for_pm2_name, load_channels_document, save_channels_document
|
||
from web2_quality_guard import build_youtube_quality_guard_report
|
||
from web2_youtube_diagnostics import build_youtube_diagnostics_report
|
||
from web2_relay_log import (
|
||
close_active_relay_sessions,
|
||
get_relay_cover_path,
|
||
get_relay_session,
|
||
import_relay_sessions_from_log,
|
||
list_relay_sessions,
|
||
relay_db_path,
|
||
sync_relay_from_recent_log,
|
||
track_process_status_transition,
|
||
)
|
||
from web2_relay_pro import (
|
||
add_relay_channel,
|
||
build_relay_live_status,
|
||
ensure_default_relay_config,
|
||
extract_url_lines_from_ini,
|
||
first_youtube_pm2_name,
|
||
channel_configured_for_monitor,
|
||
get_channel_by_id,
|
||
get_channel_detail,
|
||
get_single_youtube_watch_url,
|
||
get_youtube_watch_url,
|
||
list_channels_for_pro,
|
||
relay_pm2_name_for_channel,
|
||
materialize_relay_channel_youtube_ini,
|
||
read_url_config_for_channel,
|
||
relay_pm2_name_for_channel,
|
||
remove_relay_channel,
|
||
set_youtube_watch_url,
|
||
sync_channel_to_legacy_files,
|
||
update_channel_key,
|
||
update_channel_keys_list,
|
||
watch_url_for_relay_process,
|
||
write_url_config_for_channel,
|
||
)
|
||
from web2_youtube_preflight import run_youtube_preflight, youtube_start_preflight_message
|
||
from web2_youtube_relay_heartbeat import read_relay_heartbeat
|
||
from web2_youtube_live_store import (
|
||
close_live_event,
|
||
extract_youtube_key_suffix,
|
||
get_live_event,
|
||
ingest_browser_telemetry,
|
||
list_live_events,
|
||
live_db_path,
|
||
record_control_action,
|
||
record_process_observation,
|
||
youtube_live_summary,
|
||
)
|
||
from web2_youtube_ini import parse_youtube_ini, serialize_youtube_ini, validate_youtube_keys
|
||
from web2_host_caps import gpu_status_payload, machine_summary_payload
|
||
from web2_camera import camera_config_ready, list_capture_devices, load_camera_config, save_camera_config
|
||
from web2_pull_stream import (
|
||
load_pull_stream_config,
|
||
protocol_help,
|
||
pull_stream_config_ready,
|
||
save_pull_stream_config,
|
||
)
|
||
from web2_pocketbase import REQUIRE_PB, pb_auth_refresh_and_validate
|
||
from web2_feature_flags import (
|
||
SHOW_CAMERA_FEATURE,
|
||
SHOW_PULL_STREAM_FEATURE,
|
||
is_disabled_stream_worker_pm2,
|
||
)
|
||
|
||
|
||
@asynccontextmanager
|
||
async def lifespan(app: FastAPI):
|
||
ensure_default_relay_config(CONFIG_DIR)
|
||
yield
|
||
|
||
|
||
app = FastAPI(title="多进程录制控制台", lifespan=lifespan)
|
||
|
||
# ---------------- 配置（根目录结构） ----------------
|
||
BASE_DIR = Path(__file__).parent
|
||
CONFIG_DIR = BASE_DIR / "config"
|
||
DEFAULT_LOG_DIR = BASE_DIR / ".pm2" / "logs"
|
||
RELAY_DB = relay_db_path(BASE_DIR)
|
||
LIVE_DB = live_db_path(BASE_DIR)
|
||
_PROCESS_STATUS_CACHE: dict[str, str] = {}
|
||
_LIVE_OBSERVE_LAST: dict[str, float] = {}
|
||
YOUTUBE_EVENT_STATUS_OBSERVE_SECONDS = 5.0
|
||
_SYSTEM_METRICS_TTL_SECONDS = 10.0
|
||
_SYSTEM_METRICS_CACHE: tuple[float, dict] | None = None
|
||
_SYSTEM_METRICS_CACHE_LOCK = asyncio.Lock()
|
||
|
||
MAX_LOG_LINES = 300
|
||
|
||
|
||
def _relay_control_meta(process: str) -> dict[str, str]:
|
||
cid = relay_channel_id_from_pm2_name(process)
|
||
channel_name = ""
|
||
if cid:
|
||
d = get_channel_detail(CONFIG_DIR, cid)
|
||
channel_name = str((d or {}).get("name") or "")
|
||
cfg = _config_snapshot_for_process(process)
|
||
return {
|
||
"channel_id": cid or "",
|
||
"channel_name": channel_name,
|
||
"url_lines": cfg.get("url_lines") or _url_lines_for_process(process),
|
||
"youtube_ini_text": cfg.get("youtube_ini_text") or "",
|
||
"script": script_for_pm2(process) or "",
|
||
}
|
||
|
||
|
||
async def _record_youtube_control(process: str, action: str, *, note: str = "") -> None:
|
||
if not is_youtube_script(script_for_pm2(process) or ""):
|
||
return
|
||
meta = _relay_control_meta(process)
|
||
try:
|
||
await asyncio.to_thread(
|
||
record_control_action,
|
||
process,
|
||
action,
|
||
script=meta["script"],
|
||
url_lines=meta["url_lines"],
|
||
youtube_ini_text=meta["youtube_ini_text"],
|
||
channel_id=meta["channel_id"],
|
||
channel_name=meta["channel_name"],
|
||
note=note,
|
||
db_path=LIVE_DB,
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
async def _close_youtube_relay_sessions(process: str, *, reason: str) -> None:
|
||
try:
|
||
await asyncio.to_thread(close_active_relay_sessions, process, reason=reason, db_path=RELAY_DB)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _label_short(prefix: str) -> str:
|
||
return {
|
||
"tiktok": "TikTok",
|
||
"youtube": "YouTube",
|
||
"camera": "摄像头",
|
||
"pull": "拉流",
|
||
"obs": "OBS",
|
||
"web": "Web 控制台",
|
||
}.get(prefix, prefix)
|
||
|
||
|
||
def discover_script_entries() -> tuple:
|
||
"""扫描项目根目录自动识别入口：tiktok*.py、youtube*.py、obs*.sh / obs*.bat，以及当前本文件（Web 控制台）。进程名 = 主文件名（无扩展名）。"""
|
||
base = BASE_DIR
|
||
web_path = Path(__file__).resolve()
|
||
entries: list[dict] = []
|
||
|
||
for path in sorted(base.glob("tiktok*.py")):
|
||
if path.is_file() and path.resolve() != web_path:
|
||
entries.append({"script": path.name, "label": _label_short("tiktok")})
|
||
|
||
for path in sorted(base.glob("youtube*.py")):
|
||
if path.is_file() and path.resolve() != web_path:
|
||
entries.append({"script": path.name, "label": _label_short("youtube")})
|
||
|
||
if SHOW_CAMERA_FEATURE:
|
||
for path in sorted(base.glob("camera*.py")):
|
||
if path.is_file() and path.resolve() != web_path:
|
||
entries.append({"script": path.name, "label": _label_short("camera")})
|
||
|
||
if SHOW_PULL_STREAM_FEATURE:
|
||
for path in sorted(base.glob("pull_stream*.py")):
|
||
if path.is_file() and path.resolve() != web_path:
|
||
entries.append({"script": path.name, "label": _label_short("pull")})
|
||
|
||
for path in sorted(base.glob("obs*.sh")):
|
||
if path.is_file():
|
||
entries.append({"script": path.name, "label": _label_short("obs")})
|
||
|
||
for path in sorted(base.glob("obs*.bat")):
|
||
if path.is_file():
|
||
entries.append({"script": path.name, "label": _label_short("obs")})
|
||
|
||
entries.append({"script": web_path.name, "label": _label_short("web")})
|
||
|
||
return tuple(entries)
|
||
|
||
|
||
SCRIPT_ENTRIES = discover_script_entries()
|
||
|
||
|
||
def pm2_name_from_script(script: str) -> str:
|
||
return Path(script).stem
|
||
|
||
|
||
def _stem(script: str) -> str:
|
||
return Path(script).stem
|
||
|
||
|
||
def is_camera_script(script: str) -> bool:
|
||
return script.endswith(".py") and _stem(script).startswith("camera")
|
||
|
||
|
||
def is_pull_stream_script(script: str) -> bool:
|
||
return script.endswith(".py") and _stem(script).startswith("pull_stream")
|
||
|
||
|
||
def is_youtube_script(script: str) -> bool:
|
||
return script.endswith(".py") and _stem(script).startswith("youtube")
|
||
|
||
|
||
def is_tiktok_script(script: str) -> bool:
|
||
return script.endswith(".py") and _stem(script).startswith("tiktok")
|
||
|
||
|
||
def _build_process_monitor():
|
||
return tuple(
|
||
{
|
||
"pm2": pm2_name_from_script(e["script"]),
|
||
"script": e["script"],
|
||
"label": e["label"],
|
||
}
|
||
for e in SCRIPT_ENTRIES
|
||
)
|
||
|
||
|
||
PROCESS_MONITOR = _build_process_monitor()
|
||
VALID_PROCESSES = frozenset(p["pm2"] for p in PROCESS_MONITOR)
|
||
|
||
|
||
def relay_channel_id_from_pm2_name(name: str) -> Optional[str]:
|
||
cid = channel_id_for_pm2_name(CONFIG_DIR, name)
|
||
if cid:
|
||
return cid
|
||
prefix = "youtube2__"
|
||
if not name.startswith(prefix):
|
||
return None
|
||
legacy = name[len(prefix) :]
|
||
if get_channel_by_id(CONFIG_DIR, legacy):
|
||
return legacy
|
||
return None
|
||
|
||
|
||
def process_valid(name: str) -> bool:
|
||
if is_disabled_stream_worker_pm2(name):
|
||
return False
|
||
if name in VALID_PROCESSES:
|
||
return True
|
||
cid = relay_channel_id_from_pm2_name(name)
|
||
if not cid:
|
||
return False
|
||
return get_channel_by_id(CONFIG_DIR, cid) is not None
|
||
|
||
URL_CONFIG_PM2 = frozenset(
|
||
pm2_name_from_script(e["script"])
|
||
for e in SCRIPT_ENTRIES
|
||
if is_tiktok_script(e["script"]) or is_youtube_script(e["script"])
|
||
)
|
||
|
||
|
||
def script_for_pm2(pm2: str) -> Optional[str]:
|
||
if relay_channel_id_from_pm2_name(pm2):
|
||
return "youtube2.py"
|
||
for p in PROCESS_MONITOR:
|
||
if p["pm2"] == pm2:
|
||
return p["script"]
|
||
return None
|
||
|
||
|
||
WEB_SCRIPT_NAME = Path(__file__).name
|
||
_builtin_supervisor: Optional[BuiltinSupervisor] = None
|
||
|
||
|
||
def get_builtin_supervisor() -> BuiltinSupervisor:
|
||
global _builtin_supervisor
|
||
if _builtin_supervisor is None:
|
||
_builtin_supervisor = BuiltinSupervisor(BASE_DIR, WEB_SCRIPT_NAME, script_for_pm2)
|
||
return _builtin_supervisor
|
||
|
||
|
||
async def control_start(process: str) -> str:
|
||
if relay_channel_id_from_pm2_name(process) and not use_builtin():
|
||
return "多频道独立进程仅支持内置进程管理（请勿使用 WEB2_SUPERVISOR=pm2）。"
|
||
if use_builtin():
|
||
return await get_builtin_supervisor().start(process)
|
||
return await run_pm2_shell(f"start {process}")
|
||
|
||
|
||
async def control_stop(process: str) -> str:
|
||
if relay_channel_id_from_pm2_name(process) and not use_builtin():
|
||
return "多频道独立进程仅支持内置进程管理（请勿使用 WEB2_SUPERVISOR=pm2）。"
|
||
if use_builtin():
|
||
return await get_builtin_supervisor().stop(process)
|
||
return await run_pm2_shell(f"stop {process}")
|
||
|
||
|
||
async def control_restart(process: str) -> str:
|
||
if relay_channel_id_from_pm2_name(process) and not use_builtin():
|
||
return "多频道独立进程仅支持内置进程管理（请勿使用 WEB2_SUPERVISOR=pm2）。"
|
||
if use_builtin():
|
||
return await get_builtin_supervisor().restart(process)
|
||
return await run_pm2_shell(f"restart {process}")
|
||
|
||
|
||
async def control_full_status() -> str:
|
||
if use_builtin():
|
||
return await get_builtin_supervisor().full_status_text()
|
||
return await run_pm2_shell("status", timeout=8.0)
|
||
|
||
|
||
async def control_get_process_info(process: str) -> dict:
|
||
if use_builtin():
|
||
return await get_builtin_supervisor().get_process_info(process)
|
||
return await get_process_info_pm2(process, DEFAULT_LOG_DIR, strip_ansi_codes)
|
||
|
||
|
||
@app.middleware("http")
|
||
async def web2_token_middleware(request: Request, call_next):
|
||
"""若设置环境变量 WEB2_API_TOKEN，则除 /health 与 OpenAPI 外需携带 Bearer 或 X-API-Token。"""
|
||
# CORS 预检不带 Authorization；若在外层先于 CORSMiddleware 拦截会缺少 Access-Control-* 头
|
||
if request.method == "OPTIONS":
|
||
return await call_next(request)
|
||
token = os.environ.get("WEB2_API_TOKEN", "").strip()
|
||
if not token:
|
||
return await call_next(request)
|
||
path = request.url.path
|
||
if path == "/health" or path.startswith("/docs") or path in ("/openapi.json", "/redoc"):
|
||
return await call_next(request)
|
||
auth = request.headers.get("authorization") or ""
|
||
xtok = (request.headers.get("x-api-token") or "").strip()
|
||
if auth.startswith("Bearer ") and auth[7:].strip() == token:
|
||
return await call_next(request)
|
||
if xtok == token:
|
||
return await call_next(request)
|
||
return JSONResponse({"error": "unauthorized"}, status_code=401)
|
||
|
||
|
||
# 最后注册，保证 CORSMiddleware 在最外层，401 等响应仍带 CORS 头（供 Vite 等跨源调试）
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
allow_origins=["*"],
|
||
allow_methods=["GET", "POST", "OPTIONS"],
|
||
allow_headers=["*"],
|
||
)
|
||
app.include_router(network_router)
|
||
|
||
@app.get("/client/overview")
|
||
async def client_overview():
|
||
"""Electron 工作台：频道列表摘要。"""
|
||
return build_client_overview(CONFIG_DIR, BASE_DIR)
|
||
|
||
|
||
# ---------------- Pro 多路转播（每频道独立 URL 文件 + 去重 + 同步 legacy） ----------------
|
||
|
||
|
||
@app.get("/relay_pro/channels")
|
||
async def relay_pro_channels():
|
||
return list_channels_for_pro(CONFIG_DIR)
|
||
|
||
|
||
@app.post("/relay_pro/channel_add")
|
||
async def relay_pro_channel_add(request: Request):
|
||
try:
|
||
data = await request.json()
|
||
except Exception:
|
||
return JSONResponse({"error": "无效的 JSON"}, status_code=400)
|
||
if not isinstance(data, dict):
|
||
return JSONResponse({"error": "无效数据"}, status_code=400)
|
||
name = str(data.get("name") or "")
|
||
key = str(data.get("youtubeKey") or data.get("key") or "")
|
||
ok, msg, new_id = add_relay_channel(CONFIG_DIR, name, key)
|
||
if not ok:
|
||
return JSONResponse({"error": msg}, status_code=400)
|
||
return {"message": msg, "channelId": new_id}
|
||
|
||
|
||
@app.post("/relay_pro/channel_delete")
|
||
async def relay_pro_channel_delete(request: Request):
|
||
try:
|
||
data = await request.json()
|
||
except Exception:
|
||
return JSONResponse({"error": "无效的 JSON"}, status_code=400)
|
||
if not isinstance(data, dict):
|
||
return JSONResponse({"error": "无效数据"}, status_code=400)
|
||
cid = str(data.get("channelId") or data.get("channel_id") or "").strip()
|
||
if not cid:
|
||
return JSONResponse({"error": "缺少 channelId"}, status_code=400)
|
||
ok, msg = remove_relay_channel(CONFIG_DIR, cid)
|
||
if not ok:
|
||
return JSONResponse({"error": msg}, status_code=400)
|
||
return {"message": msg}
|
||
|
||
|
||
@app.get("/relay_pro/channel_detail")
|
||
async def relay_pro_channel_detail(channel_id: str = Query(..., description="频道 id，如 ch_0")):
|
||
d = get_channel_detail(CONFIG_DIR, channel_id)
|
||
if not d:
|
||
return JSONResponse({"error": "未找到频道"}, status_code=404)
|
||
return d
|
||
|
||
|
||
@app.get("/relay_pro/youtube_watch")
|
||
async def relay_pro_youtube_watch_get(channel_id: str = Query("single", description="频道 id；单路填 single")):
|
||
cid = str(channel_id or "single").strip() or "single"
|
||
return {"channelId": cid, "url": get_youtube_watch_url(CONFIG_DIR, cid)}
|
||
|
||
|
||
@app.post("/relay_pro/youtube_watch")
|
||
async def relay_pro_youtube_watch_post(request: Request):
|
||
try:
|
||
data = await request.json()
|
||
except Exception:
|
||
return JSONResponse({"error": "无效的 JSON"}, status_code=400)
|
||
if not isinstance(data, dict):
|
||
return JSONResponse({"error": "无效数据"}, status_code=400)
|
||
cid = str(data.get("channelId") or data.get("channel_id") or "single").strip() or "single"
|
||
url = str(data.get("url") or data.get("youtubeWatchUrl") or "")
|
||
ok, msg = set_youtube_watch_url(CONFIG_DIR, cid, url)
|
||
if not ok:
|
||
return JSONResponse({"error": msg}, status_code=400)
|
||
return {"message": msg, "channelId": cid, "url": get_youtube_watch_url(CONFIG_DIR, cid)}
|
||
|
||
|
||
@app.get("/relay_pro/url_config")
|
||
async def relay_pro_get_url_config(channel_id: str = Query(...)):
|
||
return {"content": read_url_config_for_channel(CONFIG_DIR, channel_id)}
|
||
|
||
|
||
@app.post("/relay_pro/url_config")
|
||
async def relay_pro_post_url_config(request: Request, channel_id: str = Query(...)):
|
||
try:
|
||
data = await request.json()
|
||
except Exception:
|
||
return JSONResponse({"error": "无效的 JSON"}, status_code=400)
|
||
content = data.get("content", "") if isinstance(data, dict) else ""
|
||
ok, msg = write_url_config_for_channel(CONFIG_DIR, channel_id, content)
|
||
if not ok:
|
||
return JSONResponse({"error": msg}, status_code=400)
|
||
return {"message": msg}
|
||
|
||
|
||
@app.post("/relay_pro/channel_key")
|
||
async def relay_pro_save_channel_key(request: Request):
|
||
try:
|
||
data = await request.json()
|
||
except Exception:
|
||
return JSONResponse({"error": "无效的 JSON"}, status_code=400)
|
||
if not isinstance(data, dict):
|
||
return JSONResponse({"error": "无效数据"}, status_code=400)
|
||
cid = str(data.get("channelId") or data.get("channel_id") or "").strip()
|
||
key = str(data.get("youtubeKey") or data.get("key") or "")
|
||
if not cid:
|
||
return JSONResponse({"error": "缺少 channelId"}, status_code=400)
|
||
ok, msg = update_channel_key(CONFIG_DIR, cid, key)
|
||
if not ok:
|
||
return JSONResponse({"error": msg}, status_code=400)
|
||
return {"message": msg}
|
||
|
||
|
||
@app.post("/relay_pro/channel_keys")
|
||
async def relay_pro_save_channel_keys(request: Request):
|
||
try:
|
||
data = await request.json()
|
||
except Exception:
|
||
return JSONResponse({"error": "无效的 JSON"}, status_code=400)
|
||
if not isinstance(data, dict):
|
||
return JSONResponse({"error": "无效数据"}, status_code=400)
|
||
cid = str(data.get("channelId") or data.get("channel_id") or "").strip()
|
||
keys = data.get("keys")
|
||
if not cid:
|
||
return JSONResponse({"error": "缺少 channelId"}, status_code=400)
|
||
if not isinstance(keys, list):
|
||
return JSONResponse({"error": "keys 须为数组"}, status_code=400)
|
||
try:
|
||
ai = int(data.get("activeIndex", 0))
|
||
except Exception:
|
||
ai = 0
|
||
ok, msg = update_channel_keys_list(CONFIG_DIR, cid, [str(x) for x in keys], ai)
|
||
if not ok:
|
||
return JSONResponse({"error": msg}, status_code=400)
|
||
return {"message": msg}
|
||
|
||
|
||
@app.post("/relay_pro/sync_active")
|
||
async def relay_pro_sync_active(channel_id: str = Query(...)):
|
||
try:
|
||
ok, msg = await asyncio.to_thread(sync_channel_to_legacy_files, CONFIG_DIR, channel_id)
|
||
except Exception as exc:
|
||
return JSONResponse({"error": str(exc)}, status_code=500)
|
||
if not ok:
|
||
return JSONResponse({"error": msg}, status_code=400)
|
||
return {"message": msg}
|
||
|
||
|
||
@app.get("/relay_pro/live_status")
|
||
async def relay_pro_live_status():
|
||
yt = first_youtube_pm2_name([{"script": e["script"]} for e in SCRIPT_ENTRIES])
|
||
|
||
async def read_tail(path: str, lines: int) -> str:
|
||
return await read_log_path(path, lines)
|
||
|
||
return await build_relay_live_status(
|
||
CONFIG_DIR,
|
||
BASE_DIR,
|
||
control_get_process_info,
|
||
read_tail,
|
||
yt,
|
||
)
|
||
|
||
|
||
@app.get("/host/gpu_status")
|
||
async def host_gpu_status():
|
||
return gpu_status_payload()
|
||
|
||
|
||
@app.get("/host/machine_summary")
|
||
async def host_machine_summary():
|
||
"""本机 hostname / OS / 架构等，供会员遥测写入 PocketBase。"""
|
||
return machine_summary_payload()
|
||
|
||
|
||
@app.post("/douyin/room_previews")
|
||
async def douyin_room_previews(request: Request):
|
||
try:
|
||
data = await request.json()
|
||
except Exception:
|
||
return JSONResponse({"error": "无效的 JSON"}, status_code=400)
|
||
if not isinstance(data, dict):
|
||
return JSONResponse({"error": "无效数据"}, status_code=400)
|
||
urls = data.get("urls")
|
||
if not isinstance(urls, list):
|
||
return JSONResponse({"error": "urls 须为数组"}, status_code=400)
|
||
out = []
|
||
for u in urls[:16]:
|
||
if not isinstance(u, str) or not u.strip():
|
||
continue
|
||
try:
|
||
out.append(fetch_douyin_anchor_name(u.strip()))
|
||
except Exception as exc:
|
||
out.append(
|
||
{
|
||
"ok": False,
|
||
"error": str(exc),
|
||
"url": u.strip(),
|
||
"room_id": "",
|
||
"anchor_name": "",
|
||
}
|
||
)
|
||
return {"previews": out}
|
||
|
||
|
||
@app.exception_handler(Exception)
|
||
async def web2_uncaught_exception_handler(request: Request, exc: Exception):
|
||
"""未捕获异常仍返回 JSON，便于 CORS 与前端展示。"""
|
||
return JSONResponse({"error": str(exc) or exc.__class__.__name__}, status_code=500)
|
||
|
||
|
||
# ---------------- 工具函数 ----------------
|
||
async def read_log_path(path: str, lines: int) -> str:
|
||
if not path:
|
||
return "无日志路径\n"
|
||
if "/dev/null" in path:
|
||
return "日志输出被禁用（无日志文件）\n"
|
||
log_file = Path(path)
|
||
if not log_file.exists():
|
||
return f"日志文件不存在: {path}\n"
|
||
try:
|
||
with open(log_file, "r", encoding="utf-8", errors="ignore") as f:
|
||
content_lines = f.readlines()
|
||
return "".join(content_lines[-lines:]) if content_lines else "无日志内容\n"
|
||
except Exception as e:
|
||
return f"读取日志失败 ({path}): {str(e)}\n"
|
||
|
||
|
||
def _log_age_seconds(path: str | None) -> float | None:
|
||
if not path:
|
||
return None
|
||
try:
|
||
p = Path(path)
|
||
if not p.exists():
|
||
return None
|
||
return max(0.0, time.time() - p.stat().st_mtime)
|
||
except OSError:
|
||
return None
|
||
|
||
|
||
def _ffmpeg_pids_for_process(info: dict, process_name: str) -> list[int]:
|
||
"""优先按内置 supervisor 的父子进程关系识别；兼容 PM2 时回退命令行关键字匹配。"""
|
||
pids: list[int] = []
|
||
seen: set[int] = set()
|
||
pid = info.get("pid")
|
||
if isinstance(pid, int) and pid > 0:
|
||
try:
|
||
parent = psutil.Process(pid)
|
||
for child in parent.children(recursive=True):
|
||
try:
|
||
name = (child.name() or "").lower()
|
||
if "ffmpeg" in name and child.pid not in seen:
|
||
pids.append(child.pid)
|
||
seen.add(child.pid)
|
||
except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
|
||
continue
|
||
except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
|
||
pass
|
||
if pids:
|
||
return pids
|
||
|
||
needle = (process_name or "").strip().lower()
|
||
if not needle:
|
||
return pids
|
||
for proc in psutil.process_iter(["pid", "name", "cmdline"]):
|
||
try:
|
||
name = (proc.info.get("name") or "").lower()
|
||
if "ffmpeg" not in name:
|
||
continue
|
||
flat = " ".join(str(x) for x in (proc.info.get("cmdline") or [])).lower()
|
||
env_name = ""
|
||
try:
|
||
env_name = str(proc.environ().get("LIVE_PM2_PROCESS_NAME") or "").strip().lower()
|
||
except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
|
||
env_name = ""
|
||
if env_name == needle or (needle in flat and int(proc.info["pid"]) not in seen):
|
||
pids.append(int(proc.info["pid"]))
|
||
seen.add(int(proc.info["pid"]))
|
||
except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess, ValueError):
|
||
continue
|
||
return pids
|
||
|
||
|
||
def _config_rel_for_channel(channel_id: str, kind: str) -> str:
|
||
if kind == "url":
|
||
return f"relay_pro/url_config.ch_{channel_id}.ini"
|
||
return "youtube.ini"
|
||
|
||
|
||
def _url_lines_for_process(process: str) -> str:
|
||
cid = relay_channel_id_from_pm2_name(process)
|
||
if cid:
|
||
return read_url_config_for_channel(CONFIG_DIR, cid)
|
||
p = CONFIG_DIR / "URL_config.ini"
|
||
if p.is_file():
|
||
try:
|
||
return p.read_text(encoding="utf-8")
|
||
except OSError:
|
||
pass
|
||
return ""
|
||
|
||
|
||
def _youtube_ini_text_for_process(process: str) -> str:
|
||
cid = relay_channel_id_from_pm2_name(process)
|
||
if cid:
|
||
d = get_channel_detail(CONFIG_DIR, cid)
|
||
key = (d or {}).get("youtubeKey") or ""
|
||
return f"[youtube]\nkey = {key}\n"
|
||
p = CONFIG_DIR / "youtube.ini"
|
||
if p.is_file():
|
||
try:
|
||
return p.read_text(encoding="utf-8")
|
||
except OSError:
|
||
pass
|
||
return ""
|
||
|
||
|
||
def _config_snapshot_for_process(process: str) -> dict[str, str]:
|
||
return {
|
||
"url_lines": _url_lines_for_process(process),
|
||
"youtube_ini_text": _youtube_ini_text_for_process(process),
|
||
"watch_url": watch_url_for_relay_process(CONFIG_DIR, process),
|
||
}
|
||
|
||
|
||
def _youtube_guard_targets(process: str) -> tuple[list[str], JSONResponse | None]:
|
||
ensure_default_relay_config(CONFIG_DIR)
|
||
proc = (process or "").strip()
|
||
if proc:
|
||
if not process_valid(proc):
|
||
return [], JSONResponse({"error": "无效进程名"}, status_code=400)
|
||
return [proc], None
|
||
targets: list[str] = []
|
||
for p in PROCESS_MONITOR:
|
||
sc = p.get("script") or ""
|
||
if is_youtube_script(sc):
|
||
targets.append(p["pm2"])
|
||
doc = list_channels_for_pro(CONFIG_DIR)
|
||
for ch in (doc.get("channels") or []) if doc.get("ok") else []:
|
||
cid = str(ch.get("id") or "").strip()
|
||
if not cid or not channel_configured_for_monitor(CONFIG_DIR, cid):
|
||
continue
|
||
pname = relay_pm2_name_for_channel(cid)
|
||
if pname not in targets:
|
||
targets.append(pname)
|
||
return targets, None
|
||
|
||
|
||
async def _youtube_guard_rows(targets: list[str]) -> tuple[list[dict], dict[str, dict[str, str]]]:
|
||
rows: list[dict] = []
|
||
config_by_process: dict[str, dict[str, str]] = {}
|
||
for proc in targets:
|
||
row = await _status_payload_for_process(proc)
|
||
label = proc
|
||
for p in PROCESS_MONITOR:
|
||
if p["pm2"] == proc:
|
||
label = p.get("label") or proc
|
||
break
|
||
cid = relay_channel_id_from_pm2_name(proc)
|
||
if cid:
|
||
d = get_channel_detail(CONFIG_DIR, cid)
|
||
if d:
|
||
label = str(d.get("name") or label)
|
||
row["label"] = label
|
||
hb = read_relay_heartbeat(proc, base_dir=BASE_DIR)
|
||
if hb.get("state") in {"starting", "streaming", "source_failover"}:
|
||
if hb.get("speed") is not None:
|
||
row["ffmpeg_speed"] = hb.get("speed")
|
||
if hb.get("frame") is not None:
|
||
row["ffmpeg_frame"] = hb.get("frame")
|
||
row["relay_heartbeat"] = hb
|
||
rows.append(row)
|
||
config_by_process[proc] = _config_snapshot_for_process(proc)
|
||
return rows, config_by_process
|
||
|
||
|
||
async def _status_payload_for_process(process: str, *, light: bool = False) -> dict:
|
||
info = await control_get_process_info(process)
|
||
ps = str(info.get("process_status") or "")
|
||
if ps == "not_found":
|
||
recent_log = "进程未运行或未由本机管理器启动。\n"
|
||
recent_error = ""
|
||
ffmpeg_pids: list[int] = []
|
||
log_age_seconds = None
|
||
elif ps in {"stopped", "stopping"}:
|
||
recent_log = ""
|
||
recent_error = ""
|
||
ffmpeg_pids = []
|
||
log_age_seconds = None
|
||
else:
|
||
online = ps in {"online", "running", "launching"}
|
||
read_logs = online or not light
|
||
if light and process_family(process) == "youtube":
|
||
log_lines = 80
|
||
err_lines = 24
|
||
else:
|
||
log_lines = MAX_LOG_LINES
|
||
err_lines = MAX_LOG_LINES
|
||
if read_logs:
|
||
recent_log = await read_log_path(info["out_path"], log_lines)
|
||
recent_error = await read_log_path(info["err_path"], err_lines)
|
||
else:
|
||
recent_log = ""
|
||
recent_error = ""
|
||
ffmpeg_pids = _ffmpeg_pids_for_process(info, process) if online else []
|
||
out_age = _log_age_seconds(info.get("out_path"))
|
||
err_age = _log_age_seconds(info.get("err_path"))
|
||
ages = [x for x in (out_age, err_age) if x is not None]
|
||
log_age_seconds = min(ages) if ages else None
|
||
activity = summarize_process_activity(
|
||
process,
|
||
ps,
|
||
recent_log,
|
||
recent_error,
|
||
log_age_seconds=log_age_seconds,
|
||
ffmpeg_pids=ffmpeg_pids,
|
||
)
|
||
source_meta = _youtube_source_meta(process, recent_log)
|
||
return {
|
||
"process": process,
|
||
"process_status": info["process_status"],
|
||
"recent_log": recent_log,
|
||
"recent_error": recent_error,
|
||
"log_age_seconds": log_age_seconds,
|
||
"ffmpeg_pids": activity["ffmpeg_pids"],
|
||
"ffmpeg_active": activity.get("ffmpeg_active"),
|
||
"business_status": activity["business_status"],
|
||
"business_note": activity["business_note"],
|
||
"is_pushing": activity["is_pushing"],
|
||
"ffmpeg_frame": activity.get("ffmpeg_frame"),
|
||
"ffmpeg_time_seconds": activity.get("ffmpeg_time_seconds"),
|
||
"ffmpeg_speed": activity.get("ffmpeg_speed"),
|
||
"ffmpeg_progress_stalled": activity.get("ffmpeg_progress_stalled"),
|
||
"ffmpeg_speed_low": activity.get("ffmpeg_speed_low"),
|
||
**source_meta,
|
||
}
|
||
|
||
|
||
def _youtube_source_meta(process: str, recent_log: str = "") -> dict:
|
||
sc = script_for_pm2(process)
|
||
if not sc or not is_youtube_script(sc):
|
||
return {}
|
||
return extract_douyin_source_meta(_url_lines_for_process(process), recent_log)
|
||
|
||
|
||
# ---------------- 配置路由（youtube.ini，仅 youtube） ----------------
|
||
@app.get("/get_config")
|
||
async def get_config(process: str = Query(..., description="进程名")):
|
||
sc = script_for_pm2(process)
|
||
if not sc or not is_youtube_script(sc):
|
||
return JSONResponse({"error": "仅 youtube*.py 对应进程支持 youtube.ini 编辑"}, status_code=400)
|
||
|
||
config_file = CONFIG_DIR / "youtube.ini"
|
||
if not config_file.exists():
|
||
return {"content": "# youtube.ini 文件不存在，将创建空文件\n"}
|
||
try:
|
||
content = config_file.read_text(encoding="utf-8")
|
||
return {"content": content}
|
||
except Exception as e:
|
||
return JSONResponse({"error": f"读取失败: {str(e)}"}, status_code=500)
|
||
|
||
@app.post("/save_config")
|
||
async def save_config(request: Request, process: str = Query(..., description="进程名")):
|
||
sc = script_for_pm2(process)
|
||
if not sc or not is_youtube_script(sc):
|
||
return JSONResponse({"error": "仅 youtube*.py 对应进程支持 youtube.ini 编辑"}, status_code=400)
|
||
|
||
try:
|
||
data = await request.json()
|
||
content = data.get("content", "")
|
||
except:
|
||
return JSONResponse({"error": "无效的 JSON 数据"}, status_code=400)
|
||
|
||
try:
|
||
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
|
||
write_config_text(CONFIG_DIR, "youtube.ini", content, reason="save_config")
|
||
return {"message": "配置保存成功"}
|
||
except Exception as e:
|
||
return JSONResponse({"error": f"保存失败: {str(e)}"}, status_code=500)
|
||
|
||
|
||
@app.get("/youtube/ini_model")
|
||
async def youtube_ini_model_get(process: str = Query(..., description="进程名")):
|
||
sc = script_for_pm2(process)
|
||
if not sc or not is_youtube_script(sc):
|
||
return JSONResponse({"error": "仅 youtube*.py 对应进程支持 youtube.ini"}, status_code=400)
|
||
config_file = CONFIG_DIR / "youtube.ini"
|
||
if not config_file.exists():
|
||
return {"keys": [""], "activeIndex": 0, "options": {}, "header": ""}
|
||
try:
|
||
content = config_file.read_text(encoding="utf-8")
|
||
except Exception as e:
|
||
return JSONResponse({"error": f"读取失败: {e}"}, status_code=500)
|
||
return parse_youtube_ini(content)
|
||
|
||
|
||
@app.post("/youtube/ini_model")
|
||
async def youtube_ini_model_post(request: Request, process: str = Query(..., description="进程名")):
|
||
sc = script_for_pm2(process)
|
||
if not sc or not is_youtube_script(sc):
|
||
return JSONResponse({"error": "仅 youtube*.py 对应进程支持 youtube.ini"}, status_code=400)
|
||
try:
|
||
data = await request.json()
|
||
except Exception:
|
||
return JSONResponse({"error": "无效的 JSON 数据"}, status_code=400)
|
||
if not isinstance(data, dict):
|
||
return JSONResponse({"error": "无效数据"}, status_code=400)
|
||
keys = data.get("keys")
|
||
if not isinstance(keys, list):
|
||
keys = [""]
|
||
keys = [str(x) for x in keys]
|
||
ok_keys, key_msg = validate_youtube_keys(keys)
|
||
if not ok_keys:
|
||
return JSONResponse({"error": key_msg}, status_code=400)
|
||
try:
|
||
active_index = int(data.get("activeIndex", 0))
|
||
except Exception:
|
||
active_index = 0
|
||
options = data.get("options")
|
||
if not isinstance(options, dict):
|
||
options = {}
|
||
else:
|
||
options = {str(k): str(v) for k, v in options.items()}
|
||
header = data.get("header")
|
||
if not isinstance(header, str):
|
||
header = ""
|
||
content = serialize_youtube_ini(keys, active_index, options, header)
|
||
try:
|
||
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
|
||
write_config_text(CONFIG_DIR, "youtube.ini", content, reason="ini_model")
|
||
return {"message": "配置保存成功"}
|
||
except Exception as e:
|
||
return JSONResponse({"error": f"保存失败: {str(e)}"}, status_code=500)
|
||
|
||
# ---------------- 配置路由（URL_config.ini，youtube 和 tiktok 共享） ----------------
|
||
@app.get("/get_url_config")
|
||
async def get_url_config(process: str = Query(..., description="进程名")):
|
||
if process not in URL_CONFIG_PM2:
|
||
return JSONResponse({"error": "仅 tiktok*.py / youtube*.py 对应进程支持 URL 配置编辑"}, status_code=400)
|
||
|
||
config_file = CONFIG_DIR / "URL_config.ini"
|
||
if not config_file.exists():
|
||
return {"content": "# URL_config.ini 文件不存在，将创建空文件\n"}
|
||
try:
|
||
content = config_file.read_text(encoding="utf-8")
|
||
return {"content": content}
|
||
except Exception as e:
|
||
return JSONResponse({"error": f"读取失败: {str(e)}"}, status_code=500)
|
||
|
||
@app.post("/save_url_config")
|
||
async def save_url_config(request: Request, process: str = Query(..., description="进程名")):
|
||
if process not in URL_CONFIG_PM2:
|
||
return JSONResponse({"error": "仅 tiktok*.py / youtube*.py 对应进程支持 URL 配置编辑"}, status_code=400)
|
||
|
||
try:
|
||
data = await request.json()
|
||
content = data.get("content", "")
|
||
except:
|
||
return JSONResponse({"error": "无效的 JSON 数据"}, status_code=400)
|
||
|
||
config_file = CONFIG_DIR / "URL_config.ini"
|
||
try:
|
||
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
|
||
write_config_text(CONFIG_DIR, "URL_config.ini", content, reason="save_url_config")
|
||
return {"message": "配置保存成功"}
|
||
except Exception as e:
|
||
return JSONResponse({"error": f"保存失败: {str(e)}"}, status_code=500)
|
||
|
||
def _script_entries_payload() -> list[dict]:
|
||
return [{"script": e["script"]} for e in SCRIPT_ENTRIES]
|
||
|
||
|
||
async def _relay_youtube_online_pm2_names() -> list[str]:
|
||
"""已启动的多路转播进程 youtube2__*（与单进程 youtube2 互斥）。"""
|
||
doc = list_channels_for_pro(CONFIG_DIR)
|
||
if not doc.get("ok"):
|
||
return []
|
||
out: list[str] = []
|
||
for ch in doc.get("channels") or []:
|
||
cid = str(ch.get("id") or "").strip()
|
||
if not cid:
|
||
continue
|
||
pnm = relay_pm2_name_for_channel(cid)
|
||
info = await control_get_process_info(pnm)
|
||
if str(info.get("process_status") or "") == "online":
|
||
out.append(pnm)
|
||
return out
|
||
|
||
|
||
async def _stop_legacy_youtube_if_online() -> None:
|
||
"""多路转播启动前必须停掉单进程 youtube2，否则会读全局 URL_config.ini，等同「所有线路一起推」。"""
|
||
if not use_builtin():
|
||
return
|
||
legacy = first_youtube_pm2_name(_script_entries_payload())
|
||
if not legacy:
|
||
return
|
||
info = await control_get_process_info(legacy)
|
||
if str(info.get("process_status") or "") != "online":
|
||
return
|
||
await control_stop(legacy)
|
||
await asyncio.sleep(0.5)
|
||
|
||
|
||
async def _legacy_youtube_conflict_response(process: str, sc: Optional[str]) -> JSONResponse | None:
|
||
legacy_pm2 = first_youtube_pm2_name(_script_entries_payload())
|
||
if sc != "youtube2.py" or process != legacy_pm2:
|
||
return None
|
||
online_relays = await _relay_youtube_online_pm2_names()
|
||
if not online_relays:
|
||
return None
|
||
return JSONResponse(
|
||
{
|
||
"output": (
|
||
"当前有多频道 Pro 推流进程正在运行，不能同时启动/重启单频道 youtube2；"
|
||
f"请先停止：{', '.join(online_relays)}"
|
||
)
|
||
},
|
||
status_code=400,
|
||
)
|
||
|
||
|
||
def _relay_channel_ready_message(channel_id: str) -> str | None:
|
||
d = get_channel_detail(CONFIG_DIR, channel_id)
|
||
if not d:
|
||
return "频道不存在"
|
||
if not str(d.get("youtubeKey") or "").strip():
|
||
return "请先配置该频道的 YouTube 串流密钥"
|
||
body = read_url_config_for_channel(CONFIG_DIR, channel_id)
|
||
if not extract_url_lines_from_ini(body):
|
||
return "请先为该频道填写至少一个源站直播间地址"
|
||
return None
|
||
|
||
|
||
# ---------------- PM2 控制路由 ----------------
|
||
def _camera_ready_message(process: str) -> str | None:
|
||
if not is_camera_script(script_for_pm2(process) or ""):
|
||
return None
|
||
return camera_config_ready(CONFIG_DIR)
|
||
|
||
|
||
def _pull_stream_ready_message(process: str) -> str | None:
|
||
if not is_pull_stream_script(script_for_pm2(process) or ""):
|
||
return None
|
||
return pull_stream_config_ready(CONFIG_DIR)
|
||
|
||
|
||
def _stream_worker_ready_message(process: str) -> str | None:
|
||
return _camera_ready_message(process) or _pull_stream_ready_message(process)
|
||
|
||
|
||
def _disabled_feature_response(feature: str) -> JSONResponse:
|
||
return JSONResponse({"error": f"{feature} 功能已禁用"}, status_code=403)
|
||
|
||
|
||
@app.get("/camera/devices")
|
||
async def camera_devices():
|
||
if not SHOW_CAMERA_FEATURE:
|
||
return _disabled_feature_response("摄像头")
|
||
return list_capture_devices()
|
||
|
||
|
||
@app.get("/camera/config")
|
||
async def camera_config_get():
|
||
if not SHOW_CAMERA_FEATURE:
|
||
return _disabled_feature_response("摄像头")
|
||
return {"config": load_camera_config(CONFIG_DIR)}
|
||
|
||
|
||
@app.post("/camera/config")
|
||
async def camera_config_post(request: Request):
|
||
if not SHOW_CAMERA_FEATURE:
|
||
return _disabled_feature_response("摄像头")
|
||
try:
|
||
data = await request.json()
|
||
except Exception:
|
||
return JSONResponse({"error": "无效的 JSON"}, status_code=400)
|
||
if not isinstance(data, dict):
|
||
return JSONResponse({"error": "无效数据"}, status_code=400)
|
||
patch = data.get("config") if isinstance(data.get("config"), dict) else data
|
||
ok, msg = save_camera_config(CONFIG_DIR, patch)
|
||
if not ok:
|
||
return JSONResponse({"error": msg}, status_code=400)
|
||
return {"message": msg, "config": load_camera_config(CONFIG_DIR)}
|
||
|
||
|
||
@app.get("/pull_stream/config")
|
||
async def pull_stream_config_get():
|
||
if not SHOW_PULL_STREAM_FEATURE:
|
||
return _disabled_feature_response("拉流")
|
||
return {"config": load_pull_stream_config(CONFIG_DIR)}
|
||
|
||
|
||
@app.get("/pull_stream/protocols")
|
||
async def pull_stream_protocols():
|
||
if not SHOW_PULL_STREAM_FEATURE:
|
||
return _disabled_feature_response("拉流")
|
||
return {"protocols": protocol_help()}
|
||
|
||
|
||
@app.post("/pull_stream/config")
|
||
async def pull_stream_config_post(request: Request):
|
||
if not SHOW_PULL_STREAM_FEATURE:
|
||
return _disabled_feature_response("拉流")
|
||
try:
|
||
data = await request.json()
|
||
except Exception:
|
||
return JSONResponse({"error": "无效的 JSON"}, status_code=400)
|
||
if not isinstance(data, dict):
|
||
return JSONResponse({"error": "无效数据"}, status_code=400)
|
||
patch = data.get("config") if isinstance(data.get("config"), dict) else data
|
||
ok, msg = save_pull_stream_config(CONFIG_DIR, patch)
|
||
if not ok:
|
||
return JSONResponse({"error": msg}, status_code=400)
|
||
return {"message": msg, "config": load_pull_stream_config(CONFIG_DIR)}
|
||
|
||
|
||
@app.get("/start")
|
||
async def start(request: Request, process: str = Query(..., description="进程名")):
|
||
if REQUIRE_PB:
|
||
tok = (request.headers.get("x-pb-token") or "").strip()
|
||
ok, msg, _ = await pb_auth_refresh_and_validate(tok)
|
||
if not ok:
|
||
return JSONResponse({"output": msg}, status_code=403)
|
||
if not process_valid(process):
|
||
return JSONResponse({"output": f"无效进程名: {process}"}, status_code=400)
|
||
cid = relay_channel_id_from_pm2_name(process)
|
||
sc = script_for_pm2(process)
|
||
conflict = await _legacy_youtube_conflict_response(process, sc)
|
||
if conflict is not None:
|
||
return conflict
|
||
if cid:
|
||
ready_msg = _relay_channel_ready_message(cid)
|
||
if ready_msg:
|
||
return JSONResponse({"output": ready_msg}, status_code=400)
|
||
ok, msg = materialize_relay_channel_youtube_ini(CONFIG_DIR, cid)
|
||
if not ok:
|
||
return JSONResponse({"output": msg}, status_code=400)
|
||
await _stop_legacy_youtube_if_online()
|
||
cam_msg = _stream_worker_ready_message(process)
|
||
if cam_msg:
|
||
return JSONResponse({"output": cam_msg}, status_code=400)
|
||
if is_youtube_script(sc):
|
||
preflight_msg = await youtube_start_preflight_message()
|
||
if preflight_msg:
|
||
return JSONResponse({"output": preflight_msg}, status_code=400)
|
||
ch_doc, _ = load_channels_document(CONFIG_DIR)
|
||
channels = (ch_doc or {}).get("channels") or []
|
||
runtime_errors = await asyncio.to_thread(
|
||
validate_youtube_runtime_files, CONFIG_DIR, process, channels
|
||
)
|
||
if runtime_errors:
|
||
return JSONResponse({"output": "\n".join(runtime_errors)}, status_code=400)
|
||
await asyncio.to_thread(sync_all_youtube_pro_runtime_files, CONFIG_DIR, channels)
|
||
output = await control_start(process)
|
||
await _record_youtube_control(process, "start")
|
||
return JSONResponse({"output": output})
|
||
|
||
|
||
@app.get("/stop")
|
||
async def stop(process: str = Query(..., description="进程名")):
|
||
if not process_valid(process):
|
||
return JSONResponse({"output": f"无效进程名: {process}"}, status_code=400)
|
||
|
||
# 第一步：尝试优雅停止
|
||
ctrl_output = await control_stop(process)
|
||
|
||
# 第二步：等待信号传播（通常 1~3 秒）
|
||
await asyncio.sleep(2.0)
|
||
|
||
# 第三步：录制类进程才强杀 ffmpeg（web* 控制台不杀 ffmpeg）
|
||
if not process.startswith("web"):
|
||
await force_kill_ffmpeg_cross_platform(process)
|
||
|
||
# 第四步：再等一小会儿，确认
|
||
await asyncio.sleep(1.0)
|
||
|
||
await _close_youtube_relay_sessions(process, reason="stopped")
|
||
await _record_youtube_control(process, "stop", note="manual stop")
|
||
|
||
return JSONResponse({
|
||
"message": "已执行 stop" + ("" if process.startswith("web") else " + 强杀 ffmpeg"),
|
||
"pm2_output": ctrl_output,
|
||
"note": "如果仍有残留，可多次点击停止或查看下方日志",
|
||
})
|
||
|
||
|
||
@app.get("/restart")
|
||
async def restart(request: Request, process: str = Query(..., description="进程名")):
|
||
if REQUIRE_PB:
|
||
tok = (request.headers.get("x-pb-token") or "").strip()
|
||
ok, msg, _ = await pb_auth_refresh_and_validate(tok)
|
||
if not ok:
|
||
return JSONResponse({"output": msg}, status_code=403)
|
||
if not process_valid(process):
|
||
return JSONResponse({"output": f"无效进程名: {process}"}, status_code=400)
|
||
cid = relay_channel_id_from_pm2_name(process)
|
||
sc = script_for_pm2(process)
|
||
conflict = await _legacy_youtube_conflict_response(process, sc)
|
||
if conflict is not None:
|
||
return conflict
|
||
|
||
stop_output = await control_stop(process)
|
||
await asyncio.sleep(1.0)
|
||
if not process.startswith("web"):
|
||
await force_kill_ffmpeg_cross_platform(process)
|
||
await asyncio.sleep(0.35)
|
||
|
||
await _close_youtube_relay_sessions(process, reason="restart")
|
||
await _record_youtube_control(process, "stop", note="restart pre-start")
|
||
|
||
if cid:
|
||
ready_msg = _relay_channel_ready_message(cid)
|
||
if ready_msg:
|
||
return JSONResponse({"output": f"{stop_output}\n{ready_msg}"}, status_code=400)
|
||
ok, msg = materialize_relay_channel_youtube_ini(CONFIG_DIR, cid)
|
||
if not ok:
|
||
return JSONResponse({"output": f"{stop_output}\n{msg}"}, status_code=400)
|
||
await _stop_legacy_youtube_if_online()
|
||
cam_msg = _stream_worker_ready_message(process)
|
||
if cam_msg:
|
||
return JSONResponse({"output": f"{stop_output}\n{cam_msg}"}, status_code=400)
|
||
sc = script_for_pm2(process)
|
||
if is_youtube_script(sc):
|
||
preflight_msg = await youtube_start_preflight_message()
|
||
if preflight_msg:
|
||
return JSONResponse({"output": f"{stop_output}\n{preflight_msg}"}, status_code=400)
|
||
ch_doc, _ = load_channels_document(CONFIG_DIR)
|
||
channels = (ch_doc or {}).get("channels") or []
|
||
runtime_errors = await asyncio.to_thread(
|
||
validate_youtube_runtime_files, CONFIG_DIR, process, channels
|
||
)
|
||
if runtime_errors:
|
||
return JSONResponse(
|
||
{"output": f"{stop_output}\n" + "\n".join(runtime_errors)},
|
||
status_code=400,
|
||
)
|
||
await asyncio.to_thread(sync_all_youtube_pro_runtime_files, CONFIG_DIR, channels)
|
||
start_output = await control_start(process)
|
||
await _record_youtube_control(process, "restart")
|
||
return JSONResponse({"output": f"{stop_output}\n{start_output}"})
|
||
|
||
|
||
@app.get("/status")
|
||
async def status(
|
||
process: str = Query(..., description="进程名"),
|
||
light: bool = Query(False, description="轻量模式：减少日志读取，适合轮询"),
|
||
):
|
||
if not process_valid(process):
|
||
return JSONResponse({
|
||
"raw_status": "",
|
||
"process_status": "invalid",
|
||
"recent_log": f"无效进程名: {process}",
|
||
"recent_error": ""
|
||
})
|
||
|
||
full_status = ""
|
||
if not light:
|
||
full_status_raw = await control_full_status()
|
||
full_status = strip_ansi_codes(full_status_raw)
|
||
payload = await _status_payload_for_process(process, light=light)
|
||
ps = str(payload["process_status"])
|
||
|
||
prev = _PROCESS_STATUS_CACHE.get(process)
|
||
if prev != ps:
|
||
track_process_status_transition(process, prev or "", ps, db_path=RELAY_DB)
|
||
_PROCESS_STATUS_CACHE[process] = ps
|
||
|
||
url_lines = _url_lines_for_process(process)
|
||
channel_name = ""
|
||
cid = relay_channel_id_from_pm2_name(process)
|
||
if cid:
|
||
d = get_channel_detail(CONFIG_DIR, cid)
|
||
channel_name = str((d or {}).get("name") or "")
|
||
try:
|
||
sync_relay_from_recent_log(
|
||
process,
|
||
str(payload.get("recent_log") or ""),
|
||
channel_name=channel_name,
|
||
url_lines=url_lines,
|
||
db_path=RELAY_DB,
|
||
base_dir=BASE_DIR,
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
try:
|
||
hb = read_relay_heartbeat(process, base_dir=BASE_DIR)
|
||
observe_payload = {
|
||
"process_status": ps,
|
||
"recent_log": str(payload.get("recent_log") or ""),
|
||
"recent_error": str(payload.get("recent_error") or ""),
|
||
"is_pushing": ps == "online" and bool(hb.get("state") in {"starting", "streaming", "source_failover"}),
|
||
"live_status": str(hb.get("state") or ""),
|
||
"business_status": "streaming" if hb.get("state") == "streaming" else ps,
|
||
}
|
||
if hb:
|
||
observe_payload["relay_heartbeat"] = hb
|
||
if hb.get("speed") is not None:
|
||
observe_payload["ffmpeg_speed"] = hb.get("speed")
|
||
if hb.get("frame") is not None:
|
||
observe_payload["ffmpeg_frame"] = hb.get("frame")
|
||
now_ts = time.time()
|
||
last_ob = _LIVE_OBSERVE_LAST.get(process, 0.0)
|
||
if now_ts - last_ob >= YOUTUBE_EVENT_STATUS_OBSERVE_SECONDS:
|
||
_LIVE_OBSERVE_LAST[process] = now_ts
|
||
record_process_observation(
|
||
process,
|
||
observe_payload,
|
||
url_lines=url_lines,
|
||
db_path=LIVE_DB,
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
return JSONResponse({
|
||
"raw_status": full_status,
|
||
**payload,
|
||
})
|
||
|
||
|
||
@app.get("/health")
|
||
async def health():
|
||
return {"ok": True}
|
||
|
||
|
||
@app.get("/proxy_config")
|
||
async def get_proxy_config():
|
||
"""仅作用于录制子进程的代理（不写 Windows 系统代理）。"""
|
||
return read_proxy_config()
|
||
|
||
|
||
@app.post("/proxy_config")
|
||
async def post_proxy_config(request: Request):
|
||
try:
|
||
data = await request.json()
|
||
except Exception:
|
||
return JSONResponse({"error": "无效的 JSON"}, status_code=400)
|
||
if not isinstance(data, dict):
|
||
return JSONResponse({"error": "无效数据"}, status_code=400)
|
||
write_proxy_config(
|
||
{
|
||
"enabled": bool(data.get("enabled")),
|
||
"http": str(data.get("http") or ""),
|
||
"socks": str(data.get("socks") or ""),
|
||
}
|
||
)
|
||
return {"ok": True}
|
||
|
||
|
||
@app.get("/system_metrics")
|
||
async def system_metrics(
|
||
force_refresh: bool = Query(False, description="是否跳过缓存强制重新检测"),
|
||
):
|
||
"""CPU/内存/磁盘/网卡速率与 ffmpeg 进程摘要。"""
|
||
global _SYSTEM_METRICS_CACHE
|
||
try:
|
||
now = time.time()
|
||
async with _SYSTEM_METRICS_CACHE_LOCK:
|
||
if (
|
||
not force_refresh
|
||
and _SYSTEM_METRICS_CACHE
|
||
and (now - _SYSTEM_METRICS_CACHE[0]) < _SYSTEM_METRICS_TTL_SECONDS
|
||
):
|
||
return _SYSTEM_METRICS_CACHE[1]
|
||
|
||
payload = await asyncio.to_thread(system_metrics_snapshot)
|
||
async with _SYSTEM_METRICS_CACHE_LOCK:
|
||
_SYSTEM_METRICS_CACHE = (time.time(), payload)
|
||
return payload
|
||
except Exception as e:
|
||
return JSONResponse({"error": str(e)}, status_code=500)
|
||
|
||
|
||
# ---------------- YouTube OAuth（串流密钥） ----------------
|
||
@app.get("/youtube/oauth/start")
|
||
async def youtube_oauth_start():
|
||
"""返回 Google 授权页 URL，前端应用浏览器打开。"""
|
||
try:
|
||
url, state = oauth_authorization_url()
|
||
return {"url": url, "state": state}
|
||
except Exception as e:
|
||
return JSONResponse({"error": str(e)}, status_code=400)
|
||
|
||
|
||
@app.get("/youtube/oauth/callback")
|
||
async def youtube_oauth_callback(code: str = Query(...), state: str = Query(...)):
|
||
"""Google 授权完成后的回调（须与 google_oauth.json 中 redirect_uri 一致）。"""
|
||
try:
|
||
oauth_exchange_code(code, state)
|
||
except Exception as e:
|
||
err = html.escape(str(e))
|
||
return HTMLResponse(
|
||
f"<!DOCTYPE html><html><head><meta charset='utf-8'><title>授权失败</title></head>"
|
||
f"<body style='font-family:system-ui;padding:24px;background:#1a1d24;color:#f56565;'>"
|
||
f"<p>授权失败：{err}</p><p style='color:#8b929e;font-size:13px;'>请关闭此页后重试。</p></body></html>",
|
||
status_code=400,
|
||
)
|
||
return HTMLResponse(
|
||
"<!DOCTYPE html><html><head><meta charset='utf-8'><title>授权成功</title></head>"
|
||
"<body style='font-family:system-ui;padding:24px;background:#1a1d24;color:#e8eaed;'>"
|
||
"<p>授权成功，可关闭此页并返回「直播录制助手」。</p>"
|
||
"<p style='color:#8b929e;font-size:13px;'>若浏览器未自动关闭，请手动关闭本标签页。</p>"
|
||
"</body></html>"
|
||
)
|
||
|
||
|
||
@app.get("/youtube/oauth/status")
|
||
async def youtube_oauth_status_route():
|
||
return youtube_oauth_status_snapshot()
|
||
|
||
|
||
@app.post("/youtube/stream/fetch_key")
|
||
async def youtube_stream_fetch_key(process: str = Query(..., description="进程名")):
|
||
"""调用 YouTube Live API 获取串流密钥并写入 config/youtube.ini。"""
|
||
if process not in VALID_PROCESSES:
|
||
return JSONResponse({"error": "无效进程名"}, status_code=400)
|
||
sc = script_for_pm2(process)
|
||
if not sc or not is_youtube_script(sc):
|
||
return JSONResponse({"error": "仅 youtube*.py 对应进程可使用此功能"}, status_code=400)
|
||
try:
|
||
return fetch_stream_key_and_write_ini()
|
||
except Exception as e:
|
||
return JSONResponse({"error": str(e)}, status_code=500)
|
||
|
||
|
||
# ---------------- 进程列表（供前端同步，由 discover_script_entries 自动扫描） ----------------
|
||
@app.get("/youtube/quality_guard")
|
||
async def youtube_quality_guard(process: str = Query("", description="进程名，空则汇总全部 YouTube 线路")):
|
||
targets, err = _youtube_guard_targets(process)
|
||
if err is not None:
|
||
return err
|
||
rows, config_by_process = await _youtube_guard_rows(targets)
|
||
live = await asyncio.to_thread(youtube_live_summary, db_path=LIVE_DB)
|
||
return build_youtube_quality_guard_report(
|
||
rows, live_summary=live, config_by_process=config_by_process
|
||
)
|
||
|
||
|
||
@app.get("/youtube/diagnostics")
|
||
async def youtube_diagnostics(process: str = Query("", description="进程名，空则汇总全部 YouTube 线路")):
|
||
targets, err = _youtube_guard_targets(process)
|
||
if err is not None:
|
||
return err
|
||
rows, config_by_process = await _youtube_guard_rows(targets)
|
||
return build_youtube_diagnostics_report(rows, config_by_process=config_by_process)
|
||
|
||
|
||
@app.get("/youtube/preflight")
|
||
async def youtube_preflight():
|
||
"""开播前网络体检（Google / YouTube / 抖音 + 录制代理）。"""
|
||
try:
|
||
return await run_youtube_preflight()
|
||
except Exception as e:
|
||
return JSONResponse({"error": str(e), "ok": False}, status_code=500)
|
||
|
||
|
||
@app.get("/youtube/relay/sessions")
|
||
async def youtube_relay_sessions(
|
||
process: str = Query("", description="PM2 进程名"),
|
||
q: str = Query("", description="搜索主播/URL/房间"),
|
||
date_from: str = Query("", description="起始日期 YYYY-MM-DD"),
|
||
date_to: str = Query("", description="结束日期 YYYY-MM-DD"),
|
||
status: str = Query("", description="active | ended"),
|
||
limit: int = Query(30, ge=1, le=200),
|
||
offset: int = Query(0, ge=0),
|
||
):
|
||
return list_relay_sessions(
|
||
process=process,
|
||
q=q,
|
||
date_from=date_from,
|
||
date_to=date_to,
|
||
status=status,
|
||
limit=limit,
|
||
offset=offset,
|
||
db_path=RELAY_DB,
|
||
)
|
||
|
||
|
||
@app.get("/youtube/relay/sessions/{session_id}")
|
||
async def youtube_relay_session_detail(session_id: str):
|
||
session = get_relay_session(session_id, db_path=RELAY_DB)
|
||
if not session:
|
||
return JSONResponse({"error": "会话不存在"}, status_code=404)
|
||
return {"session": session}
|
||
|
||
|
||
@app.get("/youtube/relay/sessions/{session_id}/cover")
|
||
async def youtube_relay_session_cover(session_id: int):
|
||
path = get_relay_cover_path(session_id, db_path=RELAY_DB)
|
||
if not path:
|
||
return JSONResponse({"error": "封面不存在"}, status_code=404)
|
||
return FileResponse(path, media_type="image/jpeg")
|
||
|
||
|
||
@app.get("/youtube/relay/heartbeat")
|
||
async def youtube_relay_heartbeat(process: str = Query(..., description="PM2 进程名")):
|
||
if not process_valid(process):
|
||
return JSONResponse({"error": "无效进程名"}, status_code=400)
|
||
hb = read_relay_heartbeat(process, base_dir=BASE_DIR)
|
||
return {"process": process, "heartbeat": hb, "fresh": bool(hb.get("updated_at"))}
|
||
|
||
|
||
@app.get("/youtube/live/summary")
|
||
async def youtube_live_summary_route():
|
||
try:
|
||
return youtube_live_summary(db_path=LIVE_DB)
|
||
except Exception as e:
|
||
return JSONResponse({"error": str(e)}, status_code=500)
|
||
|
||
|
||
@app.get("/youtube/live/events")
|
||
async def youtube_live_events(
|
||
process: str = Query("", description="PM2 进程名"),
|
||
status: str = Query("", description="active | ended | error"),
|
||
limit: int = Query(30, ge=1, le=200),
|
||
offset: int = Query(0, ge=0),
|
||
):
|
||
return list_live_events(
|
||
process=process,
|
||
status=status,
|
||
limit=limit,
|
||
offset=offset,
|
||
db_path=LIVE_DB,
|
||
)
|
||
|
||
|
||
@app.get("/youtube/live/events/{event_id}")
|
||
async def youtube_live_event_detail(event_id: str):
|
||
event = get_live_event(event_id, db_path=LIVE_DB)
|
||
if not event:
|
||
return JSONResponse({"error": "事件不存在"}, status_code=404)
|
||
return {"event": event}
|
||
|
||
|
||
@app.post("/youtube/live/events/{event_id}/close")
|
||
async def youtube_live_event_close(event_id: str, request: Request):
|
||
try:
|
||
data = await request.json()
|
||
except Exception:
|
||
data = {}
|
||
reason = str(data.get("reason") or "manual")
|
||
result = close_live_event(event_id, reason=reason, db_path=LIVE_DB)
|
||
if not result:
|
||
return JSONResponse({"error": "事件不存在"}, status_code=404)
|
||
return result
|
||
|
||
|
||
@app.post("/youtube/live/ingest")
|
||
async def youtube_live_ingest(request: Request):
|
||
try:
|
||
data = await request.json()
|
||
except Exception:
|
||
return JSONResponse({"error": "无效的 JSON"}, status_code=400)
|
||
if not isinstance(data, dict):
|
||
return JSONResponse({"error": "无效的 JSON"}, status_code=400)
|
||
ok, msg = verify_ingest_payload(data)
|
||
if not ok:
|
||
return JSONResponse({"error": msg}, status_code=403)
|
||
return ingest_browser_telemetry(data, db_path=LIVE_DB)
|
||
|
||
|
||
@app.get("/youtube/live/ingest_config")
|
||
async def youtube_live_ingest_config():
|
||
return ingest_config_payload()
|
||
|
||
|
||
@app.post("/youtube/relay/sessions/import")
|
||
async def youtube_relay_sessions_import(request: Request, process: str = Query(..., description="进程名")):
|
||
if not process_valid(process):
|
||
return JSONResponse({"error": "无效进程名"}, status_code=400)
|
||
replace_imported = False
|
||
log_path_override = ""
|
||
try:
|
||
body = await request.json()
|
||
if isinstance(body, dict):
|
||
replace_imported = bool(body.get("replace_imported"))
|
||
log_path_override = str(body.get("log_path") or "").strip()
|
||
except Exception:
|
||
pass
|
||
info = await control_get_process_info(process)
|
||
log_path = log_path_override or info.get("out_path") or ""
|
||
if not log_path:
|
||
return JSONResponse({"error": "无法定位 PM2 日志路径"}, status_code=400)
|
||
result = import_relay_sessions_from_log(
|
||
Path(log_path),
|
||
process=process,
|
||
url_lines=_url_lines_for_process(process),
|
||
youtube_key_suffix=extract_youtube_key_suffix(_youtube_ini_text_for_process(process)),
|
||
db_path=RELAY_DB,
|
||
replace_imported=replace_imported,
|
||
)
|
||
if result.get("error"):
|
||
return JSONResponse(result, status_code=400)
|
||
return result
|
||
|
||
|
||
@app.get("/config/backups")
|
||
async def config_backups(file: str = Query(..., description="相对 config 的路径，如 youtube.ini")):
|
||
rel = file.strip().replace("\\", "/").lstrip("/")
|
||
if ".." in rel.split("/"):
|
||
return JSONResponse({"error": "非法路径"}, status_code=400)
|
||
return {"file": rel, "backups": list_config_backups(CONFIG_DIR, rel)}
|
||
|
||
|
||
@app.post("/config/restore")
|
||
async def config_restore(request: Request):
|
||
try:
|
||
data = await request.json()
|
||
except Exception:
|
||
return JSONResponse({"error": "无效的 JSON"}, status_code=400)
|
||
rel = str(data.get("file") or "").strip().replace("\\", "/").lstrip("/")
|
||
backup_id = str(data.get("backupId") or data.get("backup_id") or "").strip()
|
||
if not rel or not backup_id or ".." in rel.split("/"):
|
||
return JSONResponse({"error": "参数无效"}, status_code=400)
|
||
ok, msg = restore_config_backup(CONFIG_DIR, rel, backup_id)
|
||
if not ok:
|
||
return JSONResponse({"error": msg}, status_code=400)
|
||
return {"message": msg}
|
||
|
||
|
||
@app.post("/config_optimize")
|
||
async def config_optimize(request: Request):
|
||
"""配置优化（参照 d2ypp2 config_governance.optimize_content）。"""
|
||
try:
|
||
data = await request.json()
|
||
except Exception:
|
||
return JSONResponse({"error": "无效的 JSON"}, status_code=400)
|
||
if not isinstance(data, dict):
|
||
return JSONResponse({"error": "无效数据"}, status_code=400)
|
||
action = str(data.get("action") or "").strip()
|
||
content = str(data.get("content") or "")
|
||
rel = str(data.get("file") or "").strip().replace("\\", "/").lstrip("/")
|
||
if ".." in rel.split("/"):
|
||
return JSONResponse({"error": "非法路径"}, status_code=400)
|
||
if not content and rel:
|
||
content = read_config_file(CONFIG_DIR, rel)
|
||
try:
|
||
result = optimize_config_content(action, rel, content)
|
||
except ValueError as exc:
|
||
return JSONResponse({"error": str(exc)}, status_code=400)
|
||
if data.get("save") and rel:
|
||
write_config_text(CONFIG_DIR, rel, result["content"], reason=action)
|
||
result["message"] = f"{result['message']}，并已保存"
|
||
return result
|
||
|
||
|
||
@app.get("/config_insights")
|
||
async def config_insights_get(file: str = Query(..., description="相对 config 的路径")):
|
||
rel = file.strip().replace("\\", "/").lstrip("/")
|
||
if ".." in rel.split("/"):
|
||
return JSONResponse({"error": "非法路径"}, status_code=400)
|
||
content = read_config_file(CONFIG_DIR, rel)
|
||
if not content and not (CONFIG_DIR / rel).is_file():
|
||
return JSONResponse({"error": "文件不存在"}, status_code=404)
|
||
return analyze_config_content(rel, content)
|
||
|
||
|
||
@app.post("/config_insights")
|
||
async def config_insights_post(request: Request):
|
||
try:
|
||
data = await request.json()
|
||
except Exception:
|
||
return JSONResponse({"error": "无效的 JSON"}, status_code=400)
|
||
rel = str(data.get("file") or "").strip().replace("\\", "/").lstrip("/")
|
||
content = str(data.get("content") or "")
|
||
if ".." in rel.split("/"):
|
||
return JSONResponse({"error": "非法路径"}, status_code=400)
|
||
return analyze_config_content(rel, content)
|
||
|
||
|
||
@app.post("/relay_pro/channels_save")
|
||
async def relay_pro_channels_save(request: Request):
|
||
try:
|
||
data = await request.json()
|
||
except Exception:
|
||
return JSONResponse({"error": "无效的 JSON"}, status_code=400)
|
||
channels = data.get("channels") if isinstance(data.get("channels"), list) else data
|
||
if not isinstance(channels, list):
|
||
return JSONResponse({"error": "channels 必须是数组"}, status_code=400)
|
||
ok, msg = save_channels_document(CONFIG_DIR, channels)
|
||
if not ok:
|
||
return JSONResponse({"error": msg}, status_code=400)
|
||
sync_errors = sync_all_youtube_pro_runtime_files(CONFIG_DIR, channels)
|
||
return {"message": msg, "runtime_errors": sync_errors}
|
||
|
||
|
||
def _process_monitor_family_matches(entry: dict, family: str | None) -> bool:
|
||
target = (family or "").strip().lower()
|
||
if not target or target == "all":
|
||
return True
|
||
script = str(entry.get("script") or "")
|
||
pm2 = str(entry.get("pm2") or "")
|
||
if target == "youtube":
|
||
return is_youtube_script(script) or pm2.startswith("youtube2__")
|
||
if target == "tiktok":
|
||
return is_tiktok_script(script)
|
||
return True
|
||
|
||
|
||
async def _process_monitor_snapshot(entry: dict, *, light: bool = True) -> dict:
|
||
pm = str(entry.get("pm2") or "").strip()
|
||
snap = await _status_payload_for_process(pm, light=light)
|
||
return {
|
||
"pm2": pm,
|
||
"script": str(entry.get("script") or ""),
|
||
"label": str(entry.get("label") or pm),
|
||
"process_status": snap.get("process_status", ""),
|
||
"business_status": snap.get("business_status", ""),
|
||
"business_note": snap.get("business_note", ""),
|
||
"is_pushing": bool(snap.get("is_pushing")),
|
||
"ffmpeg_active": snap.get("ffmpeg_active"),
|
||
"ffmpeg_speed": snap.get("ffmpeg_speed"),
|
||
"ffmpeg_frame": snap.get("ffmpeg_frame"),
|
||
"ffmpeg_time_seconds": snap.get("ffmpeg_time_seconds"),
|
||
"ffmpeg_progress_stalled": snap.get("ffmpeg_progress_stalled"),
|
||
"ffmpeg_speed_low": snap.get("ffmpeg_speed_low"),
|
||
"source_display": snap.get("source_display", ""),
|
||
"source_title": snap.get("source_title", ""),
|
||
"source_room_id": snap.get("source_room_id", ""),
|
||
"source_url": snap.get("source_url", ""),
|
||
"source_ambiguous": snap.get("source_ambiguous", False),
|
||
"source_current": snap.get("source_current", False),
|
||
}
|
||
|
||
|
||
@app.get("/process_monitor")
|
||
async def process_monitor(
|
||
family: str | None = Query(None, description="按进程族过滤：youtube/tiktok"),
|
||
light: bool = Query(False, description="轻量快照：含 business_status / is_pushing"),
|
||
):
|
||
ensure_default_relay_config(CONFIG_DIR)
|
||
entries = [
|
||
{"pm2": p["pm2"], "script": p["script"], "label": p["label"]}
|
||
for p in PROCESS_MONITOR
|
||
]
|
||
doc = list_channels_for_pro(CONFIG_DIR)
|
||
if doc.get("ok"):
|
||
for ch in doc.get("channels") or []:
|
||
cid = str(ch.get("id") or "").strip()
|
||
if not cid:
|
||
continue
|
||
pm2 = relay_pm2_name_for_channel(cid, CONFIG_DIR)
|
||
if any(e.get("pm2") == pm2 for e in entries):
|
||
continue
|
||
entries.append(
|
||
{
|
||
"pm2": pm2,
|
||
"script": "youtube2.py",
|
||
"label": f"YouTube {ch.get('name', cid)}",
|
||
}
|
||
)
|
||
family_value = (family or "").strip().lower() or None
|
||
if family_value:
|
||
entries = [e for e in entries if _process_monitor_family_matches(e, family_value)]
|
||
if not light and not family_value:
|
||
return {"entries": entries}
|
||
results = await asyncio.gather(
|
||
*(_process_monitor_snapshot(e, light=bool(light)) for e in entries),
|
||
return_exceptions=True,
|
||
)
|
||
snapshots: list[dict] = []
|
||
for entry, result in zip(entries, results):
|
||
if isinstance(result, Exception):
|
||
snapshots.append(
|
||
{
|
||
"pm2": entry["pm2"],
|
||
"script": entry.get("script", ""),
|
||
"label": entry.get("label", entry["pm2"]),
|
||
"process_status": "error",
|
||
"business_status": "error",
|
||
"business_note": "Failed to inspect process state",
|
||
"is_pushing": False,
|
||
}
|
||
)
|
||
else:
|
||
snapshots.append(result)
|
||
return {
|
||
"entries": snapshots,
|
||
"family": family_value or "all",
|
||
"light": bool(light),
|
||
}
|
||
|
||
|
||
# ---------------- 首页 ----------------
|
||
@app.get("/")
|
||
async def index():
|
||
return FileResponse(BASE_DIR / "index.html")
|