From f2d15c2ffee032b0113a47ead9cc85d06cd1b24a Mon Sep 17 00:00:00 2001 From: Roland Date: Mon, 14 Sep 2026 19:18:45 -0600 Subject: [PATCH] Transcode: verify the output before replacing the original After ffmpeg exits, the output is probed and compared with the source (container, presence of video/audio streams, codec per stream, duration within 1s or 1%) and its last 30 seconds are decoded to a null output. Any failure removes the output and keeps the original. The tail decode replaces the download-time completeness check, which grabbed a single frame near the end: one ffmpeg_video_complete (sync for the downloader, async for Jobs) now serves both, and is stricter, since every frame of the tail plus the audio must decode. Co-Authored-By: Claude Fable 5.1 --- modules/videos/common.py | 77 ++++++--- modules/videos/test/test_common.py | 35 ++++ modules/videos/test/test_transcode.py | 191 +++++++++++++++++++++- modules/videos/test/test_transcode_job.py | 26 ++- modules/videos/transcode.py | 86 ++++++++++ 5 files changed, 386 insertions(+), 29 deletions(-) diff --git a/modules/videos/common.py b/modules/videos/common.py index 84e86b75..1d8cadc7 100644 --- a/modules/videos/common.py +++ b/modules/videos/common.py @@ -149,31 +149,70 @@ def ffmpeg_poster(video_path: Path, poster_path: Path, seconds: int): return stderr -def ffmpeg_video_complete(video_path: Path, seconds: int = None) -> bool: - """Checks if video file is complete by taking screenshot from the end of the video. +# How much of a video's tail is decoded to prove the file is complete. Corruption from an +# interrupted download or write lands at the end of the file; decoding all of it would take as +# long as playing it. +TAIL_DECODE_SECONDS = 30 +TAIL_DECODE_TIMEOUT = 10 * 60 - @raise FileNotFoundError: raised if the video file does not exist. - """ + +def tail_decode_command(video_path: Path, seconds: int = None) -> Tuple[str | Path, ...]: + """ffmpeg command which decodes the last `seconds` of `video_path` to nothing, printing only + errors. `-sseof` seeks from the end; a file shorter than `seconds` is decoded whole.""" + seconds = seconds or TAIL_DECODE_SECONDS + return FFMPEG_BIN, '-v', 'error', '-sseof', f'-{int(seconds)}', '-i', video_path, '-f', 'null', '-' + + +def _tail_decode_verdict(return_code: int, stderr: bytes | str) -> Tuple[bool, str]: + errors = (stderr.decode(errors='replace') if isinstance(stderr, bytes) else (stderr or '')).strip() + ok = return_code == 0 and not errors + return ok, errors + + +def _check_video_file(video_path: Path): if not video_path.is_file(): raise FileNotFoundError(f'Video file not found: {video_path}') if not video_path.stat().st_size: raise RuntimeError(f'Video file is empty: {video_path}') + if not FFMPEG_BIN: + raise RuntimeError('ffmpeg was not found') - with tempfile.NamedTemporaryFile(suffix='.jpg', delete=False) as fh: - path = Path(fh.name) - path.unlink() - try: - seconds = seconds or extract_video_duration(video_path) - 5 - ffmpeg_poster(video_path, path, seconds) - return True - except subprocess.CalledProcessError: - return False - except TypeError: - # "N/A" duration. - return False - finally: - if path.is_file(): - path.unlink() + +def ffmpeg_video_complete(video_path: Path, seconds: int = None) -> bool: + """True if the last `seconds` of the video decode without error. + + Used after a download, and after a transcode before the original is replaced. A truncated or + corrupt file fails here because its damage is at the end; the index (moov) must also be + readable for ffmpeg to seek from the end at all. + + @raise FileNotFoundError: raised if the video file does not exist. + @raise RuntimeError: raised if the video file is empty, or ffmpeg is missing. + """ + _check_video_file(video_path) + cmd = tuple(str(i) for i in tail_decode_command(video_path, seconds)) + try: + proc = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=TAIL_DECODE_TIMEOUT) + except subprocess.TimeoutExpired: + logger.error(f'Timed out checking video is complete: {video_path}') + return False + ok, errors = _tail_decode_verdict(proc.returncode, proc.stderr) + if not ok: + logger.warning(f'Video is incomplete or corrupt: {video_path} (exit {proc.returncode}) {errors[-500:]}') + return ok + + +async def ffmpeg_video_complete_async(video_path: Path, seconds: int = None, runner=None) -> Tuple[bool, str]: + """`ffmpeg_video_complete` for async callers; returns (complete, ffmpeg's error output). + + `runner` defaults to `run_command`; a Job passes its own so ffmpeg's output lands in the Job log. + """ + _check_video_file(video_path) + runner = runner or run_command + result = await runner(tail_decode_command(video_path, seconds), cwd=video_path.parent, timeout=TAIL_DECODE_TIMEOUT) + ok, errors = _tail_decode_verdict(result.return_code, result.stderr) + if not ok: + logger.warning(f'Video is incomplete or corrupt: {video_path} (exit {result.return_code}) {errors[-500:]}') + return ok, errors def generate_video_poster(video_path: Path, seconds: int = 5) -> Tuple[Path, Optional[int]]: diff --git a/modules/videos/test/test_common.py b/modules/videos/test/test_common.py index 0e66fcf1..4045f731 100644 --- a/modules/videos/test/test_common.py +++ b/modules/videos/test/test_common.py @@ -644,3 +644,38 @@ async def test_video_modeler_does_not_hold_write_lock_during_ffprobe(async_clien assert lock_held_during_ffprobe, 'ffprobe never ran, so the test proves nothing' assert not any(lock_held_during_ffprobe), \ 'the write lock was held while ffprobe ran; every other writer waits out busy_timeout' + + +def test_ffmpeg_video_complete_real_files(test_directory, video_file): + """A whole video passes; a truncated copy fails. Uses the real ffmpeg on the example video.""" + from modules.videos.common import ffmpeg_video_complete, tail_decode_command + assert ffmpeg_video_complete(video_file) is True + + truncated = test_directory / 'truncated.mp4' + data = video_file.read_bytes() + truncated.write_bytes(data[:int(len(data) * 0.6)]) + assert ffmpeg_video_complete(truncated) is False + + empty = test_directory / 'empty.mp4' + empty.write_bytes(b'') + with pytest.raises(RuntimeError, match='empty'): + ffmpeg_video_complete(empty) + with pytest.raises(FileNotFoundError): + ffmpeg_video_complete(test_directory / 'missing.mp4') + + cmd = tail_decode_command(video_file) + assert cmd[cmd.index('-sseof') + 1] == '-30' and cmd[cmd.index('-f') + 1] == 'null' and cmd[-1] == '-' + + +@pytest.mark.asyncio +async def test_ffmpeg_video_complete_async_real_files(test_directory, video_file): + """The async variant reaches the same verdicts and returns ffmpeg's error text.""" + from modules.videos.common import ffmpeg_video_complete_async + ok, errors = await ffmpeg_video_complete_async(video_file) + assert ok is True and errors == '' + + truncated = test_directory / 'truncated.mp4' + data = video_file.read_bytes() + truncated.write_bytes(data[:int(len(data) * 0.6)]) + ok, errors = await ffmpeg_video_complete_async(truncated) + assert ok is False and errors diff --git a/modules/videos/test/test_transcode.py b/modules/videos/test/test_transcode.py index 41247c6c..b6c74310 100644 --- a/modules/videos/test/test_transcode.py +++ b/modules/videos/test/test_transcode.py @@ -5,10 +5,51 @@ from unittest import mock import pytest from modules.videos.transcode import codecs_match, get_stream_codec_names, get_transcode_target, \ - transcode_can_satisfy_codecs, transcode_video_file, TRANSCODE_VIDEO_TARGETS, TRANSCODE_AUDIO_TARGETS + transcode_can_satisfy_codecs, transcode_video_file, TRANSCODE_VIDEO_TARGETS, TRANSCODE_AUDIO_TARGETS, \ + verify_transcode_output from wrolpi.cmd import CommandResult +def make_probe(vcodec='vp9', acodec='opus', duration=10.0, format_name='matroska,webm'): + """An ffprobe result with one video and one audio stream.""" + streams = [] + if vcodec: + streams.append({'codec_type': 'video', 'codec_name': vcodec}) + if acodec: + streams.append({'codec_type': 'audio', 'codec_name': acodec}) + return {'streams': streams, 'format': {'format_name': format_name, 'duration': str(duration)}} + + +MP4 = 'mov,mp4,m4a,3gp,3g2,mj2' + + +class FakeProbes: + """What ffprobe reports for the source (a vp9/opus webm) and for the transcode output. + + The default output is a full h264/aac transcode; a test that copies a stream sets `output` + to what ffmpeg would really produce (a copied stream keeps the source codec).""" + + def __init__(self): + self.source = make_probe('vp9', 'opus', 10.0) + self.output = make_probe('h264', 'aac', 10.0, MP4) + + def __call__(self, path: pathlib.Path): + # Sync: mock.patch wraps the async target in an AsyncMock, which awaits for us. + if '.transcode.' in path.name: + output = dict(self.output) + if path.suffix != '.mp4' and output.get('format'): + output['format'] = {**output['format'], 'format_name': 'matroska,webm'} + return output + return self.source + + +@pytest.fixture(autouse=True) +def mock_probe() -> FakeProbes: + probes = FakeProbes() + with mock.patch('modules.videos.transcode.probe_for_verify', side_effect=probes): + yield probes + + def test_get_stream_codec_names(): """Embedded thumbnail streams (mjpeg/png) are not video.""" data = {'streams': [ @@ -79,6 +120,9 @@ async def test_transcode_video_file(test_directory, async_client): async def fake_run_command(cmd, **kwargs): # ffmpeg writes the temporary output file. + if cmd[-1] == '-': + # The tail-decode check: nothing to write. + return CommandResult(return_code=0, cancelled=False, stdout=b'', stderr=b'', elapsed=0) pathlib.Path(cmd[-1]).write_bytes(b'transcoded data') return CommandResult(return_code=0, cancelled=False, stdout=b'', stderr=b'', elapsed=1) @@ -92,7 +136,7 @@ async def test_transcode_video_file(test_directory, async_client): assert not sidecar.exists(), 'The stale ffprobe sidecar should be deleted' assert not (test_directory / 'video.transcode.mp4').exists(), 'The temporary file should be renamed' - cmd = mock_run.call_args[0][0] + cmd = mock_run.call_args_list[0][0][0] assert '-c:v' in cmd and cmd[cmd.index('-c:v') + 1] == 'libx264' assert '-c:a' in cmd and cmd[cmd.index('-c:a') + 1] == 'aac' # Attached-picture streams are excluded; a missing audio stream is tolerated. @@ -100,12 +144,16 @@ async def test_transcode_video_file(test_directory, async_client): @pytest.mark.asyncio -async def test_transcode_video_file_copies_matching_stream(test_directory, async_client): +async def test_transcode_video_file_copies_matching_stream(test_directory, async_client, mock_probe): """Only the mismatching stream is re-encoded; the other stream is copied.""" video_path = test_directory / 'video.mp4' video_path.write_bytes(b'fake video data') + mock_probe.output = make_probe('h264', 'opus', 10.0, MP4) # Audio was copied. async def fake_run_command(cmd, **kwargs): + if cmd[-1] == '-': + # The tail-decode check: nothing to write. + return CommandResult(return_code=0, cancelled=False, stdout=b'', stderr=b'', elapsed=0) pathlib.Path(cmd[-1]).write_bytes(b'transcoded data') return CommandResult(return_code=0, cancelled=False, stdout=b'', stderr=b'', elapsed=1) @@ -113,18 +161,22 @@ async def test_transcode_video_file_copies_matching_stream(test_directory, async result = await transcode_video_file(video_path, target_vcodec='h264', container='mp4') assert result == video_path, 'Same container: the file is replaced in place' - cmd = mock_run.call_args[0][0] + cmd = mock_run.call_args_list[0][0][0] assert cmd[cmd.index('-c:v') + 1] == 'libx264' assert cmd[cmd.index('-c:a') + 1] == 'copy' @pytest.mark.asyncio -async def test_transcode_video_file_forces_supported_container(test_directory, async_client): +async def test_transcode_video_file_forces_supported_container(test_directory, async_client, mock_probe): """webm cannot contain h264/aac; the container is forced to mp4.""" video_path = test_directory / 'video.webm' video_path.write_bytes(b'fake video data') + mock_probe.output = make_probe('h264', 'opus', 10.0, MP4) # Audio was copied. async def fake_run_command(cmd, **kwargs): + if cmd[-1] == '-': + # The tail-decode check: nothing to write. + return CommandResult(return_code=0, cancelled=False, stdout=b'', stderr=b'', elapsed=0) pathlib.Path(cmd[-1]).write_bytes(b'transcoded data') return CommandResult(return_code=0, cancelled=False, stdout=b'', stderr=b'', elapsed=1) @@ -141,6 +193,9 @@ async def test_transcode_video_file_failure_removes_tmp(test_directory, async_cl video_path.write_bytes(b'fake video data') async def fake_run_command(cmd, **kwargs): + if cmd[-1] == '-': + # The tail-decode check: nothing to write. + return CommandResult(return_code=0, cancelled=False, stdout=b'', stderr=b'', elapsed=0) pathlib.Path(cmd[-1]).write_bytes(b'partial data') return CommandResult(return_code=1, cancelled=False, stdout=b'', stderr=b'boom', elapsed=1) @@ -168,7 +223,7 @@ async def test_transcode_video_file_disk_space_guard(test_directory, async_clien @pytest.mark.asyncio -async def test_transcode_lock_serializes(test_directory, async_client): +async def test_transcode_lock_serializes(test_directory, async_client, mock_probe): """A transcode waits while `shared_ctx.transcode_lock` is held and releases it when done. This exercises the wait/release logic in-process; cross-process sharing of the lock is @@ -178,8 +233,12 @@ async def test_transcode_lock_serializes(test_directory, async_client): video_path = test_directory / 'video.webm' video_path.write_bytes(b'fake video data') + mock_probe.output = make_probe('h264', 'opus', 10.0, MP4) # Audio was copied. async def fake_run_command(cmd, **kwargs): + if cmd[-1] == '-': + # The tail-decode check: nothing to write. + return CommandResult(return_code=0, cancelled=False, stdout=b'', stderr=b'', elapsed=0) pathlib.Path(cmd[-1]).write_bytes(b'transcoded data') return CommandResult(return_code=0, cancelled=False, stdout=b'', stderr=b'', elapsed=1) @@ -203,7 +262,9 @@ async def test_transcode_lock_serializes(test_directory, async_client): task.cancel() assert result == test_directory / 'video.mp4' - mock_run.assert_called_once() + # One encode (plus the tail-decode verification of its output); the second waiter ran nothing yet. + encodes = [c for c in mock_run.call_args_list if '-progress' in c[0][0]] + assert len(encodes) == 1 # The lock was released after the transcode. assert lock.acquire(block=False), 'transcode_video_file did not release the lock' @@ -211,12 +272,16 @@ async def test_transcode_lock_serializes(test_directory, async_client): @pytest.mark.asyncio -async def test_transcode_video_file_remux(test_directory, async_client): +async def test_transcode_video_file_remux(test_directory, async_client, mock_probe): """No target codec is a remux: both streams are copied into the requested container.""" video_path = test_directory / 'video.mkv' video_path.write_bytes(b'fake video data') + mock_probe.output = make_probe('vp9', 'opus', 10.0, MP4) # Both copied. async def fake_run_command(cmd, **kwargs): + if cmd[-1] == '-': + # The tail-decode check: nothing to write. + return CommandResult(return_code=0, cancelled=False, stdout=b'', stderr=b'', elapsed=0) pathlib.Path(cmd[-1]).write_bytes(b'remuxed data') return CommandResult(return_code=0, cancelled=False, stdout=b'', stderr=b'', elapsed=1) @@ -225,6 +290,114 @@ async def test_transcode_video_file_remux(test_directory, async_client): assert result == test_directory / 'video.mp4' and result.is_file() assert not video_path.exists() - cmd = mock_run.call_args[0][0] + cmd = mock_run.call_args_list[0][0][0] assert cmd[cmd.index('-c:v') + 1] == 'copy' and cmd[cmd.index('-c:a') + 1] == 'copy' assert '+faststart' in cmd + + + +# ------------------------------------------------------------------ output verification + +def test_verify_transcode_output_ok(): + source = make_probe('vp9', 'opus', 100.0) + output = make_probe('h264', 'opus', 99.6, MP4) # Audio copied; a frame trimmed. + verify_transcode_output(source, output, target_vcodec='h264', target_acodec=None, container='mp4') + + +def test_verify_transcode_output_remux_ok(): + source = make_probe('vp9', 'opus', 100.0) + output = make_probe('vp9', 'opus', 100.0, MP4) + verify_transcode_output(source, output, target_vcodec=None, target_acodec=None, container='mp4') + + +def test_verify_transcode_output_no_audio_ok(): + """A silent source has no audio to demand of the output.""" + source = make_probe('vp9', None, 100.0) + output = make_probe('h264', None, 100.0, MP4) + verify_transcode_output(source, output, target_vcodec='h264', target_acodec=None, container='mp4') + + +@pytest.mark.parametrize('output,message', [ + (make_probe('h264', 'aac', 50.0, MP4), 'duration'), # Truncated. + (make_probe('h264', None, 100.0, MP4), 'audio'), # Lost the audio stream. + (make_probe(None, 'aac', 100.0, MP4), 'video'), # Lost the video stream. + (make_probe('vp9', 'aac', 100.0, MP4), 'video codec'), # Not re-encoded. + (make_probe('h264', 'opus', 100.0, MP4), 'audio codec'), # Not re-encoded. + (make_probe('h264', 'aac', 100.0, 'matroska,webm'), 'container'), # Wrong container. + ({}, 'probe'), # ffprobe could not read it. +]) +def test_verify_transcode_output_rejects(output, message): + source = make_probe('vp9', 'opus', 100.0) + with pytest.raises(RuntimeError, match=message): + verify_transcode_output(source, output, target_vcodec='h264', target_acodec='aac', container='mp4') + + +def test_verify_transcode_output_copied_stream_must_match_source(): + """A copied stream must come out with the source's codec.""" + source = make_probe('vp9', 'opus', 100.0) + output = make_probe('h264', 'aac', 100.0, MP4) + with pytest.raises(RuntimeError, match='audio codec'): + verify_transcode_output(source, output, target_vcodec='h264', target_acodec=None, container='mp4') + + +def test_verify_transcode_output_duration_tolerance(): + """One second, or one percent, whichever is larger.""" + source = make_probe('vp9', 'opus', 10.0) + verify_transcode_output(source, make_probe('h264', 'aac', 9.1, MP4), 'h264', 'aac', 'mp4') + with pytest.raises(RuntimeError, match='duration'): + verify_transcode_output(source, make_probe('h264', 'aac', 8.9, MP4), 'h264', 'aac', 'mp4') + long_source = make_probe('vp9', 'opus', 40000.0) + verify_transcode_output(long_source, make_probe('h264', 'aac', 39700.0, MP4), 'h264', 'aac', 'mp4') + with pytest.raises(RuntimeError, match='duration'): + verify_transcode_output(long_source, make_probe('h264', 'aac', 39500.0, MP4), 'h264', 'aac', 'mp4') + + +@pytest.mark.asyncio +async def test_transcode_video_file_rejects_truncated_output(test_directory, async_client, mock_probe): + """ffmpeg exits 0 but the output is short: the original is kept, the output is removed.""" + video_path = test_directory / 'video.webm' + video_path.write_bytes(b'fake video data') + + mock_probe.output = make_probe('h264', 'aac', 4.0, MP4) # Source is 10s. + + async def fake_run_command(cmd, **kwargs): + if cmd[-1] != '-': + pathlib.Path(cmd[-1]).write_bytes(b'partial data') + return CommandResult(return_code=0, cancelled=False, stdout=b'', stderr=b'', elapsed=1) + + with mock.patch('modules.videos.transcode.run_command', side_effect=fake_run_command): + with pytest.raises(RuntimeError, match='duration'): + await transcode_video_file(video_path, target_vcodec='h264', target_acodec='aac') + + assert video_path.read_bytes() == b'fake video data', 'The original must survive' + assert not (test_directory / 'video.mp4').exists() + assert not (test_directory / 'video.transcode.mp4').exists(), 'The bad output is removed' + + +@pytest.mark.asyncio +async def test_transcode_video_file_rejects_undecodable_tail(test_directory, async_client): + """The last seconds of the output must decode cleanly, or the original is kept.""" + video_path = test_directory / 'video.webm' + video_path.write_bytes(b'fake video data') + tail_cmds = [] + + async def fake_run_command(cmd, **kwargs): + if cmd[-1] == '-': + tail_cmds.append(cmd) + return CommandResult(return_code=0, cancelled=False, stdout=b'', + stderr=b'[h264 @ 0x1] Invalid NAL unit size', elapsed=1) + pathlib.Path(cmd[-1]).write_bytes(b'transcoded data') + return CommandResult(return_code=0, cancelled=False, stdout=b'', stderr=b'', elapsed=1) + + with mock.patch('modules.videos.transcode.run_command', side_effect=fake_run_command): + with pytest.raises(RuntimeError, match='decode'): + await transcode_video_file(video_path, target_vcodec='h264', target_acodec='aac') + + assert video_path.read_bytes() == b'fake video data' + assert not (test_directory / 'video.transcode.mp4').exists() + # The check decodes only the tail of the temporary output, discarding the frames. + assert len(tail_cmds) == 1 + cmd = tail_cmds[0] + assert '-sseof' in cmd and cmd[cmd.index('-sseof') + 1].startswith('-') + assert str(cmd[cmd.index('-i') + 1]).endswith('video.transcode.mp4') + assert cmd[cmd.index('-f') + 1] == 'null' diff --git a/modules/videos/test/test_transcode_job.py b/modules/videos/test/test_transcode_job.py index 0486189b..7c711cf2 100644 --- a/modules/videos/test/test_transcode_job.py +++ b/modules/videos/test/test_transcode_job.py @@ -1,5 +1,6 @@ """Tests for the transcode Job and its API (see `wrolpi.jobs`).""" import json +import pathlib import shutil from unittest import mock @@ -12,6 +13,23 @@ from wrolpi.cmd import CommandResult from wrolpi.files.models import FileGroup from wrolpi.vars import PROJECT_DIR +MP4 = 'mov,mp4,m4a,3gp,3g2,mj2' + + +@pytest.fixture(autouse=True) +def mock_probe(): + """The output probe agrees with what the Job asked for (targets or copied source codecs).""" + + async def probe(path: pathlib.Path): + # The test video is h264/aac; a copied stream keeps that, a target replaces it. + fmt = MP4 if path.suffix == '.mp4' else 'matroska,webm' + return {'streams': [{'codec_type': 'video', 'codec_name': 'h264'}, + {'codec_type': 'audio', 'codec_name': 'aac'}], + 'format': {'format_name': fmt, 'duration': '5.3'}} + + with mock.patch('modules.videos.transcode.probe_for_verify', side_effect=probe): + yield + def test_make_ffmpeg_progress_parser(): """`-progress pipe:1` lines are converted to whole-percent progress, reported only on change.""" @@ -46,6 +64,9 @@ async def test_transcode_video_job(test_session, test_directory, async_client, v # ffmpeg reports progress on stdout, then writes the temporary output file. The output # must be a real video: the refresh would otherwise drop the Video model (mimetype changed). stdout_callback('out_time_us=5000000') + if cmd[-1] == '-': + # The tail-decode check: nothing to write. + return CommandResult(return_code=0, cancelled=False, stdout=b'', stderr=b'', elapsed=0) shutil.copy(PROJECT_DIR / 'test/big_buck_bunny_720p_1mb.mp4', cmd[-1]) return CommandResult(return_code=0, cancelled=False, stdout=b'', stderr=b'ffmpeg version fake', elapsed=1) @@ -88,6 +109,9 @@ async def test_transcode_video_job_remux(test_session, test_directory, async_cli test_session.commit() async def fake_run_command(cmd, stdout_callback=None, **kwargs): + if cmd[-1] == '-': + # The tail-decode check: nothing to write. + return CommandResult(return_code=0, cancelled=False, stdout=b'', stderr=b'', elapsed=0) shutil.copy(PROJECT_DIR / 'test/big_buck_bunny_720p_1mb.mp4', cmd[-1]) return CommandResult(return_code=0, cancelled=False, stdout=b'', stderr=b'', elapsed=1) @@ -96,7 +120,7 @@ async def test_transcode_video_job_remux(test_session, test_directory, async_cli record = await jobs.wait_for_job(job_id) assert record['status'] == jobs.COMPLETE, record - cmd = mock_run.call_args[0][0] + cmd = mock_run.call_args_list[0][0][0] assert cmd[cmd.index('-c:v') + 1] == 'copy' and cmd[cmd.index('-c:a') + 1] == 'copy' assert '+faststart' in cmd assert not video_path.exists() diff --git a/modules/videos/transcode.py b/modules/videos/transcode.py index 1f01eeec..73a396ab 100644 --- a/modules/videos/transcode.py +++ b/modules/videos/transcode.py @@ -4,6 +4,7 @@ import pathlib import shutil from typing import List, Optional, Tuple, Callable +from modules.videos.common import ffprobe_json, ffmpeg_video_complete_async from wrolpi.captions import FFMPEG_BIN from wrolpi.cmd import run_command from wrolpi.common import logger, get_relative_to_media_directory, wrol_mode_enabled @@ -47,6 +48,12 @@ TRANSCODE_CONTAINERS = ('mp4', 'mkv') TRANSCODE_TIMEOUT = 4 * 60 * 60 # A long video on a Raspberry Pi can take hours. +# Output verification (see `verify_transcode_output`): the output's duration may differ from the +# source by this much before it is considered truncated. Muxers trim a frame or two; a truncated +# file is off by minutes. +DURATION_TOLERANCE_SECONDS = 1.0 +DURATION_TOLERANCE_RATIO = 0.01 + # Transcoding writes a whole new copy of the video next to the original. MINIMUM_FREE_SPACE_RATIO = 1.5 @@ -146,6 +153,75 @@ async def transcode_video_file(video_path: pathlib.Path, lock.release() +async def probe_for_verify(path: pathlib.Path) -> dict: + """ffprobe a file for `verify_transcode_output`. Separate so tests can replace it.""" + return await ffprobe_json(path) + + +def _format_names(probe: dict) -> List[str]: + return str((probe.get('format') or {}).get('format_name') or '').split(',') + + +def _duration(probe: dict) -> Optional[float]: + try: + return float((probe.get('format') or {}).get('duration')) + except (TypeError, ValueError): + return None + + +def verify_transcode_output(source: dict, output: dict, target_vcodec: Optional[str], + target_acodec: Optional[str], container: str): + """Compare ffprobe results of the source and the transcoded output; the output must be a + complete rendition of the source before the source is replaced by it. + + Checks: the output parses (its index was written), it is in the requested container, it has a + video stream (and an audio stream when the source has one), each stream's codec is the target + or, when copied, the source's, and the duration matches within a tolerance. + + @raise RuntimeError: describing the first failed check. + """ + if not output or not output.get('streams'): + raise RuntimeError('Transcode output could not be probed (no streams); its index may be missing') + + if container not in _format_names(output): + raise RuntimeError(f'Transcode output container is {_format_names(output)}, expected {container}') + + source_video = get_stream_codec_names(source, 'video') + source_audio = get_stream_codec_names(source, 'audio') + output_video = get_stream_codec_names(output, 'video') + output_audio = get_stream_codec_names(output, 'audio') + + if not output_video: + raise RuntimeError('Transcode output has no video stream') + if source_audio and not output_audio: + raise RuntimeError('Transcode output has no audio stream, but the source has one') + + expected_video = target_vcodec or (source_video[0] if source_video else None) + if expected_video and output_video[0] != expected_video: + raise RuntimeError(f'Transcode output video codec is {output_video[0]}, expected {expected_video}') + expected_audio = target_acodec or (source_audio[0] if source_audio else None) + if output_audio and expected_audio and output_audio[0] != expected_audio: + raise RuntimeError(f'Transcode output audio codec is {output_audio[0]}, expected {expected_audio}') + + source_duration, output_duration = _duration(source), _duration(output) + if source_duration is not None: + if output_duration is None: + raise RuntimeError('Transcode output has no duration') + tolerance = max(DURATION_TOLERANCE_SECONDS, DURATION_TOLERANCE_RATIO * source_duration) + if abs(source_duration - output_duration) > tolerance: + raise RuntimeError( + f'Transcode output duration is {output_duration:.1f}s, source is {source_duration:.1f}s') + + +async def _verify_tail_decodes(path: pathlib.Path, runner: Callable): + """The same completeness check a download gets (`ffmpeg_video_complete`), as a hard failure. + + @raise RuntimeError: when the tail of `path` does not decode cleanly.""" + ok, errors = await ffmpeg_video_complete_async(path, runner=runner) + if not ok: + raise RuntimeError(f'Transcode output failed to decode: {errors[-2000:] or "ffmpeg exited non-zero"}') + + def make_ffmpeg_progress_parser(duration: Optional[float], set_progress: Callable[[float], None]) \ -> Callable[[str], None]: """Return a callback for `ffmpeg -progress pipe:1` stdout lines which reports percent complete. @@ -207,8 +283,11 @@ async def _transcode_video_file(video_path: pathlib.Path, target_vcodec: Optiona '-nostats', '-progress', 'pipe:1', str(tmp_path)) logger.warning(f'Transcoding {video_path} to {final_path} (video={target_vcodec}, audio={target_acodec})') + # Probed before ffmpeg so verification compares against the file as it was. + source_probe = await probe_for_verify(video_path) try: job = get_current_job() + runner = job.run_command if job else run_command if job: # Inside a Job: ffmpeg's output lands in the Job log and progress is reported. result = await job.run_command(cmd, cwd=video_path.parent, timeout=TRANSCODE_TIMEOUT, @@ -219,6 +298,13 @@ async def _transcode_video_file(video_path: pathlib.Path, target_vcodec: Optiona raise RuntimeError( f'ffmpeg exited with {result.return_code} while transcoding {video_path}:' f'\n{result.stderr.decode()[-2000:]}') + + # The original is only replaced by a complete output: compare probes, then prove the tail + # decodes. A failure here leaves the original untouched. + logger.info(f'Verifying transcode output {tmp_path}') + output_probe = await probe_for_verify(tmp_path) + verify_transcode_output(source_probe, output_probe, target_vcodec, target_acodec, container) + await _verify_tail_decodes(tmp_path, runner) except Exception: tmp_path.unlink(missing_ok=True) raise