from fastapi import FastAPI, Query, Request 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 CONFIG_DIR = BASE_DIR / "config" DEFAULT_LOG_DIR = BASE_DIR / ".pm2" / "logs" MAX_LOG_LINES = 300 # 支持的进程列表 VALID_PROCESSES = {"tiktok", "youtube", "obs"} # 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="进程名")): if process != "youtube": return JSONResponse({"error": "仅 youtube 支持此配置编辑"}, 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="进程名")): if process != "youtube": return JSONResponse({"error": "仅 youtube 支持此配置编辑"}, 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 ["youtube", "tiktok"]: return JSONResponse({"error": "仅 youtube 和 tiktok 支持此配置编辑"}, 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 ["youtube", "tiktok"]: return JSONResponse({"error": "仅 youtube 和 tiktok 支持此配置编辑"}, 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 await force_kill_ffmpeg(process) # 第四步：再等一小会儿，确认 await asyncio.sleep(1.0) return JSONResponse({ "message": "已执行 stop + 强杀 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}") # 重启后也建议清理一次残留（视情况可选） 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, }) # ---------------- 首页 ---------------- @app.get("/") async def index(): return FileResponse(BASE_DIR / "index.html")