417 lines
14 KiB
Python
417 lines
14 KiB
Python
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import sys
|
||
from contextlib import asynccontextmanager
|
||
from pathlib import Path
|
||
from typing import Optional
|
||
|
||
from fastapi import FastAPI, Query, Request
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
from fastapi.responses import FileResponse, JSONResponse
|
||
from fastapi.staticfiles import StaticFiles
|
||
|
||
from src.web_process_backend import WebProcessBackend, force_kill_ffmpeg, strip_ansi_codes
|
||
|
||
# ---------------- 配置（根目录结构） ----------------
|
||
BASE_DIR = Path(__file__).parent.resolve()
|
||
CONFIG_DIR = BASE_DIR / "config"
|
||
DEFAULT_LOG_DIR = BASE_DIR / ".pm2" / "logs"
|
||
CONSOLE_OUT = BASE_DIR / "web-console" / "out"
|
||
|
||
MAX_LOG_LINES = 300
|
||
|
||
process_backend = WebProcessBackend(BASE_DIR, DEFAULT_LOG_DIR)
|
||
|
||
|
||
def _label_short(prefix: str) -> str:
|
||
return {
|
||
"tiktok": "TikTok",
|
||
"youtube": "YouTube",
|
||
"douyin_youtube": "抖音→YouTube",
|
||
"obs": "OBS",
|
||
"web": "Web 控制台",
|
||
}.get(prefix, prefix)
|
||
|
||
|
||
def _sort_script_entries(entries: list[dict], web_script_name: str) -> list[dict]:
|
||
"""下拉列表顺序：抖音→YouTube → YouTube → TikTok → OBS → Web 控制台（默认首选 YouTube 业务）。"""
|
||
|
||
def sort_key(e: dict) -> tuple:
|
||
script = e["script"]
|
||
st = Path(script).stem.lower()
|
||
if st.startswith("douyin_youtube"):
|
||
return (0, script)
|
||
if script.endswith(".py") and st.startswith("youtube"):
|
||
return (1, script)
|
||
if st.startswith("tiktok"):
|
||
return (2, script)
|
||
if st.startswith("obs"):
|
||
return (3, script)
|
||
if script == web_script_name or st == "web":
|
||
return (99, script)
|
||
return (50, script)
|
||
|
||
return sorted(entries, key=sort_key)
|
||
|
||
|
||
def discover_script_entries() -> tuple:
|
||
"""扫描根目录：tiktok*.py、youtube*.py、douyin_youtube*.py；OBS 在 Windows 用 obs*.bat/.cmd，在类 Unix 用 obs*.sh。"""
|
||
base = BASE_DIR
|
||
web_path = Path(__file__).resolve()
|
||
entries: list[dict] = []
|
||
seen_stems: set[str] = set()
|
||
|
||
def add(script_name: str, label_key: str) -> None:
|
||
stem = Path(script_name).stem
|
||
if stem in seen_stems:
|
||
return
|
||
seen_stems.add(stem)
|
||
entries.append({"script": script_name, "label": _label_short(label_key)})
|
||
|
||
for path in sorted(base.glob("tiktok*.py")):
|
||
if path.is_file() and path.resolve() != web_path:
|
||
add(path.name, "tiktok")
|
||
|
||
for path in sorted(base.glob("youtube*.py")):
|
||
if path.is_file() and path.resolve() != web_path:
|
||
add(path.name, "youtube")
|
||
|
||
for path in sorted(base.glob("douyin_youtube*.py")):
|
||
if path.is_file() and path.resolve() != web_path:
|
||
add(path.name, "douyin_youtube")
|
||
|
||
if sys.platform == "win32":
|
||
for path in sorted(list(base.glob("obs*.bat")) + list(base.glob("obs*.cmd"))):
|
||
if path.is_file():
|
||
add(path.name, "obs")
|
||
else:
|
||
for path in sorted(base.glob("obs*.sh")):
|
||
if path.is_file():
|
||
add(path.name, "obs")
|
||
|
||
entries.append({"script": web_path.name, "label": _label_short("web")})
|
||
entries = _sort_script_entries(entries, web_path.name)
|
||
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:
|
||
if not script.endswith(".py"):
|
||
return False
|
||
s = _stem(script).lower()
|
||
return s.startswith("youtube") or s.startswith("douyin_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
|
||
|
||
|
||
@asynccontextmanager
|
||
async def lifespan(app: FastAPI):
|
||
await process_backend.refresh_mode()
|
||
yield
|
||
|
||
|
||
app = FastAPI(title="多进程录制控制台", lifespan=lifespan)
|
||
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
allow_origins=["*"],
|
||
allow_methods=["GET", "POST"],
|
||
allow_headers=["*"],
|
||
)
|
||
|
||
|
||
async def read_log_path(path: Optional[str], lines: int) -> str:
|
||
if not path:
|
||
return "无日志路径\n"
|
||
norm = path.replace("\\", "/").lower()
|
||
if "/dev/null" in path or norm.endswith("/nul") or norm == "nul":
|
||
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 OSError as e:
|
||
return f"读取日志失败 ({path}): {str(e)}\n"
|
||
|
||
|
||
async def get_process_info(process: str) -> dict:
|
||
return await process_backend.parse_pm2_describe(process)
|
||
|
||
|
||
# ---------------- 服务信息（前端可展示 PM2 / 本地后端） ----------------
|
||
@app.get("/health")
|
||
async def health():
|
||
return {"status": "ok"}
|
||
|
||
|
||
@app.get("/server_info")
|
||
async def server_info(refresh: bool = Query(False, description="为 true 时重新检测 PM2 是否可用")):
|
||
if refresh:
|
||
await process_backend.refresh_mode()
|
||
else:
|
||
await process_backend.ensure_mode()
|
||
return {
|
||
"platform": sys.platform,
|
||
"process_backend": process_backend.mode,
|
||
"console_built": (CONSOLE_OUT / "index.html").is_file(),
|
||
"message": "PM2 可用时使用 PM2；否则使用本地 PID 表（无需 npm i -g pm2）。",
|
||
"deployment": {
|
||
"runtime_env_example": "config/runtime.env.example",
|
||
"runtime_env_active": (CONFIG_DIR / "runtime.env").is_file(),
|
||
"docker_compose_default_port": 8101,
|
||
"process_dropdown_order": "douyin_youtube → youtube → tiktok → obs → web",
|
||
},
|
||
}
|
||
|
||
|
||
# ---------------- 配置路由（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 / 抖音→YouTube 类脚本支持 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 OSError 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 / 抖音→YouTube 类脚本支持 youtube.ini 编辑"}, status_code=400
|
||
)
|
||
|
||
try:
|
||
data = await request.json()
|
||
content = data.get("content", "")
|
||
except Exception:
|
||
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 OSError as e:
|
||
return JSONResponse({"error": f"保存失败: {str(e)}"}, status_code=500)
|
||
|
||
|
||
# ---------------- 配置路由（URL_config.ini） ----------------
|
||
@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 / douyin_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 OSError 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 / douyin_youtube*.py 对应进程支持 URL 配置编辑",
|
||
},
|
||
status_code=400,
|
||
)
|
||
|
||
try:
|
||
data = await request.json()
|
||
content = data.get("content", "")
|
||
except Exception:
|
||
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 OSError as e:
|
||
return JSONResponse({"error": f"保存失败: {str(e)}"}, status_code=500)
|
||
|
||
|
||
def _not_found_hint(process: str) -> str:
|
||
if process_backend.mode == "local":
|
||
return (
|
||
f"进程「{process}」未在本地注册表中；请点击「启动」由控制台直接拉起脚本，"
|
||
"或安装 PM2 后先用 pm2 start 注册同名进程。"
|
||
)
|
||
return "进程未注册到 PM2（可能被 pm2 delete 删除或从未保存）"
|
||
|
||
|
||
# ---------------- 进程控制 ----------------
|
||
@app.get("/start")
|
||
async def start(process: str = Query(..., description="进程名")):
|
||
if process not in VALID_PROCESSES:
|
||
return JSONResponse({"output": f"无效进程名: {process}"}, status_code=400)
|
||
sc = script_for_pm2(process)
|
||
if not sc:
|
||
return JSONResponse({"output": "内部错误：找不到脚本映射"}, status_code=500)
|
||
output = await process_backend.start(process, sc)
|
||
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 process_backend.stop(process)
|
||
|
||
await asyncio.sleep(2.0)
|
||
|
||
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)
|
||
sc = script_for_pm2(process)
|
||
if not sc:
|
||
return JSONResponse({"output": "内部错误：找不到脚本映射"}, status_code=500)
|
||
output = await process_backend.restart(process, sc)
|
||
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 process_backend.full_status_raw()
|
||
full_status = strip_ansi_codes(full_status_raw)
|
||
info = await get_process_info(process)
|
||
recent_log = await read_log_path(info.get("out_path"), MAX_LOG_LINES)
|
||
recent_error = await read_log_path(info.get("err_path"), MAX_LOG_LINES)
|
||
|
||
if info["process_status"] == "not_found":
|
||
recent_log = _not_found_hint(process) + "\n"
|
||
recent_error = ""
|
||
|
||
return JSONResponse(
|
||
{
|
||
"raw_status": full_status,
|
||
"process_status": info["process_status"],
|
||
"recent_log": recent_log,
|
||
"recent_error": recent_error,
|
||
}
|
||
)
|
||
|
||
|
||
@app.get("/process_monitor")
|
||
async def process_monitor():
|
||
await process_backend.ensure_mode()
|
||
return {
|
||
"entries": [
|
||
{"pm2": p["pm2"], "script": p["script"], "label": p["label"]}
|
||
for p in PROCESS_MONITOR
|
||
],
|
||
"process_backend": process_backend.mode,
|
||
}
|
||
|
||
|
||
@app.get("/favicon.ico", include_in_schema=False)
|
||
async def favicon():
|
||
ico = CONSOLE_OUT / "favicon.ico"
|
||
if ico.is_file():
|
||
return FileResponse(ico)
|
||
return JSONResponse({}, status_code=404)
|
||
|
||
|
||
@app.get("/")
|
||
async def index():
|
||
next_index = CONSOLE_OUT / "index.html"
|
||
if next_index.is_file():
|
||
return FileResponse(next_index)
|
||
return FileResponse(BASE_DIR / "index.html")
|
||
|
||
|
||
_next_dir = CONSOLE_OUT / "_next"
|
||
if _next_dir.is_dir():
|
||
app.mount("/_next", StaticFiles(directory=str(_next_dir)), name="next_static")
|