JMR-devandClaude Opus 5
3bff193f92
Three defects found by truncating a real download and killing one mid-flight. Matroska and WebM write the duration into the header, so ffprobe reports the full 212.8s of a video.mkv cut off after a kilobyte -- exit 0, plausible answer. verify_video therefore returned OK, the corrupt file was never re-downloaded, ffmpeg extracted the 0.02s of audio it could find, and the run reported "1 ok" with an empty transcript. The size recorded at download time is the only evidence the bytes are still there, so it is now checked whenever it is known rather than only as a fallback when the duration is unreadable. The audio stage now takes the expected duration and rejects an extraction that does not match it. ffmpeg exits 0 on a truncated container, so without this a damaged source yields a confident transcript of near-silence, which is a worse outcome than a failed job. It compares against the video's own probed duration rather than state.duration_s, which can come from playlist metadata. aria2 saves its control file every 60s by default. Since that file is what a resume reads, a kill -9 inside the first minute preserved a control file recording zero completed pieces: measured 0/13 on the sample, so the "resume" re-downloaded the lot while reporting a partial. At --auto-save-interval=20 the same kill preserves 2/13 pieces and the resumed download is byte-identical to a clean one. The duration tolerance moves to media/ffmpeg.py, which both callers already import, instead of being restated per call site. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
191 lines
7.7 KiB
Python
191 lines
7.7 KiB
Python
from __future__ import annotations
|
|
|
|
import shutil
|
|
import subprocess
|
|
from typing import TYPE_CHECKING
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from ccn_transcribe import errors
|
|
from ccn_transcribe.media import ffmpeg
|
|
|
|
if TYPE_CHECKING:
|
|
from pathlib import Path
|
|
|
|
|
|
class TestProbeDuration:
|
|
def test_reads_duration_of_a_real_file(self, sine_wav: Path) -> None:
|
|
assert ffmpeg.probe_duration(sine_wav) == pytest.approx(1.0, abs=0.05)
|
|
|
|
def test_raises_on_a_non_media_file(self, not_media: Path) -> None:
|
|
with pytest.raises(errors.ProbeError):
|
|
ffmpeg.probe_duration(not_media)
|
|
|
|
def test_raises_on_a_missing_file(self, tmp_path: Path) -> None:
|
|
with pytest.raises(errors.ProbeError):
|
|
ffmpeg.probe_duration(tmp_path / "nope.wav")
|
|
|
|
def test_returns_none_when_the_container_declares_no_duration(
|
|
self, sine_wav: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
# Some webm/mkv muxes report format.duration=N/A with no stream duration.
|
|
def na(*_a: object, **_k: object) -> str:
|
|
return "N/A"
|
|
|
|
monkeypatch.setattr(ffmpeg, "probe_field", na)
|
|
assert ffmpeg.probe_duration(sine_wav) is None
|
|
|
|
|
|
class TestHasAudioStream:
|
|
def test_true_for_audio(self, sine_wav: Path) -> None:
|
|
assert ffmpeg.has_audio_stream(sine_wav) is True
|
|
|
|
def test_false_for_a_video_without_audio(self, silent_video: Path) -> None:
|
|
assert ffmpeg.has_audio_stream(silent_video) is False
|
|
|
|
def test_false_for_a_non_media_file(self, not_media: Path) -> None:
|
|
assert ffmpeg.has_audio_stream(not_media) is False
|
|
|
|
|
|
class TestCommandBuilders:
|
|
def test_flac_command_always_forces_s16(self, tmp_path: Path) -> None:
|
|
# The FLAC encoder accepts only s16/s32; AAC/Opus sources decode to fltp,
|
|
# so leaving the sample format to filter-graph negotiation can fail.
|
|
cmd = ffmpeg.flac_command(tmp_path / "in.mkv", tmp_path / "out.flac")
|
|
assert cmd[cmd.index("-sample_fmt") + 1] == "s16"
|
|
|
|
def test_flac_command_maps_the_first_audio_track_only(self, tmp_path: Path) -> None:
|
|
cmd = ffmpeg.flac_command(tmp_path / "in.mkv", tmp_path / "out.flac")
|
|
assert cmd[cmd.index("-map") + 1] == "0:a:0"
|
|
assert "-vn" in cmd
|
|
|
|
def test_source_profile_does_not_resample(self, tmp_path: Path) -> None:
|
|
cmd = ffmpeg.flac_command(tmp_path / "in.mkv", tmp_path / "out.flac")
|
|
assert "-ar" not in cmd
|
|
assert "-ac" not in cmd
|
|
|
|
def test_whisper_profile_resamples_to_16k_mono(self, tmp_path: Path) -> None:
|
|
cmd = ffmpeg.flac_command(tmp_path / "i.mkv", tmp_path / "o.flac", profile="whisper")
|
|
assert cmd[cmd.index("-ar") + 1] == "16000"
|
|
assert cmd[cmd.index("-ac") + 1] == "1"
|
|
|
|
def test_rejects_an_unknown_profile(self, tmp_path: Path) -> None:
|
|
with pytest.raises(ValueError, match="profile"):
|
|
ffmpeg.flac_command(tmp_path / "i.mkv", tmp_path / "o.flac", profile="nope")
|
|
|
|
def test_decode_command_requests_16k_mono_float32(self, tmp_path: Path) -> None:
|
|
cmd = ffmpeg.decode_command(tmp_path / "a.flac")
|
|
assert cmd[cmd.index("-f") + 1] == "f32le"
|
|
assert cmd[cmd.index("-ar") + 1] == "16000"
|
|
assert cmd[cmd.index("-ac") + 1] == "1"
|
|
|
|
def test_probe_command_omits_stream_selection_when_not_given(self, tmp_path: Path) -> None:
|
|
assert "-select_streams" not in ffmpeg.probe_command(tmp_path / "a", "format=duration")
|
|
|
|
def test_probe_command_includes_stream_selection_when_given(self, tmp_path: Path) -> None:
|
|
cmd = ffmpeg.probe_command(tmp_path / "a", "stream=index", "a:0")
|
|
assert cmd[cmd.index("-select_streams") + 1] == "a:0"
|
|
|
|
|
|
class TestExtractFlac:
|
|
def test_produces_a_playable_flac(self, sine_wav: Path, tmp_path: Path) -> None:
|
|
out = tmp_path / "audio.flac"
|
|
ffmpeg.extract_flac(sine_wav, out)
|
|
assert ffmpeg.probe_duration(out) == pytest.approx(1.0, abs=0.05)
|
|
|
|
def test_preserves_source_rate_and_channels_by_default(
|
|
self, sine_wav: Path, tmp_path: Path
|
|
) -> None:
|
|
out = tmp_path / "audio.flac"
|
|
ffmpeg.extract_flac(sine_wav, out)
|
|
assert ffmpeg.probe_field(out, "stream=sample_rate", "a:0") == "44100"
|
|
assert ffmpeg.probe_field(out, "stream=channels", "a:0") == "2"
|
|
|
|
def test_whisper_profile_downmixes(self, sine_wav: Path, tmp_path: Path) -> None:
|
|
out = tmp_path / "audio.flac"
|
|
ffmpeg.extract_flac(sine_wav, out, profile="whisper")
|
|
assert ffmpeg.probe_field(out, "stream=sample_rate", "a:0") == "16000"
|
|
assert ffmpeg.probe_field(out, "stream=channels", "a:0") == "1"
|
|
|
|
def test_rejects_a_source_with_no_audio(self, silent_video: Path, tmp_path: Path) -> None:
|
|
with pytest.raises(errors.NoAudioStreamError):
|
|
ffmpeg.extract_flac(silent_video, tmp_path / "a.flac")
|
|
|
|
def test_raises_decode_error_when_ffmpeg_fails(
|
|
self, not_media: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
# Bypass the audio-stream gate so the ffmpeg failure itself is exercised.
|
|
def yes(_p: Path) -> bool:
|
|
return True
|
|
|
|
monkeypatch.setattr(ffmpeg, "has_audio_stream", yes)
|
|
with pytest.raises(errors.DecodeError):
|
|
ffmpeg.extract_flac(not_media, tmp_path / "a.flac")
|
|
|
|
|
|
class TestDecode16kMono:
|
|
def test_returns_float32_at_16k(self, sine_wav: Path) -> None:
|
|
pcm = ffmpeg.decode_16k_mono(sine_wav)
|
|
assert pcm.dtype == np.float32
|
|
assert len(pcm) == pytest.approx(16000, abs=800)
|
|
|
|
def test_audio_is_not_silent(self, sine_wav: Path) -> None:
|
|
assert float(np.abs(ffmpeg.decode_16k_mono(sine_wav)).max()) > 0.1
|
|
|
|
def test_raises_decode_error_on_garbage(self, not_media: Path) -> None:
|
|
with pytest.raises(errors.DecodeError):
|
|
ffmpeg.decode_16k_mono(not_media)
|
|
|
|
|
|
class TestPreflight:
|
|
def test_passes_when_binaries_exist(self) -> None:
|
|
ffmpeg.require_binaries()
|
|
|
|
def test_reports_the_real_flac_encoder_state(self) -> None:
|
|
assert ffmpeg.has_flac_encoder() is True
|
|
|
|
def test_raises_when_a_binary_is_missing(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
def missing(_name: str, _mode: int = 0, _path: str | None = None) -> str | None:
|
|
return None
|
|
|
|
monkeypatch.setattr(shutil, "which", missing)
|
|
with pytest.raises(errors.PreflightError, match="ffmpeg"):
|
|
ffmpeg.require_binaries()
|
|
|
|
def test_raises_when_the_flac_encoder_is_absent(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
def no_flac() -> bool:
|
|
return False
|
|
|
|
monkeypatch.setattr(ffmpeg, "has_flac_encoder", no_flac)
|
|
with pytest.raises(errors.PreflightError, match="flac"):
|
|
ffmpeg.require_binaries()
|
|
|
|
def test_a_missing_binary_surfaces_as_preflight_not_oserror(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
def boom(*_a: object, **_k: object) -> None:
|
|
raise FileNotFoundError("ffprobe")
|
|
|
|
monkeypatch.setattr(subprocess, "run", boom)
|
|
with pytest.raises(errors.PreflightError):
|
|
ffmpeg.has_flac_encoder()
|
|
|
|
|
|
class TestDurationsMatch:
|
|
def test_an_exact_match(self) -> None:
|
|
assert ffmpeg.durations_match(212.861, 212.861) is True
|
|
|
|
def test_within_the_relative_tolerance(self) -> None:
|
|
assert ffmpeg.durations_match(3600.0, 3610.0) is True
|
|
|
|
def test_beyond_the_relative_tolerance(self) -> None:
|
|
assert ffmpeg.durations_match(3600.0, 3700.0) is False
|
|
|
|
def test_short_clips_get_an_absolute_floor(self) -> None:
|
|
# 0.5% of two seconds is 10ms; rounding in container metadata exceeds that.
|
|
assert ffmpeg.durations_match(2.0, 2.9) is True
|
|
|
|
def test_a_near_empty_extraction_never_matches(self) -> None:
|
|
assert ffmpeg.durations_match(0.02, 212.861) is False
|