JMR-dev/audio-scribe
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat(stages): add the download stage with retry and error classification

Browse Source 
yt-dlp messages are matched against a regex table kept as data, so the
classification is unit-testable without the network and can be amended when
YouTube changes its wording. Only network failures and aria2 process failures
retry; a private video will still be private on the third attempt, and disk
exhaustion is fatal rather than per-job.

The retry loop is ours because it has to be: ExternalFD.real_download calls the
downloader exactly once and turns a nonzero exit into report_error(), so
yt-dlp's own `retries` never covers aria2c. Since --lowest-speed-limit
deliberately makes aria2c exit nonzero on a stall, without this loop that stall
would end the job instead of resuming it -- the .aria2 control file survives, so
each attempt picks up where the last stopped.

A missing aria2c downgrades to the native downloader with a warning rather than
failing the run.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
JMR-dev
2026-09-13 14:48:09 -05:00
co-authored by Claude Opus 5
parent 3d511f8fea
commit cc1aa9a269
2 changed files with 610 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ccn_transcribe/stages/download.py
+262
View File
@@ -0,0 +1,262 @@
"""The download stage.
yt-dlp classification lives in a regex table so it can be unit-tested without the
network and amended when YouTube changes its wording.
"""
from __future__ import annotations
import logging
import re
import shutil
import time
from collections.abc import Callable
from typing import TYPE_CHECKING, Any
from ccn_transcribe import errors, paths
from ccn_transcribe.media import ytdlp_opts
if TYPE_CHECKING:
from collections.abc import Sequence
from pathlib import Path
from ccn_transcribe.paths import JobPaths
log = logging.getLogger(__name__)
YdlFactory = Callable[[dict[str, Any]], Any]
# (pattern, error class, hint). Order matters: first match wins.
SOURCE_PATTERNS: tuple[tuple[re.Pattern[str], type[errors.SourceError], str], ...] = (
(
re.compile(r"private video|sign in if you.{0,20}been granted access", re.I),
errors.PrivateVideoError,
"This video is private. Supply cookies: --cookies-from-browser firefox",
),
(
re.compile(r"sign in to confirm your age|age.?restricted|inappropriate for some", re.I),
errors.AgeGatedError,
"Age-restricted. Use --cookies-from-browser with a signed-in account.",
),
(
re.compile(r"members-only|available to (this channel's )?members", re.I),
errors.MembersOnlyError,
"Members-only content. Needs cookies from a subscribed account.",
),
(
re.compile(r"confirm you.{0,5}re not a bot|not a bot", re.I),
errors.BotCheckError,
"YouTube bot check. Try --cookies-from-browser, or check that a JS runtime "
"is available: ccn-transcribe doctor",
),
(
re.compile(
r"geo.?restricted|geo.?block|available in your country|"
r"not available from your location|blocked it in your",
re.I,
),
errors.GeoBlockedError,
"Geo-blocked from this location.",
),
(
re.compile(
r"http error 4(0[34]|10)|video unavailable|does not exist|been removed|"
r"is not available",
re.I,
),
errors.NotFoundError,
"Video not available (removed, or the URL is wrong).",
),
(
re.compile(
r"unable to download|connection reset|timed out|timeout|"
r"temporary failure|http error 5\d\d|network is unreachable",
re.I,
),
errors.NetworkError,
"Network problem; this is retryable.",
),
)
# aria2c process exit codes worth explaining. 9 is promoted to a fatal DiskFull.
ARIA2_EXITS: dict[int, tuple[str, str]] = {
1: ("unknown error", "See logs/aria2.log in the job directory."),
2: ("timed out", "The server stopped responding; retrying usually clears it."),
3: ("resource not found", "Format URLs are signed and expire (~6h); re-run to re-extract."),
5: ("aborted: too slow", "Hit --lowest-speed-limit. Lower it for a genuinely slow link."),
6: ("network problem", "Retryable."),
7: ("unfinished downloads remained", "The .aria2 control file allows a resume."),
8: ("server does not support resume", "Restarting the download from scratch."),
22: ("HTTP 4xx response header", "Often a 403 from too many connections: --aria2c-conns 4."),
24: ("HTTP authorization failed", "Needs cookies: --cookies-from-browser."),
}
def classify(message: str, job_id: str | None = None) -> errors.CcnError:
"""Map a yt-dlp message onto an actionable error."""
if "no space left" in message.lower():
return errors.DiskFullError(message, hint="Free some space and re-run.")
for pattern, cls, hint in SOURCE_PATTERNS:
if pattern.search(message):
return cls(message, job_id=job_id, hint=hint)
found = re.search(r"exited with code (\d+)", message)
if found:
code = int(found.group(1))
if code == 9:
return errors.DiskFullError(message, hint="aria2c ran out of disk space.")
reason, hint = ARIA2_EXITS.get(code, ("failed", "See logs/aria2.log."))
return errors.Aria2Error(
f"aria2c {reason} (exit {code})", aria2_exit_code=code, job_id=job_id, hint=hint
)
return errors.DownloadError(message, job_id=job_id, hint="Re-run with -v for detail.")
def default_factory(options: dict[str, Any]) -> Any:
from yt_dlp import YoutubeDL # pylint: disable=import-outside-toplevel
return YoutubeDL(options) # pyright: ignore[reportArgumentType]
def extract_info(
url: str,
config: ytdlp_opts.DownloadConfig = ytdlp_opts.DEFAULT_DOWNLOAD,
*,
factory: YdlFactory | None = None,
) -> dict[str, Any]:
"""Metadata only, no download."""
options = ytdlp_opts.ydl_opts("-", config)
options.update({"skip_download": True, "writeinfojson": False, "extract_flat": "in_playlist"})
make = factory or default_factory
try:
with make(options) as ydl:
info: dict[str, Any] | None = ydl.extract_info(url, download=False)
except Exception as exc: # pylint: disable=broad-exception-caught
raise classify(str(exc)) from exc
if info is None:
raise errors.NotFoundError(f"no metadata for {url}", hint="Check the URL.")
return info
def entries_for(info: dict[str, Any]) -> list[dict[str, Any]]:
"""Flatten a playlist into one record per video."""
raw = info.get("entries")
if raw is None:
return [info]
return [entry for entry in raw if entry]
def identify(entry: dict[str, Any], url: str) -> str:
extractor = entry.get("extractor_key") or entry.get("ie_key") or ""
video_id = entry.get("id")
if extractor and video_id:
return paths.job_id(str(extractor), str(video_id))
return paths.url_job_id(url)
def find_video(job: JobPaths) -> Path | None:
"""The downloaded media file, whatever extension yt-dlp settled on."""
for candidate in sorted(job.media_dir.glob("video.*")):
if candidate.suffix not in (".part", ".aria2", ".json", ".ytdl"):
return candidate
return None
def download_options(job: JobPaths, config: ytdlp_opts.DownloadConfig) -> dict[str, Any]:
return ytdlp_opts.ydl_opts(
str(job.media_dir / "video.%(ext)s"),
config,
aria2_log=job.logs_dir / "aria2.log",
)
def download(
job: JobPaths,
url: str,
config: ytdlp_opts.DownloadConfig = ytdlp_opts.DEFAULT_DOWNLOAD,
*,
factory: YdlFactory | None = None,
attempts: int = 3,
backoff: float = 2.0,
sleep: Callable[[float], None] = time.sleep,
) -> Path:
"""Download to the job directory, retrying transient failures ourselves.
yt-dlp's ExternalFD calls the downloader exactly once and turns a nonzero exit
into report_error(); its `retries` option does not cover external downloaders.
Since --lowest-speed-limit deliberately makes aria2c exit nonzero on a stall,
without this loop that stall would become a job failure instead of a resume.
"""
job.ensure()
options = download_options(job, config)
make = factory or default_factory
last: errors.CcnError | None = None
for attempt in range(1, attempts + 1):
try:
with make(options) as ydl:
ydl.download([url])
except Exception as exc: # pylint: disable=broad-exception-caught
failure = classify(str(exc), job_id=job.root.name)
if isinstance(failure, errors.FatalError) or not is_retryable(failure):
raise failure from exc
last = failure
if attempt < attempts:
delay = backoff ** (attempt - 1)
log.warning(
"download attempt %d/%d failed (%s); retrying in %.0fs",
attempt,
attempts,
failure,
delay,
)
sleep(delay)
continue
found = find_video(job)
if found is not None:
return found
last = errors.IncompleteDownloadError(
"yt-dlp reported success but produced no media file",
job_id=job.root.name,
hint="Check the job's logs, or re-run with -v.",
)
break
raise last or errors.DownloadError("download failed", job_id=job.root.name)
def is_retryable(failure: errors.CcnError) -> bool:
# A private video will still be private on the third try, and continuing past
# ENOSPC only makes more corrupt artifacts.
if isinstance(failure, errors.FatalError):
return False
return failure.retryable or isinstance(failure, (errors.Aria2Error, errors.DownloadError))
def aria2c_available(which: Callable[[str], str | None] | None = None) -> bool:
lookup = which or shutil.which
return lookup("aria2c") is not None
def resolve_aria2c(requested: bool, *, available: bool) -> bool:
"""Never fail because aria2c is missing; fall back and say so."""
if requested and not available:
log.warning("aria2c not found; using yt-dlp's native downloader")
return False
return requested
def describe_entry(entry: dict[str, Any]) -> dict[str, Any]:
return {
"id": entry.get("id"),
"title": entry.get("title"),
"url": entry.get("webpage_url") or entry.get("url"),
"duration": entry.get("duration"),
"uploader": entry.get("uploader"),
"upload_date": entry.get("upload_date"),
}
def summarize(entries: Sequence[dict[str, Any]]) -> str:
return f"{len(entries)} video{'s' if len(entries) != 1 else ''}"
tests/test_stages_download.py
+348
View File
@@ -0,0 +1,348 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Any
import pytest
from ccn_transcribe import errors, paths
from ccn_transcribe.media import ytdlp_opts
from ccn_transcribe.stages import download
if TYPE_CHECKING:
from pathlib import Path
@pytest.fixture
def job(tmp_path: Path) -> paths.JobPaths:
j = paths.Workspace(tmp_path).job("youtube-abc")
j.ensure()
return j
class FakeYdl:
"""Stands in for YoutubeDL; records calls and can be told to fail."""
def __init__(
self,
options: dict[str, Any],
*,
raises: Exception | None = None,
info: dict[str, Any] | None = None,
on_download: Any = None,
) -> None:
self.options = options
self.raises = raises
self.info = info
self.on_download = on_download
self.downloads: list[list[str]] = []
def __enter__(self) -> FakeYdl:
return self
def __exit__(self, *_exc: object) -> None:
return
def extract_info( # pylint: disable=unused-argument
self, url: str, download: bool = True
) -> dict[str, Any] | None:
if self.raises:
raise self.raises
return self.info
def download(self, urls: list[str]) -> None:
self.downloads.append(urls)
if self.on_download is not None:
self.on_download(len(self.downloads))
if self.raises:
raise self.raises
def factory_for(**kw: Any) -> Any:
made: list[FakeYdl] = []
def make(options: dict[str, Any]) -> FakeYdl:
ydl = FakeYdl(options, **kw)
made.append(ydl)
return ydl
make.made = made # type: ignore[attr-defined]
return make
class TestClassify:
@pytest.mark.parametrize(
("message", "expected"),
[
(
"ERROR: Private video. Sign in if you've been granted access",
errors.PrivateVideoError,
),
("Sign in to confirm your age", errors.AgeGatedError),
("This video is available to members-only", errors.MembersOnlyError),
("Sign in to confirm you're not a bot", errors.BotCheckError),
(
"The uploader has not made this video available in your country",
errors.GeoBlockedError,
),
("ERROR: Video unavailable", errors.NotFoundError),
("HTTP Error 404: Not Found", errors.NotFoundError),
("HTTP Error 503: Service Unavailable", errors.NetworkError),
("Unable to download webpage: timed out", errors.NetworkError),
],
)
def test_maps_messages_to_actionable_errors(
self, message: str, expected: type[errors.JobError]
) -> None:
assert isinstance(download.classify(message), expected)
def test_every_source_error_carries_a_hint(self) -> None:
assert download.classify("Private video").hint
def test_the_bot_check_hint_points_at_doctor(self) -> None:
# A missing JS runtime presents as a bot check, so this is the likely cause.
hint = download.classify("Sign in to confirm you're not a bot").hint or ""
assert "doctor" in hint
def test_only_network_errors_are_retryable(self) -> None:
assert download.classify("timed out").retryable is True
assert download.classify("Private video").retryable is False
def test_no_space_becomes_a_fatal_disk_full(self) -> None:
# Continuing after ENOSPC just produces more corrupt artifacts.
assert isinstance(download.classify("No space left on device"), errors.DiskFullError)
def test_an_unrecognized_message_is_still_a_job_error(self) -> None:
failure = download.classify("something entirely new")
assert isinstance(failure, errors.DownloadError)
assert failure.hint
def test_the_job_id_is_attached(self) -> None:
failure = download.classify("Private video", job_id="youtube-x")
assert isinstance(failure, errors.JobError)
assert failure.job_id == "youtube-x"
class TestAria2ExitCodes:
def test_a_stall_abort_is_explained(self) -> None:
failure = download.classify("aria2c exited with code 5")
assert isinstance(failure, errors.Aria2Error)
assert failure.aria2_exit_code == 5
assert "lowest-speed-limit" in (failure.hint or "")
def test_expired_urls_are_explained(self) -> None:
assert "expire" in (download.classify("aria2c exited with code 3").hint or "")
def test_a_403_suggests_fewer_connections(self) -> None:
assert "aria2c-conns" in (download.classify("aria2c exited with code 22").hint or "")
def test_out_of_disk_is_fatal_not_per_job(self) -> None:
assert isinstance(download.classify("aria2c exited with code 9"), errors.DiskFullError)
def test_an_unknown_exit_code_still_classifies(self) -> None:
failure = download.classify("aria2c exited with code 77")
assert isinstance(failure, errors.Aria2Error)
assert failure.aria2_exit_code == 77
class TestExtractInfo:
def test_returns_metadata(self) -> None:
make = factory_for(info={"id": "abc", "title": "T"})
assert download.extract_info("u", factory=make)["id"] == "abc"
def test_does_not_download(self) -> None:
make = factory_for(info={"id": "abc"})
download.extract_info("u", factory=make)
assert make.made[0].options["skip_download"] is True
def test_flattens_playlists_during_enumeration(self) -> None:
make = factory_for(info={"id": "abc"})
download.extract_info("u", factory=make)
assert make.made[0].options["extract_flat"] == "in_playlist"
def test_no_metadata_raises_not_found(self) -> None:
with pytest.raises(errors.NotFoundError):
download.extract_info("u", factory=factory_for(info=None))
def test_errors_are_classified(self) -> None:
make = factory_for(raises=RuntimeError("Private video"))
with pytest.raises(errors.PrivateVideoError):
download.extract_info("u", factory=make)
class TestEntries:
def test_a_single_video_is_one_entry(self) -> None:
assert download.entries_for({"id": "a"}) == [{"id": "a"}]
def test_a_playlist_expands(self) -> None:
info = {"entries": [{"id": "a"}, {"id": "b"}]}
assert [e["id"] for e in download.entries_for(info)] == ["a", "b"]
def test_dead_entries_are_dropped(self) -> None:
# ignoreerrors leaves a None where an unavailable video was.
info = {"entries": [{"id": "a"}, None, {"id": "b"}]}
assert len(download.entries_for(info)) == 2
def test_an_empty_playlist_is_empty(self) -> None:
assert download.entries_for({"entries": []}) == []
class TestIdentify:
def test_uses_extractor_and_id(self) -> None:
assert download.identify({"extractor_key": "Youtube", "id": "abc"}, "u") == "youtube-abc"
def test_accepts_ie_key(self) -> None:
assert download.identify({"ie_key": "Youtube", "id": "abc"}, "u") == "youtube-abc"
def test_falls_back_to_the_url_when_unidentifiable(self) -> None:
assert download.identify({}, "https://a.test/x").startswith("url-")
class TestFindVideo:
def test_finds_the_media_file(self, job: paths.JobPaths) -> None:
(job.media_dir / "video.mkv").write_bytes(b"x")
assert download.find_video(job) == job.media_dir / "video.mkv"
def test_ignores_part_and_control_files(self, job: paths.JobPaths) -> None:
(job.media_dir / "video.mkv.part").write_bytes(b"x")
(job.media_dir / "video.mkv.part.aria2").write_bytes(b"x")
assert download.find_video(job) is None
def test_ignores_the_info_json(self, job: paths.JobPaths) -> None:
(job.media_dir / "video.info.json").write_text("{}")
assert download.find_video(job) is None
def test_none_when_nothing_downloaded(self, job: paths.JobPaths) -> None:
assert download.find_video(job) is None
class TestDownload:
def _succeed(self, job: paths.JobPaths) -> Any:
def on_download(_n: int) -> None:
(job.media_dir / "video.mkv").write_bytes(b"data")
return factory_for(on_download=on_download)
def test_returns_the_downloaded_file(self, job: paths.JobPaths) -> None:
assert download.download(job, "u", factory=self._succeed(job)).name == "video.mkv"
def test_writes_the_aria2_log_into_the_job(self, job: paths.JobPaths) -> None:
make = self._succeed(job)
download.download(job, "u", ytdlp_opts.DownloadConfig(use_aria2c=True), factory=make)
args = make.made[0].options["external_downloader_args"]["aria2c"]
assert any(str(job.logs_dir) in a for a in args)
def test_retries_a_transient_failure(self, job: paths.JobPaths) -> None:
# yt-dlp does not retry external downloader failures, so this loop is the
# only thing standing between a stall-abort and a failed job.
attempts: list[int] = []
def on_download(_n: int) -> None:
attempts.append(len(attempts) + 1)
if len(attempts) < 2:
raise RuntimeError("aria2c exited with code 5")
(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(attempts) == 2
def test_gives_up_after_the_attempt_cap(self, job: paths.JobPaths) -> None:
make = factory_for(raises=RuntimeError("aria2c exited with code 5"))
with pytest.raises(errors.Aria2Error):
download.download(job, "u", factory=make, attempts=3, sleep=lambda _s: None)
assert len(make.made) == 3
def test_does_not_retry_a_permanent_failure(self, job: paths.JobPaths) -> None:
# A private video will still be private on the third try.
make = factory_for(raises=RuntimeError("Private video"))
with pytest.raises(errors.PrivateVideoError):
download.download(job, "u", factory=make, sleep=lambda _s: None)
assert len(make.made) == 1
def test_does_not_retry_a_fatal_error(self, job: paths.JobPaths) -> None:
make = factory_for(raises=RuntimeError("No space left on device"))
with pytest.raises(errors.DiskFullError):
download.download(job, "u", factory=make, sleep=lambda _s: None)
assert len(make.made) == 1
def test_success_without_a_file_is_an_error(self, job: paths.JobPaths) -> None:
with pytest.raises(errors.IncompleteDownloadError):
download.download(job, "u", factory=factory_for(), sleep=lambda _s: None)
def test_backoff_grows(self, job: paths.JobPaths) -> None:
delays: list[float] = []
make = factory_for(raises=RuntimeError("timed out"))
with pytest.raises(errors.JobError):
download.download(job, "u", factory=make, attempts=3, backoff=2.0, sleep=delays.append)
assert delays == [1.0, 2.0]
class TestAria2Availability:
def test_detects_the_real_binary(self) -> None:
assert download.aria2c_available() is True
def test_reports_absence(self) -> None:
assert download.aria2c_available(which=lambda _n: None) is False
def test_missing_aria2c_downgrades_rather_than_failing(
self, caplog: pytest.LogCaptureFixture
) -> None:
assert download.resolve_aria2c(requested=True, available=False) is False
assert "native downloader" in caplog.text
def test_available_aria2c_is_used(self) -> None:
assert download.resolve_aria2c(requested=True, available=True) is True
def test_opting_out_is_respected(self) -> None:
assert download.resolve_aria2c(requested=False, available=True) is False
class TestDescribe:
def test_extracts_the_fields_the_json_output_records(self) -> None:
got = download.describe_entry(
{
"id": "a",
"title": "T",
"webpage_url": "u",
"duration": 12.0,
"uploader": "U",
"upload_date": "20260913",
}
)
assert got == {
"id": "a",
"title": "T",
"url": "u",
"duration": 12.0,
"uploader": "U",
"upload_date": "20260913",
}
def test_falls_back_to_the_flat_url(self) -> None:
assert download.describe_entry({"url": "u"})["url"] == "u"
def test_summarize_pluralizes(self) -> None:
assert download.summarize([{}]) == "1 video"
assert download.summarize([{}, {}]) == "2 videos"
def test_the_default_factory_builds_a_real_youtube_dl() -> None:
ydl = download.default_factory({"quiet": True, "skip_download": True})
assert hasattr(ydl, "extract_info")
class TestRetryPolicy:
def test_network_failures_are_retried(self) -> None:
assert download.is_retryable(download.classify("timed out")) is True
def test_aria2_stalls_are_retried(self) -> None:
assert download.is_retryable(download.classify("aria2c exited with code 5")) is True
def test_permanent_source_failures_are_not(self) -> None:
assert download.is_retryable(download.classify("Private video")) is False
def test_fatal_errors_are_never_retried(self) -> None:
# Continuing past ENOSPC only produces more corrupt artifacts.
assert download.is_retryable(errors.DiskFullError("full")) is False
assert download.is_retryable(errors.SchemaTooNewError("newer")) is False