's'

src/async_thread_loop.py

@@ -0,0 +1,18 @@
+"""Per-thread asyncio event loop reuse (avoids asyncio.run() overhead in recorder threads)."""
+from __future__ import annotations
+
+import asyncio
+import threading
+from typing import Any, Coroutine, TypeVar
+
+_T = TypeVar("_T")
+_tls = threading.local()
+
+
+def run_async_coro(coro: Coroutine[Any, Any, _T]) -> _T:
+    loop: asyncio.AbstractEventLoop | None = getattr(_tls, "loop", None)
+    if loop is None or loop.is_closed():
+        loop = asyncio.new_event_loop()
+        _tls.loop = loop
+    asyncio.set_event_loop(loop)
+    return loop.run_until_complete(coro)

src/hub_routes.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import json
+from pathlib import Path
 from typing import Any, Optional
 
 from fastapi import APIRouter
@@ -19,22 +20,30 @@ from src.control_plane import (
 )
 from src.live_config import business_config, load_hub_config_snapshot, services_config_list
 from src.web_process_backend import WebProcessBackend
+from src.youtube_ini_sync import write_youtube_ini_stream_key
 
 router = APIRouter(prefix="/hub", tags=["Live Hub"])
 
 _process_backend: WebProcessBackend | None = None
 _monitor_entries: list[dict[str, Any]] = []
+_repo_root: Path | None = None
 
 
-def configure_hub(process_backend: WebProcessBackend, monitor_entries: list[dict[str, Any]]) -> None:
-    global _process_backend, _monitor_entries
+def configure_hub(
+    process_backend: WebProcessBackend,
+    monitor_entries: list[dict[str, Any]],
+    repo_root: Optional[Path] = None,
+) -> None:
+    global _process_backend, _monitor_entries, _repo_root
     _process_backend = process_backend
     _monitor_entries = list(monitor_entries)
+    _repo_root = repo_root.resolve() if repo_root is not None else None
 
 
 class BusinessDouyinPatch(BaseModel):
     streams: Optional[dict[str, str]] = None
     ffmpeg: Optional[dict[str, Any]] = None
+    sync_youtube_ini: Optional[bool] = None
 
 
 @router.get("/dashboard")
