's'

web.py

@@ -0,0 +1,353 @@
+from fastapi import FastAPI, Query, Request
+from fastapi.responses import FileResponse, JSONResponse
+from pathlib import Path
+import asyncio
+import re
+from typing import Optional
+from fastapi.middleware.cors import CORSMiddleware
+
+app = FastAPI(title="多进程录制控制台")
+
+# ---------------- 配置（根目录结构） ----------------
+BASE_DIR = Path(__file__).parent
+CONFIG_DIR = BASE_DIR / "config"
+DEFAULT_LOG_DIR = BASE_DIR / ".pm2" / "logs"
+
+MAX_LOG_LINES = 300
+
+
+def _label_short(prefix: str) -> str:
+    return {"tiktok": "TikTok", "youtube": "YouTube", "obs": "OBS", "web": "Web 控制台"}.get(prefix, prefix)
+
+
+def discover_script_entries() -> tuple:
+    """扫描项目根目录自动识别入口：tiktok*.py、youtube*.py、obs*.sh，以及当前本文件（Web 控制台）。PM2 名 = 主文件名（无扩展名）。"""
+    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")})
+
+    for path in sorted(base.glob("obs*.sh")):
+        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_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)
+
+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]:
+    for p in PROCESS_MONITOR:
+        if p["pm2"] == pm2:
+            return p["script"]
+    return None
+
+# CORS
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_methods=["GET", "POST"],
+    allow_headers=["*"],
+)
+
+# ---------------- 工具函数 ----------------
+def strip_ansi_codes(text: str) -> str:
+    return re.sub(r'\x1b\[[0-9;]*m', '', text)
+
+
+async def run_pm2(cmd: str, timeout: float = 15.0) -> str:
+    process = await asyncio.create_subprocess_shell(
+        f"pm2 {cmd}",
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+    try:
+        stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout)
+        result = (stdout or stderr).decode(errors="ignore").strip()
+        return result if result else "命令执行完成（无输出）"
+    except asyncio.TimeoutError:
+        process.kill()
+        await process.wait()
+        return "ERROR: PM2 命令超时，已强制终止"
+
+
+async def force_kill_ffmpeg(process_name: str):
+    """尝试强杀与该进程相关的 ffmpeg"""
+    cmds = [
+        # 优先杀带有进程名的 ffmpeg（最精准）
+        f"pkill -f 'ffmpeg.*{process_name}' || true",
+        # 再杀所有 ffmpeg（兜底，比较暴力）
+        "killall -9 ffmpeg 2>/dev/null || true",
+    ]
+
+    for cmd in cmds:
+        proc = await asyncio.create_subprocess_shell(
+            cmd,
+            stdout=asyncio.subprocess.DEVNULL,
+            stderr=asyncio.subprocess.DEVNULL,
+        )
+        await proc.wait()
+        await asyncio.sleep(0.4)  # 稍微间隔，避免竞争
+
+
+async def read_log_path(path: str, lines: int) -> str:
+    if not path:
+        return "无日志路径\n"
+    if "/dev/null" in path:
+        return "日志输出被禁用（PM2 配置为无日志）\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"
+
+
+async def get_process_info(process: str) -> dict:
+    describe_raw = await run_pm2(f"describe {process}", timeout=10.0)
+    describe_clean = strip_ansi_codes(describe_raw)
+
+    if any(x in describe_clean.lower() for x in ["no such process", "unknown process", "not found"]) or not describe_clean.strip():
+        return {
+            "process_status": "not_found",
+            "out_path": None,
+            "err_path": None,
+        }
+
+    status = "unknown"
+    out_path = None
+    err_path = None
+
+    for line in describe_raw.splitlines():
+        line_clean = strip_ansi_codes(line)
+        if line_clean.count('│') < 2:
+            continue
+        parts = [p.strip() for p in line_clean.split('│') if p.strip()]
+        if len(parts) >= 2:
+            key = parts[0].lower()
+            value = parts[1]
+            if key == "status":
+                status = value.lower()
+            elif key == "out_log_path":
+                out_path = value
+            elif key == "err_log_path":
+                err_path = value
+
+    if not out_path:
+        out_path = str(DEFAULT_LOG_DIR / f"{process}-out.log")
+    if not err_path:
+        err_path = str(DEFAULT_LOG_DIR / f"{process}-error.log")
+
+    return {
+        "process_status": status,
+        "out_path": out_path,
+        "err_path": err_path,
+    }
+
+# ---------------- 配置路由（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)
+    
+    config_file = CONFIG_DIR / "youtube.ini"
+    try:
+        CONFIG_DIR.mkdir(parents=True, exist_ok=True)
+        config_file.write_text(content, encoding="utf-8")
+        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)
+        config_file.write_text(content, encoding="utf-8")
+        return {"message": "配置保存成功"}
+    except Exception as e:
+        return JSONResponse({"error": f"保存失败: {str(e)}"}, status_code=500)
+
+# ---------------- PM2 控制路由 ----------------
+@app.get("/start")
+async def start(process: str = Query(..., description="进程名")):
+    if process not in VALID_PROCESSES:
+        return JSONResponse({"output": f"无效进程名: {process}"}, status_code=400)
+    output = await run_pm2(f"start {process}")
+    return JSONResponse({"output": output})
+
+
+@app.get("/stop")
+async def stop(process: str = Query(..., description="进程名")):
+    if process not in VALID_PROCESSES:
+        return JSONResponse({"output": f"无效进程名: {process}"}, status_code=400)
+
+    # 第一步：尝试优雅停止
+    pm2_output = await run_pm2(f"stop {process}")
+
+    # 第二步：等待 PM2 信号传播（通常 1~3 秒）
+    await asyncio.sleep(2.0)
+
+    # 第三步：录制类进程才强杀 ffmpeg（web* 控制台不杀 ffmpeg）
+    if not process.startswith("web"):
+        await force_kill_ffmpeg(process)
+
+    # 第四步：再等一小会儿，确认
+    await asyncio.sleep(1.0)
+
+    return JSONResponse({
+        "message": "已执行 stop" + ("" if process.startswith("web") else " + 强杀 ffmpeg"),
+        "pm2_output": pm2_output,
+        "note": "如果仍有残留，可多次点击停止或检查 pm2 logs"
+    })
+
+
+@app.get("/restart")
+async def restart(process: str = Query(..., description="进程名")):
+    if process not in VALID_PROCESSES:
+        return JSONResponse({"output": f"无效进程名: {process}"}, status_code=400)
+    output = await run_pm2(f"restart {process}")
+    if not process.startswith("web"):
+        await force_kill_ffmpeg(process)
+    return JSONResponse({"output": output})
+
+
+@app.get("/status")
+async def status(process: str = Query(..., description="进程名")):
+    if process not in VALID_PROCESSES:
+        return JSONResponse({
+            "raw_status": "",
+            "process_status": "invalid",
+            "recent_log": f"无效进程名: {process}",
+            "recent_error": ""
+        })
+
+    full_status_raw = await run_pm2("status", timeout=8.0)
+    full_status = strip_ansi_codes(full_status_raw)
+    info = await get_process_info(process)
+    recent_log = await read_log_path(info["out_path"], MAX_LOG_LINES)
+    recent_error = await read_log_path(info["err_path"], MAX_LOG_LINES)
+
+    if info["process_status"] == "not_found":
+        recent_log = "进程未注册到 PM2（可能被 pm2 delete 删除或从未保存）\n"
+        recent_error = ""
+
+    return JSONResponse({
+        "raw_status": full_status,
+        "process_status": info["process_status"],
+        "recent_log": recent_log,
+        "recent_error": recent_error,
+    })
+
+
+# ---------------- 进程列表（供前端同步，由 discover_script_entries 自动扫描） ----------------
+@app.get("/process_monitor")
+async def process_monitor():
+    return {
+        "entries": [
+            {"pm2": p["pm2"], "script": p["script"], "label": p["label"]}
+            for p in PROCESS_MONITOR
+        ]
+    }
+
+
+# ---------------- 首页 ----------------
+@app.get("/")
+async def index():
+    return FileResponse(BASE_DIR / "index.html")