diff --git a/src/ccn_transcribe/stages/download.py b/src/ccn_transcribe/stages/download.py new file mode 100644 index 0000000..9b8b329 --- /dev/null +++ b/src/ccn_transcribe/stages/download.py @@ -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 ''}" diff --git a/tests/test_stages_download.py b/tests/test_stages_download.py new file mode 100644 index 0000000..74e76fd --- /dev/null +++ b/tests/test_stages_download.py @@ -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