353 lines
12 KiB
Python
353 lines
12 KiB
Python
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") |