143 lines
4.2 KiB
Python
143 lines
4.2 KiB
Python
from fastapi import FastAPI
|
||
from fastapi.responses import FileResponse, JSONResponse
|
||
from pathlib import Path
|
||
import asyncio
|
||
import re
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
|
||
app = FastAPI(title="HDMI 录制控制台")
|
||
|
||
# ---------------- 配置 ----------------
|
||
BOT_NAME = "hdmi"
|
||
BASE_DIR = Path(__file__).parent
|
||
DEFAULT_LOG_DIR = BASE_DIR.parent / ".pm2" / "logs"
|
||
MAX_LOG_LINES = 300
|
||
|
||
# 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_hdmi_info() -> dict:
|
||
describe_raw = await run_pm2(f"describe {BOT_NAME}")
|
||
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"{BOT_NAME}-out.log")
|
||
if not err_path:
|
||
err_path = str(DEFAULT_LOG_DIR / f"{BOT_NAME}-error.log")
|
||
|
||
return {
|
||
"process_status": status,
|
||
"out_path": out_path,
|
||
"err_path": err_path,
|
||
}
|
||
|
||
# ---------------- 路由 ----------------
|
||
@app.get("/start")
|
||
async def start():
|
||
output = await run_pm2(f"start {BOT_NAME}")
|
||
return JSONResponse({"output": output})
|
||
|
||
@app.get("/stop")
|
||
async def stop():
|
||
output = await run_pm2(f"stop {BOT_NAME}")
|
||
return JSONResponse({"output": output})
|
||
|
||
@app.get("/restart")
|
||
async def restart():
|
||
output = await run_pm2(f"restart {BOT_NAME}")
|
||
return JSONResponse({"output": output})
|
||
|
||
@app.get("/status")
|
||
async def status():
|
||
full_status_raw = await run_pm2("status")
|
||
full_status = strip_ansi_codes(full_status_raw) # 去除颜色代码，前端显示更干净
|
||
|
||
info = await get_hdmi_info()
|
||
|
||
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")
|