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")