Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 23 additions & 3 deletions media_shrinker.py
Original file line number Diff line number Diff line change
Expand Up @@ -923,9 +923,29 @@ def _execute_segment_conversion(
plan = opus_plan
output_size = first_result.stat().st_size

output_duration = _probe_output_duration(
first_result, ffprobe_path=ffprobe_path, output_size=output_size
)
try:
output_duration = _probe_output_duration(
first_result, ffprobe_path=ffprobe_path, output_size=output_size
)
except MediaShrinkerError as exc:
return _discard_invalid_generated_output(
source,
first_result,
{
"source_path": source,
"output_path": first_result,
"original_size_bytes": original_size,
"output_size_bytes": output_size,
"strategy": plan.strategy,
"segment_index": segment.index,
"segment_count": segment.total_segments,
"start_seconds": segment.start_seconds,
"duration_seconds": None,
},
status="duration_mismatch",
message=f"Generated output has no usable duration: {exc}",
protected_sources=resolved_protected_sources,
)
preserve_file_attributes(source, first_result)
common_fields = {
"source_path": source,
Expand Down
62 changes: 62 additions & 0 deletions tests/test_media_shrinker.py
Original file line number Diff line number Diff line change
Expand Up @@ -636,6 +636,68 @@ def test_convert_segment_deletes_generated_output_when_duration_mismatches_expec
self.assertIsNone(result.output_path)
self.assertFalse((output_dir / "source.wav.flac").exists())

def test_convert_segment_discards_output_that_has_no_probeable_duration(
self,
) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
source = root / "source.wav"
output_dir = root / "out"
fake_ffmpeg = root / "fake_ffmpeg.py"
source.write_bytes(b"source")
fake_ffmpeg.write_text(
"#!/usr/bin/env python3\n"
"import pathlib, sys\n"
"pathlib.Path(sys.argv[-1]).write_bytes(b'junk')\n",
encoding="utf-8",
)
fake_ffmpeg.chmod(0o755)
source_probe = MediaProbe(60.0, 1_000, "pcm_s16le", 1_411_200, False, "wav")
original_probe_media = media_shrinker.probe_media

def fake_probe_media(
path: Path,
*,
ffprobe_path: str = "ffprobe",
source_size: int | None = None,
) -> MediaProbe:
# The source probes fine, but the generated output cannot be
# probed for a usable duration (e.g. a truncated/empty file that
# ffmpeg still exits 0 for). Real probe_media raises here.
if path.suffix == ".wav":
return source_probe
raise MediaShrinkerError(f"{path} has no usable duration")

try:
media_shrinker.probe_media = fake_probe_media
result = media_shrinker._convert_segment(
source,
rel_source=Path("source.wav"),
probe=source_probe,
segment=MediaSegment(
index=1,
start_seconds=0.0,
duration_seconds=60.0,
total_segments=1,
),
output_dir=output_dir,
target_bytes=1_900_000_000,
original_size=source.stat().st_size,
ffmpeg_path=str(fake_ffmpeg),
ffprobe_path="fake_ffprobe",
prefer_flac=True,
ffmpeg_threads=None,
overwrite=False,
max_segment_duration_seconds=14_400.0,
)
finally:
media_shrinker.probe_media = original_probe_media

self.assertEqual(result.status, "duration_mismatch")
self.assertIsNone(result.output_path)
self.assertIsNone(result.duration_seconds)
self.assertFalse((output_dir / "source.wav.flac").exists())

def test_existing_output_with_wrong_duration_is_replaced_not_skipped(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
Expand Down
Loading