grok

web.py

@@ -1,71 +1,142 @@
 from fastapi import FastAPI
 from fastapi.responses import FileResponse, JSONResponse
-from fastapi.staticfiles import StaticFiles
-import subprocess
 from pathlib import Path
+import asyncio
+import re
 from fastapi.middleware.cors import CORSMiddleware
 
-app = FastAPI()
+app = FastAPI(title="HDMI 录制控制台")
 
 # ---------------- 配置 ----------------
-BOT_NAME = "hdmi"  # PM2 进程名
+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=["*"],
+    allow_methods=["GET"],
     allow_headers=["*"],
 )
 
-# 静态文件
-app.mount("/static", StaticFiles(directory=BASE_DIR), name="static")
+# ---------------- 工具 ----------------
+def strip_ansi_codes(text: str) -> str:
+    return re.sub(r'\x1b\[[0-9;]*m', '', text)
 
-# ---------------- PM2 控制 ----------------
-def pm2_cmd(cmd: str):
+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:
-        out = subprocess.check_output(f"pm2 {cmd}", shell=True, stderr=subprocess.STDOUT, text=True)
-        return out
-    except subprocess.CalledProcessError as e:
-        return e.output
+        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 命令超时"
 
-def read_log(log_type="out", lines=50):
-    log_file = BASE_DIR.parent / f".pm2/logs/{BOT_NAME}-{log_type}.log"
+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") as f:
-            content = f.readlines()[-lines:]
-            return "".join(content)
-    except FileNotFoundError:
-        return f"未找到日志文件: {log_file}"
+        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")
-def start():
-    out = pm2_cmd(f"start {BOT_NAME}")
-    return JSONResponse({"action": "start", "output": out})
+async def start():
+    output = await run_pm2(f"start {BOT_NAME}")
+    return JSONResponse({"output": output})
 
 @app.get("/stop")
-def stop():
-    out = pm2_cmd(f"stop {BOT_NAME} --no-autorestart")
-    return JSONResponse({"action": "stop", "output": out})
+async def stop():
+    output = await run_pm2(f"stop {BOT_NAME}")
+    return JSONResponse({"output": output})
 
 @app.get("/restart")
-def restart():
-    out = pm2_cmd(f"restart {BOT_NAME}")
-    return JSONResponse({"action": "restart", "output": out})
+async def restart():
+    output = await run_pm2(f"restart {BOT_NAME}")
+    return JSONResponse({"output": output})
 
 @app.get("/status")
-def status():
-    status = pm2_cmd(f"status {BOT_NAME}")
-    log = read_log("out", 50)
-    error_log = read_log("error", 50)
+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({
-        "status": status,
-        "recent_log": log,
-        "recent_error": error_log
+        "raw_status": full_status,
+        "process_status": info["process_status"],
+        "recent_log": recent_log,
+        "recent_error": recent_error,
     })
 
-# ---------------- HTML ----------------
 @app.get("/")
-def index():
+async def index():
     return FileResponse(BASE_DIR / "index.html")