's'

web.py

@@ -0,0 +1,146 @@
+from fastapi import FastAPI, Query
+from fastapi.responses import FileResponse, JSONResponse
+from pathlib import Path
+import asyncio
+import re
+from fastapi.middleware.cors import CORSMiddleware
+
+app = FastAPI(title="多进程录制控制台")
+
+# ---------------- 配置 ----------------
+BASE_DIR = Path(__file__).parent
+DEFAULT_LOG_DIR = BASE_DIR.parent / ".pm2" / "logs"
+MAX_LOG_LINES = 300
+
+# 支持的进程列表（可扩展）
+VALID_PROCESSES = {"tiktok", "youtube", "obs"}
+
+# CORS
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_methods=["GET"],
+    allow_headers=["*"],
+)
+
+# ---------------- 工具 ----------------
+def strip_ansi_codes(text: str) -> str:
+    return re.sub(r'\x1b\[[0-9;]*m', '', text)
+
+async def run_pm2(cmd: str) -> 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=15.0)
+        return (stdout or stderr).decode(errors="ignore").strip()
+    except asyncio.TimeoutError:
+        process.kill()
+        return "ERROR: PM2 命令超时"
+
+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}")
+    describe_clean = strip_ansi_codes(describe_raw)
+    if ("No such process" in describe_clean or
+        "Unknown process" in describe_clean or
+        "not found" in describe_clean.lower() 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
+    # 备用默认路径（仅当 describe 未提供时）
+    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,
+    }
+
+# ---------------- 路由 ----------------
+@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)
+    output = await run_pm2(f"stop {process}")
+    return JSONResponse({"output": output})
+
+@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}")
+    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")
+    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,
+    })
+
+@app.get("/")
+async def index():
+    return FileResponse(BASE_DIR / "index.html")