eric/gitlab-instance-0a899031_douyinyoutube
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Safer ffmpeg kill, POST process/service actions, ARM encoder from hardware.env, filter_threads scaling

Browse Source 
Made-with: Cursor
This commit is contained in:
eric
2026-03-28 12:03:13 -05:00
parent f4794e720a
commit 9b6570d06e
7 changed files with 204 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

86
douyin_youtube_ffplay.py
View File

@@ -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,

4
main.py
View File

@@ -94,10 +94,6 @@ def signal_handler(_signal, _frame):
signal.signal(signal.SIGTERM, signal_handler)
import shutil
# 全局变量（只初始化一次）
_last_lines_count = 0

18
scripts/hardware_probe.py
View File

@@ -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")

86
src/web_process_backend.py
View File

@@ -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:

16
web-console/app/page.tsx
View File

@@ -480,9 +480,11 @@ export default function Home() {
if (!currentProcess) return;
setBusy(`process:${action}`);
try {
const data = await fetchJson<ProcessStatusResponse>(
`/${action}?process=${encodeURIComponent(currentProcess)}`,
);
const data = await fetchJson<ProcessStatusResponse>(`/${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<ServiceItem>(
`/service_action?service=${encodeURIComponent(serviceId)}&action=${encodeURIComponent(action)}`,
);
const data = await fetchJson<ServiceItem>("/service_action", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ service: serviceId, action }),
});
notify(data.message || "OK");
await refreshServices();
} catch (reason) {

1
web-console/middleware.ts
View File

@@ -42,6 +42,7 @@ export function middleware(request: NextRequest) {
return NextResponse.rewrite(dest);
}
/** Dev 代理：含 GET/POST 等同名 API（进程与服务控制已优先使用 POST） */
export const config = {
matcher: [
"/process_monitor",

38
web.py
View File

@@ -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: