Coverage for src/ph/downloader.py: 97.4%
181 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-28 08:17 +0000
1"""Native HTTP and selective BitTorrent download engine."""
3import logging
4import os
5import re
6import ssl
7from collections.abc import Callable, Mapping, Sequence
8from datetime import UTC, datetime
9from email.utils import parsedate_to_datetime
10from functools import partial
11from pathlib import Path
12from urllib.error import HTTPError, URLError
13from urllib.parse import unquote, urlparse
14from urllib.request import Request, urlopen
16from ph.bittorrent import (
17 BitTorrentCancelled,
18 BitTorrentError,
19 BitTorrentSettings,
20 TorrentFileChoice,
21 TorrentSelectionRequired,
22 download_torrent_file,
23)
24from ph.models import DownloadResult, MediaDownload
25from ph.store import USER_AGENT
27LOGGER = logging.getLogger(__name__)
29type ProgressCallback = Callable[[str, int, int | None], None]
30type CancelCallback = Callable[[], bool]
33class DownloadError(RuntimeError):
34 """A download could not be completed."""
37class DownloadCancelled(DownloadError):
38 """The user cancelled an in-progress download."""
41class DownloadRateLimited(DownloadError):
42 """The remote store temporarily refused another download."""
44 def __init__(self, retry_after_seconds: float | None = None) -> None:
45 super().__init__("The store returned HTTP 429 Too Many Requests.")
46 self.retry_after_seconds = retry_after_seconds
49class DownloadSelectionRequired(DownloadError):
50 """A changed torrent needs an explicit file choice from the user."""
52 def __init__(self, error: TorrentSelectionRequired) -> None:
53 super().__init__(str(error))
54 self.torrent_url = error.torrent_url
55 self.expected_filename = error.expected_filename
56 self.catalogue_index = error.catalogue_index
57 self.candidates: tuple[TorrentFileChoice, ...] = error.candidates
58 self.total_files = error.total_files
61def _safe_filename(value: str) -> str:
62 name = Path(unquote(value)).name
63 name = re.sub(r"[\\/:*?\"<>|\x00-\x1f]", "_", name).strip(". ")
64 return name or "download.bin"
67def _filename_from_headers(headers: Mapping[str, str], url: str) -> str:
68 disposition = headers.get("Content-Disposition", "")
69 match = re.search(r"filename\*?=(?:UTF-8''|\")?([^\";]+)", disposition, re.IGNORECASE)
70 if match:
71 return _safe_filename(match.group(1).strip())
72 return _safe_filename(urlparse(url).path)
75def download_files(
76 downloads: Sequence[str | MediaDownload],
77 directory: Path,
78 referer: str,
79 timeout_seconds: float = 30.0,
80 progress: ProgressCallback | None = None,
81 cancelled: CancelCallback | None = None,
82 bittorrent_settings: BitTorrentSettings | None = None,
83 *,
84 resume: bool = False,
85) -> list[DownloadResult]:
86 """Download direct URLs or selected torrent files without external tools."""
88 if not downloads:
89 LOGGER.warning("Download requested with an empty media list")
90 raise DownloadError("The download list is empty.")
91 LOGGER.info("Downloading %d media item(s) into %s", len(downloads), directory)
92 directory.mkdir(parents=True, exist_ok=True)
93 resolved = [_as_media_download(download) for download in downloads]
94 results: list[DownloadResult] = []
95 for download in resolved:
96 _raise_if_cancelled(cancelled)
97 if download.torrent_file_index is None:
98 LOGGER.debug("Starting direct media download")
99 results.extend(
100 _download_with_urllib(
101 [download.url],
102 directory,
103 referer,
104 timeout_seconds,
105 progress,
106 cancelled,
107 resume,
108 )
109 )
110 continue
111 if not download.expected_filename:
112 raise DownloadError("The torrent download is missing its expected filename.")
113 destination = (
114 directory / download.expected_filename
115 if resume
116 else _unique_download_path(directory / download.expected_filename)
117 )
118 LOGGER.debug(
119 "Starting torrent media download file=%r index=%d",
120 download.expected_filename,
121 download.torrent_file_index,
122 )
123 try:
124 download_torrent_file(
125 download.url,
126 download.torrent_file_index,
127 download.expected_filename,
128 destination,
129 referer,
130 timeout_seconds,
131 partial(progress, download.expected_filename) if progress is not None else None,
132 cancelled,
133 bittorrent_settings,
134 download.torrent_file_path,
135 resume=resume,
136 )
137 except BitTorrentCancelled as error:
138 LOGGER.info("Torrent download cancelled")
139 raise DownloadCancelled("Download cancelled.") from error
140 except TorrentSelectionRequired as error:
141 LOGGER.warning("Torrent file selection requires user confirmation: %s", error)
142 raise DownloadSelectionRequired(error) from error
143 except BitTorrentError as error:
144 LOGGER.error("Torrent download failed: %s", error)
145 raise DownloadError(str(error)) from error
146 results.append(DownloadResult(download.url, destination))
147 LOGGER.info("Completed %d media download(s)", len(results))
148 return results
151def _as_media_download(download: str | MediaDownload) -> MediaDownload:
152 return download if isinstance(download, MediaDownload) else MediaDownload(download)
155def _download_with_urllib(
156 urls: Sequence[str],
157 directory: Path,
158 referer: str,
159 timeout_seconds: float,
160 progress: ProgressCallback | None,
161 cancelled: CancelCallback | None,
162 resume: bool = False,
163) -> list[DownloadResult]:
164 results: list[DownloadResult] = []
165 ssl_context = ssl.create_default_context()
166 for url in urls:
167 _raise_if_cancelled(cancelled)
168 resumable_partial = _resumable_partial(directory) if resume else None
169 resume_offset = resumable_partial.stat().st_size if resumable_partial is not None else 0
170 headers = {"User-Agent": USER_AGENT, "Referer": referer}
171 if resume_offset:
172 headers["Range"] = f"bytes={resume_offset}-"
173 request = Request(url, headers=headers)
174 try:
175 with urlopen(
176 request,
177 timeout=timeout_seconds,
178 context=ssl_context,
179 ) as response:
180 response_status = getattr(response, "status", None)
181 range_accepted = resume_offset > 0 and response_status == 206
182 filename = _filename_from_headers(response.headers, response.geturl())
183 if range_accepted and resumable_partial is not None:
184 partial = resumable_partial
185 destination = partial.with_name(partial.name.removesuffix(".part"))
186 else:
187 destination = (
188 directory / filename
189 if resume
190 else _unique_download_path(directory / filename)
191 )
192 partial = destination.with_name(destination.name + ".part")
193 if resumable_partial is not None and resumable_partial != partial: 193 ↛ 194line 193 didn't jump to line 194 because the condition on line 193 was never true
194 resumable_partial.unlink(missing_ok=True)
195 total_header = response.headers.get("Content-Length")
196 response_total = (
197 int(total_header) if total_header and total_header.isdigit() else None
198 )
199 downloaded = resume_offset if range_accepted else 0
200 total = (
201 _content_range_total(response.headers.get("Content-Range"))
202 if range_accepted
203 else response_total
204 )
205 if range_accepted and total is None and response_total is not None: 205 ↛ 206line 205 didn't jump to line 206 because the condition on line 205 was never true
206 total = resume_offset + response_total
207 try:
208 if progress and downloaded:
209 progress(filename, downloaded, total)
210 with partial.open("ab" if range_accepted else "wb") as output:
211 while True:
212 _raise_if_cancelled(cancelled)
213 chunk = response.read(1024 * 256)
214 if not chunk:
215 break
216 output.write(chunk)
217 downloaded += len(chunk)
218 if progress:
219 progress(filename, downloaded, total)
220 _raise_if_cancelled(cancelled)
221 os.replace(str(partial), str(destination))
222 except BaseException:
223 if not resume: 223 ↛ 225line 223 didn't jump to line 225 because the condition on line 223 was always true
224 partial.unlink(missing_ok=True)
225 raise
226 results.append(DownloadResult(url=url, path=destination))
227 except HTTPError as error:
228 if error.code == 416 and resumable_partial is not None:
229 LOGGER.warning(
230 "Server rejected HTTP resume; restarting %s",
231 resumable_partial.name.removesuffix(".part"),
232 )
233 resumable_partial.unlink(missing_ok=True)
234 results.extend(
235 _download_with_urllib(
236 [url],
237 directory,
238 referer,
239 timeout_seconds,
240 progress,
241 cancelled,
242 resume=True,
243 )
244 )
245 continue
246 if error.code == 429:
247 retry_after = _retry_after_seconds(error.headers.get("Retry-After"))
248 LOGGER.warning(
249 "HTTP download rate limited status=429 retry_after_seconds=%s",
250 retry_after,
251 )
252 raise DownloadRateLimited(retry_after) from error
253 LOGGER.error("HTTP download returned status=%d", error.code)
254 raise DownloadError("Download returned HTTP %d." % error.code) from error
255 except (URLError, TimeoutError, OSError) as error:
256 reason = getattr(error, "reason", error)
257 LOGGER.error("HTTP download failed: %s", reason)
258 raise DownloadError(f"Download failed: {reason}") from error
259 return results
262def _resumable_partial(directory: Path) -> Path | None:
263 partials = tuple(path for path in directory.glob("*.part") if path.is_file())
264 return partials[0] if len(partials) == 1 else None
267def _content_range_total(value: str | None) -> int | None:
268 if not value:
269 return None
270 match = re.fullmatch(r"bytes\s+\d+-\d+/(\d+|\*)", value.strip(), re.IGNORECASE)
271 if match is None or match.group(1) == "*":
272 return None
273 return int(match.group(1))
276def _retry_after_seconds(value: str | None) -> float | None:
277 if value is None:
278 return None
279 normalized = value.strip()
280 if normalized.isdigit():
281 return float(normalized)
282 try:
283 retry_at = parsedate_to_datetime(normalized)
284 except TypeError, ValueError, OverflowError:
285 return None
286 if retry_at.tzinfo is None: 286 ↛ 288line 286 didn't jump to line 288 because the condition on line 286 was always true
287 retry_at = retry_at.replace(tzinfo=UTC)
288 return max(0.0, (retry_at - datetime.now(UTC)).total_seconds())
291def _raise_if_cancelled(cancelled: CancelCallback | None) -> None:
292 if cancelled is not None and cancelled():
293 raise DownloadCancelled("Download cancelled.")
296def _unique_download_path(destination: Path) -> Path:
297 partial = destination.with_name(destination.name + ".part")
298 if not destination.exists() and not partial.exists():
299 return destination
300 suffixes = "".join(destination.suffixes)
301 stem = destination.name[: -len(suffixes)] if suffixes else destination.name
302 counter = 2
303 while True:
304 candidate = destination.with_name(f"{stem} ({counter}){suffixes}")
305 partial = candidate.with_name(candidate.name + ".part")
306 if not candidate.exists() and not partial.exists():
307 return candidate
308 counter += 1