From ab8ad7abb37dc461a9a0267f70dc166290ce3367 Mon Sep 17 00:00:00 2001 From: Ark Deliev Date: Sat, 1 Aug 2026 08:31:07 +0000 Subject: [PATCH 1/2] fix: stop daily OOM crash loop and hours-long post-Phase-1 refilter The service has been OOM-killed daily since Jul 16: every cycle Phase 2 picked the same four multi-hour videos first, four concurrent model.transcribe() calls grew past 31 GB RSS (no swap), and the kernel killed the process before any translations completed. Between Phase 1 and Phase 2 the job list was also re-filtered by re-statting all 143k jobs sequentially on NFS, adding 6-12 h of dead time per cycle. - Re-filter only the jobs Phase 1 touched (a translation becomes newly needed only when a fresh ru.vtt appears), in parallel, deduped against the already-queued translation jobs - Sort both phase queues smallest video first: concurrent workers hold similarly-sized jobs (bounded combined peak RSS) and steady progress happens before the multi-hour videos - Serialize long videos: anything over --long-video-minutes (default 45) takes an exclusive semaphore so at most one long transcribe runs at a time, in both phases - Record Phase 1 failures in the manifest with video_mtime; audio extraction failures (corrupt containers) are permanent and skipped on later cycles until the video file changes - the 93 known-broken videos stop retrying every cycle - Decode ffmpeg/ffprobe output with errors=replace: corrupt containers emit non-UTF-8 bytes on stderr and the UnicodeDecodeError was masking the real extraction error - Probe video metadata inside the Phase 2 try block so a corrupt video yields an error record instead of an exception escaping the worker --- src/python/tools/archive_transcriber.py | 183 ++++++++++++++++++++- tests/test_archive_transcriber.py | 206 ++++++++++++++++++++++++ 2 files changed, 381 insertions(+), 8 deletions(-) diff --git a/src/python/tools/archive_transcriber.py b/src/python/tools/archive_transcriber.py index 7e627fe..7388ed7 100644 --- a/src/python/tools/archive_transcriber.py +++ b/src/python/tools/archive_transcriber.py @@ -135,6 +135,19 @@ def signal_handler(signum: int, frame: Optional[FrameType]) -> None: # CPU-bound; a wide thread pool turns hours of sequential stats into minutes. STARTUP_STAT_THREADS = 64 +# Peak RSS during model.transcribe() grows with audio duration (~3 GB for a +# 27-minute video on faster-whisper 1.2.1); four workers on multi-hour videos +# exceeded the 31 GB of this box (no swap) and got the service OOM-killed +# daily. Videos above the --long-video-minutes threshold therefore hold this +# semaphore so only one of them transcribes at a time. +LONG_VIDEO_GATE = threading.Semaphore(1) + +# Error types that will not succeed on retry while the input is unchanged +# (e.g. FFmpeg cannot extract audio from a corrupt container). Jobs whose +# last manifest record carries one of these and an unchanged video mtime are +# excluded from Phase 1 instead of failing again every cycle. +PERMANENT_ERROR_TYPES = {"audio_extraction"} + class SegmentLike(Protocol): """Protocol for transcription segment objects.""" @@ -169,6 +182,7 @@ class ManifestRecord(TypedDict, total=False): phase: str worker_info: str processing_mode: str + video_mtime: Optional[float] # Register signal handlers for graceful shutdown and pause @@ -575,7 +589,7 @@ def probe_video_metadata(video_path: Path) -> VideoMetadata: ] try: - result = subprocess.run(command, capture_output=True, text=True, check=True) + result = subprocess.run(command, capture_output=True, text=True, errors="replace", check=True) data = cast(Dict[str, Any], json.loads(result.stdout or "{}")) except (subprocess.CalledProcessError, json.JSONDecodeError) as exc: LOGGER.warning("Failed to probe metadata for %s: %s", video_path, exc) @@ -1185,6 +1199,70 @@ def _mtime(path: Path) -> Optional[float]: return need_transcription, need_translation +def known_permanent_failure(job: VideoJob, manifest: ManifestProtocol) -> bool: + """True if the video's last manifest record is a permanent error and the video is unchanged. + + Prevents unfixable inputs (corrupt containers FFmpeg cannot read) from + being retried every cycle. A changed video mtime (file replaced/repaired) + makes the job eligible again. + """ + record = manifest.get(job.video_path) + if record is None or record.get("status") != "error": + return False + if record.get("error_type") not in PERMANENT_ERROR_TYPES: + return False + recorded_mtime = record.get("video_mtime") + if recorded_mtime is None: + return False + try: + return job.video_path.stat().st_mtime == float(recorded_mtime) + except OSError: + return False + + +def sort_jobs_by_size(jobs: List[VideoJob]) -> List[VideoJob]: + """Order jobs smallest video first (size is a cheap NFS-side proxy for duration). + + Concurrent workers then always hold similarly-sized jobs, which bounds + their combined transcribe peak RSS, and the multi-hour videos that need + the LONG_VIDEO_GATE run last — after the bulk of the queue has completed — + instead of monopolizing the head of every cycle. + """ + + def _size(job: VideoJob) -> int: + try: + return job.video_path.stat().st_size + except OSError: + return 0 + + with ThreadPoolExecutor(max_workers=STARTUP_STAT_THREADS) as executor: + try: + sizes = list(executor.map(_size, jobs, chunksize=16)) + except BaseException: + executor.shutdown(wait=False, cancel_futures=True) + raise + order = sorted(range(len(jobs)), key=sizes.__getitem__) + return [jobs[i] for i in order] + + +def acquire_long_video_slot(duration: Optional[float], threshold_minutes: float, video_path: Path) -> bool: + """Block until this video may transcribe; True if the exclusive long-video slot was taken. + + The caller must release LONG_VIDEO_GATE when done iff this returns True. + Videos shorter than the threshold (or an unknown duration, or a disabled + threshold <= 0) never wait and never take the slot. + """ + if threshold_minutes <= 0 or not duration or duration < threshold_minutes * 60: + return False + LOGGER.info( + "[LongVideo] %s is %.0f min long; waiting for exclusive long-video slot", + video_path, + duration / 60, + ) + LONG_VIDEO_GATE.acquire() + return True + + def extract_audio(video_path: Path, sample_rate: int) -> Path: tmp_file = tempfile.NamedTemporaryFile(suffix=".wav", delete=False) tmp_file_path = Path(tmp_file.name) @@ -1206,7 +1284,10 @@ def extract_audio(video_path: Path, sample_rate: int) -> Path: ] LOGGER.debug("Running FFmpeg: %s", " ".join(command)) - result = subprocess.run(command, capture_output=True, text=True) + # errors="replace": corrupt containers make FFmpeg emit non-UTF-8 bytes on + # stderr; a strict decode would raise UnicodeDecodeError here and mask the + # actual extraction failure. + result = subprocess.run(command, capture_output=True, text=True, errors="replace") if result.returncode != 0: stderr_preview = (result.stderr or "").splitlines()[-5:] raise RuntimeError( @@ -1253,7 +1334,7 @@ def detect_audio_start_time( ] try: - result = subprocess.run(cmd, capture_output=True, text=True, timeout=30) + result = subprocess.run(cmd, capture_output=True, text=True, errors="replace", timeout=30) # Parse FFmpeg output for silence end times silence_end_times: List[float] = [] for line in result.stderr.split("\n"): @@ -1558,8 +1639,17 @@ def process_transcription_only(job: VideoJob, args: argparse.Namespace, quiet: b filter_words: List[str] = load_filter_words() audio_path: Optional[Path] = None + holds_long_slot = False + long_video_minutes = float(getattr(args, "long_video_minutes", 0) or 0) try: + if long_video_minutes > 0: + try: + duration_hint = probe_video_metadata(job.video_path).duration + except Exception: + duration_hint = None # broken container: extract_audio below reports the real error + holds_long_slot = acquire_long_video_slot(duration_hint, long_video_minutes, job.video_path) + audio_path = extract_audio(job.video_path, args.sample_rate) LOGGER.debug("[Transcription] Loading model %s...", args.model) model: Any = get_model(args) @@ -1606,17 +1696,29 @@ def process_transcription_only(job: VideoJob, args: argparse.Namespace, quiet: b except Exception as exc: LOGGER.error("[Transcription] Failed %s: %s", job.video_path, exc) + # Audio-extraction failures are input defects that retrying cannot fix; + # record the video mtime so the job is skipped until the file changes. + is_extract_failure = isinstance(exc, RuntimeError) and str(exc).startswith("FFmpeg failed") + video_mtime: Optional[float] = None + try: + video_mtime = job.video_path.stat().st_mtime + except OSError: + pass return { "video_path": str(job.video_path), "ru_vtt": str(job.ru_vtt), "status": "error", "phase": "transcription", "error": str(exc), + "error_type": "audio_extraction" if is_extract_failure else "processing", + "video_mtime": video_mtime, "processed_at": human_time(), "worker_info": get_worker_info(), } finally: + if holds_long_slot: + LONG_VIDEO_GATE.release() if audio_path and audio_path.exists(): try: audio_path.unlink() @@ -1639,10 +1741,16 @@ def process_translation_only( return skipped filter_words: List[str] = load_filter_words() - metadata = probe_video_metadata(job.video_path) audio_path: Optional[Path] = None + holds_long_slot = False + long_video_minutes = float(getattr(args, "long_video_minutes", 0) or 0) try: + # Probe inside the try: a corrupt container must produce an error + # record, not an exception that escapes the worker thread. + metadata = probe_video_metadata(job.video_path) + holds_long_slot = acquire_long_video_slot(metadata.duration, long_video_minutes, job.video_path) + # Read existing Russian VTT for segment count comparison ru_content = job.ru_vtt.read_text(encoding="utf-8") if job.ru_vtt.exists() else "" ru_cues: List[Any] = parse_vtt_content(ru_content) if ru_content else [] @@ -1764,6 +1872,8 @@ def __init__(self, text: str) -> None: return error_record finally: + if holds_long_slot: + LONG_VIDEO_GATE.release() if audio_path and audio_path.exists(): try: audio_path.unlink() @@ -1867,6 +1977,15 @@ def parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace: help="Comma-separated list of video extensions to include", ) parser.add_argument("--workers", type=int, default=1, help="Number of worker threads for processing") + parser.add_argument( + "--long-video-minutes", + type=float, + default=45.0, + help=( + "Videos longer than this many minutes are transcribed one at a time " + "to bound peak RAM (transcribe RSS grows with duration); 0 disables the gate" + ), + ) parser.add_argument( "--gpus", type=str, @@ -2110,6 +2229,19 @@ def check_job(j: VideoJob) -> Tuple[bool, bool]: LOGGER.warning("Interrupted during filtering") return 130 + # Drop inputs already recorded as permanently broken (corrupt containers + # fail identically every cycle until the file itself is replaced). + retryable = [j for j in transcription_jobs if not known_permanent_failure(j, manifest)] + if len(retryable) != len(transcription_jobs): + LOGGER.info( + "Skipping %d videos with recorded permanent failures (unchanged since last attempt)", + len(transcription_jobs) - len(retryable), + ) + transcription_jobs = retryable + + # Smallest first: bulk progress up front, OOM-prone giants last (and gated). + transcription_jobs = sort_jobs_by_size(transcription_jobs) + LOGGER.info( "Two-phase mode: %d transcriptions needed, %d translations needed (from %d total videos)", len(transcription_jobs), @@ -2155,6 +2287,10 @@ def update_progress(record: ManifestRecord) -> None: video_path, record.get("error", "unknown"), ) + # Persist failures so known_permanent_failure() can exclude + # unfixable inputs from the next cycle. + if record.get("status") == "error": + manifest.append(record) if progress_bar is not None: cast(Any, progress_bar).set_postfix(ok=phase1_successes, fail=phase1_failures, refresh=False) progress_bar.update(1) @@ -2194,13 +2330,44 @@ def update_progress(record: ManifestRecord) -> None: total_successes += phase1_successes total_failures += phase1_failures - # Re-evaluate translation jobs after transcription phase - # (newly transcribed files may now be ready for translation) - translation_jobs = [j for j in all_jobs if needs_translation(j, ttml_enabled)] - LOGGER.info("After Phase 1: %d translations now needed", len(translation_jobs)) + # Re-evaluate translation needs after transcription, but only for the + # jobs Phase 1 actually touched: a translation becomes newly needed + # only when a fresh ru.vtt appeared. Re-statting all jobs here (and + # sequentially at that) used to add hours of dead time per cycle on + # a 143k-video NFS archive. + already_queued = {j.video_path for j in translation_jobs} + recheck = [j for j in transcription_jobs if j.video_path not in already_queued] + newly_ready: List[VideoJob] = [] + + def recheck_job(j: VideoJob) -> bool: + return needs_translation(j, ttml_enabled) + + try: + with ThreadPoolExecutor(max_workers=STARTUP_STAT_THREADS) as executor: + try: + for j, need_tl in zip(recheck, executor.map(recheck_job, recheck, chunksize=16)): + if need_tl: + newly_ready.append(j) + except BaseException: + executor.shutdown(wait=False, cancel_futures=True) + raise + except KeyboardInterrupt: + LOGGER.warning("Interrupted during post-transcription re-filter") + return 130 + + translation_jobs.extend(newly_ready) + LOGGER.info( + "After Phase 1: %d translations now needed (%d from new transcriptions)", + len(translation_jobs), + len(newly_ready), + ) else: LOGGER.info("=== PHASE 1: No transcriptions needed ===") + # Smallest first (see sort_jobs_by_size): steady progress before the + # multi-hour videos, and concurrent workers hold similarly-sized jobs. + translation_jobs = sort_jobs_by_size(translation_jobs) + # Phase 2: Translations if translation_jobs: LOGGER.info( diff --git a/tests/test_archive_transcriber.py b/tests/test_archive_transcriber.py index 0c8ddc5..be27d5b 100644 --- a/tests/test_archive_transcriber.py +++ b/tests/test_archive_transcriber.py @@ -576,3 +576,209 @@ def test_matches_separate_checks_across_all_states(self): assert phase_needs(job, ttml_enabled) == expected, ( f"mismatch for present={present} stale_ru={stale_ru} ttml={ttml_enabled}" ) + + +def _make_job(tmp: Path, stem: str = "video"): + return VideoJob( + video_path=tmp / f"{stem}_1080p.mp4", + normalized_name=f"{stem}.mp4", + ru_vtt=tmp / f"{stem}.ru.vtt", + en_vtt=tmp / f"{stem}.en.vtt", + ttml=tmp / f"{stem}.ttml", + smil=tmp / f"{stem}.smil", + ) + + +class TestKnownPermanentFailure: + """Videos with recorded permanent errors must not be retried while unchanged.""" + + def _manifest(self, tmp: Path): + return Manifest(tmp / "manifest.jsonl") + + def _error_record(self, job, video_mtime, error_type="audio_extraction", status="error"): + return { + "video_path": str(job.video_path), + "status": status, + "phase": "transcription", + "error": "FFmpeg failed: return code 234", + "error_type": error_type, + "video_mtime": video_mtime, + "processed_at": "2026-08-01 00:00:00", + } + + def test_skips_unchanged_video_with_extraction_error(self): + with tempfile.TemporaryDirectory() as tmpdir: + tmp = Path(tmpdir) + job = _make_job(tmp) + job.video_path.write_bytes(b"corrupt") + manifest = self._manifest(tmp) + manifest.append(self._error_record(job, job.video_path.stat().st_mtime)) + + assert archive_transcriber.known_permanent_failure(job, manifest) is True + + def test_retries_when_video_changed(self): + with tempfile.TemporaryDirectory() as tmpdir: + tmp = Path(tmpdir) + job = _make_job(tmp) + job.video_path.write_bytes(b"corrupt") + manifest = self._manifest(tmp) + manifest.append(self._error_record(job, job.video_path.stat().st_mtime - 100.0)) + + assert archive_transcriber.known_permanent_failure(job, manifest) is False + + def test_retries_transient_error_types(self): + with tempfile.TemporaryDirectory() as tmpdir: + tmp = Path(tmpdir) + job = _make_job(tmp) + job.video_path.write_bytes(b"ok") + manifest = self._manifest(tmp) + manifest.append(self._error_record(job, job.video_path.stat().st_mtime, error_type="processing")) + + assert archive_transcriber.known_permanent_failure(job, manifest) is False + + def test_retries_without_record_or_mtime(self): + with tempfile.TemporaryDirectory() as tmpdir: + tmp = Path(tmpdir) + job = _make_job(tmp) + job.video_path.write_bytes(b"ok") + manifest = self._manifest(tmp) + assert archive_transcriber.known_permanent_failure(job, manifest) is False + + manifest.append(self._error_record(job, None)) + assert archive_transcriber.known_permanent_failure(job, manifest) is False + + def test_success_record_never_blocks(self): + with tempfile.TemporaryDirectory() as tmpdir: + tmp = Path(tmpdir) + job = _make_job(tmp) + job.video_path.write_bytes(b"ok") + manifest = self._manifest(tmp) + manifest.append( + self._error_record(job, job.video_path.stat().st_mtime, status="success"), + ) + assert archive_transcriber.known_permanent_failure(job, manifest) is False + + +class TestSortJobsBySize: + """Phase queues run smallest video first; unreadable paths sort first and fail fast.""" + + def test_orders_smallest_first(self): + with tempfile.TemporaryDirectory() as tmpdir: + tmp = Path(tmpdir) + jobs = [] + for stem, size in (("big", 3000), ("small", 10), ("mid", 500)): + job = _make_job(tmp, stem) + job.video_path.write_bytes(b"x" * size) + jobs.append(job) + + ordered = archive_transcriber.sort_jobs_by_size(jobs) + assert [j.normalized_name for j in ordered] == ["small.mp4", "mid.mp4", "big.mp4"] + + def test_missing_file_sorts_first(self): + with tempfile.TemporaryDirectory() as tmpdir: + tmp = Path(tmpdir) + present = _make_job(tmp, "present") + present.video_path.write_bytes(b"x" * 100) + missing = _make_job(tmp, "missing") + + ordered = archive_transcriber.sort_jobs_by_size([present, missing]) + assert ordered[0] is missing + + def test_empty_list(self): + assert archive_transcriber.sort_jobs_by_size([]) == [] + + +class TestAcquireLongVideoSlot: + """Only videos over the threshold take the exclusive slot; others never wait.""" + + def _release_if(self, acquired: bool) -> None: + if acquired: + archive_transcriber.LONG_VIDEO_GATE.release() + + def test_short_video_does_not_take_slot(self): + acquired = archive_transcriber.acquire_long_video_slot(10 * 60, 45.0, Path("v.mp4")) + self._release_if(acquired) + assert acquired is False + + def test_long_video_takes_and_holds_slot(self): + acquired = archive_transcriber.acquire_long_video_slot(120 * 60, 45.0, Path("v.mp4")) + try: + assert acquired is True + # Slot is exclusive while held + assert archive_transcriber.LONG_VIDEO_GATE.acquire(blocking=False) is False + finally: + self._release_if(acquired) + # And free again after release + assert archive_transcriber.LONG_VIDEO_GATE.acquire(blocking=False) is True + archive_transcriber.LONG_VIDEO_GATE.release() + + def test_disabled_threshold_never_gates(self): + acquired = archive_transcriber.acquire_long_video_slot(999 * 60, 0, Path("v.mp4")) + self._release_if(acquired) + assert acquired is False + + def test_unknown_duration_never_gates(self): + for duration in (None, 0.0): + acquired = archive_transcriber.acquire_long_video_slot(duration, 45.0, Path("v.mp4")) + self._release_if(acquired) + assert acquired is False + + +class TestTranscriptionErrorRecord: + """Extraction failures must be recorded as permanent with the video mtime.""" + + def _args(self): + import argparse + + return argparse.Namespace( + sample_rate=16000, + model="large-v3", + beam_size=5, + source_language="ru", + vad_filter=True, + trim_silence=False, + verbose=False, + long_video_minutes=0, + ) + + def test_ffmpeg_failure_marked_permanent(self, monkeypatch): + with tempfile.TemporaryDirectory() as tmpdir: + tmp = Path(tmpdir) + job = _make_job(tmp) + job.video_path.write_bytes(b"corrupt") + # Valid transcoder SMIL so the precheck passes + job.smil.write_text( + "" + "" + ) + + def boom(video_path, sample_rate): + raise RuntimeError(f"FFmpeg failed for {video_path}: return code 234") + + monkeypatch.setattr(archive_transcriber, "extract_audio", boom) + record = archive_transcriber.process_transcription_only(job, self._args(), quiet=True) + + assert record["status"] == "error" + assert record["error_type"] == "audio_extraction" + assert record["video_mtime"] == job.video_path.stat().st_mtime + + def test_other_failure_not_marked_permanent(self, monkeypatch): + with tempfile.TemporaryDirectory() as tmpdir: + tmp = Path(tmpdir) + job = _make_job(tmp) + job.video_path.write_bytes(b"ok") + job.smil.write_text( + "" + "" + ) + + def boom(video_path, sample_rate): + raise ValueError("CUDA hiccup") + + monkeypatch.setattr(archive_transcriber, "extract_audio", boom) + record = archive_transcriber.process_transcription_only(job, self._args(), quiet=True) + + assert record["status"] == "error" + assert record["error_type"] == "processing" From d23e078cdfdb5c976ec879b3b46536db5bbf0c12 Mon Sep 17 00:00:00 2001 From: Ark Deliev Date: Sat, 1 Aug 2026 08:41:05 +0000 Subject: [PATCH 2/2] review: bound the long-video gate, persist Phase 1 skips, cover re-filter dedup - LONG_VIDEO_GATE is a BoundedSemaphore so an unbalanced release raises instead of silently allowing two concurrent long videos - Persist every non-success Phase 1 record, not just errors: invalid-SMIL skips now survive restarts as skip_record_for_invalid_smil documents - Add a run_two_phase integration test covering the post-Phase-1 re-filter: a video needing both phases must not be queued for translation twice (verified to fail without the dedup guard) --- src/python/tools/archive_transcriber.py | 10 ++- tests/test_archive_transcriber.py | 98 +++++++++++++++++++++++++ 2 files changed, 104 insertions(+), 4 deletions(-) diff --git a/src/python/tools/archive_transcriber.py b/src/python/tools/archive_transcriber.py index 7388ed7..6e928d9 100644 --- a/src/python/tools/archive_transcriber.py +++ b/src/python/tools/archive_transcriber.py @@ -140,7 +140,9 @@ def signal_handler(signum: int, frame: Optional[FrameType]) -> None: # exceeded the 31 GB of this box (no swap) and got the service OOM-killed # daily. Videos above the --long-video-minutes threshold therefore hold this # semaphore so only one of them transcribes at a time. -LONG_VIDEO_GATE = threading.Semaphore(1) +# Bounded so an unbalanced release() raises instead of silently widening the +# gate and letting two long videos transcribe at once. +LONG_VIDEO_GATE = threading.BoundedSemaphore(1) # Error types that will not succeed on retry while the input is unchanged # (e.g. FFmpeg cannot extract audio from a corrupt container). Jobs whose @@ -2288,9 +2290,9 @@ def update_progress(record: ManifestRecord) -> None: record.get("error", "unknown"), ) # Persist failures so known_permanent_failure() can exclude - # unfixable inputs from the next cycle. - if record.get("status") == "error": - manifest.append(record) + # unfixable inputs from the next cycle, and so invalid-SMIL + # skips survive restarts as the skip_record contract promises. + manifest.append(record) if progress_bar is not None: cast(Any, progress_bar).set_postfix(ok=phase1_successes, fail=phase1_failures, refresh=False) progress_bar.update(1) diff --git a/tests/test_archive_transcriber.py b/tests/test_archive_transcriber.py index be27d5b..24d5084 100644 --- a/tests/test_archive_transcriber.py +++ b/tests/test_archive_transcriber.py @@ -2,6 +2,7 @@ """Unit tests for archive_transcriber.py core functionality.""" import importlib +import os import re import sys import tempfile @@ -10,6 +11,8 @@ from typing import Callable, Optional from unittest import mock +import pytest + # Mock faster_whisper before importing archive_transcriber sys.modules["faster_whisper"] = mock.MagicMock() # On 3.11+, typing.Required exists; alias typing_extensions to typing to avoid @@ -723,6 +726,101 @@ def test_unknown_duration_never_gates(self): self._release_if(acquired) assert acquired is False + def test_unbalanced_release_fails_loudly(self): + # BoundedSemaphore: an over-release must raise rather than silently + # widening the gate to allow two concurrent long videos. + with pytest.raises(ValueError): + archive_transcriber.LONG_VIDEO_GATE.release() + + +class TestTwoPhaseQueueing: + """Phase 2 must pick up newly transcribed videos exactly once, without re-statting everything.""" + + def _args(self, tmp: Path): + import argparse + + return argparse.Namespace( + max_files=None, + force_scan=True, + no_ttml=True, + workers=1, + progress=False, + verbose=False, + long_video_minutes=0, + ) + + def _make_video(self, tmp: Path, stem: str, *, ru=False, en=False, smil=False, stale_ru=False): + video = tmp / f"{stem}_1080p.mp4" + video.write_bytes(b"x" * 1000) + stamp = 2000000 if stale_ru else 1000000 + os.utime(video, (stamp, stamp)) + for flag, suffix in ((ru, "ru.vtt"), (en, "en.vtt"), (smil, "smil")): + if flag: + artifact = tmp / f"{stem}.{suffix}" + artifact.write_text("x") + # stale_ru: artifacts predate the video, so it needs both phases + os.utime(artifact, (1000000, 1000000)) + return video + + def test_newly_transcribed_video_translated_once(self, monkeypatch): + with tempfile.TemporaryDirectory() as tmpdir: + tmp = Path(tmpdir) + archive = tmp / "archive" + archive.mkdir() + + # A: nothing yet -> transcribe, then translate + self._make_video(archive, "a") + # B: fully done -> neither phase + self._make_video(archive, "b", ru=True, en=True, smil=True) + # C: transcript only -> translate only + self._make_video(archive, "c", ru=True, smil=True) + # D: stale transcript, no en.vtt -> needs BOTH phases. It is already + # in the Phase 2 queue before Phase 1 runs, so the post-Phase-1 + # re-filter must not queue it a second time. + self._make_video(archive, "d", ru=True, smil=True, stale_ru=True) + + transcribed: list[str] = [] + translated: list[str] = [] + + def fake_transcribe(job, args, quiet=False): + transcribed.append(job.normalized_name) + job.ru_vtt.write_text("WEBVTT\n") # now eligible for translation + return { + "video_path": str(job.video_path), + "status": "success", + "phase": "transcription", + "processed_at": "2026-08-01 00:00:00", + } + + def fake_translate(job, args, manifest, quiet=False): + translated.append(job.normalized_name) + return { + "video_path": str(job.video_path), + "status": "success", + "phase": "translation", + "processed_at": "2026-08-01 00:00:00", + } + + monkeypatch.setattr(archive_transcriber, "process_transcription_only", fake_transcribe) + monkeypatch.setattr(archive_transcriber, "process_translation_only", fake_translate) + + manifest = Manifest(tmp / "manifest.jsonl") + rc = archive_transcriber.run_two_phase( + self._args(tmp), + archive, + None, + manifest, + [".mp4"], + tmp / "scan_cache.json", + ) + + assert rc == 0 + assert sorted(transcribed) == ["a.mp4", "d.mp4"] + # c and d were queued up front, a became eligible after Phase 1 — + # each exactly once, and d is not duplicated by the re-filter + assert sorted(translated) == ["a.mp4", "c.mp4", "d.mp4"] + assert len(translated) == 3 + class TestTranscriptionErrorRecord: """Extraction failures must be recorded as permanent with the video mtime."""