diff --git a/src/python/tools/archive_transcriber.py b/src/python/tools/archive_transcriber.py
index 7e627fe..6e928d9 100644
--- a/src/python/tools/archive_transcriber.py
+++ b/src/python/tools/archive_transcriber.py
@@ -135,6 +135,21 @@ 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.
+# 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
+# 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 +184,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 +591,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 +1201,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 +1286,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 +1336,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 +1641,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 +1698,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 +1743,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 +1874,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 +1979,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 +2231,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 +2289,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, 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)
@@ -2194,13 +2332,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..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
@@ -576,3 +579,304 @@ 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
+
+ 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."""
+
+ 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"