from __future__ import annotations import json import os import subprocess import sys import tempfile import types import unittest from pathlib import Path from unittest.mock import patch PROJECT_DEV = Path(__file__).resolve().parents[1] if str(PROJECT_DEV) not in sys.path: sys.path.insert(0, str(PROJECT_DEV)) import transcribe_media class FakeSegment: def __init__(self, start: float, end: float, text: str) -> None: self.start = start self.end = end self.text = text class FakeModel: def __init__(self) -> None: self.calls: list[tuple[str, dict[str, object]]] = [] def transcribe(self, audio: str, **options: object): self.calls.append((audio, options)) info = types.SimpleNamespace(language="zh", language_probability=0.99) segments = iter( [ FakeSegment(3.12, 8.64, " 大家好。 "), FakeSegment(9.0, 12.5, "开始测试。"), ] ) return segments, info class ScriptedModel: def __init__(self, responses, *, fail_on_call: int | None = None) -> None: self.responses = responses self.fail_on_call = fail_on_call self.calls: list[tuple[str, dict[str, object]]] = [] def transcribe(self, audio: str, **options: object): self.calls.append((audio, options)) call_number = len(self.calls) if self.fail_on_call == call_number: raise RuntimeError("synthetic GPU failure") info = types.SimpleNamespace(language="zh", language_probability=0.98) return iter(self.responses[call_number - 1]), info class RecordingRunner: def __init__( self, *, has_audio: bool = True, ffmpeg_ok: bool = True, audio_duration: float = 60.0, ) -> None: self.has_audio = has_audio self.ffmpeg_ok = ffmpeg_ok self.audio_duration = audio_duration self.commands: list[list[str]] = [] def __call__(self, command): command = list(command) self.commands.append(command) executable = Path(command[0]).name.lower() if executable.startswith("ffprobe"): if "format=duration" in command: stdout = f"{self.audio_duration}\n" else: stdout = "0\n" if self.has_audio else "" return subprocess.CompletedProcess(command, 0, stdout, "") if executable.startswith("ffmpeg"): if not self.ffmpeg_ok: return subprocess.CompletedProcess(command, 1, "", "synthetic failure") Path(command[-1]).write_bytes(b"fLaC synthetic") return subprocess.CompletedProcess(command, 0, "", "") raise AssertionError(f"unexpected executable: {command[0]}") class TranscribeMediaTests(unittest.TestCase): def setUp(self) -> None: self.executable_patch = patch( "transcribe_media.shutil.which", side_effect=lambda name: f"C:/tools/{name}.exe" ) self.executable_patch.start() def tearDown(self) -> None: self.executable_patch.stop() def test_srt_timestamp_rounding(self) -> None: self.assertEqual(transcribe_media._format_srt_timestamp(3661.2346), "01:01:01,235") self.assertEqual(transcribe_media._format_srt_timestamp(-1), "00:00:00,000") def test_end_to_end_writes_four_outputs_without_touching_source(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) video = root / "会议录像.mp4" source_bytes = b"synthetic video bytes" video.write_bytes(source_bytes) runner = RecordingRunner() model = FakeModel() runtime = transcribe_media.RuntimeInfo(1, ("float16", "int8_float16")) outputs, actual_runtime = transcribe_media.transcribe_video( video, language="zh", runner=runner, model_loader=lambda: (model, runtime), ) self.assertEqual(video.read_bytes(), source_bytes) self.assertEqual(actual_runtime, runtime) self.assertEqual(outputs.directory, root / "会议录像.transcript") for output in outputs.files: self.assertTrue(output.is_file(), output) self.assertIn("[00:00:03] 大家好。", outputs.text.read_text(encoding="utf-8")) self.assertIn("00:00:03,120 --> 00:00:08,640", outputs.srt.read_text(encoding="utf-8")) payload = json.loads(outputs.json.read_text(encoding="utf-8")) self.assertEqual(payload["source"], video.name) self.assertEqual(payload["device"], "cuda") self.assertEqual(payload["compute_type"], "float16") self.assertTrue(payload["vad_filter"]) self.assertEqual(len(payload["segments"]), 2) self.assertEqual(model.calls[0][1], {"vad_filter": True, "language": "zh"}) ffmpeg_command = next( command for command in runner.commands if Path(command[0]).name.lower().startswith("ffmpeg") ) self.assertIn("-n", ffmpeg_command) self.assertEqual( ffmpeg_command[ffmpeg_command.index("-threads") + 1], str(transcribe_media.MEDIA_DECODE_THREADS), ) self.assertLess(ffmpeg_command.index("-threads"), ffmpeg_command.index("-i")) self.assertEqual(ffmpeg_command[ffmpeg_command.index("-map") + 1], "0:a:0") self.assertEqual(ffmpeg_command[ffmpeg_command.index("-ac") + 1], "1") self.assertEqual(ffmpeg_command[ffmpeg_command.index("-ar") + 1], "16000") self.assertEqual(ffmpeg_command[ffmpeg_command.index("-c:a") + 1], "flac") def test_default_process_runner_uses_below_normal_creation_flags(self) -> None: completed = subprocess.CompletedProcess(["tool"], 0, "", "") with patch.object(transcribe_media, "_is_windows", return_value=True), patch.object( transcribe_media.subprocess, "run", return_value=completed ) as run: self.assertIs(transcribe_media._run_process(["tool"]), completed) self.assertEqual( run.call_args.kwargs["creationflags"], getattr(subprocess, "BELOW_NORMAL_PRIORITY_CLASS", 0x00004000), ) @unittest.skipUnless(os.name == "nt", "Windows named mutex and priority are required") def test_named_mutex_blocks_second_process_and_restores_priority(self) -> None: import ctypes kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) kernel32.GetCurrentProcess.restype = ctypes.c_void_p kernel32.GetPriorityClass.argtypes = (ctypes.c_void_p,) kernel32.GetPriorityClass.restype = ctypes.c_uint32 process_handle = kernel32.GetCurrentProcess() original_priority = int(kernel32.GetPriorityClass(process_handle)) child_code = ( "import sys\n" f"sys.path.insert(0, {str(PROJECT_DEV)!r})\n" "import transcribe_media as module\n" "try:\n" " with module.media_host_guard():\n" " pass\n" "except module.MediaHostGuardError:\n" " raise SystemExit(23)\n" ) with transcribe_media.media_host_guard(): self.assertEqual(int(kernel32.GetPriorityClass(process_handle)), 0x00004000) busy = subprocess.run([sys.executable, "-c", child_code], check=False) self.assertEqual(busy.returncode, 23) self.assertEqual(int(kernel32.GetPriorityClass(process_handle)), original_priority) free = subprocess.run([sys.executable, "-c", child_code], check=False) self.assertEqual(free.returncode, 0) @unittest.skipUnless(os.name == "nt", "Windows named mutex is required") def test_both_real_clis_reject_busy_before_path_or_external_actions(self) -> None: scripts = ( (PROJECT_DEV / "transcribe_media.py", 1), (PROJECT_DEV / "extract_ppt_slides.py", 2), ) with transcribe_media.media_host_guard(): for script, expected_code in scripts: with self.subTest(script=script.name): result = subprocess.run( [sys.executable, str(script), "definitely-missing-media.mp4"], capture_output=True, text=True, encoding="utf-8", errors="replace", env={**os.environ, "PYTHONIOENCODING": "utf-8"}, check=False, ) self.assertEqual(result.returncode, expected_code) self.assertIn("已有重型媒体任务运行", result.stderr) self.assertNotIn("视频文件不存在", result.stderr) self.assertNotIn("输入视频不存在", result.stderr) def test_win32_guard_failures_close_every_created_handle(self) -> None: def kernel(**values): defaults = { "CreateMutexW": unittest.mock.Mock(return_value=101), "WaitForSingleObject": unittest.mock.Mock(return_value=0), "ReleaseMutex": unittest.mock.Mock(return_value=True), "CloseHandle": unittest.mock.Mock(return_value=True), "GetCurrentProcess": unittest.mock.Mock(return_value=202), "GetPriorityClass": unittest.mock.Mock(return_value=0x20), "SetPriorityClass": unittest.mock.Mock(return_value=True), } defaults.update(values) return types.SimpleNamespace(**defaults) scenarios = ( ("wait", kernel(WaitForSingleObject=unittest.mock.Mock(return_value=0x102))), ("get", kernel(GetPriorityClass=unittest.mock.Mock(return_value=0))), ("set", kernel(SetPriorityClass=unittest.mock.Mock(return_value=False))), ) for name, fake in scenarios: with self.subTest(name=name), patch.object( transcribe_media, "_is_windows", return_value=True ), patch.object(transcribe_media, "_windows_kernel32", return_value=fake), patch.object( transcribe_media, "_win32_last_error", return_value=5 ): with self.assertRaises(transcribe_media.MediaHostGuardError): with transcribe_media.media_host_guard(): self.fail("guard must not yield") fake.CloseHandle.assert_called_once_with(101) if name == "wait": fake.ReleaseMutex.assert_not_called() else: fake.ReleaseMutex.assert_called_once_with(101) create_failed = kernel(CreateMutexW=unittest.mock.Mock(return_value=0)) with patch.object(transcribe_media, "_is_windows", return_value=True), patch.object( transcribe_media, "_windows_kernel32", return_value=create_failed ), patch.object(transcribe_media, "_win32_last_error", return_value=5): with self.assertRaises(transcribe_media.MediaHostGuardError): with transcribe_media.media_host_guard(): self.fail("guard must not yield") create_failed.CloseHandle.assert_not_called() def test_existing_output_is_not_overwritten(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) video = root / "input.mp4" video.write_bytes(b"video") output = root / "result" output.mkdir() existing = output / "input.txt" existing.write_text("keep", encoding="utf-8") with self.assertRaisesRegex( transcribe_media.MediaTranscriptionError, "默认不会覆盖" ): transcribe_media.transcribe_video( video, output, runner=RecordingRunner(), model_loader=lambda: self.fail("model should not load"), ) self.assertEqual(existing.read_text(encoding="utf-8"), "keep") def test_video_without_audio_fails_before_model_load(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: video = Path(temp_dir) / "silent.mp4" video.write_bytes(b"video") with self.assertRaisesRegex( transcribe_media.MediaTranscriptionError, "不包含可用音轨" ): transcribe_media.transcribe_video( video, runner=RecordingRunner(has_audio=False), model_loader=lambda: self.fail("model should not load"), ) def test_ffmpeg_failure_leaves_no_completed_output(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) video = root / "broken.mp4" video.write_bytes(b"video") with self.assertRaisesRegex( transcribe_media.MediaTranscriptionError, "FFmpeg 提取第一条音轨失败" ): transcribe_media.transcribe_video( video, runner=RecordingRunner(ffmpeg_ok=False), model_loader=lambda: self.fail("model should not load"), ) self.assertFalse((root / "broken.transcript").exists()) self.assertEqual(list(root.glob(".broken.transcribe-*")), []) def test_four_hour_chunk_plan_has_expected_overlap_and_edges(self) -> None: chunks = transcribe_media._plan_audio_chunks(14_400.0) self.assertEqual(len(chunks), 12) self.assertEqual( chunks[0], transcribe_media.AudioChunk(1, 12, 0.0, 1200.0, 0.0, 1205.0), ) self.assertEqual( chunks[1], transcribe_media.AudioChunk(2, 12, 1200.0, 2400.0, 1195.0, 2405.0), ) self.assertEqual( chunks[-1], transcribe_media.AudioChunk( 12, 12, 13200.0, 14400.0, 13195.0, 14400.0 ), ) self.assertEqual(transcribe_media._plan_audio_chunks(1200.0), ()) def test_chunk_midpoint_has_one_owner_and_global_timestamps(self) -> None: first, second = transcribe_media._plan_audio_chunks(2400.0 + 1.0)[:2] boundary = transcribe_media.TranscriptSegment(1198.0, 1202.0, "边界") first_result = transcribe_media._project_chunk_segments( (boundary,), first, 2401.0 ) second_local = transcribe_media.TranscriptSegment(3.0, 7.0, "边界") second_result = transcribe_media._project_chunk_segments( (second_local,), second, 2401.0 ) self.assertEqual(first_result, ()) self.assertEqual( second_result, (transcribe_media.TranscriptSegment(1198.0, 1202.0, "边界"),), ) def test_long_audio_reuses_one_model_and_first_detected_language(self) -> None: responses = [ [FakeSegment(10.0, 12.0, "第一块"), FakeSegment(1198.0, 1202.0, "丢弃")], [FakeSegment(3.0, 7.0, "边界归第二块")], [FakeSegment(4.0, 6.0, "末块")], ] with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) video = root / "long.mp4" source_bytes = b"long synthetic video" video.write_bytes(source_bytes) runner = RecordingRunner(audio_duration=2401.0) model = ScriptedModel(responses) runtime = transcribe_media.RuntimeInfo(1, ("float16",)) load_count = 0 def load_model(): nonlocal load_count load_count += 1 return model, runtime outputs, _ = transcribe_media.transcribe_video( video, runner=runner, model_loader=load_model, ) self.assertEqual(load_count, 1) self.assertEqual(len(model.calls), 3) self.assertEqual(model.calls[0][1], {"vad_filter": True}) self.assertEqual( model.calls[1][1], {"vad_filter": True, "language": "zh"} ) self.assertEqual( model.calls[2][1], {"vad_filter": True, "language": "zh"} ) payload = json.loads(outputs.json.read_text(encoding="utf-8")) self.assertEqual(payload["language"], "zh") self.assertEqual( [(item["start"], item["end"], item["text"]) for item in payload["segments"]], [ (10.0, 12.0, "第一块"), (1198.0, 1202.0, "边界归第二块"), (2399.0, 2401.0, "末块"), ], ) self.assertEqual(video.read_bytes(), source_bytes) ffmpeg_commands = [ command for command in runner.commands if Path(command[0]).name.lower().startswith("ffmpeg") ] self.assertEqual(len(ffmpeg_commands), 4) self.assertEqual(list(root.glob(".long.transcribe-*")), []) def test_long_audio_failure_cleans_chunk_staging_and_outputs(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) video = root / "long-fail.mp4" video.write_bytes(b"video") runner = RecordingRunner(audio_duration=2401.0) model = ScriptedModel( [[FakeSegment(1, 2, "ok")], [FakeSegment(1, 2, "never")]], fail_on_call=2, ) with self.assertRaisesRegex( transcribe_media.MediaTranscriptionError, "GPU 语音转写失败" ): transcribe_media.transcribe_video( video, runner=runner, model_loader=lambda: ( model, transcribe_media.RuntimeInfo(1, ("float16",)), ), ) self.assertFalse((root / "long-fail.transcript").exists()) self.assertEqual(list(root.glob(".long-fail.transcribe-*")), []) self.assertEqual(len(model.calls), 2) def test_commit_keyboard_interrupt_rolls_back_after_each_partial_move(self) -> None: for interrupt_after in (1, 2, 3): with self.subTest(interrupt_after=interrupt_after): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) video = root / f"interrupt-{interrupt_after}.mp4" source_bytes = b"video" video.write_bytes(source_bytes) output = root / "result" runner = RecordingRunner() model = FakeModel() original_replace = type(video).replace interrupt = KeyboardInterrupt(f"after move {interrupt_after}") move_count = 0 def replace_then_interrupt(path, destination): nonlocal move_count result = original_replace(path, destination) move_count += 1 if move_count == interrupt_after: raise interrupt return result with patch.object( type(video), "replace", new=replace_then_interrupt ): with self.assertRaises(KeyboardInterrupt) as caught: transcribe_media.transcribe_video( video, output, runner=runner, model_loader=lambda: ( model, transcribe_media.RuntimeInfo(1, ("float16",)), ), ) self.assertIs(caught.exception, interrupt) self.assertFalse(output.exists()) self.assertEqual( list(root.glob(f".{video.stem}.transcribe-*")), [] ) self.assertEqual(video.read_bytes(), source_bytes) def test_commit_oserror_is_wrapped_and_partial_output_is_rolled_back(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) video = root / "ordinary-error.mp4" video.write_bytes(b"video") output = root / "result" original_replace = type(video).replace move_count = 0 def fail_before_second_move(path, destination): nonlocal move_count move_count += 1 if move_count == 2: raise OSError("synthetic move failure") return original_replace(path, destination) with patch.object(type(video), "replace", new=fail_before_second_move): with self.assertRaisesRegex( transcribe_media.MediaTranscriptionError, "提交输出文件失败，已回滚已写结果", ): transcribe_media.transcribe_video( video, output, runner=RecordingRunner(), model_loader=lambda: ( FakeModel(), transcribe_media.RuntimeInfo(1, ("float16",)), ), ) self.assertFalse(output.exists()) self.assertEqual(list(root.glob(".ordinary-error.transcribe-*")), []) def test_cuda_model_loader_never_requests_cpu(self) -> None: calls: list[tuple[str, dict[str, object]]] = [] class WhisperModel: def __init__(self, name: str, **kwargs: object) -> None: calls.append((name, kwargs)) fake_ctranslate2 = types.ModuleType("ctranslate2") fake_ctranslate2.get_cuda_device_count = lambda: 1 fake_ctranslate2.get_supported_compute_types = lambda device: {"float16", "int8"} fake_faster_whisper = types.ModuleType("faster_whisper") fake_faster_whisper.WhisperModel = WhisperModel with patch.dict( sys.modules, { "ctranslate2": fake_ctranslate2, "faster_whisper": fake_faster_whisper, }, ): _, runtime = transcribe_media._load_cuda_model() self.assertEqual(runtime.cuda_device_count, 1) self.assertEqual(calls, [("large-v3", {"device": "cuda", "compute_type": "float16"})]) if __name__ == "__main__": unittest.main()