feat(external): zip-parity provenance/tagging + thorough worker logging

Operator-requested: a worker download must be tagged + provenance-associated
exactly like an extracted zip, and the path must log well (we won't get it right
first try).

- _route_files now mirrors download_service._phase3_persist branch-for-branch:
  imported/superseded → collect member_image_ids+image_id (provenance-linked via
  the synthesized sidecar, same as extracted-zip members) → caller enqueues
  tag_and_embed + generate_thumbnail; attached → drop on-disk original, and warn
  on an UNEXTRACTED archive (#718 symptom); skipped duplicate → unlink; failed →
  unlink + warn.
- Logging at every stage: start (link/host/post/artist/attempt/url), requeue,
  fetch result (files/bytes) or fetch failure, per-file import decision, dead-
  letter transitions, and done (files/images/duration).
- Parity test: an archive downloaded by the worker is extracted, provenance-
  linked to the SAME post, and tag_and_embed+generate_thumbnail are queued for
  exactly the member images.

Refs FC #830.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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

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