diff --git a/src/python/tools/archive_transcriber.py b/src/python/tools/archive_transcriber.py index 4f06ecb..7e627fe 100644 --- a/src/python/tools/archive_transcriber.py +++ b/src/python/tools/archive_transcriber.py @@ -131,6 +131,10 @@ def signal_handler(signum: int, frame: Optional[FrameType]) -> None: LOGGER = logging.getLogger("archive_transcriber") +# Startup job discovery is NFS-latency-bound (stat/exists round trips), not +# CPU-bound; a wide thread pool turns hours of sequential stats into minutes. +STARTUP_STAT_THREADS = 64 + class SegmentLike(Protocol): """Protocol for transcription segment objects.""" @@ -1002,41 +1006,58 @@ def discover_video_jobs( skipped_count = 0 last_log_time = time.time() - for candidates in grouped.values(): - best_path = select_best_variant(candidates) - if not best_path: - continue - normalized_name = normalise_variant_name(best_path) - ru_vtt, en_vtt, ttml_path, smil_path = build_output_artifacts( - best_path, normalized_name, input_root, output_root + def build_job(candidates: List[Path]) -> Optional[VideoJob]: + """Return the job for a variant group, or None if empty/already processed.""" + try: + best_path = select_best_variant(candidates) + if not best_path: + return None + normalized_name = normalise_variant_name(best_path) + ru_vtt, en_vtt, ttml_path, smil_path = build_output_artifacts( + best_path, normalized_name, input_root, output_root + ) + if should_skip(best_path, ru_vtt, en_vtt, ttml_path, smil_path, force, ttml_enabled): + LOGGER.debug("Skipping already processed %s", best_path) + return None + except OSError as exc: + LOGGER.warning("Skipping group %s: stat failed during discovery: %s", candidates[0], exc) + return None + return VideoJob( + video_path=best_path, + normalized_name=normalized_name, + ru_vtt=ru_vtt, + en_vtt=en_vtt, + ttml=ttml_path, + smil=smil_path, ) - if should_skip(best_path, ru_vtt, en_vtt, ttml_path, smil_path, force, ttml_enabled): - LOGGER.debug("Skipping already processed %s", best_path) - skipped_count += 1 - else: - jobs.append( - VideoJob( - video_path=best_path, - normalized_name=normalized_name, - ru_vtt=ru_vtt, - en_vtt=en_vtt, - ttml=ttml_path, - smil=smil_path, - ) - ) + # The per-group checks are NFS-latency-bound; run them in parallel threads + # (stat/exists release the GIL) instead of one round trip at a time. + with ThreadPoolExecutor(max_workers=STARTUP_STAT_THREADS) as executor: + try: + for job in executor.map(build_job, grouped.values(), chunksize=16): + if job is not None: + jobs.append(job) + else: + skipped_count += 1 - processed_groups += 1 - current_time = time.time() - if current_time - last_log_time >= 5: - LOGGER.info( - "Building job list... processed %d/%d groups (%d to process, %d already done)", - processed_groups, - len(grouped), - len(jobs), - skipped_count, - ) - last_log_time = current_time + processed_groups += 1 + current_time = time.time() + if current_time - last_log_time >= 5: + LOGGER.info( + "Building job list... processed %d/%d groups (%d to process, %d already done)", + processed_groups, + len(grouped), + len(jobs), + skipped_count, + ) + last_log_time = current_time + except BaseException: + # Ctrl+C (or a worker error) must not wait for the queued backlog: + # executor.map submits every group up front, and the context + # manager's shutdown would otherwise drain all of them first. + executor.shutdown(wait=False, cancel_futures=True) + raise LOGGER.info( "Job list complete! %d videos to process (%d already done)", @@ -1048,19 +1069,34 @@ def discover_video_jobs( def select_best_variant(candidates: List[Path]) -> Optional[Path]: - best_path: Optional[Path] = None - best_score = (-1, -1) + """Pick the highest-resolution variant, falling back to file size on ties. + Stat calls are only made when the filename resolution is ambiguous — on a + 143k-group archive over NFS this avoids ~700k round trips at startup. + """ + if not candidates: + return None + if len(candidates) == 1: + return candidates[0] + + by_priority: Dict[int, List[Path]] = {} for path in candidates: - resolution = extract_resolution(path.name) - priority = resolution or 0 + priority = extract_resolution(path.name) or 0 + by_priority.setdefault(priority, []).append(path) + + contenders = by_priority[max(by_priority)] + if len(contenders) == 1: + return contenders[0] + + best_path: Optional[Path] = None + best_size = -1 + for path in contenders: try: size = path.stat().st_size except OSError: size = 0 - score = (priority, size) - if score > best_score: - best_score = score + if size > best_size: + best_size = size best_path = path return best_path @@ -1120,6 +1156,35 @@ def needs_translation(job: VideoJob, ttml_enabled: bool) -> bool: return True +def phase_needs(job: VideoJob, ttml_enabled: bool) -> Tuple[bool, bool]: + """Combined needs_transcription/needs_translation check with one stat per file. + + Semantically identical to calling the two functions separately, but stats + each artifact exactly once (5 round trips max instead of ~11) — this check + runs once per video across a 143k-video archive on NFS at every startup. + """ + + def _mtime(path: Path) -> Optional[float]: + try: + return path.stat().st_mtime + except OSError: + return None + + ru_mtime = _mtime(job.ru_vtt) + if ru_mtime is None: + return True, False # needs transcription; can't translate without it + + video_mtime = _mtime(job.video_path) + need_transcription = video_mtime is None or ru_mtime < video_mtime + + required = [_mtime(job.en_vtt), _mtime(job.smil)] + if ttml_enabled: + required.append(_mtime(job.ttml)) + need_translation = any(m is None or m < ru_mtime for m in required) + + return need_transcription, need_translation + + 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) @@ -2013,21 +2078,34 @@ def run_two_phase( transcription_jobs: List[VideoJob] = [] translation_jobs: List[VideoJob] = [] last_log_time = time.time() + + def check_job(j: VideoJob) -> Tuple[bool, bool]: + return phase_needs(j, ttml_enabled) + try: - for i, j in enumerate(all_jobs): - if needs_transcription(j): - transcription_jobs.append(j) - if needs_translation(j, ttml_enabled): - translation_jobs.append(j) - if time.time() - last_log_time >= 5: - LOGGER.info( - " Filtering: checked %d/%d (%d transcription, %d translation)", - i + 1, - len(all_jobs), - len(transcription_jobs), - len(translation_jobs), - ) - last_log_time = time.time() + # NFS-latency-bound; parallel threads keep job order via executor.map + with ThreadPoolExecutor(max_workers=STARTUP_STAT_THREADS) as executor: + try: + for i, (j, (need_ts, need_tl)) in enumerate( + zip(all_jobs, executor.map(check_job, all_jobs, chunksize=16)) + ): + if need_ts: + transcription_jobs.append(j) + if need_tl: + translation_jobs.append(j) + if time.time() - last_log_time >= 5: + LOGGER.info( + " Filtering: checked %d/%d (%d transcription, %d translation)", + i + 1, + len(all_jobs), + len(transcription_jobs), + len(translation_jobs), + ) + last_log_time = time.time() + except BaseException: + # Don't drain the queued backlog on interrupt (see discover_video_jobs) + executor.shutdown(wait=False, cancel_futures=True) + raise except KeyboardInterrupt: LOGGER.warning("Interrupted during filtering") return 130 diff --git a/tests/test_archive_transcriber.py b/tests/test_archive_transcriber.py index b05900c..0c8ddc5 100644 --- a/tests/test_archive_transcriber.py +++ b/tests/test_archive_transcriber.py @@ -534,3 +534,45 @@ def test_sample_rate_preserved(self): ar_index = cmd.index("-ar") assert cmd[ar_index + 1] == "22050" print("✓ test_sample_rate_preserved passed") + + +class TestPhaseNeeds: + """phase_needs must match needs_transcription/needs_translation exactly.""" + + def _job(self, tmp: Path): + return VideoJob( + video_path=tmp / "video_1080p.mp4", + normalized_name="video.mp4", + ru_vtt=tmp / "video.ru.vtt", + en_vtt=tmp / "video.en.vtt", + ttml=tmp / "video.ttml", + smil=tmp / "video.smil", + ) + + def test_matches_separate_checks_across_all_states(self): + import itertools + import os + + phase_needs = archive_transcriber.phase_needs + needs_transcription = archive_transcriber.needs_transcription + needs_translation = archive_transcriber.needs_translation + + # Every combination of artifact presence, with fresh and stale mtimes + names = ["video_1080p.mp4", "video.ru.vtt", "video.en.vtt", "video.ttml", "video.smil"] + for present in itertools.product([False, True], repeat=len(names)): + for stale_ru in (False, True): + with tempfile.TemporaryDirectory() as tmpdir: + tmp = Path(tmpdir) + for name, exists in zip(names, present): + if exists: + (tmp / name).write_text("x") + os.utime(tmp / name, (2000000, 2000000)) + ru = tmp / "video.ru.vtt" + if stale_ru and ru.exists(): + os.utime(ru, (1000000, 1000000)) # older than everything + job = self._job(tmp) + for ttml_enabled in (False, True): + expected = (needs_transcription(job), needs_translation(job, ttml_enabled)) + assert phase_needs(job, ttml_enabled) == expected, ( + f"mismatch for present={present} stale_ru={stale_ru} ttml={ttml_enabled}" + )