refactor(native-ingest): extract native_ingest_common + BaseNativeDownloader (#899 DRY 1/3)
DRY pass commit 1 (process #594). Consolidate the helpers + download plumbing the Patreon and SubscribeStar adapters had duplicated (SubscribeStar was importing patreon privates — wrong owner). New backend/app/services/ native_ingest_common.py is the neutral home for: - make_session (was _load_session ×2), retry_after_seconds + 429 constants, sanitize_segment, basename_from_url, post_dir_name, MediaOutcome / PostRecordOutcome. - BaseNativeDownloader: the shared streaming GET (transient-retry + Range-resume) and validation/quarantine. Patreon + SubscribeStar downloaders now subclass it; each keeps only what differs (Patreon's Mux/yt-dlp video branch + detail-fetch enrichment; SubscribeStar nothing extra). Behavior preserved exactly; the divergence-bug risk (a fix to one _fetch_to_file not reaching the other) is gone. - Folds in #899 L2: a quarantine now log.warning's path+reason (was counted only). post_dir_name merges both date handlers (accepts trailing-Z and pre-parsed ISO). Tests repointed to the single source at every consumer (rule 93 / §8b parity): patreon_client/downloader, subscribestar_native. Exception-trio consolidation + base _failure_result (2/3) and the remaining ingest_core logging (3/3) follow. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -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 <img> + external <a href> 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=<have>-`. 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
|
||||
|
||||
Reference in New Issue
Block a user