"""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('<>:"/\\|?*') # -- shared exception taxonomy -------------------------------------------- # Every native client raises one of these (platform subclasses keep an # isinstance-distinct platform name AND the semantic Auth/Drift class), so the # base Ingester._failure_result can map them platform-agnostically. class NativeIngestError(Exception): """Base for a native-ingest client failure. `status_code` carries the HTTP status when the failure was an HTTP response (None for transport/parse); `retry_after` carries the server's 429 Retry-After hint so the cooldown can match it (plan #708 B1).""" def __init__( self, message: str, *, status_code: int | None = None, retry_after: float | None = None, ): super().__init__(message) self.status_code = status_code self.retry_after = retry_after class NativeAuthError(NativeIngestError): """Authentication/authorization failure — expired/missing credential or an insufficient tier. The fix is rotating the credential, NOT updating the ingester. Maps to error_type 'auth_error'.""" class NativeDriftError(NativeIngestError): """A response did not match the shape the ingester depends on (JSON:API field set, or scraped HTML structure). Fail loud so the import step flags 'the platform changed' instead of silently importing nothing. Maps to API_DRIFT.""" # -- 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 # -- membership roster seam (shared dataclass, #387 C2/C7) ----------------- @dataclass class Membership: """One membership the ACCOUNT holds, as the roster needs it (#387 C2). Lives HERE rather than in the platform module that first produced it, for the same reason `PostRecordOutcome` does: it is the seam's contract, not Patreon's. C7 moved it — while it sat in `patreon_client` a second platform would have had to import its contract from the first platform's module, which inverts the dependency and is how a "portable" seam quietly becomes Patreon-shaped. Deliberately not a raw upstream row: the sweep should not have to know that a tier lives behind a JSON:API `reward` relationship, and `platform_membership` should not gain columns because one platform shapes things a certain way. `status` carries the PLATFORM's own word, verbatim and unmapped (`active_patron`, `former_patron`, ...). Deciding what it means is the read site's job — `membership_roster.has_paid_access` — precisely so an unrecognised word records as evidence rather than as a decision. `is_free_member` is SEPARATE from status and must stay that way. Patreon expresses a free follow as this boolean rather than as a status value, so "does the account pay for this" is `status == "active_patron" and not is_free_member` — a question the status string alone cannot answer. NOTE: the C0 capture contains no ACTIVE free member, so the two fields are perfectly correlated in that sample; the separation is what the schema says, not something the sample proves. A platform that lacks a field supplies the empty answer, never a guess: no tiers -> `[]`, no pledge -> `amount_cents=None` (absent stays distinguishable from zero — "free" and "we don't know" are different answers), no vanity -> None and identity falls back to the URL tail. """ campaign_id: str display_name: str | None url: str | None vanity: str | None status: str | None is_free_member: bool tier_names: list[str] amount_cents: int | None currency: str | None # Everything the roster did not model, kept so a later question can be # answered without another authenticated round-trip. Scoped to the # membership's own attributes plus the creator's — never the raw page, # which is where the card/address resources live. details: dict # -- 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