Native Patreon ingester + download-engine ownership (plans #697, #703–#708) #72

Merged
bvandeusen merged 26 commits from dev into main 2026-06-06 12:57:31 -04:00
2 changed files with 99 additions and 5 deletions
Showing only changes of commit 402086c34c - Show all commits
+22 -3
View File
@@ -301,18 +301,30 @@ class PatreonDownloader:
def _fetch_to_file(self, url: str, dest: Path) -> None: def _fetch_to_file(self, url: str, dest: Path) -> None:
"""Stream a non-video URL to `dest` via the (stubbable) session, retrying """Stream a non-video URL to `dest` via the (stubbable) session, retrying
TRANSIENT failures within the same pass (plan #705 #8). 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 / Retried (backoff): transport blips (connection reset / timeout /
truncated stream — incl. mid-download), HTTP 429 (honoring Retry-After), truncated stream — incl. mid-download), HTTP 429 (honoring Retry-After),
and 5xx. Failed fast (no retry → HTTPError → per-item error → dead-letter and 5xx. Failed fast (no retry → HTTPError → per-item error → dead-letter
path): 4xx other than 429 (404 gone, 403 forbidden) — re-fetching a path): 4xx other than 429 (404 gone, 403 forbidden) — re-fetching a
permanent failure is pointless. 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 attempt = 0
while True: while True:
have = dest.stat().st_size if dest.exists() else 0
headers = {"Range": f"bytes={have}-"} if have > 0 else None
try: try:
resp = self.session.get(url, stream=True, timeout=_TIMEOUT_SECONDS) resp = self.session.get(
url, stream=True, timeout=_TIMEOUT_SECONDS, headers=headers,
)
if (resp.status_code == 429 or resp.status_code >= 500) \ if (resp.status_code == 429 or resp.status_code >= 500) \
and attempt < _MAX_MEDIA_RETRIES: and attempt < _MAX_MEDIA_RETRIES:
attempt += 1 attempt += 1
@@ -324,10 +336,17 @@ class PatreonDownloader:
) )
time.sleep(delay) time.sleep(delay)
continue 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 # 2xx → ok; 4xx-non-429 (or an exhausted 429/5xx) → HTTPError
# (permanent for this pass) → not caught below → per-item error. # (permanent for this pass) → not caught below → per-item error.
resp.raise_for_status() resp.raise_for_status()
with open(dest, "wb") as fh: # 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): for chunk in resp.iter_content(chunk_size=_CHUNK):
if chunk: if chunk:
fh.write(chunk) fh.write(chunk)
+77 -2
View File
@@ -50,7 +50,7 @@ class _FakeSession:
self.payloads = payloads or {} self.payloads = payloads or {}
self.calls: list[str] = [] self.calls: list[str] = []
def get(self, url, stream=False, timeout=None): def get(self, url, stream=False, timeout=None, headers=None):
self.calls.append(url) self.calls.append(url)
return _FakeResponse(self.payloads.get(url, _PNG_BYTES)) return _FakeResponse(self.payloads.get(url, _PNG_BYTES))
@@ -63,7 +63,7 @@ class _SeqSession:
self._responses = list(responses) self._responses = list(responses)
self.calls: list[str] = [] self.calls: list[str] = []
def get(self, url, stream=False, timeout=None): def get(self, url, stream=False, timeout=None, headers=None):
self.calls.append(url) self.calls.append(url)
item = self._responses.pop(0) item = self._responses.pop(0)
if isinstance(item, Exception): if isinstance(item, Exception):
@@ -71,6 +71,26 @@ class _SeqSession:
return item return item
class _FlakyResp:
"""A streaming response that yields its payload then optionally raises a
mid-stream transport error (for the Range-resume tests, plan #708 B5)."""
headers: dict = {}
def __init__(self, payload, status_code=200, die=False):
self._payload = payload
self.status_code = status_code
self._die = die
def raise_for_status(self):
return None
def iter_content(self, chunk_size=65536):
yield self._payload
if self._die:
raise requests.exceptions.ChunkedEncodingError("cut mid-stream")
def _post(post_id="1001", title="My Post Title", published="2026-05-01T12:00:00Z"): def _post(post_id="1001", title="My Post Title", published="2026-05-01T12:00:00Z"):
return { return {
"id": post_id, "id": post_id,
@@ -442,3 +462,58 @@ def test_exhausted_transient_becomes_error(tmp_path, monkeypatch):
dl = _dl(tmp_path, session, monkeypatch) dl = _dl(tmp_path, session, monkeypatch)
outcomes = dl.download_post(_post(), [_img("a.png")], "artist-x") outcomes = dl.download_post(_post(), [_img("a.png")], "artist-x")
assert outcomes[0].status == "error" assert outcomes[0].status == "error"
def test_partial_download_resumes_via_range(tmp_path, monkeypatch):
"""plan #708 B5: a mid-download transport cut resumes from the bytes already
on disk via a Range request, not a refetch from zero."""
full = bytes(i % 256 for i in range(1000))
split = 400
class _ResumeSession:
def __init__(self):
self.headers_seen = []
def get(self, url, stream=False, timeout=None, headers=None):
self.headers_seen.append(headers)
if len(self.headers_seen) == 1:
return _FlakyResp(full[:split], die=True) # partial then cut
start = int(headers["Range"].split("=")[1].rstrip("-"))
return _FlakyResp(full[start:], status_code=206)
monkeypatch.setattr(pd_mod.time, "sleep", lambda *_a: None)
session = _ResumeSession()
dl = PatreonDownloader(
images_root=tmp_path, cookies_path=None, validate=False, session=session,
)
outcomes = dl.download_post(_post(), [_img("a.png")], "artist-x")
assert outcomes[0].status == "downloaded"
assert outcomes[0].path.read_bytes() == full
assert session.headers_seen[0] is None # first GET: no Range
assert session.headers_seen[1] == {"Range": "bytes=400-"} # resume from offset
def test_range_ignored_by_server_refetches_clean(tmp_path, monkeypatch):
"""plan #708 B5: if the server ignores Range (200, not 206) on the resume, we
truncate and start clean — never append into a corrupt double-write."""
full = b"ABCD" * 250 # 1000 bytes
split = 400
class _NoRangeSession:
def __init__(self):
self.n = 0
def get(self, url, stream=False, timeout=None, headers=None):
self.n += 1
if self.n == 1:
return _FlakyResp(full[:split], die=True)
return _FlakyResp(full, status_code=200) # ignores Range, whole file
monkeypatch.setattr(pd_mod.time, "sleep", lambda *_a: None)
dl = PatreonDownloader(
images_root=tmp_path, cookies_path=None, validate=False,
session=_NoRangeSession(),
)
outcomes = dl.download_post(_post(), [_img("a.png")], "artist-x")
assert outcomes[0].status == "downloaded"
assert outcomes[0].path.read_bytes() == full # clean, not full[:400] + full