feat(provenance): capture which archive an extracted image came from (#87)

Images pulled out of a .zip/.rar previously kept no record of WHICH archive
they came from — the member->archive link was computed during extraction and
discarded, leaving only image->post. So the provenance modal could only scope
attachments to the whole post, showing every archive a 'High Resolution Files'
bundle carried instead of the one a given file lives in.

- ImageProvenance.from_attachment_id: nullable FK -> post_attachment.id
  (SET NULL), migration 0055.
- importer: _import_archive stamps from_attachment_id on every member's
  provenance row for the post (new + superseded + deduped members), resolving
  the archive's own PostAttachment by (post, sha). Post-pass UPDATE, NULL-only
  and idempotent, so it doesn't touch the dedup/supersede branches and the
  backfill is safe to re-run. Nested members link to the outer stored archive.
- provenance_service.for_image: when the originating post's provenance row
  records from_attachment_id, return ONLY that archive; else fall back to the
  primary-post scoping from 068def2.
- ProvenancePanel: heading pluralizes ('Attachment' for a single file).
- Backfill: re-running reextract_archive_attachments (ArchiveReextractCard)
  routes through _import_archive and stamps existing rows — no new code.

Tests: capture stamps on fresh import, nested-archive attribution, per-post
archive on dedup; for_image filters to the containing archive; reextract
backfill stamps the link.

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

alembic/versions/0055_image_provenance_from_attachment.py

@@ -0,0 +1,55 @@
+"""image_provenance: from_attachment_id (which archive an image was extracted from)
+
+Milestone #87. When an image is pulled out of a .zip/.rar, record WHICH archive
+PostAttachment it came from, so the provenance UI can show the single archive a
+file lives inside instead of every attachment on the post. Nullable FK with
+ON DELETE SET NULL — a loose (non-archive) download leaves it NULL, and deleting
+the archive attachment forgets the linkage without destroying the (image, post)
+provenance edge. Existing rows are NULL until the reextract backfill stamps them.
+
+Revision ID: 0055
+Revises: 0054
+Create Date: 2026-06-22
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0055"
+down_revision: Union[str, None] = "0054"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "image_provenance",
+        sa.Column("from_attachment_id", sa.Integer(), nullable=True),
+    )
+    op.create_index(
+        "ix_image_provenance_from_attachment_id",
+        "image_provenance",
+        ["from_attachment_id"],
+    )
+    op.create_foreign_key(
+        "fk_image_provenance_from_attachment",
+        "image_provenance",
+        "post_attachment",
+        ["from_attachment_id"],
+        ["id"],
+        ondelete="SET NULL",
+    )
+
+
+def downgrade() -> None:
+    op.drop_constraint(
+        "fk_image_provenance_from_attachment",
+        "image_provenance",
+        type_="foreignkey",
+    )
+    op.drop_index(
+        "ix_image_provenance_from_attachment_id",
+        table_name="image_provenance",
+    )
+    op.drop_column("image_provenance", "from_attachment_id")

backend/app/models/image_provenance.py

@@ -41,6 +41,16 @@ class ImageProvenance(Base):
     source_id: Mapped[int | None] = mapped_column(
         ForeignKey("source.id", ondelete="SET NULL"), nullable=True, index=True
     )
+    # The archive PostAttachment this image was extracted FROM, when it came
+    # out of a .zip/.rar rather than as a loose file (milestone #87). Lets the
+    # provenance UI show the exact archive a file lives inside instead of every
+    # attachment on the post. NULL for loose downloads and pre-backfill rows.
+    # SET NULL so deleting the archive attachment never destroys the (image,
+    # post) edge — it just forgets which archive it came from.
+    from_attachment_id: Mapped[int | None] = mapped_column(
+        ForeignKey("post_attachment.id", ondelete="SET NULL"),
+        nullable=True, index=True,
+    )
     captured_metadata: Mapped[dict | None] = mapped_column(JSON, nullable=True)
     captured_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), nullable=False, server_default=func.now()

backend/app/services/importer.py

@@ -17,7 +17,7 @@ from enum import StrEnum
 from pathlib import Path
 
 from PIL import Image
-from sqlalchemy import select
+from sqlalchemy import select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
@@ -506,6 +506,12 @@ class Importer:
         artist_use = artist if artist is not None else self._resolve_artist(source)
         post = self._post_for_sidecar(source, artist_use)
         member_ids: list[int] = []
+        # Every member image touched (new + superseded + deduped), so the
+        # from_attachment_id stamp below covers files that already existed in the
+        # library and were merely re-linked to this post — those matter most
+        # (the HR copy a bundle re-ships). Separate from member_ids, which is
+        # the NEWLY-imported subset feeding the ImportResult contract.
+        member_record_ids: set[int] = set()
         # Per-outcome tally so the "no images" reason names the ACTUAL cause
         # (#718): nested-archive packs, all-deduped (benign), unsupported formats,
         # or failed/corrupt members — instead of one catch-all string.
@@ -516,11 +522,19 @@ class Importer:
         self._collect_archive_members(
             source, attribution=source, source_row=source_row,
             depth=0, member_ids=member_ids, counts=counts,
+            member_record_ids=member_record_ids,
         )
         # Preserve the archive itself (links to the same Post/Artist).
         self._capture_attachment(
             source, post=post, artist=artist_use, resolved=True
         )
+        # Stamp each member's provenance row for THIS post with the archive it
+        # came out of (milestone #87). Done as a post-pass rather than threaded
+        # through _import_media/_apply_sidecar so the many dedup/supersede
+        # branches stay untouched. NULL-only so a re-extract never re-stamps and
+        # the backfill (reextract task → this same path) is idempotent. Nested
+        # members link to this OUTER archive — the only one stored as a blob.
+        self._stamp_member_archive(post.id, source, member_record_ids)
         if member_ids:
             return ImportResult(
                 status="imported", image_id=member_ids[0],
@@ -555,6 +569,7 @@ class Importer:
         self, archive_path: Path, *, attribution: Path,
         source_row: Source | None, depth: int,
         member_ids: list[int], counts: dict,
+        member_record_ids: set[int],
     ) -> None:
         """Extract `archive_path` and import its image/video members, RECURSING
         into nested archives (#718). Members attribute to `attribution` — the
@@ -590,6 +605,7 @@ class Importer:
                             member_path, attribution=attribution,
                             source_row=source_row, depth=depth + 1,
                             member_ids=member_ids, counts=counts,
+                            member_record_ids=member_record_ids,
                         )
                         continue
                     counts["media"] += 1
@@ -601,10 +617,16 @@ class Importer:
                     )
                     if res.status in ("imported", "superseded") and res.image_id:
                         member_ids.append(res.image_id)
