From 819947c01fbf5f32cf8745344b74378769e111f4 Mon Sep 17 00:00:00 2001 From: Jason Ross Date: Sun, 13 Sep 2026 16:01:06 -0500 Subject: [PATCH] fix(download): repair a corrupt video and discard unresumable partials Two cases the resume plan already routed to the download stage, but which the stage could not actually fix. yt-dlp's overwrites default is off (existing_video_file passes default_overwrite=False), so an existing target is reported as "already downloaded" and skipped. A truncated video.mkv was therefore planned for re-download, never re-downloaded, and re-probed as the same corrupt file -- a job that reports success on media it could not read. A .part without its .aria2 control file is a sparse file, not a valid prefix, because aria2 writes its segments out of order. continuedl would resume from its length and produce a video that silently stops being the source partway through. Re-downloading is the cheaper mistake, so orphans go unconditionally; a .part with its control file is left for aria2 to resume, and the sweep runs once before the retry loop rather than between attempts. find_video now matches only the merged output. Per-format streams from a failed merge sort ahead of video.mkv and hold one stream each, so it could return a video-only file to the audio stage. Co-Authored-By: Claude Opus 5 (1M context) --- src/ccn_transcribe/jobs/verify.py | 4 +- src/ccn_transcribe/stages/download.py | 56 ++++++++++++- tests/test_stages_download.py | 109 ++++++++++++++++++++++++++ 3 files changed, 164 insertions(+), 5 deletions(-) diff --git a/src/ccn_transcribe/jobs/verify.py b/src/ccn_transcribe/jobs/verify.py index a15e7e9..1615791 100644 --- a/src/ccn_transcribe/jobs/verify.py +++ b/src/ccn_transcribe/jobs/verify.py @@ -44,8 +44,8 @@ def _resolve(job: JobPaths, artifact: Artifact) -> Path | None: def _part_files(job: JobPaths) -> Iterator[Path]: for directory in (job.media_dir, job.tmp_dir): - if directory.is_dir(): - yield from directory.glob("*.part") + # glob yields nothing for a directory that does not exist; no guard needed. + yield from directory.glob("*.part") def _resumable_partial_exists(job: JobPaths) -> bool: diff --git a/src/ccn_transcribe/stages/download.py b/src/ccn_transcribe/stages/download.py index 9b8b329..d3d6bc0 100644 --- a/src/ccn_transcribe/stages/download.py +++ b/src/ccn_transcribe/stages/download.py @@ -17,7 +17,7 @@ from ccn_transcribe import errors, paths from ccn_transcribe.media import ytdlp_opts if TYPE_CHECKING: - from collections.abc import Sequence + from collections.abc import Iterator, Sequence from pathlib import Path from ccn_transcribe.paths import JobPaths @@ -155,13 +155,59 @@ def identify(entry: dict[str, Any], url: str) -> str: def find_video(job: JobPaths) -> Path | None: - """The downloaded media file, whatever extension yt-dlp settled on.""" + """The merged media file, whatever extension yt-dlp settled on. + + Only the file yt-dlp's output template names directly. A failed merge leaves + per-format streams like video.f251.webm behind, which sort first and hold one + stream each; returning one would silently transcribe half the source. + """ for candidate in sorted(job.media_dir.glob("video.*")): - if candidate.suffix not in (".part", ".aria2", ".json", ".ytdl"): + if candidate.stem == "video" and candidate.suffix not in (".part", ".aria2", ".ytdl"): return candidate return None +def _part_files(job: JobPaths) -> Iterator[Path]: + for directory in (job.media_dir, job.tmp_dir): + # glob yields nothing for a directory that does not exist; no guard needed. + yield from directory.glob("*.part") + + +def clear_stale_parts(job: JobPaths) -> list[Path]: + """Delete .part files that cannot be resumed, and say which. + + aria2 writes its 16 segments out of order, so without the .aria2 control file + a .part is a sparse file full of holes rather than a valid prefix. yt-dlp's + continuedl would resume from its length and produce a video that plays for a + while and then isn't the source any more -- a corrupt transcript, not an error. + Re-downloading is the cheaper mistake, so an orphan goes even when the native + downloader (which does write sequentially) could have resumed it. + """ + removed: list[Path] = [] + for part in sorted(_part_files(job)): + if part.with_suffix(part.suffix + ".aria2").exists(): + continue + part.unlink() + removed.append(part) + log.warning("discarded unresumable partial download %s", part.name) + return removed + + +def discard_video(job: JobPaths) -> Path | None: + """Delete the merged file so yt-dlp will actually fetch it again. + + yt-dlp's overwrites default is off, so it reports an existing target as + "already downloaded" and skips it; a truncated video.mkv would otherwise never + be repaired. Per-format streams are left alone -- yt-dlp re-merges complete + ones without re-downloading, so clearing them is a pessimization. + """ + video = find_video(job) + if video is not None: + video.unlink() + log.info("discarded %s so it can be downloaded again", video.name) + return video + + def download_options(job: JobPaths, config: ytdlp_opts.DownloadConfig) -> dict[str, Any]: return ytdlp_opts.ydl_opts( str(job.media_dir / "video.%(ext)s"), @@ -188,6 +234,10 @@ def download( without this loop that stall would become a job failure instead of a resume. """ job.ensure() + # Before the loop, not inside it: a failed attempt leaves a .part with a live + # control file, which is exactly the progress the next attempt resumes from. + clear_stale_parts(job) + discard_video(job) options = download_options(job, config) make = factory or default_factory diff --git a/tests/test_stages_download.py b/tests/test_stages_download.py index 74e76fd..cac51bf 100644 --- a/tests/test_stages_download.py +++ b/tests/test_stages_download.py @@ -214,6 +214,115 @@ class TestFindVideo: def test_none_when_nothing_downloaded(self, job: paths.JobPaths) -> None: assert download.find_video(job) is None + def test_ignores_unmerged_per_format_files(self, job: paths.JobPaths) -> None: + # A failed merge leaves video.f251.webm behind; it sorts before video.mkv + # and is a single stream, so returning it would transcribe half the source. + (job.media_dir / "video.f251.webm").write_bytes(b"audio only") + assert download.find_video(job) is None + + +class TestClearStaleParts: + def test_removes_a_part_with_no_control_file(self, job: paths.JobPaths) -> None: + stale = job.media_dir / "video.mkv.part" + stale.write_bytes(b"sparse") + assert download.clear_stale_parts(job) == [stale] + assert not stale.exists() + + def test_keeps_a_part_that_aria2_can_resume(self, job: paths.JobPaths) -> None: + part = job.media_dir / "video.mkv.part" + part.write_bytes(b"sparse") + (job.media_dir / "video.mkv.part.aria2").write_bytes(b"ctrl") + assert download.clear_stale_parts(job) == [] + assert part.exists() + + def test_sweeps_the_tmp_directory_too(self, job: paths.JobPaths) -> None: + stale = job.tmp_dir / "video.mkv.part" + stale.write_bytes(b"sparse") + assert download.clear_stale_parts(job) == [stale] + + def test_nothing_to_do_is_not_an_error(self, job: paths.JobPaths) -> None: + assert download.clear_stale_parts(job) == [] + + +class TestDiscardVideo: + def test_removes_the_merged_file(self, job: paths.JobPaths) -> None: + video = job.media_dir / "video.mkv" + video.write_bytes(b"truncated") + assert download.discard_video(job) == video + assert not video.exists() + + def test_leaves_per_format_files_alone(self, job: paths.JobPaths) -> None: + # yt-dlp re-merges complete per-format files without re-downloading them; + # deleting them turns a cheap remux into a full re-download. + intermediate = job.media_dir / "video.f137.mp4" + intermediate.write_bytes(b"video stream") + assert download.discard_video(job) is None + assert intermediate.exists() + + def test_nothing_to_discard(self, job: paths.JobPaths) -> None: + assert download.discard_video(job) is None + + +class TestDownloadClearsStaleState: + def test_a_corrupt_video_is_unlinked_before_yt_dlp_runs(self, job: paths.JobPaths) -> None: + # yt-dlp's overwrites default is off: it reports an existing target as + # "already downloaded" and skips, so a truncated file is never repaired. + video = job.media_dir / "video.mkv" + video.write_bytes(b"truncated") + seen: list[bool] = [] + + def on_download(_n: int) -> None: + seen.append(video.exists()) + video.write_bytes(b"complete") + + download.download(job, "u", factory=factory_for(on_download=on_download)) + assert seen == [False] + assert video.read_bytes() == b"complete" + + def test_an_orphan_part_is_unlinked_before_yt_dlp_runs(self, job: paths.JobPaths) -> None: + stale = job.media_dir / "video.mkv.part" + stale.write_bytes(b"sparse") + seen: list[bool] = [] + + def on_download(_n: int) -> None: + seen.append(stale.exists()) + (job.media_dir / "video.mkv").write_bytes(b"data") + + download.download(job, "u", factory=factory_for(on_download=on_download)) + assert seen == [False] + + def test_a_resumable_part_survives(self, job: paths.JobPaths) -> None: + part = job.media_dir / "video.mkv.part" + part.write_bytes(b"sparse") + (job.media_dir / "video.mkv.part.aria2").write_bytes(b"ctrl") + seen: list[bool] = [] + + def on_download(_n: int) -> None: + seen.append(part.exists()) + (job.media_dir / "video.mkv").write_bytes(b"data") + + download.download(job, "u", factory=factory_for(on_download=on_download)) + assert seen == [True] + + def test_the_sweep_happens_once_not_per_attempt(self, job: paths.JobPaths) -> None: + # aria2 writes its control file during the failed attempt; sweeping again + # between retries would delete the very progress the retry exists to keep. + part = job.media_dir / "video.mkv.part" + calls: list[int] = [] + + def on_download(_n: int) -> None: + calls.append(len(calls) + 1) + if len(calls) < 2: + part.write_bytes(b"sparse") + raise RuntimeError("aria2c exited with code 5") + assert part.exists() + (job.media_dir / "video.mkv").write_bytes(b"data") + + download.download( + job, "u", factory=factory_for(on_download=on_download), sleep=lambda _s: None + ) + assert len(calls) == 2 + class TestDownload: def _succeed(self, job: paths.JobPaths) -> Any: