From 9b6570d06e520e3f44d150ece774af352e6b9097 Mon Sep 17 00:00:00 2001 From: eric Date: Sat, 28 Mar 2026 12:03:13 -0500 Subject: [PATCH] Safer ffmpeg kill, POST process/service actions, ARM encoder from hardware.env, filter_threads scaling Made-with: Cursor --- douyin_youtube_ffplay.py | 86 ++++++++++++++++++++++++++++++-------- main.py | 4 -- scripts/hardware_probe.py | 18 ++++++++ src/web_process_backend.py | 86 +++++++++++++++++++++++++++++++------- web-console/app/page.tsx | 16 ++++--- web-console/middleware.ts | 1 + web.py | 38 ++++++++++++++++- 7 files changed, 204 insertions(+), 45 deletions(-) diff --git a/douyin_youtube_ffplay.py b/douyin_youtube_ffplay.py index 6a655df..4317531 100644 --- a/douyin_youtube_ffplay.py +++ b/douyin_youtube_ffplay.py @@ -157,24 +157,74 @@ def start_douyin_youtube_ffplay( stream_title = sanitize_title(f"{anchor_name}{('_' + clean_title) if clean_title else ''}_{timestamp_str}") print(f"[INFO] {rec_info} 开始推流到 YouTube: {stream_title}") - # ====================== NVENC 检测（符合 YouTube 推荐比特率）====================== + # ====================== 视频编码：hardware.env（ARM）→ NVENC → libx264 ====================== + def _load_simple_env(path: str) -> dict[str, str]: + out: dict[str, str] = {} + if not os.path.isfile(path): + return out + with open(path, "r", encoding="utf-8", errors="ignore") as fh: + for line in fh: + line = line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + k, _, v = line.partition("=") + key = k.strip() + if key.startswith("#"): + continue + out[key] = v.strip().strip('"').strip("'") + return out + codec_v = "libx264" - # ultrafast+zerolatency：在软编下尽量让 speed≥1.0；画质略降、直播可接受 - video_args = ["-preset", "ultrafast", "-tune", "zerolatency", "-profile:v", "high", "-level", "4.0", - "-b:v", b_v, "-maxrate", maxrate, "-bufsize", bufsize] - try: - if platform.system().lower() in ["windows", "linux"]: - check = subprocess.run(["nvidia-smi"], capture_output=True, timeout=5) - if check.returncode == 0: - enc = subprocess.run(["ffmpeg", "-hide_banner", "-encoders"], capture_output=True, text=True, timeout=5) - if "h264_nvenc" in (enc.stdout or ""): - codec_v = "h264_nvenc" - video_args = ["-profile:v", "high", "-level", "4.0", - "-tune", "ll", "-rc", "cbr", "-b:v", b_v, - "-maxrate", maxrate, "-bufsize", bufsize] - print("[INFO] 检测到 NVIDIA GPU → 使用 NVENC 硬件加速") - except Exception as e: - print(f"[WARN] NVENC 检测失败，回退软件编码: {e}") + video_args = [ + "-preset", "ultrafast", "-tune", "zerolatency", "-profile:v", "high", "-level", "4.0", + "-b:v", b_v, "-maxrate", maxrate, "-bufsize", bufsize, + ] + + _config_dir = os.path.dirname(os.path.abspath(config_file)) + _hw_env = _load_simple_env(os.path.join(_config_dir, "hardware.env")) + _want_enc = (_hw_env.get("LIVE_VIDEO_ENCODER") or "").strip() + _hw_enc_active = False + if _want_enc and platform.system().lower() == "linux": + try: + enc = subprocess.run( + ["ffmpeg", "-hide_banner", "-encoders"], + capture_output=True, + text=True, + timeout=6, + ) + blob = (enc.stdout or "") + (enc.stderr or "") + if _want_enc in blob and _want_enc.startswith("h264_"): + codec_v = _want_enc + video_args = ["-b:v", b_v, "-maxrate", maxrate, "-bufsize", bufsize] + _hw_enc_active = True + print(f"[INFO] 使用 config/hardware.env LIVE_VIDEO_ENCODER={codec_v}（板载硬件编码）") + except Exception as e: + print(f"[WARN] LIVE_VIDEO_ENCODER 校验失败: {e}") + + if not _hw_enc_active: + try: + if platform.system().lower() in ["windows", "linux"]: + check = subprocess.run(["nvidia-smi"], capture_output=True, timeout=5) + if check.returncode == 0: + enc = subprocess.run( + ["ffmpeg", "-hide_banner", "-encoders"], + capture_output=True, + text=True, + timeout=5, + ) + if "h264_nvenc" in (enc.stdout or ""): + codec_v = "h264_nvenc" + video_args = [ + "-profile:v", "high", "-level", "4.0", + "-tune", "ll", "-rc", "cbr", "-b:v", b_v, + "-maxrate", maxrate, "-bufsize", bufsize, + ] + print("[INFO] 检测到 NVIDIA GPU → 使用 NVENC 硬件加速") + except Exception as e: + print(f"[WARN] NVENC 检测失败，回退软件编码: {e}") + + _cpu_n = os.cpu_count() or 4 + _filter_threads = str(min(4, max(1, _cpu_n // 2))) # ====================== 开源音频处理：背景音乐弱化 + 防版权 ====================== _script_dir = os.path.dirname(os.path.abspath(__file__)) @@ -236,7 +286,7 @@ def start_douyin_youtube_ffplay( "-err_detect", "ignore_err", "-max_delay", "500000", "-thread_queue_size", "2048", - "-filter_threads", "4", + "-filter_threads", _filter_threads, "-headers", douyin_headers, "-re", "-i", real_url, diff --git a/main.py b/main.py index b7cd0be..2417f86 100644 --- a/main.py +++ b/main.py @@ -94,10 +94,6 @@ def signal_handler(_signal, _frame): signal.signal(signal.SIGTERM, signal_handler) - -import shutil - - # 全局变量（只初始化一次） _last_lines_count = 0 diff --git a/scripts/hardware_probe.py b/scripts/hardware_probe.py index 785577e..9cae8ff 100644 --- a/scripts/hardware_probe.py +++ b/scripts/hardware_probe.py @@ -145,15 +145,33 @@ def build_profile() -> dict: return profile +def suggest_video_encoder(profile: dict) -> str: + """在 Linux 上根据 ffmpeg 能力与探测结果给出可选硬件编码器名（供 douyin→YouTube 读取 LIVE_VIDEO_ENCODER）。""" + text = run_text(["ffmpeg", "-hide_banner", "-encoders"]) + if not text: + return "" + hwaccels = set(profile.get("ffmpeg_hwaccels") or []) + if "h264_rkmpp" in text and Path("/dev/mpp_service").exists(): + return "h264_rkmpp" + if "h264_v4l2m2m" in text and "v4l2m2m" in hwaccels: + return "h264_v4l2m2m" + return "" + + def write_outputs(profile: dict) -> None: CONFIG_DIR.mkdir(parents=True, exist_ok=True) JSON_PATH.write_text(json.dumps(profile, indent=2, ensure_ascii=False), encoding="utf-8") + enc = suggest_video_encoder(profile) env_lines = [ f"LIVE_HW_DECODER={profile['recommendation']['video_decoder']}", f"LIVE_HDMI_PLAYER={profile['recommendation']['hdmi_player']}", f"LIVE_BROWSER_HWACCEL={profile['recommendation']['browser_hwaccel']}", f"LIVE_STABILITY_MODE={profile['recommendation']['stability_mode']}", ] + if enc: + env_lines.append(f"LIVE_VIDEO_ENCODER={enc}") + else: + env_lines.append("# LIVE_VIDEO_ENCODER= # optional: h264_v4l2m2m / h264_rkmpp when ffmpeg supports it") ENV_PATH.write_text("\n".join(env_lines) + "\n", encoding="utf-8") diff --git a/src/web_process_backend.py b/src/web_process_backend.py index 7ad2614..21f2dfa 100644 --- a/src/web_process_backend.py +++ b/src/web_process_backend.py @@ -12,7 +12,7 @@ import subprocess import sys import time from pathlib import Path -from typing import Any, Optional +from typing import Any, Iterable, Optional BACKEND_STATE_VERSION = 1 @@ -49,10 +49,58 @@ async def run_pm2(cmd: str, timeout: float = 15.0) -> str: return text -async def force_kill_ffmpeg(process_name: str) -> None: - """尽量结束与本业务相关的 ffmpeg（Linux 精准 + 兜底；Windows 与 Linux 兜底策略一致：结束 ffmpeg 进程）。""" +def _linux_ffmpeg_pids_for_stack(process_name: str, project_root: Path | None) -> list[int]: + """仅匹配与本项目直播栈相关的 ffmpeg（命令行含 YouTube 推流、抖音拉流或项目路径），避免 killall 误伤整机其它 ffmpeg。""" + markers: list[str] = [ + process_name, + "a.rtmp.youtube.com", + "rtmps://a.rtmp.youtube.com", + "live.douyin.com", + "pull-flv", + "douyincdn.com", + ] + if project_root is not None: + try: + markers.append(str(project_root.resolve())) + except OSError: + pass + markers = [m for m in markers if m] + seen: set[int] = set() + proc_root = Path("/proc") + if not proc_root.is_dir(): + return [] + for entry in proc_root.iterdir(): + if not entry.name.isdigit(): + continue + try: + pid = int(entry.name) + except ValueError: + continue + cmdline = entry / "cmdline" + try: + raw = cmdline.read_bytes() + except OSError: + continue + if not raw or b"ffmpeg" not in raw.lower(): + continue + cmd = raw.replace(b"\0", b" ").decode("utf-8", "replace") + if any(m in cmd for m in markers): + seen.add(pid) + return sorted(seen) + + +def _signal_pids(pids: Iterable[int], sig: int) -> None: + for pid in pids: + try: + os.kill(pid, sig) + except (ProcessLookupError, PermissionError): + pass + + +async def force_kill_ffmpeg(process_name: str, project_root: Path | None = None) -> None: + """结束与本业务相关的 ffmpeg：Linux 按 /proc 命令行匹配 + 可选 pkill；不再使用 killall -9 ffmpeg。""" if sys.platform == "win32": - _ = process_name # 保留参数供将来按命令行筛选；当前与 Linux killall 行为对齐 + _ = process_name proc2 = await asyncio.create_subprocess_shell( "taskkill /IM ffmpeg.exe /F /T 2>nul || exit /b 0", stdout=asyncio.subprocess.DEVNULL, @@ -61,17 +109,25 @@ async def force_kill_ffmpeg(process_name: str) -> None: await proc2.wait() return - for cmd in ( - f"pkill -f 'ffmpeg.*{process_name}' || true", - "killall -9 ffmpeg 2>/dev/null || true", - ): - proc = await asyncio.create_subprocess_shell( - cmd, - stdout=asyncio.subprocess.DEVNULL, - stderr=asyncio.subprocess.DEVNULL, - ) - await proc.wait() - await asyncio.sleep(0.4) + def collect() -> list[int]: + return _linux_ffmpeg_pids_for_stack(process_name, project_root) + + pids = await asyncio.to_thread(collect) + if pids: + await asyncio.to_thread(_signal_pids, pids, signal.SIGTERM) + await asyncio.sleep(0.6) + remaining = await asyncio.to_thread(collect) + if remaining: + await asyncio.to_thread(_signal_pids, remaining, signal.SIGKILL) + await asyncio.sleep(0.2) + + safe = re.sub(r"[^a-zA-Z0-9_.-]", "", process_name) or "x" + proc = await asyncio.create_subprocess_shell( + f"pkill -f 'ffmpeg.*{safe}' || true", + stdout=asyncio.subprocess.DEVNULL, + stderr=asyncio.subprocess.DEVNULL, + ) + await proc.wait() class LocalProcessRegistry: diff --git a/web-console/app/page.tsx b/web-console/app/page.tsx index 276225c..c711b8d 100644 --- a/web-console/app/page.tsx +++ b/web-console/app/page.tsx @@ -480,9 +480,11 @@ export default function Home() { if (!currentProcess) return; setBusy(`process:${action}`); try { - const data = await fetchJson( - `/${action}?process=${encodeURIComponent(currentProcess)}`, - ); + const data = await fetchJson(`/${action}`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ process: currentProcess }), + }); notify(String(data.output || data.pm2_output || data.message || "OK")); await refreshProcess(); } catch (reason) { @@ -495,9 +497,11 @@ export default function Home() { const serviceAction = async (serviceId: string, action: "start" | "restart" | "stop") => { setBusy(`${serviceId}:${action}`); try { - const data = await fetchJson( - `/service_action?service=${encodeURIComponent(serviceId)}&action=${encodeURIComponent(action)}`, - ); + const data = await fetchJson("/service_action", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ service: serviceId, action }), + }); notify(data.message || "OK"); await refreshServices(); } catch (reason) { diff --git a/web-console/middleware.ts b/web-console/middleware.ts index 8ba2787..f8fbccc 100644 --- a/web-console/middleware.ts +++ b/web-console/middleware.ts @@ -42,6 +42,7 @@ export function middleware(request: NextRequest) { return NextResponse.rewrite(dest); } +/** Dev 代理：含 GET/POST 等同名 API（进程与服务控制已优先使用 POST） */ export const config = { matcher: [ "/process_monitor", diff --git a/web.py b/web.py index 92ff360..c1b3c4f 100644 --- a/web.py +++ b/web.py @@ -6,6 +6,8 @@ from contextlib import asynccontextmanager from pathlib import Path from typing import Optional +from pydantic import BaseModel, Field + from fastapi import FastAPI, Query, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse, JSONResponse, Response @@ -315,6 +317,14 @@ async def run_service_action( return JSONResponse({"error": str(exc)}, status_code=400) +@app.post("/service_action") +async def run_service_action_post(body: ServiceActionPostBody): + try: + return await service_action(body.service, body.action) + except (KeyError, ValueError) as exc: + return JSONResponse({"error": str(exc)}, status_code=400) + + @app.get("/managed_roots") async def get_managed_roots(): return {"roots": list_root_summaries()} @@ -465,6 +475,15 @@ async def save_url_config(request: Request, process: str = Query(..., descriptio return JSONResponse({"error": f"保存失败: {str(e)}"}, status_code=500) +class ProcessNameBody(BaseModel): + process: str = Field(..., min_length=1, max_length=160) + + +class ServiceActionPostBody(BaseModel): + service: str = Field(..., min_length=1, max_length=64) + action: str = Field(..., pattern="^(start|stop|restart)$") + + def _not_found_hint(process: str) -> str: if process_backend.mode == "local": return ( @@ -486,6 +505,11 @@ async def start(process: str = Query(..., description="进程名")): return JSONResponse({"output": output}) +@app.post("/start") +async def start_post(body: ProcessNameBody): + return await start(process=body.process) + + @app.get("/stop") async def stop(process: str = Query(..., description="进程名")): if process not in VALID_PROCESSES: @@ -496,7 +520,7 @@ async def stop(process: str = Query(..., description="进程名")): await asyncio.sleep(2.0) if not process.startswith("web"): - await force_kill_ffmpeg(process) + await force_kill_ffmpeg(process, BASE_DIR) await asyncio.sleep(1.0) @@ -509,6 +533,11 @@ async def stop(process: str = Query(..., description="进程名")): ) +@app.post("/stop") +async def stop_post(body: ProcessNameBody): + return await stop(process=body.process) + + @app.get("/restart") async def restart(process: str = Query(..., description="进程名")): if process not in VALID_PROCESSES: @@ -518,10 +547,15 @@ async def restart(process: str = Query(..., description="进程名")): return JSONResponse({"output": "内部错误：找不到脚本映射"}, status_code=500) output = await process_backend.restart(process, sc) if not process.startswith("web"): - await force_kill_ffmpeg(process) + await force_kill_ffmpeg(process, BASE_DIR) return JSONResponse({"output": output}) +@app.post("/restart") +async def restart_post(body: ProcessNameBody): + return await restart(process=body.process) + + @app.get("/status") async def status(process: str = Query(..., description="进程名")): if process not in VALID_PROCESSES: