feat(maintenance): re-extract archive attachments + link to post — #713 part 2

Existing PostAttachments that are actually archives (filed opaquely before the
magic-byte gate) need extracting retroactively. cleanup_service.
reextract_archive_attachments scans PostAttachments, magic-detects the archives,
and for each reconstructs the post's sidecar from the DB + re-runs attach_in_place
in a temp dir — so the members extract and re-link to the SAME post via
find_or_create_post (source_id + external_post_id). Idempotent (members dedupe by
sha256). Enqueues thumbnail+ML for new members.

Wired as a maintenance-queue Celery task (tasks/admin) + POST
/api/admin/maintenance/reextract-archives (202) + a "Re-extract archive
attachments" card in Settings → Maintenance.

Test: a zip stored under a mangled extension-less name extracts + links its
member to the post via ImageProvenance, and a second run is a no-op (idempotent).

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

backend/app/api/admin.py

@@ -289,3 +289,14 @@ async def trigger_vacuum():
 
     vacuum_analyze.delay()
     return jsonify({"status": "queued"}), 202
+
+
+@admin_bp.route("/maintenance/reextract-archives", methods=["POST"])
+async def trigger_reextract_archives():
+    """Operator-triggered re-extract (#713): PostAttachments that are actually
+    archives but were filed opaquely (pre magic-byte gate) get extracted and
+    their members linked to the post. Idempotent; runs on the maintenance queue."""
+    from ..tasks.admin import reextract_archive_attachments_task
+
+    async_result = reextract_archive_attachments_task.delay()
+    return jsonify({"task_id": async_result.id, "status": "queued"}), 202

backend/app/services/cleanup_service.py

@@ -11,6 +11,7 @@ the one-and-done GS/IR migration tooling.)
 """
 from __future__ import annotations
 
+import logging
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
 from typing import Any
@@ -22,6 +23,8 @@ from ..models import Artist, ImageRecord, LibraryAuditRun, Tag
 from ..models.series_page import SeriesPage
 from ..models.tag import image_tag
 
+log = logging.getLogger(__name__)
+
 
 def project_artist_cascade(session: Session, *, slug: str) -> dict:
     """Read-only projection of what delete_artist_cascade would touch.
@@ -665,3 +668,114 @@ def cancel_audit_run(session: Session, *, audit_id: int) -> None:
         .where(LibraryAuditRun.status == "running")
         .values(status="cancelled", finished_at=datetime.now(UTC))
     )
+
+
+# -- archive-attachment re-extraction (#713 part 2) ------------------------
+
+_ARCHIVE_EXT_FOR_FORMAT = {"zip": ".zip", "rar": ".rar", "7z": ".7z"}
+
+
+def _reextract_archive_to_post(importer, archive_path: Path, post, source_row, artist) -> list[int]:
+    """Extract one stored archive and link its members to `post`.
+
+    The stored attachment has no adjacent sidecar (it lives in the sha-addressed
+    attachment store), so reconstruct the post's sidecar from the DB and re-run
+    the normal import path: `attach_in_place` extracts the members and
+    `_apply_sidecar` → `find_or_create_post` re-attaches them to the SAME Post
+    (matched by source_id + external_post_id). Returns the new member image ids.
+    """
+    import json
+    import shutil
+    import tempfile
+
+    from .archive_extractor import detect_archive_format
+
+    fmt = detect_archive_format(archive_path)
+    ext = _ARCHIVE_EXT_FOR_FORMAT.get(fmt or "", ".zip")
+    sidecar = {
+        "category": source_row.platform if source_row is not None
+        else (post.raw_metadata or {}).get("category"),
+        "id": post.external_post_id,
+        "title": post.post_title or "",
+        "content": post.description or "",
+        "published_at": post.post_date.isoformat() if post.post_date else None,
+        "url": post.post_url,
+    }
+    with tempfile.TemporaryDirectory(prefix="fc_reextract_") as td:
+        tmp = Path(td) / f"archive{ext}"  # clean ext → is_archive + find_sidecar
+        shutil.copy2(archive_path, tmp)
+        tmp.with_suffix(".json").write_text(json.dumps(sidecar))
+        res = importer.attach_in_place(tmp, artist=artist, source=source_row)
+        return list(res.member_image_ids or [])
+
+
+def reextract_archive_attachments(session: Session, *, images_root: Path) -> dict:
+    """Re-process existing PostAttachments that are ACTUALLY archives but were
+    filed opaquely before #713 part 1 (extension-only is_archive missed mangled /
+    extension-less Patreon attachment names). For each: extract the members,
+    import them, and link them to the attachment's post.
+
+    Idempotent — members dedupe by sha256, the archive dedupes by sha — so it's
+    safe to run repeatedly. Returns a summary dict for task_run.metadata.
+    """
+    from ..models import ImportSettings, Post, PostAttachment, Source
+    from ..tasks.ml import tag_and_embed
+    from ..tasks.thumbnail import generate_thumbnail
+    from .archive_extractor import is_archive
+    from .importer import Importer
+    from .thumbnailer import Thumbnailer
+
+    summary = {
+        "scanned": 0, "archives": 0, "members_imported": 0,
+        "posts_touched": 0, "skipped_no_post": 0, "errors": 0,
+    }
+    settings = ImportSettings.load_sync(session)
+    importer = Importer(
+        session=session, images_root=images_root, import_root=images_root,
+        thumbnailer=Thumbnailer(images_root=images_root), settings=settings,
+    )
+
+    attachments = session.execute(
+        select(PostAttachment).order_by(PostAttachment.id)
+    ).scalars().all()
+    enqueue_ids: list[int] = []
+    for att in attachments:
+        summary["scanned"] += 1
+        stored = Path(att.path)
+        try:
+            if not stored.is_file() or not is_archive(stored):
+                continue
+        except OSError:
+            continue
+        summary["archives"] += 1
+        if att.post_id is None:
+            summary["skipped_no_post"] += 1
+            continue
+        post = session.get(Post, att.post_id)
+        if post is None:
+            summary["skipped_no_post"] += 1
+            continue
+        artist = session.get(Artist, att.artist_id) if att.artist_id else None
+        source_row = session.get(Source, post.source_id) if post.source_id else None
+        try:
+            ids = _reextract_archive_to_post(importer, stored, post, source_row, artist)
+            session.commit()
+        except Exception as exc:  # one bad archive must not strand the rest
+            session.rollback()
+            summary["errors"] += 1
+            log.warning("re-extract failed for attachment %s: %s", att.id, exc)
+            continue
+        if ids:
+            summary["members_imported"] += len(ids)
+            summary["posts_touched"] += 1
+            enqueue_ids.extend(ids)
+
+    # Thumbnails + ML for the newly-imported members (best-effort; off the
+    # critical path — a Redis hiccup must not fail the whole re-extract).
+    for img_id in enqueue_ids:
+        try:
+            generate_thumbnail.delay(img_id)
+            tag_and_embed.delay(img_id)
+        except Exception as exc:
+            log.warning("re-extract enqueue failed for image %s: %s", img_id, exc)
+    return summary

backend/app/tasks/admin.py

@@ -55,3 +55,21 @@ def bulk_delete_images_task(self, *, image_ids: list[int]) -> dict:
         return cleanup_service.delete_images(
             session, image_ids=image_ids, images_root=IMAGES_ROOT,
         )
+
+
+@celery.task(
+    name="backend.app.tasks.admin.reextract_archive_attachments_task",
+    bind=True,
+    autoretry_for=(OperationalError, DBAPIError),
+    retry_backoff=15, retry_backoff_max=180, max_retries=1,
+    soft_time_limit=1800, time_limit=2400,  # 30 min / 40 min
+)
+def reextract_archive_attachments_task(self) -> dict:
+    """Wraps cleanup_service.reextract_archive_attachments (#713 part 2):
+    re-extract PostAttachments that are actually archives but were filed
+    opaquely before the magic-byte gate, and link their members to the post."""
+    SessionLocal = _sync_session_factory()
+    with SessionLocal() as session:
+        return cleanup_service.reextract_archive_attachments(
+            session, images_root=IMAGES_ROOT,
+        )