diff --git a/backend/app/services/native_ingest_common.py b/backend/app/services/native_ingest_common.py new file mode 100644 index 0000000..4317e9a --- /dev/null +++ b/backend/app/services/native_ingest_common.py @@ -0,0 +1,310 @@ +"""Shared primitives for the native-ingest platform adapters (Patreon, +SubscribeStar, …) — the single home for logic the per-platform client/downloader +modules would otherwise each copy. + +DRY pass 2026-06-17 (#899): these used to live in `patreon_*` with the +SubscribeStar modules importing patreon privates (wrong owner + sibling-coupling). +They're platform-agnostic, so they live here and both adapters import them. The +per-platform modules keep only what genuinely differs (feed parsing, the media +shape, Patreon's Mux/yt-dlp video branch + detail-fetch enrichment). + +FC runs on a plain-HTTP homelab; nothing here uses a secure-context Web API. +""" + +from __future__ import annotations + +import contextlib +import http.cookiejar +import json +import logging +import os +import time +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from urllib.parse import urlsplit + +import requests + +from ..utils.paths import filehash_from_url, safe_ext +from .file_validator import is_validatable, quarantine_file, validate_file + +log = logging.getLogger(__name__) + +_USER_AGENT = ( + "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36" +) + +# 429 backoff (plan #703): ride out a transient API rate-limit instead of failing +# the whole walk. Honor the server's Retry-After; else exponential, capped. +_MAX_429_RETRIES = 3 +_BACKOFF_BASE_SECONDS = 2.0 +_BACKOFF_CAP_SECONDS = 30.0 + +# Media-download tuning (shared by every platform downloader). +_TIMEOUT_SECONDS = 120.0 +_CHUNK = 1 << 16 +_MAX_MEDIA_RETRIES = 3 +_TRANSIENT_TRANSPORT_EXC = ( + requests.ConnectionError, + requests.Timeout, + requests.exceptions.ChunkedEncodingError, +) + +_TITLE_MAX = 40 +# Windows/gallery-dl path-restrict forbidden set + path separators. +_FORBIDDEN = set('<>:"/\\|?*') + + +# -- HTTP session ---------------------------------------------------------- + +def make_session( + cookies_path: str | Path | None, + *, + accept: str = "*/*", + extra_headers: dict | None = None, +) -> requests.Session: + """Build a requests.Session loaded with the Netscape cookies.txt + CredentialService materializes. `accept` sets the Accept header (the JSON:API + vs HTML feed differ); `extra_headers` adds platform headers (e.g. + X-Requested-With). Missing/unparseable cookies log a warning, never fail.""" + session = requests.Session() + headers = {"User-Agent": _USER_AGENT, "Accept": accept} + if extra_headers: + headers.update(extra_headers) + session.headers.update(headers) + if cookies_path and os.path.isfile(str(cookies_path)): + try: + jar = http.cookiejar.MozillaCookieJar(str(cookies_path)) + jar.load(ignore_discard=True, ignore_expires=True) + session.cookies = jar # type: ignore[assignment] + except (OSError, http.cookiejar.LoadError) as exc: + log.warning("Could not load cookies from %s: %s", cookies_path, exc) + return session + + +def retry_after_seconds( + resp: requests.Response, + attempt: int, + *, + base: float = _BACKOFF_BASE_SECONDS, + cap: float = _BACKOFF_CAP_SECONDS, +) -> float: + """Backoff for a 429: the numeric Retry-After header if present, else + exponential base·2^(attempt-1), both capped.""" + header = resp.headers.get("Retry-After") + if header: + try: + return min(float(header), cap) + except (TypeError, ValueError): + pass + return min(base * (2 ** max(0, attempt - 1)), cap) + + +# -- filename / path helpers ----------------------------------------------- + +def sanitize_segment(name: str) -> str: + """Make `name` safe for one filesystem path segment: replace separators, the + Windows-forbidden set, and control chars with `_`; strip trailing dots/spaces + (gallery-dl path-restrict). Never empty (falls back to `_`).""" + out = ["_" if (ch in _FORBIDDEN or ord(ch) < 32) else ch for ch in name] + cleaned = "".join(out).rstrip(". ") + return cleaned or "_" + + +def basename_from_url(url: str) -> str: + """Derive a sane filename from a URL when the media has no name: path basename + with a junk-extension guard (safe_ext), bounded stem; falls back to the URL's + content hash, then "file".""" + path = urlsplit(url).path + base = os.path.basename(path) + if base: + ext = safe_ext(base) + stem = base[: -len(Path(base).suffix)] if Path(base).suffix else base + stem = stem[:120] or "file" + return f"{stem}{ext}" + return filehash_from_url(url) or "file" + + +def post_dir_name(post: dict) -> str: + """`__` matching gallery-dl's layout (date prefix + omitted when published_at is missing/unparseable; title is empty for platforms + with no title field). Accepts both ISO and trailing-`Z` published_at.""" + post_id = str(post.get("id") or "") + attrs = post.get("attributes") or {} + title = attrs.get("title") + title40 = (title if isinstance(title, str) else "")[:_TITLE_MAX] + published = attrs.get("published_at") + date_prefix = None + if isinstance(published, str) and published: + s = published.strip() + if s.endswith("Z"): + s = s[:-1] + "+00:00" + try: + date_prefix = f"{datetime.fromisoformat(s):%Y-%m-%d}" + except ValueError: + date_prefix = None + raw = f"{date_prefix}_{post_id}_{title40}" if date_prefix else f"{post_id}_{title40}" + return sanitize_segment(raw) + + +# -- per-item download outcomes (shared dataclasses) ----------------------- + +@dataclass +class MediaOutcome: + """Per-media result of a download_post pass. status ∈ downloaded / + skipped_seen / skipped_disk / quarantined / error. `path` is the on-disk file + (downloaded / skipped_disk), the quarantine dest (quarantined), or None; + `error` is the failure/validation reason (error/quarantined) else None.""" + + media: object + status: str + path: Path | None + error: str | None + + +@dataclass +class PostRecordOutcome: + """Result of write_post_record — mirrors the MediaOutcome contract so the core + reports per-post handling. `path` is the _post.json sidecar (None when the post + had no id); the rest is the captured body's shape for the run log.""" + + path: Path | None + post_type: str | None + title: str | None + body_chars: int + + +# -- base downloader (shared fetch/validate plumbing) ---------------------- + +class BaseNativeDownloader: + """Shared download plumbing for native-platform downloaders: the streaming + GET (transient-retry + Range-resume) and file validation/quarantine. Platform + downloaders subclass this and implement `download_post` / `write_post_record` + / the per-media sidecar (and any platform-specific fetch, e.g. Patreon's + Mux/yt-dlp video branch). PURE: no DB; the seen-skip is an injected predicate. + """ + + def __init__( + self, + images_root: Path, + cookies_path: str | None = None, + *, + platform: str, + validate: bool = True, + rate_limit: float = 0.0, + session: requests.Session | None = None, + ): + self.images_root = Path(images_root) + self.cookies_path = str(cookies_path) if cookies_path else None + self.platform = platform + self._validate = validate + self._rate_limit = rate_limit or 0.0 + self.session = session if session is not None else make_session(cookies_path) + + # -- download seams ---------------------------------------------------- + + def _fetch_get(self, url: str, dest: Path) -> Path: + """Stream `url` to a .part then atomic-rename to `dest`.""" + part = dest.with_name(dest.name + ".part") + try: + self._fetch_to_file(url, part) + except Exception: + with contextlib.suppress(OSError): + part.unlink() + raise + os.replace(part, dest) + return dest + + def _fetch_to_file(self, url: str, dest: Path) -> None: + """Stream a URL to `dest`, retrying TRANSIENT failures (transport blips, + 429 honoring Retry-After, 5xx) with backoff + resume-from-disk (Range); + failing fast on permanent 4xx (404/403). Resume: a retry with bytes on + disk asks `Range: bytes=-`; 206 → append, 200 → restart clean, 416 → + already complete. The caller stages into a `.part` so a non-range server + never corrupts the output.""" + attempt = 0 + while True: + have = dest.stat().st_size if dest.exists() else 0 + headers = {"Range": f"bytes={have}-"} if have > 0 else None + try: + resp = self.session.get( + url, stream=True, timeout=_TIMEOUT_SECONDS, headers=headers, + ) + if (resp.status_code == 429 or resp.status_code >= 500) \ + and attempt < _MAX_MEDIA_RETRIES: + attempt += 1 + delay = retry_after_seconds(resp, attempt) + log.warning( + "%s media transient HTTP %d (%s) — backing off %.1fs " + "(retry %d/%d)", + self.platform, resp.status_code, url, delay, attempt, + _MAX_MEDIA_RETRIES, + ) + time.sleep(delay) + continue + if have > 0 and resp.status_code == 416: + return + resp.raise_for_status() + mode = "ab" if (have > 0 and resp.status_code == 206) else "wb" + with open(dest, mode) as fh: + for chunk in resp.iter_content(chunk_size=_CHUNK): + if chunk: + fh.write(chunk) + return + except _TRANSIENT_TRANSPORT_EXC as exc: + if attempt >= _MAX_MEDIA_RETRIES: + raise + attempt += 1 + delay = min(2.0 * (2 ** (attempt - 1)), _BACKOFF_CAP_SECONDS) + log.warning( + "%s media transport error (%s) — backing off %.1fs " + "(retry %d/%d): %s", + self.platform, url, delay, attempt, _MAX_MEDIA_RETRIES, exc, + ) + time.sleep(delay) + + # -- validation -------------------------------------------------------- + + def _validate_path( + self, path: Path, artist_slug: str, source_url: str | None = None + ) -> tuple[str | None, Path | None]: + """Validate a freshly-written file; quarantine if bad (shared + file_validator move + provenance sidecar). Returns (reason, + quarantine_dest) when quarantined, else (None, None). Logs the quarantine + so a corrupt file is visible in the worker logs, not just counted (#899 + L2).""" + if not self._validate or not is_validatable(path): + return None, None + try: + result = validate_file(path) + except Exception as exc: + log.warning("Validator raised on %s: %s", path, exc) + return None, None + if result.ok: + return None, None + dest = quarantine_file( + self.images_root, path, artist_slug, self.platform, + url=source_url, result=result, + ) + reason = result.reason or "validation failed" + log.warning( + "%s quarantined %s (%s) — %s", + self.platform, dest or path, artist_slug, reason, + ) + return reason, (dest or path) + + # -- sidecar (per-media, minimal) -------------------------------------- + + def _write_minimal_sidecar( + self, post: dict, media_path: Path, *, source_url: str | None = None + ) -> Path: + """Post-first per-media sidecar (#856): image identity ONLY + (category/id/source_url). The post body/links live solely in _post.json.""" + data: dict = {"category": self.platform, "id": str(post.get("id") or "")} + if source_url: + data["source_url"] = source_url + sidecar_path = media_path.with_suffix(".json") + sidecar_path.write_text(json.dumps(data, indent=2)) + return sidecar_path diff --git a/backend/app/services/patreon_client.py b/backend/app/services/patreon_client.py index 907ce73..b5264b7 100644 --- a/backend/app/services/patreon_client.py +++ b/backend/app/services/patreon_client.py @@ -26,9 +26,7 @@ FC runs on a plain-HTTP homelab; nothing here uses a secure-context Web API. from __future__ import annotations -import http.cookiejar import logging -import os import re import time from collections.abc import Iterator @@ -39,46 +37,20 @@ from urllib.parse import parse_qs, urlsplit import requests -from ..utils.paths import filehash_from_url, safe_ext +from ..utils.paths import filehash_from_url from ..utils.prosemirror import post_body_html +from .native_ingest_common import ( + _MAX_429_RETRIES, + basename_from_url, + make_session, + retry_after_seconds, +) log = logging.getLogger(__name__) _POSTS_URL = "https://www.patreon.com/api/posts" -_USER_AGENT = ( - "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " - "(KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36" -) _TIMEOUT_SECONDS = 30.0 -# 429 backoff (plan #703): ride out a transient API rate-limit instead of -# failing the whole walk (which would stamp RATE_LIMITED → platform-wide -# cooldown → every Patreon source dark). Honor the server's `Retry-After`; -# otherwise exponential base·2^(n-1), capped. Only after the retries are -# exhausted does the 429 propagate as terminal RATE_LIMITED. -_MAX_429_RETRIES = 3 -_BACKOFF_BASE_SECONDS = 2.0 -_BACKOFF_CAP_SECONDS = 30.0 - - -def _retry_after_seconds( - resp: requests.Response, - attempt: int, - *, - base: float = _BACKOFF_BASE_SECONDS, - cap: float = _BACKOFF_CAP_SECONDS, -) -> float: - """Backoff delay for a 429: the `Retry-After` seconds header if present and - numeric, else exponential `base·2^(attempt-1)`, both capped. (HTTP-date form - of Retry-After is rare here and falls through to exponential.)""" - header = resp.headers.get("Retry-After") - if header: - try: - return min(float(header), cap) - except (TypeError, ValueError): - pass - return min(base * (2 ** max(0, attempt - 1)), cap) - # JSON:API request contract (observed from real traffic — see module plan). _INCLUDE = ( "campaign,access_rules,attachments,attachments_media,audio,images,media," @@ -168,49 +140,12 @@ class MediaItem: post_id: str -def _load_session(cookies_path: str | Path | None) -> requests.Session: - session = requests.Session() - session.headers.update( - { - "User-Agent": _USER_AGENT, - "Accept": "application/vnd.api+json", - } - ) - if cookies_path and os.path.isfile(str(cookies_path)): - try: - jar = http.cookiejar.MozillaCookieJar(str(cookies_path)) - jar.load(ignore_discard=True, ignore_expires=True) - session.cookies = jar # type: ignore[assignment] - except (OSError, http.cookiejar.LoadError) as exc: - log.warning("Could not load Patreon cookies from %s: %s", cookies_path, exc) - return session - - def _filehash(url: str) -> str | None: # Delegate to the shared extractor (utils.paths) so capture-time persistence # and render-time inline-image matching use the EXACT same identity. return filehash_from_url(url) -def _basename_from_url(url: str) -> str: - """Derive a sane filename from a URL when the media has no file_name. - - Strips query/fragment, takes the path basename, and drops a junk - extension (the importer._safe_ext gotcha) so we never write base64 noise - as a name. Falls back to the filehash, then to "file". - """ - path = urlsplit(url).path - base = os.path.basename(path) - if base: - ext = safe_ext(base) - stem = base[: -len(Path(base).suffix)] if Path(base).suffix else base - # Keep the stem bounded; URL-encoded stems can be enormous. - stem = stem[:120] or "file" - return f"{stem}{ext}" - fh = _filehash(url) - return fh or "file" - - def parse_cursor_from_url(url: str | None) -> str | None: """Extract the `page[cursor]` query param from a links.next URL.""" if not url: @@ -238,7 +173,7 @@ class PatreonClient: max_retries: int = _MAX_429_RETRIES, ): self.cookies_path = str(cookies_path) if cookies_path else None - self._session = _load_session(cookies_path) + self._session = make_session(cookies_path, accept="application/vnd.api+json") # Politeness: seconds to sleep before each /api/posts page fetch (paces # the rate-limited API endpoint). 0 = no pacing. plan #703. self._request_sleep = request_sleep or 0.0 @@ -283,7 +218,7 @@ class PatreonClient: # through to the terminal RATE_LIMITED raise below. if resp.status_code == 429 and attempt < self._max_retries: attempt += 1 - delay = _retry_after_seconds(resp, attempt) + delay = retry_after_seconds(resp, attempt) log.warning( "Patreon 429 (campaign_id=%s) — backing off %.1fs (retry %d/%d)", campaign_id, delay, attempt, self._max_retries, @@ -416,7 +351,7 @@ class PatreonClient: # uses. A genuine schema change shows up as no URL (above) or a media id # absent from `included` (caller), not a missing name. file_name = attrs.get("file_name") - filename = file_name if isinstance(file_name, str) and file_name else _basename_from_url(url) + filename = file_name if isinstance(file_name, str) and file_name else basename_from_url(url) return MediaItem( url=url, filename=filename, @@ -462,7 +397,7 @@ class PatreonClient: items.append( MediaItem( url=large_url, - filename=_basename_from_url(large_url), + filename=basename_from_url(large_url), kind="image_large", filehash=_filehash(large_url), post_id=post_id, @@ -481,7 +416,7 @@ class PatreonClient: filename = ( pf_name if isinstance(pf_name, str) and pf_name - else _basename_from_url(pf_url) + else basename_from_url(pf_url) ) items.append( MediaItem( @@ -503,7 +438,7 @@ class PatreonClient: items.append( MediaItem( url=src, - filename=_basename_from_url(src), + filename=basename_from_url(src), kind="content", filehash=_filehash(src), post_id=post_id, diff --git a/backend/app/services/patreon_downloader.py b/backend/app/services/patreon_downloader.py index 2502d14..47132e7 100644 --- a/backend/app/services/patreon_downloader.py +++ b/backend/app/services/patreon_downloader.py @@ -27,46 +27,33 @@ FC runs on a plain-HTTP homelab; nothing here uses a secure-context Web API. from __future__ import annotations -import contextlib import json import logging import os import subprocess import time from collections.abc import Callable -from dataclasses import dataclass -from datetime import datetime from pathlib import Path from urllib.parse import urlsplit import requests from ..utils.prosemirror import post_body_html -from .file_validator import is_validatable, quarantine_file, validate_file -from .patreon_client import ( +from .native_ingest_common import ( _BACKOFF_CAP_SECONDS, - _load_session, - _retry_after_seconds, + _MAX_MEDIA_RETRIES, + BaseNativeDownloader, + MediaOutcome, + PostRecordOutcome, + post_dir_name, + sanitize_segment, ) log = logging.getLogger(__name__) -_TITLE_MAX = 40 +# yt-dlp subprocess wall-clock per attempt (video only; the shared HTTP fetch +# budgets live in BaseNativeDownloader). _TIMEOUT_SECONDS = 120.0 -_CHUNK = 1 << 16 -# Retry a media GET that hits a TRANSIENT failure within the same pass (plan -# #705 #8): a transport blip (connection reset / timeout / truncated stream), a -# 429, or a 5xx. PERMANENT failures (404 gone, 403 forbidden) fail fast straight -# to the error/dead-letter path — no point re-fetching them. Keeps a momentary -# network hiccup from becoming a per-item error that waits for the next walk. -_MAX_MEDIA_RETRIES = 3 -# requests transport errors worth retrying (vs. an HTTPError, which is a real -# server response and is classified by status code). -_TRANSIENT_TRANSPORT_EXC = ( - requests.ConnectionError, - requests.Timeout, - requests.exceptions.ChunkedEncodingError, -) # Referer/Origin yt-dlp must send for Mux-hosted Patreon video. Mux's JWT # playback policy checks Referer/Origin on every request, so yt-dlp must send @@ -78,26 +65,6 @@ _VIDEO_HEADERS = { "Origin": "https://www.patreon.com", } -# Characters Windows/gallery-dl path-restrict forbids, plus path separators. -_FORBIDDEN = set('<>:"/\\|?*') - - -def _sanitize(name: str) -> str: - """Make `name` safe for a single filesystem path segment. - - Replaces path separators, the Windows-forbidden set <>:"/\\|?* and control - characters with `_`, then strips trailing dots/spaces (gallery-dl - path-restrict behavior). Never returns empty (falls back to "_"). - """ - out = [] - for ch in name: - if ch in _FORBIDDEN or ord(ch) < 32: - out.append("_") - else: - out.append(ch) - cleaned = "".join(out).rstrip(". ") - return cleaned or "_" - def _is_video_url(url: str) -> bool: parts = urlsplit(url) @@ -106,73 +73,12 @@ def _is_video_url(url: str) -> bool: return parts.path.lower().endswith(".m3u8") -def _post_dir_name(post: dict) -> str: - """Build the post directory name matching gallery-dl's layout.""" - post_id = str(post.get("id") or "") - attrs = post.get("attributes") or {} - title = attrs.get("title") - title = title if isinstance(title, str) else "" - title40 = title[:_TITLE_MAX] - - published = attrs.get("published_at") - date_prefix = None - if isinstance(published, str) and published: - s = published.strip() - if s.endswith("Z"): - s = s[:-1] + "+00:00" - try: - dt = datetime.fromisoformat(s) - except ValueError: - dt = None - if dt is not None: - date_prefix = f"{dt:%Y-%m-%d}" - - if date_prefix: - raw = f"{date_prefix}_{post_id}_{title40}" - else: - raw = f"{post_id}_{title40}" - return _sanitize(raw) - - -@dataclass -class MediaOutcome: - """Per-media result of a download_post pass. - - status is one of: "downloaded", "skipped_seen", "skipped_disk", - "quarantined", "error". `path` is the final on-disk path for "downloaded" - (the actual yt-dlp output for video), the path that already existed for - "skipped_disk", or the _quarantine destination for "quarantined"; None for - "skipped_seen" and (usually) "error". `error` carries the failure/validation - reason for "error"/"quarantined", else None. - """ - - media: object # MediaItem (avoid importing the name for a bare annotation) - status: str - path: Path | None - error: str | None - - -@dataclass -class PostRecordOutcome: - """Result of write_post_record — mirrors the download_post → MediaOutcome - contract so the engine reports per-post handling without re-reading the post. - `path` is the _post.json sidecar (None when the post had no id); the rest is - the captured body's shape (post_type + final char count) for the run log. - """ - - path: Path | None - post_type: str | None - title: str | None - body_chars: int - - -class PatreonDownloader: - """Download resolved Patreon media to gallery-dl's on-disk layout. - - PURE: no DB. The HTTP session and the yt-dlp invocation are injectable seams - so tests run without network or a real subprocess: - - pass `session=` to stub `session.get`, or monkeypatch `_fetch_to_file`. - - monkeypatch `_run_ytdlp` to avoid spawning yt-dlp. +class PatreonDownloader(BaseNativeDownloader): + """Download resolved Patreon media to gallery-dl's on-disk layout. Subclasses + BaseNativeDownloader for the shared streaming GET (transient-retry + + Range-resume) and validation/quarantine; adds the Mux/HLS yt-dlp video branch + and the detail-fetch body enrichment. PURE: no DB. `_run_ytdlp` is + monkeypatchable and the HTTP session is the injectable `session=` seam. """ def __init__( @@ -185,23 +91,16 @@ class PatreonDownloader: session: requests.Session | None = None, content_fetcher: Callable[[str], str | None] | None = None, ): - self.images_root = Path(images_root) - self.cookies_path = str(cookies_path) if cookies_path else None - self._validate = validate + super().__init__( + images_root, cookies_path, platform="patreon", + validate=validate, rate_limit=rate_limit, session=session, + ) # Best-effort enrichment seam: (post_id) -> full HTML body, or None. The # feed endpoint often omits `content`; the adapter wires this to # PatreonClient.fetch_post_detail_content so the sidecar captures the # real body (formatting + inline + external links). # None in unit tests / when enrichment isn't wanted. self._content_fetcher = content_fetcher - # Politeness: seconds to sleep before each actual media download (paces - # the CDN; honors ImportSettings.download_rate_limit_seconds, the same - # value gallery-dl used as its between-downloads `sleep`). 0 = no pacing. - # Applied only to real downloads, not to seen/disk skips. plan #703. - self._rate_limit = rate_limit or 0.0 - # Build a cookie-loaded session the same way patreon_client does, so the - # CDN GETs carry the creator's auth. - self.session = session if session is not None else _load_session(cookies_path) # -- public ------------------------------------------------------------ @@ -234,7 +133,7 @@ class PatreonDownloader: would tier-1 skip it — so the engine can backfill source_filehash for inline-image localization. Genuinely-missing seen media is NOT refetched. """ - post_dir = self.images_root / artist_slug / "patreon" / _post_dir_name(post) + post_dir = self.images_root / artist_slug / "patreon" / post_dir_name(post) outcomes: list[MediaOutcome] = [] for i, media in enumerate(media_items, start=1): @@ -279,7 +178,7 @@ class PatreonDownloader: return MediaOutcome(media=media, status="skipped_seen", path=None, error=None) nn = f"{index:02d}" - final_name = _sanitize(f"{nn}_{media.filename}") + final_name = sanitize_segment(f"{nn}_{media.filename}") media_path = post_dir / final_name # tier-2: already on disk. @@ -333,87 +232,9 @@ class PatreonDownloader: self._write_sidecar(post, out_path, source_url=media.url) return MediaOutcome(media=media, status="downloaded", path=out_path, error=None) - # -- download seams ---------------------------------------------------- - - def _fetch_get(self, url: str, dest: Path) -> Path: - """Stream `url` to a .part file then atomic-rename to `dest`. - - Thin wrapper over `_fetch_to_file` so tests can stub either the whole - GET path (`_fetch_to_file`) or just `session.get`. - """ - part = dest.with_name(dest.name + ".part") - try: - self._fetch_to_file(url, part) - except Exception: - with contextlib.suppress(OSError): - part.unlink() - raise - os.replace(part, dest) - return dest - - def _fetch_to_file(self, url: str, dest: Path) -> None: - """Stream a non-video URL to `dest` via the (stubbable) session, retrying - TRANSIENT failures within the same pass (plan #705 #8) and RESUMING from - the bytes already on disk via a Range request when a retry follows a - mid-download cut (plan #708 B5). - - Retried (backoff): transport blips (connection reset / timeout / - truncated stream — incl. mid-download), HTTP 429 (honoring Retry-After), - and 5xx. Failed fast (no retry → HTTPError → per-item error → dead-letter - path): 4xx other than 429 (404 gone, 403 forbidden) — re-fetching a - permanent failure is pointless. - - Resume: on a retry, if bytes already landed in `dest`, ask for the rest - with `Range: bytes=-`. A 206 means the server honored it → append; a - 200 means it ignored it (served the whole file) → start clean. The caller - (_fetch_get) stages into a `.part`, so a non-range server never corrupts - the output — the worst case is re-downloading from zero, as before. - """ - attempt = 0 - while True: - have = dest.stat().st_size if dest.exists() else 0 - headers = {"Range": f"bytes={have}-"} if have > 0 else None - try: - resp = self.session.get( - url, stream=True, timeout=_TIMEOUT_SECONDS, headers=headers, - ) - if (resp.status_code == 429 or resp.status_code >= 500) \ - and attempt < _MAX_MEDIA_RETRIES: - attempt += 1 - delay = _retry_after_seconds(resp, attempt) - log.warning( - "Patreon media transient HTTP %d (%s) — backing off " - "%.1fs (retry %d/%d)", - resp.status_code, url, delay, attempt, _MAX_MEDIA_RETRIES, - ) - time.sleep(delay) - continue - # A Range that starts at/past EOF (we already have the whole file) - # comes back 416 — the bytes we kept ARE the file. - if have > 0 and resp.status_code == 416: - return - # 2xx → ok; 4xx-non-429 (or an exhausted 429/5xx) → HTTPError - # (permanent for this pass) → not caught below → per-item error. - resp.raise_for_status() - # 206 → server honored the Range; append after the kept bytes. - # Anything else (200) → it served the whole file → start clean. - mode = "ab" if (have > 0 and resp.status_code == 206) else "wb" - with open(dest, mode) as fh: - for chunk in resp.iter_content(chunk_size=_CHUNK): - if chunk: - fh.write(chunk) - return - except _TRANSIENT_TRANSPORT_EXC as exc: - if attempt >= _MAX_MEDIA_RETRIES: - raise # exhausted → terminal error outcome - attempt += 1 - delay = min(2.0 * (2 ** (attempt - 1)), _BACKOFF_CAP_SECONDS) - log.warning( - "Patreon media transport error (%s) — backing off %.1fs " - "(retry %d/%d): %s", - url, delay, attempt, _MAX_MEDIA_RETRIES, exc, - ) - time.sleep(delay) + # -- video (Mux/HLS via yt-dlp) ---------------------------------------- + # The plain-GET streaming path (_fetch_get / _fetch_to_file) and + # _validate_path are inherited from BaseNativeDownloader. def _run_ytdlp(self, url: str, dest: Path, headers: dict) -> Path | None: """Invoke yt-dlp to fetch a Mux/HLS stream to (around) `dest`. @@ -495,35 +316,6 @@ class PatreonDownloader: return cand return None - # -- validation -------------------------------------------------------- - - def _validate_path( - self, path: Path, artist_slug: str, source_url: str | None = None - ) -> tuple[str | None, Path | None]: - """Validate a freshly-written file; quarantine if bad. - - Uses the shared `file_validator.quarantine_file` — same move + provenance - sidecar gallery-dl writes (the native path used to skip the sidecar; that - parity gap is closed here). Returns `(reason, quarantine_dest)` when - quarantined (dest is the original path if the move itself failed), else - `(None, None)` (ok / not validatable / disabled). plan #704: the dest is - surfaced so the run reports a real quarantined-paths list. - """ - if not self._validate or not is_validatable(path): - return None, None - try: - result = validate_file(path) - except Exception as exc: - log.warning("Validator raised on %s: %s", path, exc) - return None, None - if result.ok: - return None, None - dest = quarantine_file( - self.images_root, path, artist_slug, "patreon", - url=source_url, result=result, - ) - return (result.reason or "validation failed"), (dest or path) - # -- sidecar ----------------------------------------------------------- def _write_sidecar( @@ -614,7 +406,7 @@ class PatreonDownloader: return PostRecordOutcome( path=None, post_type=post_type, title=title, body_chars=0, ) - post_dir = self.images_root / artist_slug / "patreon" / _post_dir_name(post) + post_dir = self.images_root / artist_slug / "patreon" / post_dir_name(post) post_dir.mkdir(parents=True, exist_ok=True) path = self._write_sidecar_data(post, post_dir / "_post.json") # _write_sidecar_data has by now memoized any detail-fetched body onto diff --git a/backend/app/services/subscribestar_client.py b/backend/app/services/subscribestar_client.py index c4a6031..2565089 100644 --- a/backend/app/services/subscribestar_client.py +++ b/backend/app/services/subscribestar_client.py @@ -27,10 +27,8 @@ plain-HTTP homelab; nothing here uses a secure-context Web API. from __future__ import annotations -import http.cookiejar import json import logging -import os import re import time from collections.abc import Iterator @@ -42,16 +40,20 @@ from urllib.parse import urljoin, urlsplit import requests -from ..utils.paths import filehash_from_url, safe_ext -from .patreon_client import _MAX_429_RETRIES, _retry_after_seconds +from ..utils.paths import filehash_from_url +from .native_ingest_common import ( + _MAX_429_RETRIES, + basename_from_url, + make_session, + retry_after_seconds, +) log = logging.getLogger(__name__) -_USER_AGENT = ( - "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " - "(KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36" -) _TIMEOUT_SECONDS = 30.0 +# Accept header for the HTML feed + JSON "load more" + the XHR marker. +_ACCEPT = "text/html,application/xhtml+xml,application/json,*/*" +_XHR_HEADERS = {"X-Requested-With": "XMLHttpRequest"} # A post block opens with this wrapper; we slice the page between consecutive # occurrences (regex can't match the balanced close of nested divs, so chunk- @@ -122,26 +124,6 @@ class MediaItem: media_id: str -def _load_session(cookies_path: str | Path | None) -> requests.Session: - session = requests.Session() - session.headers.update( - { - "User-Agent": _USER_AGENT, - # HTML feed + JSON "load more"; let the server pick. - "Accept": "text/html,application/xhtml+xml,application/json,*/*", - "X-Requested-With": "XMLHttpRequest", - } - ) - if cookies_path and os.path.isfile(str(cookies_path)): - try: - jar = http.cookiejar.MozillaCookieJar(str(cookies_path)) - jar.load(ignore_discard=True, ignore_expires=True) - session.cookies = jar # type: ignore[assignment] - except (OSError, http.cookiejar.LoadError) as exc: - log.warning("Could not load SubscribeStar cookies from %s: %s", cookies_path, exc) - return session - - def _split_creator_url(campaign_id: str) -> tuple[str, str]: """`campaign_id` is the creator URL → (base, slug). @@ -167,17 +149,6 @@ def _parse_ss_datetime(text: str) -> str | None: return None -def _basename_from_url(url: str) -> str: - path = urlsplit(url).path - base = os.path.basename(path) - if base: - ext = safe_ext(base) - stem = base[: -len(Path(base).suffix)] if Path(base).suffix else base - stem = stem[:120] or "file" - return f"{stem}{ext}" - return "file" - - class SubscribeStarClient: """Synchronous SubscribeStar HTML-scrape read client. Construct with a path to a Netscape cookies.txt (the same file CredentialService.get_cookies_path @@ -191,7 +162,9 @@ class SubscribeStarClient: max_retries: int = _MAX_429_RETRIES, ): self.cookies_path = str(cookies_path) if cookies_path else None - self._session = _load_session(cookies_path) + self._session = make_session( + cookies_path, accept=_ACCEPT, extra_headers=_XHR_HEADERS + ) self._request_sleep = request_sleep or 0.0 self._max_retries = max_retries @@ -210,7 +183,7 @@ class SubscribeStarClient: ) from exc if resp.status_code == 429 and attempt < self._max_retries: attempt += 1 - delay = _retry_after_seconds(resp, attempt) + delay = retry_after_seconds(resp, attempt) log.warning( "SubscribeStar 429 (%s) — backing off %.1fs (retry %d/%d)", url, delay, attempt, self._max_retries, @@ -341,7 +314,7 @@ class SubscribeStarClient: url = urljoin(base + "/", rel) media_id = str(it.get("id") or "") name = it.get("original_filename") - filename = name if isinstance(name, str) and name else _basename_from_url(url) + filename = name if isinstance(name, str) and name else basename_from_url(url) items.append( MediaItem( url=url, diff --git a/backend/app/services/subscribestar_downloader.py b/backend/app/services/subscribestar_downloader.py index 94982e0..9845a3b 100644 --- a/backend/app/services/subscribestar_downloader.py +++ b/backend/app/services/subscribestar_downloader.py @@ -22,62 +22,31 @@ FC runs on a plain-HTTP homelab; nothing here uses a secure-context Web API. from __future__ import annotations -import contextlib import json import logging -import os import time from collections.abc import Callable -from datetime import datetime from pathlib import Path import requests -from .file_validator import is_validatable, quarantine_file, validate_file -from .patreon_downloader import ( +from .native_ingest_common import ( + BaseNativeDownloader, MediaOutcome, PostRecordOutcome, - _sanitize, + post_dir_name, + sanitize_segment, ) -from .subscribestar_client import _MAX_429_RETRIES, _load_session, _retry_after_seconds log = logging.getLogger(__name__) -_TITLE_MAX = 40 -_TIMEOUT_SECONDS = 120.0 -_CHUNK = 1 << 16 -_MAX_MEDIA_RETRIES = 3 -_BACKOFF_CAP_SECONDS = 30.0 -_TRANSIENT_TRANSPORT_EXC = ( - requests.ConnectionError, - requests.Timeout, - requests.exceptions.ChunkedEncodingError, -) - -def _post_dir_name(post: dict) -> str: - """`__` matching gallery-dl's subscribestar - layout (title is empty for SubscribeStar → trailing underscore). Date prefix - omitted when published_at is missing/unparseable.""" - post_id = str(post.get("id") or "") - attrs = post.get("attributes") or {} - title = attrs.get("title") - title40 = (title if isinstance(title, str) else "")[:_TITLE_MAX] - published = attrs.get("published_at") - date_prefix = None - if isinstance(published, str) and published: - try: - date_prefix = f"{datetime.fromisoformat(published):%Y-%m-%d}" - except ValueError: - date_prefix = None - raw = f"{date_prefix}_{post_id}_{title40}" if date_prefix else f"{post_id}_{title40}" - return _sanitize(raw) - - -class SubscribeStarDownloader: - """Download resolved SubscribeStar media to gallery-dl's on-disk layout. PURE: - no DB. The HTTP session is an injectable seam (pass `session=` or monkeypatch - `_fetch_to_file`) so tests run without network.""" +class SubscribeStarDownloader(BaseNativeDownloader): + """Download resolved SubscribeStar media to gallery-dl's on-disk layout. + Subclasses BaseNativeDownloader for the shared streaming GET (transient-retry + + Range-resume) and validation/quarantine. No video branch (SubscribeStar serves + files directly via /post_uploads) and no detail-fetch (the body is already in + the feed HTML). PURE: no DB.""" def __init__( self, @@ -87,14 +56,11 @@ class SubscribeStarDownloader: validate: bool = True, rate_limit: float = 0.0, session: requests.Session | None = None, - max_retries: int = _MAX_429_RETRIES, ): - self.images_root = Path(images_root) - self.cookies_path = str(cookies_path) if cookies_path else None - self._validate = validate - self._rate_limit = rate_limit or 0.0 - self._max_retries = max_retries - self.session = session if session is not None else _load_session(cookies_path) + super().__init__( + images_root, cookies_path, platform="subscribestar", + validate=validate, rate_limit=rate_limit, session=session, + ) # -- public ------------------------------------------------------------ @@ -111,7 +77,7 @@ class SubscribeStarDownloader: """Download every media item of one post; return per-item outcomes. Mirrors PatreonDownloader.download_post (two-tier skip, mid-post time-box, recapture surfacing) minus the video branch.""" - post_dir = self.images_root / artist_slug / "subscribestar" / _post_dir_name(post) + post_dir = self.images_root / artist_slug / "subscribestar" / post_dir_name(post) outcomes: list[MediaOutcome] = [] for i, media in enumerate(media_items, start=1): if should_stop(): @@ -151,7 +117,7 @@ class SubscribeStarDownloader: return MediaOutcome(media=media, status="skipped_seen", path=None, error=None) nn = f"{index:02d}" - media_path = post_dir / _sanitize(f"{nn}_{media.filename}") + media_path = post_dir / sanitize_segment(f"{nn}_{media.filename}") if media_path.exists(): # tier-2: already on disk return MediaOutcome( @@ -174,85 +140,8 @@ class SubscribeStarDownloader: self._write_sidecar(post, out_path, source_url=media.url) return MediaOutcome(media=media, status="downloaded", path=out_path, error=None) - # -- download seams ---------------------------------------------------- - - def _fetch_get(self, url: str, dest: Path) -> Path: - """Stream `url` to a .part file then atomic-rename to `dest`.""" - part = dest.with_name(dest.name + ".part") - try: - self._fetch_to_file(url, part) - except Exception: - with contextlib.suppress(OSError): - part.unlink() - raise - os.replace(part, dest) - return dest - - def _fetch_to_file(self, url: str, dest: Path) -> None: - """Stream a URL to `dest`, retrying TRANSIENT failures (transport blips, - 429, 5xx) with backoff + resume-from-disk (Range), failing fast on - permanent 4xx. Mirrors PatreonDownloader._fetch_to_file.""" - attempt = 0 - while True: - have = dest.stat().st_size if dest.exists() else 0 - headers = {"Range": f"bytes={have}-"} if have > 0 else None - try: - resp = self.session.get( - url, stream=True, timeout=_TIMEOUT_SECONDS, headers=headers, - ) - if (resp.status_code == 429 or resp.status_code >= 500) \ - and attempt < _MAX_MEDIA_RETRIES: - attempt += 1 - delay = _retry_after_seconds(resp, attempt) - log.warning( - "SubscribeStar media transient HTTP %d (%s) — backing off " - "%.1fs (retry %d/%d)", - resp.status_code, url, delay, attempt, _MAX_MEDIA_RETRIES, - ) - time.sleep(delay) - continue - if have > 0 and resp.status_code == 416: - return # Range past EOF — we already have the whole file - resp.raise_for_status() - mode = "ab" if (have > 0 and resp.status_code == 206) else "wb" - with open(dest, mode) as fh: - for chunk in resp.iter_content(chunk_size=_CHUNK): - if chunk: - fh.write(chunk) - return - except _TRANSIENT_TRANSPORT_EXC as exc: - if attempt >= _MAX_MEDIA_RETRIES: - raise - attempt += 1 - delay = min(2.0 * (2 ** (attempt - 1)), _BACKOFF_CAP_SECONDS) - log.warning( - "SubscribeStar media transport error (%s) — backing off %.1fs " - "(retry %d/%d): %s", - url, delay, attempt, _MAX_MEDIA_RETRIES, exc, - ) - time.sleep(delay) - - # -- validation -------------------------------------------------------- - - def _validate_path( - self, path: Path, artist_slug: str, source_url: str | None = None - ) -> tuple[str | None, Path | None]: - """Validate a freshly-written file; quarantine if bad. Mirrors the Patreon - path (shared file_validator).""" - if not self._validate or not is_validatable(path): - return None, None - try: - result = validate_file(path) - except Exception as exc: - log.warning("Validator raised on %s: %s", path, exc) - return None, None - if result.ok: - return None, None - dest = quarantine_file( - self.images_root, path, artist_slug, "subscribestar", - url=source_url, result=result, - ) - return (result.reason or "validation failed"), (dest or path) + # The plain-GET streaming path (_fetch_get / _fetch_to_file) and + # _validate_path are inherited from BaseNativeDownloader. # -- sidecar ----------------------------------------------------------- @@ -307,7 +196,7 @@ class SubscribeStarDownloader: return PostRecordOutcome( path=None, post_type=post_type, title=title, body_chars=0, ) - post_dir = self.images_root / artist_slug / "subscribestar" / _post_dir_name(post) + post_dir = self.images_root / artist_slug / "subscribestar" / post_dir_name(post) post_dir.mkdir(parents=True, exist_ok=True) path = self._write_sidecar_data(post, post_dir / "_post.json") body = attrs.get("content") diff --git a/tests/test_patreon_client.py b/tests/test_patreon_client.py index 2106237..2f4a990 100644 --- a/tests/test_patreon_client.py +++ b/tests/test_patreon_client.py @@ -12,13 +12,13 @@ import pytest import requests from backend.app.services import patreon_client as pc_mod +from backend.app.services.native_ingest_common import retry_after_seconds from backend.app.services.patreon_client import ( MediaItem, PatreonAPIError, PatreonAuthError, PatreonClient, PatreonDriftError, - _retry_after_seconds, parse_cursor_from_url, ) @@ -288,19 +288,19 @@ def test_verify_auth_inconclusive_on_network(monkeypatch): def test_retry_after_seconds_honors_numeric_header(): - assert _retry_after_seconds(_FakeResp(429, headers={"Retry-After": "7"}), 1) == 7.0 + assert retry_after_seconds(_FakeResp(429, headers={"Retry-After": "7"}), 1) == 7.0 def test_retry_after_seconds_exponential_without_header(): resp = _FakeResp(429) - assert _retry_after_seconds(resp, 1) == 2.0 - assert _retry_after_seconds(resp, 2) == 4.0 - assert _retry_after_seconds(resp, 3) == 8.0 + assert retry_after_seconds(resp, 1) == 2.0 + assert retry_after_seconds(resp, 2) == 4.0 + assert retry_after_seconds(resp, 3) == 8.0 def test_retry_after_seconds_caps(): - assert _retry_after_seconds(_FakeResp(429), 10) == 30.0 - assert _retry_after_seconds(_FakeResp(429, headers={"Retry-After": "999"}), 1) == 30.0 + assert retry_after_seconds(_FakeResp(429), 10) == 30.0 + assert retry_after_seconds(_FakeResp(429, headers={"Retry-After": "999"}), 1) == 30.0 def _client_with_sequence(monkeypatch, responses): diff --git a/tests/test_patreon_downloader.py b/tests/test_patreon_downloader.py index 5162364..a26ffb5 100644 --- a/tests/test_patreon_downloader.py +++ b/tests/test_patreon_downloader.py @@ -14,13 +14,14 @@ import pytest import requests from backend.app.services import patreon_downloader as pd_mod -from backend.app.services.patreon_client import MediaItem -from backend.app.services.patreon_downloader import ( +from backend.app.services.native_ingest_common import ( MediaOutcome, - PatreonDownloader, PostRecordOutcome, - _sanitize, + post_dir_name, + sanitize_segment, ) +from backend.app.services.patreon_client import MediaItem +from backend.app.services.patreon_downloader import PatreonDownloader from backend.app.utils.sidecar import find_sidecar # Minimal valid PNG so the file_validator passes on the happy path. @@ -414,10 +415,10 @@ def test_validation_off_keeps_bytes(tmp_path): def test_sanitize_helper(): - assert _sanitize('a/b') == "a_b" - assert _sanitize('x<>:"|?*y') == "x_______y" - assert _sanitize("trail. ") == "trail" - assert _sanitize("") == "_" + assert sanitize_segment('a/b') == "a_b" + assert sanitize_segment('x<>:"|?*y') == "x_______y" + assert sanitize_segment("trail. ") == "trail" + assert sanitize_segment("") == "_" def test_media_outcome_shape(): @@ -613,7 +614,7 @@ def test_media_sidecar_is_minimal_post_first(tmp_path): post["attributes"]["content"] = "" # even an empty feed body isn't fetched here dl.download_post(post, [_img("a.png"), _img("b.png")], "artist-x") - post_dir = tmp_path / "artist-x" / "patreon" / pd_mod._post_dir_name(post) + post_dir = tmp_path / "artist-x" / "patreon" / post_dir_name(post) sc = find_sidecar(post_dir / "01_a.png") assert sc is not None data = json.loads(sc.read_text()) diff --git a/tests/test_subscribestar_native.py b/tests/test_subscribestar_native.py index 5bc0e24..c84cfd7 100644 --- a/tests/test_subscribestar_native.py +++ b/tests/test_subscribestar_native.py @@ -14,6 +14,7 @@ from pathlib import Path import requests from backend.app.services.gallery_dl import DownloadResult, ErrorType +from backend.app.services.native_ingest_common import post_dir_name from backend.app.services.subscribestar_client import ( MediaItem, SubscribeStarAPIError, @@ -22,10 +23,7 @@ from backend.app.services.subscribestar_client import ( SubscribeStarDriftError, _parse_ss_datetime, ) -from backend.app.services.subscribestar_downloader import ( - SubscribeStarDownloader, - _post_dir_name, -) +from backend.app.services.subscribestar_downloader import SubscribeStarDownloader from backend.app.services.subscribestar_ingester import ( SubscribeStarIngester, _ledger_key, @@ -210,7 +208,7 @@ def _downloader(tmp_path: Path, session=None, validate=True): def test_post_dir_name_empty_title_matches_gallery_dl(): # SubscribeStar has no title → "__" (matches gallery-dl's layout so # existing downloads dedup on disk at cutover). - assert _post_dir_name(_post("111")) == "2026-05-01_111_" + assert post_dir_name(_post("111")) == "2026-05-01_111_" def test_download_post_layout_and_outcomes(tmp_path):