+                        member_record_ids.add(res.image_id)
                     elif res.status == "skipped" and res.skip_reason in (
                         SkipReason.duplicate_hash, SkipReason.duplicate_phash
                     ):
                         counts["deduped"] += 1
+                        # A deduped member still links provenance to this post
+                        # (enrich-on-duplicate); record it so its archive origin
+                        # gets stamped too.
+                        if res.image_id:
+                            member_record_ids.add(res.image_id)
                     else:
                         counts["failed"] += 1
         except Exception as exc:  # noqa: BLE001 — defensive per level; keep going
@@ -613,6 +635,39 @@ class Importer:
                 archive_path.name, depth, exc,
             )
 
+    def _stamp_member_archive(
+        self, post_id: int | None, archive_source: Path, member_record_ids: set[int],
+    ) -> None:
+        """Record which archive each extracted member came from (milestone #87).
+
+        Resolves the archive's own PostAttachment (by post + sha — it was just
+        captured) and stamps from_attachment_id on every member's provenance row
+        FOR THIS POST. NULL-only, so re-extracting the same archive (the backfill
+        path) never overwrites and stays idempotent. No-op when the archive isn't
+        post-attached (filesystem import with no post) or yielded no members.
+        """
+        if post_id is None or not member_record_ids:
+            return
+        sha = _sha256_of(archive_source)
+        att_id = self.session.execute(
+            select(PostAttachment.id).where(
+                PostAttachment.post_id == post_id,
+                PostAttachment.sha256 == sha,
+            )
+        ).scalar_one_or_none()
+        if att_id is None:
+            return
+        self.session.execute(
+            update(ImageProvenance)
+            .where(
+                ImageProvenance.image_record_id.in_(member_record_ids),
+                ImageProvenance.post_id == post_id,
+                ImageProvenance.from_attachment_id.is_(None),
+            )
+            .values(from_attachment_id=att_id)
+        )
+        self.session.commit()
+
     @staticmethod
     def _video_aspect_matches(w, h, cw, ch) -> bool:
         """True when two (w,h) pairs share an aspect ratio within tolerance.

backend/app/services/provenance_service.py

@@ -66,6 +66,10 @@ class ProvenanceService:
         ).scalars().all()
         return [_attachment_dict(a) for a in rows]
 
+    async def _attachment_by_id(self, attachment_id: int) -> list[dict]:
+        att = await self.session.get(PostAttachment, attachment_id)
+        return [_attachment_dict(att)] if att is not None else []
+
     async def for_image(self, image_id: int) -> dict | None:
         rec = await self.session.get(ImageRecord, image_id)
         if rec is None:
@@ -85,20 +89,31 @@ class ProvenanceService:
         )
         rows = (await self.session.execute(stmt)).all()
         post_ids = [ip.post_id for ip, _p, _s, _a in rows]
-        # Scope attachments to the image's ORIGINATING post only, not every
-        # post it's pHash-linked to. An image dupe gets a provenance row per
-        # post it reappears in (enrich-on-duplicate), and one of those is
-        # often a "High Resolution Files" mega-bundle carrying dozens of
-        # unrelated archives — aggregating across all linked posts ballooned
-        # the panel with files that have nothing to do with this image.
+        # Prefer the EXACT archive this file came out of (milestone #87): if the
+        # originating post's provenance row records from_attachment_id, the image
+        # was extracted from that one .zip/.rar, so show only it — not the dozens
+        # of unrelated archives a "High Resolution Files" bundle post carries.
+        from_att_id = next(
+            (
+                ip.from_attachment_id
+                for ip, _p, _s, _a in rows
+                if ip.post_id == rec.primary_post_id
+                and ip.from_attachment_id is not None
+            ),
+            None,
+        )
+        if from_att_id is not None:
+            attachments = await self._attachment_by_id(from_att_id)
+        else:
+            # No recorded containing archive (loose download, or pre-backfill):
+            # scope to the originating post only, not every pHash-linked post.
             # primary_post_id is the post this file was actually captured from;
-        # fall back to all linked posts only when it's unset (older rows /
-        # filesystem imports). NB: FC stores archives as opaque blobs and
-        # never records which archive an extracted image came from, so we
-        # cannot scope tighter than the post — see milestone for the
-        # image->archive capture work.
+            # fall back to all linked posts when it's unset (older rows /
+            # filesystem imports).
             attach_post_ids = (
-            [rec.primary_post_id] if rec.primary_post_id is not None else post_ids
+                [rec.primary_post_id]
+                if rec.primary_post_id is not None
+                else post_ids
             )
             attachments = await self._attachments_for_posts(attach_post_ids)
         return {

frontend/src/components/modal/ProvenancePanel.vue

@@ -70,7 +70,9 @@
     </div>
 
     <div v-if="attachments.length" class="fc-prov__attach">
-      <h4 class="fc-prov__attach-title">Attachments ({{ attachments.length }})</h4>
+      <h4 class="fc-prov__attach-title">
+        {{ attachments.length === 1 ? 'Attachment' : `Attachments (${attachments.length})` }}
+      </h4>
       <!-- Scroll-capped: a single post can carry dozens of archives (HR
            bundle posts), which previously ballooned the panel past the
            viewport. Mirror the cards' independent-scroll treatment. -->

tests/test_importer_archive.py

@@ -164,6 +164,86 @@ def test_archive_all_deduped_is_benign_not_flagged(importer, import_layout):
     assert provs == 4  # 2 images × 2 posts (enrich-on-duplicate)
 
 
+def test_archive_members_record_containing_archive(importer, import_layout):
+    """Milestone #87: each extracted member's provenance row records the archive
+    PostAttachment it came out of, so provenance can show the one archive a file
+    lives inside."""
+    import_root, _ = import_layout
+    importer.settings.phash_threshold = 0
+    arc = import_root / "Bob" / "set.cbz"
+    arc.parent.mkdir(parents=True, exist_ok=True)
+    with zipfile.ZipFile(arc, "w") as zf:
+        zf.writestr("a.jpg", _split_bytes("v"))
+        zf.writestr("b.jpg", _split_bytes("h"))
+    arc.with_suffix(".cbz.json").write_text(json.dumps(
+        {"category": "patreon", "id": "777", "title": "Set"}))
+
+    importer.import_one(arc)
+    att = importer.session.execute(select(PostAttachment)).scalar_one()
+    provs = importer.session.execute(select(ImageProvenance)).scalars().all()
+    assert len(provs) == 2
+    assert all(p.from_attachment_id == att.id for p in provs)
+
+
+def test_nested_archive_member_records_outer_archive(importer, import_layout):
+    """Nested members link to the OUTER stored archive — the only one persisted
+    as a PostAttachment (the inner archive lives in a tempdir)."""
+    import_root, _ = import_layout
+    importer.settings.phash_threshold = 0
+    inner = io.BytesIO()
+    with zipfile.ZipFile(inner, "w") as zf:
+        zf.writestr("p1.jpg", _split_bytes("v"))
+        zf.writestr("p2.jpg", _split_bytes("h"))
+    arc = import_root / "Nessie" / "hr.cbz"
+    arc.parent.mkdir(parents=True, exist_ok=True)
+    with zipfile.ZipFile(arc, "w") as zf:
+        zf.writestr("chapter.zip", inner.getvalue())
+    arc.with_suffix(".cbz.json").write_text(json.dumps(
+        {"category": "patreon", "id": "900", "title": "HR Pack"}))
+
+    importer.import_one(arc)
+    att = importer.session.execute(select(PostAttachment)).scalar_one()
+    provs = importer.session.execute(select(ImageProvenance)).scalars().all()
+    assert len(provs) == 2
+    assert all(p.from_attachment_id == att.id for p in provs)
+
+
+def test_deduped_archive_member_records_each_posts_archive(importer, import_layout):
+    """A member re-shipped in a second post's archive gets its NEW provenance row
+    (for the second post) stamped with the SECOND archive — the UPDATE-of-existing
+    path the reextract backfill relies on. Each post's rows point at its own
+    archive."""
+    import_root, _ = import_layout
+    importer.settings.phash_threshold = 0
+    pv, ph = _split_bytes("v"), _split_bytes("h")
+    first = import_root / "Dup" / "a.cbz"
+    first.parent.mkdir(parents=True, exist_ok=True)
+    with zipfile.ZipFile(first, "w") as zf:
+        zf.writestr("a.jpg", pv)
+        zf.writestr("b.jpg", ph)
+    first.with_suffix(".cbz.json").write_text(json.dumps(
+        {"category": "patreon", "id": "111", "title": "First"}))
+    importer.import_one(first)
+
+    second = import_root / "Dup" / "b.cbz"  # different post, same two images
+    with zipfile.ZipFile(second, "w") as zf:
+        zf.writestr("a.jpg", pv)
+        zf.writestr("b.jpg", ph)
+    second.with_suffix(".cbz.json").write_text(json.dumps(
+        {"category": "patreon", "id": "222", "title": "Second"}))
+    importer.import_one(second)
+
+    posts = {p.external_post_id: p.id for p in
+             importer.session.execute(select(Post)).scalars().all()}
+    atts = {a.post_id: a.id for a in
+            importer.session.execute(select(PostAttachment)).scalars().all()}
+    provs = importer.session.execute(select(ImageProvenance)).scalars().all()
+    assert len(provs) == 4  # 2 images × 2 posts
+    for p in provs:
+        assert p.from_attachment_id == atts[p.post_id]
+    assert atts[posts["111"]] != atts[posts["222"]]
+
+
 def test_corrupt_archive_still_stored(importer, import_layout):
     import_root, _ = import_layout
     arc = import_root / "Bob" / "broken.zip"

tests/test_provenance_service.py

@@ -165,6 +165,37 @@ async def test_for_image_attachments_scoped_to_primary_post(db):
     assert names == ["keep.zip"]
 
 
+@pytest.mark.asyncio
+async def test_for_image_attachments_filtered_to_containing_archive(db):
+    # Milestone #87: when the originating post's provenance row records which
+    # archive the file came from, show ONLY that archive — not the post's other
+    # attachments.
+    rec = await _seed_image(db)
+    a1, s1, post = await _seed_post(db, artist_name="Arc", slug="arc",
+                                    platform="patreon", ext_id="500")
+    rec.primary_post_id = post.id
+    att_in = PostAttachment(
+        post_id=post.id, artist_id=a1.id, sha256="e" + "0" * 63,
+        path="/images/attachments/e00/in.cbz", original_filename="in.cbz",
+        ext=".cbz", mime="application/zip", size_bytes=9,
+    )
+    att_other = PostAttachment(
+        post_id=post.id, artist_id=a1.id, sha256="f" + "0" * 63,
+        path="/images/attachments/f00/other.rar", original_filename="other.rar",
+        ext=".rar", mime="application/x-rar", size_bytes=9,
+    )
+    db.add(att_in)
+    db.add(att_other)
+    await db.flush()
+    db.add(ImageProvenance(image_record_id=rec.id, post_id=post.id,
+                           source_id=s1.id, from_attachment_id=att_in.id))
+    await db.flush()
+
+    payload = await ProvenanceService(db).for_image(rec.id)
+    names = [a["original_filename"] for a in payload["attachments"]]
+    assert names == ["in.cbz"]
+
+
 @pytest.mark.asyncio
 async def test_for_image_attachments_fallback_to_all_posts_when_no_primary(db):
     # No primary_post_id (older rows / filesystem imports) → preserve the

tests/test_reextract_archives.py

@@ -74,10 +74,13 @@ def test_reextract_links_archive_members_to_post(db_sync, tmp_path, monkeypatch)
     with zipfile.ZipFile(arc, "w") as zf:
         zf.writestr("inside.jpg", _jpeg("red"))
     sha = hashlib.sha256(arc.read_bytes()).hexdigest()
-    db_sync.add(PostAttachment(
+    att = PostAttachment(
         post_id=post.id, artist_id=artist.id, sha256=sha, path=str(arc),
         original_filename=arc.name, ext="", size_bytes=arc.stat().st_size,
-    ))
+    )
+    db_sync.add(att)
+    db_sync.flush()
+    att_id = att.id
     db_sync.commit()
 
     summary = cleanup_service.reextract_archive_attachments(
@@ -94,6 +97,9 @@ def test_reextract_links_archive_members_to_post(db_sync, tmp_path, monkeypatch)
     ).scalars().all()
     assert len(prov) == 1
     assert prov[0].image_record_id == images[0].id
+    # Milestone #87: re-extraction stamps which archive the member came from —
+    # this is the backfill path for pre-existing opaque archives.
+    assert prov[0].from_attachment_id == att_id
 
     # Idempotent — a second run imports nothing new (member dedups by sha256).
     again = cleanup_service.reextract_archive_attachments(