From 90ae151acde57210a151954b1396aa9373a3bb50 Mon Sep 17 00:00:00 2001
From: Cai <cai@nbcai.cc>
Date: Wed, 12 Aug 2026 17:37:06 +0800
Subject: [PATCH] research(defense): complete topic priorities and peer comparisons
---
dev/project-dev/transcribe_media.py | 118 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-
1 files changed, 116 insertions(+), 2 deletions(-)
diff --git a/dev/project-dev/transcribe_media.py b/dev/project-dev/transcribe_media.py
index 1b065bd..ae7cc8e 100644
--- a/dev/project-dev/transcribe_media.py
+++ b/dev/project-dev/transcribe_media.py
@@ -3,6 +3,7 @@
from __future__ import annotations
import argparse
+from contextlib import contextmanager
import json
import math
import os
@@ -21,10 +22,17 @@
SHORT_AUDIO_LIMIT_SECONDS = 1200.0
CHUNK_CORE_SECONDS = 1200.0
CHUNK_OVERLAP_SECONDS = 5.0
+MEDIA_DECODE_THREADS = 4
+MEDIA_FILTER_THREADS = 2
+MEDIA_MUTEX_NAME = r"Local\MBXMediaHeavyTaskV1"
class MediaTranscriptionError(RuntimeError):
"""A user-facing error that should stop the command without a traceback."""
+
+
+class MediaHostGuardError(RuntimeError):
+ """The shared Windows heavy-media guard could not be acquired safely."""
@dataclass(frozen=True)
@@ -78,6 +86,106 @@
ModelLoader = Callable[[], tuple[Any, RuntimeInfo]]
+def _is_windows() -> bool:
+ return os.name == "nt"
+
+
+def _subprocess_creation_flags() -> int:
+ if not _is_windows():
+ return 0
+ return int(getattr(subprocess, "BELOW_NORMAL_PRIORITY_CLASS", 0x00004000))
+
+
+def _windows_kernel32() -> Any:
+ import ctypes
+ from ctypes import wintypes
+
+ kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
+ kernel32.CreateMutexW.argtypes = (ctypes.c_void_p, wintypes.BOOL, wintypes.LPCWSTR)
+ kernel32.CreateMutexW.restype = wintypes.HANDLE
+ kernel32.WaitForSingleObject.argtypes = (wintypes.HANDLE, wintypes.DWORD)
+ kernel32.WaitForSingleObject.restype = wintypes.DWORD
+ kernel32.ReleaseMutex.argtypes = (wintypes.HANDLE,)
+ kernel32.ReleaseMutex.restype = wintypes.BOOL
+ kernel32.CloseHandle.argtypes = (wintypes.HANDLE,)
+ kernel32.CloseHandle.restype = wintypes.BOOL
+ kernel32.GetCurrentProcess.restype = wintypes.HANDLE
+ kernel32.GetPriorityClass.argtypes = (wintypes.HANDLE,)
+ kernel32.GetPriorityClass.restype = wintypes.DWORD
+ kernel32.SetPriorityClass.argtypes = (wintypes.HANDLE, wintypes.DWORD)
+ kernel32.SetPriorityClass.restype = wintypes.BOOL
+ return kernel32
+
+
+def _win32_last_error() -> int:
+ import ctypes
+
+ return int(ctypes.get_last_error())
+
+
+@contextmanager
+def media_host_guard() -> Iterable[None]:
+ """Serialize heavy media CLIs and lower the current Windows process priority."""
+ if not _is_windows():
+ yield
+ return
+
+ kernel32 = _windows_kernel32()
+
+ wait_object_0 = 0x00000000
+ wait_abandoned = 0x00000080
+ wait_timeout = 0x00000102
+ below_normal_priority_class = 0x00004000
+ handle = kernel32.CreateMutexW(None, False, MEDIA_MUTEX_NAME)
+ if not handle:
+ raise MediaHostGuardError(
+ f"无法创建重型媒体互斥（Win32={_win32_last_error()}）。"
+ )
+ acquired = False
+ process_handle = kernel32.GetCurrentProcess()
+ original_priority = 0
+ priority_lowered = False
+ try:
+ wait_result = int(kernel32.WaitForSingleObject(handle, 0))
+ if wait_result == wait_timeout:
+ raise MediaHostGuardError(
+ "已有重型媒体任务运行，拒绝并行启动；请等待其结束后重试。"
+ )
+ if wait_result not in (wait_object_0, wait_abandoned):
+ raise MediaHostGuardError(
+ f"无法获取重型媒体互斥（Win32 wait={wait_result}）。"
+ )
+ acquired = True
+ original_priority = int(kernel32.GetPriorityClass(process_handle))
+ if original_priority == 0:
+ raise MediaHostGuardError(
+ f"无法读取当前进程优先级（Win32={_win32_last_error()}）。"
+ )
+ if not kernel32.SetPriorityClass(process_handle, below_normal_priority_class):
+ raise MediaHostGuardError(
+ f"无法将当前进程设为低于正常优先级（Win32={_win32_last_error()}）。"
+ )
+ priority_lowered = True
+ yield
+ finally:
+ if priority_lowered and not kernel32.SetPriorityClass(process_handle, original_priority):
+ print(
+ "警告：重型媒体任务结束后无法恢复原进程优先级"
+ f"（Win32={_win32_last_error()}）。",
+ file=sys.stderr,
+ )
+ if acquired and not kernel32.ReleaseMutex(handle):
+ print(
+ f"警告：释放重型媒体互斥失败（Win32={_win32_last_error()}）。",
+ file=sys.stderr,
+ )
+ if not kernel32.CloseHandle(handle):
+ print(
+ f"警告：关闭重型媒体互斥句柄失败（Win32={_win32_last_error()}）。",
+ file=sys.stderr,
+ )
+
+
def _run_process(command: Sequence[str]) -> subprocess.CompletedProcess[str]:
return subprocess.run(
list(command),
@@ -86,6 +194,7 @@
text=True,
encoding="utf-8",
errors="replace",
+ creationflags=_subprocess_creation_flags(),
)
@@ -182,6 +291,8 @@
"error",
"-nostdin",
"-n",
+ "-threads",
+ str(MEDIA_DECODE_THREADS),
"-i",
str(video),
"-map",
@@ -275,6 +386,8 @@
"-n",
"-ss",
f"{chunk.audio_start:.6f}",
+ "-threads",
+ str(MEDIA_DECODE_THREADS),
"-i",
str(full_audio),
"-t",
@@ -636,8 +749,9 @@
def main(argv: Sequence[str] | None = None) -> int:
args = _build_parser().parse_args(argv)
try:
- outputs, runtime = transcribe_video(args.video, args.output, args.language)
- except MediaTranscriptionError as exc:
+ with media_host_guard():
+ outputs, runtime = transcribe_video(args.video, args.output, args.language)
+ except (MediaTranscriptionError, MediaHostGuardError) as exc:
print(f"错误：{exc}", file=sys.stderr)
return 1
except KeyboardInterrupt as exc:
--
Gitblit v1.9.3