377 lines
14 KiB
Python
377 lines
14 KiB
Python
"""
|
||
进程编排：默认使用内置 Supervisor（Windows / 免 PM2）。
|
||
设置环境变量 WEB2_SUPERVISOR=pm2 时使用 PM2（与旧版 Linux 工作流兼容）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import os
|
||
import re
|
||
import shutil
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
from typing import Any, Callable, Optional
|
||
|
||
import psutil
|
||
|
||
from web2_proxy_config import merge_child_env
|
||
|
||
|
||
def use_builtin() -> bool:
|
||
return os.environ.get("WEB2_SUPERVISOR", "builtin").strip().lower() != "pm2"
|
||
|
||
|
||
def strip_ansi_codes(text: str) -> str:
|
||
return re.sub(r"\x1b\[[0-9;]*m", "", text)
|
||
|
||
|
||
def _looks_like_null_device(path: str | Path | None) -> bool:
|
||
if not path:
|
||
return True
|
||
norm = str(path).replace("\\", "/").lower()
|
||
return "/dev/null" in norm or norm.endswith("/nul") or norm == "nul"
|
||
|
||
|
||
def _truncate_text_log(path: str | Path | None) -> None:
|
||
"""对齐 sh2：每次 start 清空日志，避免旧 mtime 触发 stale/心跳过期。"""
|
||
if _looks_like_null_device(path):
|
||
return
|
||
log_path = Path(str(path))
|
||
try:
|
||
log_path.parent.mkdir(parents=True, exist_ok=True)
|
||
with log_path.open("w", encoding="utf-8", errors="ignore"):
|
||
pass
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def _child_env_for_process(process_name: str, base_dir: Path) -> dict[str, str]:
|
||
extra: dict[str, str] = {
|
||
"LIVE_PM2_PROCESS_NAME": process_name,
|
||
"PYTHONUNBUFFERED": "1",
|
||
"PYTHONIOENCODING": "utf-8",
|
||
}
|
||
relay_ch = ""
|
||
try:
|
||
from web2_relay_pro import channel_id_for_pm2_name
|
||
|
||
relay_ch = channel_id_for_pm2_name(base_dir / "config", process_name)
|
||
except Exception:
|
||
relay_ch = ""
|
||
if not relay_ch and process_name.startswith("youtube2__"):
|
||
relay_ch = process_name[len("youtube2__") :]
|
||
if relay_ch:
|
||
extra["RELAY_CHANNEL_ID"] = relay_ch
|
||
return merge_child_env(extra)
|
||
|
||
|
||
class BuiltinSupervisor:
|
||
"""用本地 PID + 日志文件管理子进程，不依赖 Node/PM2。"""
|
||
|
||
def __init__(self, base_dir: Path, console_script: str, script_resolver: Callable[[str], Optional[str]]):
|
||
self.base_dir = base_dir.resolve()
|
||
self.console_script = console_script
|
||
self.console_stem = Path(console_script).stem
|
||
self._script_for = script_resolver
|
||
self.log_dir = self.base_dir / ".pm2" / "logs"
|
||
self.log_dir.mkdir(parents=True, exist_ok=True)
|
||
self.state_path = self.base_dir / ".pm2" / "supervisor_state.json"
|
||
self._io_lock = asyncio.Lock()
|
||
|
||
def _read_state(self) -> dict[str, Any]:
|
||
if not self.state_path.exists():
|
||
return {}
|
||
try:
|
||
return json.loads(self.state_path.read_text(encoding="utf-8"))
|
||
except Exception:
|
||
return {}
|
||
|
||
def _write_state(self, state: dict[str, Any]) -> None:
|
||
self.state_path.parent.mkdir(parents=True, exist_ok=True)
|
||
self.state_path.write_text(json.dumps(state, indent=2, ensure_ascii=False), encoding="utf-8")
|
||
|
||
def _pid_alive(self, pid: int | None) -> bool:
|
||
if pid is None or pid <= 0:
|
||
return False
|
||
try:
|
||
return psutil.Process(pid).is_running()
|
||
except (psutil.NoSuchProcess, psutil.AccessDenied):
|
||
return False
|
||
|
||
def _log_paths(self, name: str) -> tuple[Path, Path]:
|
||
out_p = self.log_dir / f"{name}-out.log"
|
||
err_p = self.log_dir / f"{name}-error.log"
|
||
return out_p, err_p
|
||
|
||
def _build_cmd(self, script: str) -> Optional[list[str]]:
|
||
path = self.base_dir / script
|
||
if not path.is_file():
|
||
return None
|
||
ext = path.suffix.lower()
|
||
if ext == ".py":
|
||
return [sys.executable, str(path)]
|
||
if ext == ".bat":
|
||
comspec = os.environ.get("COMSPEC") or "cmd.exe"
|
||
return [comspec, "/c", str(path)]
|
||
if ext == ".sh":
|
||
bash = shutil.which("bash") or shutil.which("sh")
|
||
if not bash:
|
||
return None
|
||
return [bash, str(path)]
|
||
return None
|
||
|
||
async def start(self, process_name: str) -> str:
|
||
if process_name == self.console_stem:
|
||
return "控制台已由本程序托管，无需单独启动。"
|
||
|
||
script = self._script_for(process_name)
|
||
if not script:
|
||
return f"未知进程名: {process_name}"
|
||
|
||
argv = self._build_cmd(script)
|
||
if argv is None:
|
||
path = self.base_dir / script
|
||
if not path.is_file():
|
||
return f"脚本不存在: {script}"
|
||
if script.endswith(".sh"):
|
||
return "未找到 bash/sh，无法运行 .sh（可改用 obs*.bat 或安装 Git for Windows）"
|
||
return f"无法构建启动命令: {script}"
|
||
|
||
async with self._io_lock:
|
||
state = self._read_state()
|
||
entry = state.get(process_name, {})
|
||
pid = entry.get("pid")
|
||
if self._pid_alive(pid):
|
||
return f"进程已在运行中（PID {pid}）。"
|
||
|
||
out_p, err_p = self._log_paths(process_name)
|
||
out_p.parent.mkdir(parents=True, exist_ok=True)
|
||
_truncate_text_log(out_p)
|
||
_truncate_text_log(err_p)
|
||
out_f = open(out_p, "a", encoding="utf-8", errors="ignore")
|
||
err_f = open(err_p, "a", encoding="utf-8", errors="ignore")
|
||
try:
|
||
proc = await asyncio.create_subprocess_exec(
|
||
*argv,
|
||
cwd=str(self.base_dir),
|
||
stdout=out_f,
|
||
stderr=err_f,
|
||
creationflags=0,
|
||
env=_child_env_for_process(process_name, self.base_dir),
|
||
)
|
||
except Exception as e:
|
||
out_f.close()
|
||
err_f.close()
|
||
return f"启动失败: {e}"
|
||
out_f.close()
|
||
err_f.close()
|
||
|
||
pid = proc.pid
|
||
state[process_name] = {
|
||
"pid": pid,
|
||
"script": script,
|
||
"out_log": str(out_p),
|
||
"err_log": str(err_p),
|
||
"started_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
|
||
}
|
||
self._write_state(state)
|
||
return f"已启动 {script}（PID {pid}），日志见 .pm2/logs"
|
||
|
||
async def stop(self, process_name: str) -> str:
|
||
if process_name == self.console_stem:
|
||
return "请关闭桌面应用窗口以结束控制台。"
|
||
|
||
async with self._io_lock:
|
||
state = self._read_state()
|
||
entry = state.get(process_name, {})
|
||
pid = entry.get("pid")
|
||
|
||
if not self._pid_alive(pid):
|
||
state.pop(process_name, None)
|
||
self._write_state(state)
|
||
return "进程未在运行（或已退出）。"
|
||
|
||
try:
|
||
parent = psutil.Process(pid)
|
||
children = parent.children(recursive=True)
|
||
for c in children:
|
||
try:
|
||
c.terminate()
|
||
except (psutil.NoSuchProcess, psutil.AccessDenied):
|
||
pass
|
||
try:
|
||
parent.terminate()
|
||
except (psutil.NoSuchProcess, psutil.AccessDenied):
|
||
pass
|
||
gone, alive = psutil.wait_procs([parent] + children, timeout=5)
|
||
for p in alive:
|
||
try:
|
||
p.kill()
|
||
except (psutil.NoSuchProcess, psutil.AccessDenied):
|
||
pass
|
||
except psutil.NoSuchProcess:
|
||
pass
|
||
|
||
state.pop(process_name, None)
|
||
self._write_state(state)
|
||
return f"已停止进程（原 PID {pid}）。"
|
||
|
||
async def restart(self, process_name: str) -> str:
|
||
if process_name == self.console_stem:
|
||
return "控制台由桌面应用托管，请关闭窗口后重新打开应用。"
|
||
out = await self.stop(process_name)
|
||
await asyncio.sleep(0.5)
|
||
out2 = await self.start(process_name)
|
||
return f"{out}\n{out2}"
|
||
|
||
async def full_status_text(self) -> str:
|
||
lines = [
|
||
"┌──────────────────┬────────────┬────────────────────────────────────────┐",
|
||
"│ 进程名 │ 状态 │ 说明 │",
|
||
"├──────────────────┼────────────┼────────────────────────────────────────┤",
|
||
]
|
||
async with self._io_lock:
|
||
state = self._read_state()
|
||
lines.append(
|
||
f"│ {self.console_stem:<16} │ {'online':<10} │ 控制台（本机 uvicorn） │"
|
||
)
|
||
for name, entry in sorted(state.items()):
|
||
pid = entry.get("pid")
|
||
alive = self._pid_alive(pid)
|
||
st = "online" if alive else "stopped"
|
||
scr = (entry.get("script") or "")[:38]
|
||
lines.append(f"│ {name:<16} │ {st:<10} │ {scr:<38} │")
|
||
lines.append(
|
||
"└──────────────────┴────────────┴────────────────────────────────────────┘"
|
||
)
|
||
return "\n".join(lines)
|
||
|
||
async def get_process_info(self, process_name: str) -> dict[str, Any]:
|
||
if process_name == self.console_stem:
|
||
out_p, err_p = self._log_paths(process_name)
|
||
return {
|
||
"process_status": "online",
|
||
"out_path": str(out_p),
|
||
"err_path": str(err_p),
|
||
}
|
||
|
||
async with self._io_lock:
|
||
state = self._read_state()
|
||
entry = state.get(process_name, {})
|
||
pid = entry.get("pid")
|
||
out_path = entry.get("out_log")
|
||
err_path = entry.get("err_log")
|
||
|
||
if not out_path:
|
||
out_path = str(self._log_paths(process_name)[0])
|
||
if not err_path:
|
||
err_path = str(self._log_paths(process_name)[1])
|
||
|
||
if not self._pid_alive(pid):
|
||
return {
|
||
"process_status": "stopped" if entry else "not_found",
|
||
"pid": pid if entry else None,
|
||
"out_path": out_path,
|
||
"err_path": err_path,
|
||
}
|
||
|
||
return {
|
||
"process_status": "online",
|
||
"pid": pid,
|
||
"out_path": out_path,
|
||
"err_path": err_path,
|
||
}
|
||
|
||
|
||
async def run_pm2_shell(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_cross_platform(process_name: str) -> None:
|
||
"""按 LIVE_PM2_PROCESS_NAME 环境变量或命令行匹配业务进程名，结束相关 ffmpeg。"""
|
||
|
||
def _sync_kill() -> None:
|
||
needle = (process_name or "").strip().lower()
|
||
for proc in psutil.process_iter(["pid", "name", "cmdline"]):
|
||
try:
|
||
name = (proc.info.get("name") or "").lower()
|
||
if "ffmpeg" not in name:
|
||
continue
|
||
cmdline = proc.info.get("cmdline") or []
|
||
flat = " ".join(str(x) for x in cmdline).lower()
|
||
env_name = ""
|
||
try:
|
||
env_name = str(proc.environ().get("LIVE_PM2_PROCESS_NAME") or "").strip().lower()
|
||
except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
|
||
env_name = ""
|
||
if needle:
|
||
if env_name != needle and needle not in flat:
|
||
continue
|
||
p = psutil.Process(proc.info["pid"])
|
||
p.kill()
|
||
except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
|
||
continue
|
||
|
||
await asyncio.to_thread(_sync_kill)
|
||
await asyncio.sleep(0.35)
|
||
|
||
|
||
async def get_process_info_pm2(
|
||
process: str,
|
||
default_log_dir: Path,
|
||
strip_fn: Callable[[str], str],
|
||
) -> dict[str, Any]:
|
||
describe_raw = await run_pm2_shell(f"describe {process}", timeout=10.0)
|
||
describe_clean = strip_fn(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_fn(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,
|
||
}
|