@@ -142,4 +151,15 @@ async def hub_business_douyin_patch(body: BusinessDouyinPatch) -> dict[str, str]
         cur_ff.update(body.ffmpeg)
         current["ffmpeg"] = cur_ff
     path.write_text(json.dumps(current, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
-    return {"message": "business/douyinyoutube.json updated"}
+    msg = "business/douyinyoutube.json updated"
+    if body.sync_youtube_ini and _repo_root is not None:
+        streams = current.get("streams") if isinstance(current.get("streams"), dict) else {}
+        raw_key = streams.get("youtube_rtmp_key")
+        key_val = raw_key.strip() if isinstance(raw_key, str) else ""
+        if key_val:
+            msg = f"{msg}; {write_youtube_ini_stream_key(_repo_root, key_val)}"
+        else:
+            msg = f"{msg}; sync_youtube_ini skipped (no youtube_rtmp_key)"
+    elif body.sync_youtube_ini and _repo_root is None:
+        msg = f"{msg}; sync_youtube_ini skipped (repo root not configured)"
+    return {"message": msg}

src/web_process_backend.py

@@ -49,8 +49,7 @@ async def run_pm2(cmd: str, timeout: float = 15.0) -> str:
     return text
 
 
-def _linux_ffmpeg_pids_for_stack(process_name: str, project_root: Path | None) -> list[int]:
-    """仅匹配与本项目直播栈相关的 ffmpeg（命令行含 YouTube 推流、抖音拉流或项目路径），避免 killall 误伤整机其它 ffmpeg。"""
+def _ffmpeg_cmdline_markers(process_name: str, project_root: Path | None) -> list[str]:
     markers: list[str] = [
         process_name,
         "a.rtmp.youtube.com",
@@ -64,7 +63,12 @@ def _linux_ffmpeg_pids_for_stack(process_name: str, project_root: Path | None) -
             markers.append(str(project_root.resolve()))
         except OSError:
             pass
-    markers = [m for m in markers if m]
+    return [m for m in markers if m]
+
+
+def _linux_ffmpeg_pids_for_stack(process_name: str, project_root: Path | None) -> list[int]:
+    """仅匹配与本项目直播栈相关的 ffmpeg（命令行含 YouTube 推流、抖音拉流或项目路径），避免 killall 误伤整机其它 ffmpeg。"""
+    markers = _ffmpeg_cmdline_markers(process_name, project_root)
     seen: set[int] = set()
     proc_root = Path("/proc")
     if not proc_root.is_dir():
@@ -97,16 +101,45 @@ def _signal_pids(pids: Iterable[int], sig: int) -> None:
             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
-        proc2 = await asyncio.create_subprocess_shell(
-            "taskkill /IM ffmpeg.exe /F /T 2>nul || exit /b 0",
-            stdout=asyncio.subprocess.DEVNULL,
-            stderr=asyncio.subprocess.DEVNULL,
+def _windows_kill_stack_ffmpeg(process_name: str, project_root: Path | None) -> None:
+    """仅结束命令行命中业务标记的 ffmpeg.exe，避免 taskkill /IM 误杀其它 FFmpeg。"""
+    markers = _ffmpeg_cmdline_markers(process_name, project_root)
+    if not markers:
+        return
+    env = os.environ.copy()
+    env["DOUYIN_FFMPEG_MARKERS"] = json.dumps(markers, ensure_ascii=False)
+    ps = r"""
+$raw = $env:DOUYIN_FFMPEG_MARKERS
+if (-not $raw) { exit 0 }
+$m = $raw | ConvertFrom-Json
+foreach ($proc in Get-CimInstance Win32_Process -Filter "Name = 'ffmpeg.exe'") {
+  $cmd = $proc.CommandLine
+  if (-not $cmd) { continue }
+  foreach ($x in $m) {
+    if ($cmd -like ('*' + $x + '*')) {
+      Stop-Process -Id $proc.ProcessId -Force -ErrorAction SilentlyContinue
+      break
+    }
+  }
+}
+"""
+    try:
+        subprocess.run(
+            ["powershell", "-NoProfile", "-NonInteractive", "-Command", ps],
+            env=env,
+            capture_output=True,
+            text=True,
+            timeout=45,
+            check=False,
         )
-        await proc2.wait()
+    except (OSError, subprocess.TimeoutExpired):
+        pass
+
+
+async def force_kill_ffmpeg(process_name: str, project_root: Path | None = None) -> None:
+    """结束与本业务相关的 ffmpeg：Linux 按 /proc 命令行匹配 + 可选 pkill；Windows 按命令行标记筛选。"""
+    if sys.platform == "win32":
+        await asyncio.to_thread(_windows_kill_stack_ffmpeg, process_name, project_root)
         return
 
     def collect() -> list[int]:

src/youtube_ini_sync.py

@@ -0,0 +1,24 @@
+"""Write YouTube stream key into repo config/youtube.ini while preserving other keys."""
+from __future__ import annotations
+
+import configparser
+from pathlib import Path
+
+
+def write_youtube_ini_stream_key(repo_root: Path, stream_key_or_url: str) -> str:
+    root = repo_root.resolve()
+    cfg_path = root / "config" / "youtube.ini"
+    value = (stream_key_or_url or "").strip()
+    if not value:
+        return "sync skipped: empty youtube_rtmp_key"
+
+    cfg_path.parent.mkdir(parents=True, exist_ok=True)
+    parser = configparser.ConfigParser()
+    if cfg_path.is_file():
+        parser.read(cfg_path, encoding="utf-8")
+    if not parser.has_section("youtube"):
+        parser.add_section("youtube")
+    parser.set("youtube", "key", value)
+    with cfg_path.open("w", encoding="utf-8") as fh:
+        parser.write(fh)
+    return f"wrote {cfg_path}"