's'

web.py

@@ -1,142 +0,0 @@
-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")