diff --git a/backend/app/tasks/external.py b/backend/app/tasks/external.py index af9ddd4..e5f243d 100644 --- a/backend/app/tasks/external.py +++ b/backend/app/tasks/external.py @@ -124,6 +124,7 @@ def fetch_external_link(self, link_id: int, _serialize_waits: int = 0) -> dict: settings = ImportSettings.load_sync(session) if not _host_enabled(settings, link.host): + log.info("extdl skip (host disabled): link=%s host=%s", link_id, link.host) link.status = "skipped" link.last_error = f"{link.host} disabled" link.completed_at = datetime.now(UTC) @@ -131,6 +132,10 @@ def fetch_external_link(self, link_id: int, _serialize_waits: int = 0) -> dict: return {"link_id": link_id, "skipped": "host disabled"} host, url, attempts = link.host, link.url, link.attempts + log.info( + "extdl start: link=%s host=%s post=%s artist=%s attempt=%d url=%s", + link_id, host, post.id, artist.slug, attempts + 1, url, + ) # Per-host serialize: one fetch per host at a time. If busy, requeue. lock = None @@ -149,11 +154,16 @@ def fetch_external_link(self, link_id: int, _serialize_waits: int = 0) -> dict: .values(status="pending") ) session.commit() + log.info( + "extdl requeue (host %s busy): link=%s wait=%d", host, link_id, _serialize_waits, + ) if _serialize_waits < _MAX_SERIALIZE_WAITS: fetch_external_link.apply_async( (link_id,), {"_serialize_waits": _serialize_waits + 1}, countdown=_SERIALIZE_COUNTDOWN, ) + else: + log.warning("extdl giving up requeue (host %s busy): link=%s", host, link_id) return {"link_id": link_id, "requeued": "host busy"} started = time.monotonic() @@ -163,10 +173,18 @@ def fetch_external_link(self, link_id: int, _serialize_waits: int = 0) -> dict: with SessionLocal() as session: link = session.get(ExternalLink, link_id) if not result.ok: + log.warning( + "extdl fetch failed: link=%s host=%s error=%s", + link_id, host, result.error, + ) _record_failure(link, attempts, result.error or "fetch failed") session.commit() return {"link_id": link_id, "error": result.error} + log.info( + "extdl fetched: link=%s host=%s files=%d bytes=%d", + link_id, host, len(result.files), result.bytes, + ) # Re-load post/artist/source attached to THIS session before handing # them to the importer (the claim block's instances are detached). post = session.get(Post, link.post_id) @@ -178,6 +196,11 @@ def fetch_external_link(self, link_id: int, _serialize_waits: int = 0) -> dict: link.completed_at = datetime.now(UTC) link.duration_seconds = time.monotonic() - started session.commit() + log.info( + "extdl done: link=%s host=%s post=%s files=%d image(s)=%d dur=%.1fs", + link_id, host, post.id, len(result.files), len(image_ids), + link.duration_seconds, + ) # Thumbnails + ML for any newly-attached images (mirrors the download # path). Lazy import to dodge a task-module import cycle. @@ -210,13 +233,23 @@ def _record_failure(link: ExternalLink, attempts: int, error: str) -> None: link.last_error = error[:1000] link.status = "dead" if nxt >= DEAD_LETTER_THRESHOLD else "failed" link.completed_at = datetime.now(UTC) + log.warning( + "extdl link %s -> %s (attempt %d/%d): %s", + link.id, link.status, nxt, DEAD_LETTER_THRESHOLD, error[:200], + ) def _route_files(session, files, post, platform, artist, source) -> list[int]: - """Import each fetched file via the existing pipeline (in the artist tree, so - art stays in place); return new image ids for thumb/ML enqueue. A captured - archive/attachment is copied into its store, so its on-disk original is - removed (mirrors download_service); art images stay.""" + """Import each fetched file through the SAME pipeline extracted-zip members + use — importer.attach_in_place — so a downloaded archive is extracted to + ImageRecords and provenance-linked to the post (via the synthesized sidecar), + a non-art file is captured as a PostAttachment, and an art image is attached + in place. Returns new image ids so the caller enqueues thumbnail + ML + (tagging) for them, exactly as download_service does for downloaded media. + + Result handling mirrors download_service._phase3_persist branch-for-branch + (duplicate cleanup, the unextracted-archive warning of #718) and logs every + decision — this path is new and the operator expects to iterate on it.""" settings = ImportSettings.load_sync(session) importer = Importer( session=session, @@ -235,10 +268,41 @@ def _route_files(session, files, post, platform, artist, source) -> list[int]: if result.image_id is not None and result.image_id not in ids: ids.append(result.image_id) image_ids.extend(ids) - elif result.status in ("attached", "failed"): - # Copied into the attachment store (or failed) — drop the on-disk - # original so we don't keep two copies / an orphan. + log.info( + "extdl import: post=%s file=%s status=%s -> %d image(s) " + "(provenance-linked, queued for tagging)", + post.id, f.name, result.status, len(ids), + ) + elif result.status == "attached": + # Non-art / archive captured as a PostAttachment (copied into the + # store) — drop the on-disk original. An archive captured WITHOUT + # extracting any image carries the reason (the recurring "zip but no + # images" symptom, #718) — surface it loudly. + if result.error: + log.warning( + "extdl archive captured UNEXTRACTED: post=%s file=%s reason=%s", + post.id, f.name, result.error, + ) + else: + log.info("extdl attachment captured: post=%s file=%s", post.id, f.name) f.unlink(missing_ok=True) + elif result.status == "skipped": + reason = result.skip_reason.value if result.skip_reason else None + log.info( + "extdl skipped: post=%s file=%s reason=%s", post.id, f.name, reason, + ) + if reason in ("duplicate_hash", "duplicate_phash"): + f.unlink(missing_ok=True) # canonical copy already in the library + elif result.status == "failed": + log.warning( + "extdl import FAILED: post=%s file=%s error=%s", + post.id, f.name, result.error, + ) + f.unlink(missing_ok=True) + else: # 'refreshed' or any future status — work happened, no new images + log.info( + "extdl import: post=%s file=%s status=%s", post.id, f.name, result.status, + ) # The synthesized sidecar has done its job — don't litter the tree. Path(str(f) + ".json").unlink(missing_ok=True) return image_ids diff --git a/tests/test_external_worker.py b/tests/test_external_worker.py index 0b3bb7b..7bde94e 100644 --- a/tests/test_external_worker.py +++ b/tests/test_external_worker.py @@ -1,15 +1,39 @@ """Integration tests for the external-link download worker (tasks/external).""" +import io +import zipfile + import pytest +from PIL import Image from sqlalchemy import func, select import backend.app.tasks.external as ext -from backend.app.models import Artist, ExternalLink, Post, PostAttachment, Source +from backend.app.models import ( + Artist, + ExternalLink, + ImageProvenance, + Post, + PostAttachment, + Source, +) from backend.app.services.external_fetch import FetchResult pytestmark = pytest.mark.integration +def _structured_jpeg_bytes(): + # A non-uniform image so it survives the import filters (a solid color would + # phash-collapse / fail thresholds). + im = Image.new("RGB", (256, 256)) + px = im.load() + for y in range(256): + for x in range(256): + px[x, y] = (x, y, (x + y) % 256) + buf = io.BytesIO() + im.save(buf, "JPEG") + return buf.getvalue() + + class _FakeLock: def acquire(self, blocking=False): return True @@ -136,3 +160,44 @@ def test_sweep_enqueues_pending_and_retryable(db_sync, monkeypatch): out = ext.sweep_external_links() assert out["enqueued"] == 2 assert set(enqueued) == {l1.id, l2.id} + + +def test_downloaded_archive_gets_provenance_and_tagging(db_sync, tmp_path, monkeypatch): + """An archive downloaded by the worker is extracted, provenance-linked to the + post, AND queued for tagging + thumbnails — exactly like an extracted zip on + the normal download path.""" + post, link = _seed(db_sync, host="mega") + link_id = link.id + + monkeypatch.setattr(ext, "IMAGES_ROOT", tmp_path) + monkeypatch.setattr(ext, "_redis", lambda: _FakeRedis()) + + def fake_fetch(host, url, dest_dir, *, timeout, should_stop=lambda: False): + dest_dir.mkdir(parents=True, exist_ok=True) + zpath = dest_dir / "pack.zip" + with zipfile.ZipFile(zpath, "w") as z: + z.writestr("art.jpg", _structured_jpeg_bytes()) + return FetchResult(files=[zpath], bytes=zpath.stat().st_size) + + monkeypatch.setattr(ext, "fetch_external", fake_fetch) + + tagged, thumbed = [], [] + import backend.app.tasks.ml as ml_mod + import backend.app.tasks.thumbnail as thumb_mod + monkeypatch.setattr(ml_mod.tag_and_embed, "delay", lambda i: tagged.append(i)) + monkeypatch.setattr(thumb_mod.generate_thumbnail, "delay", lambda i: thumbed.append(i)) + + out = ext.fetch_external_link(link_id) + assert out["images"] >= 1 + + db_sync.expire_all() + assert db_sync.get(ExternalLink, link_id).status == "downloaded" + # Extracted member is provenance-linked to the SAME post (as extracted zips). + prov = db_sync.execute( + select(ImageProvenance).where(ImageProvenance.post_id == post.id) + ).scalars().all() + assert len(prov) >= 1 + member_ids = {p.image_record_id for p in prov} + # Tagging + thumbnails queued for exactly those member images. + assert set(tagged) == member_ids + assert set(thumbed) == member_ids