feat(attachments): importer dispatch — archive extract + non-media capture

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

tests/test_importer.py

@@ -117,13 +117,14 @@ def test_transparent_filter(importer, import_layout):
 
 
 def test_unsupported_extension(importer, import_layout):
+    # FC-2d-iii: non-media is no longer skipped — it's captured as a
+    # PostAttachment so nothing a post contained is lost.
     import_root, _ = import_layout
     src = import_root / "Alice" / "notes.txt"
     src.parent.mkdir(parents=True, exist_ok=True)
     src.write_text("hello")
     result = importer.import_one(src)
-    assert result.status == "skipped"
-    assert result.skip_reason == SkipReason.invalid_image
+    assert result.status == "attached"
 
 
 def test_root_level_file_has_no_artist(importer, import_layout):

tests/test_importer_archive.py

@@ -0,0 +1,102 @@
+"""FC-2d-iii: archive → import media members + store the archive."""
+
+import io
+import json
+import zipfile
+
+import pytest
+from PIL import Image
+from sqlalchemy import func, select
+
+from backend.app.models import (
+    ImageProvenance,
+    ImageRecord,
+    ImportSettings,
+    Post,
+    PostAttachment,
+)
+from backend.app.services.importer import Importer
+from backend.app.services.thumbnailer import Thumbnailer
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture
+def import_layout(tmp_path):
+    import_root = tmp_path / "import"
+    images_root = tmp_path / "images"
+    import_root.mkdir()
+    images_root.mkdir()
+    return import_root, images_root
+
+
+@pytest.fixture
+def importer(db_sync, import_layout):
+    import_root, images_root = import_layout
+    settings = db_sync.execute(
+        select(ImportSettings).where(ImportSettings.id == 1)
+    ).scalar_one()
+    return Importer(
+        session=db_sync, images_root=images_root, import_root=import_root,
+        thumbnailer=Thumbnailer(images_root=images_root), settings=settings,
+    )
+
+
+def _jpeg_bytes(color):
+    buf = io.BytesIO()
+    Image.new("RGB", (40, 40), color).save(buf, "JPEG")
+    return buf.getvalue()
+
+
+def test_archive_imports_members_and_stores_archive(importer, import_layout):
+    import_root, _ = import_layout
+    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", _jpeg_bytes((200, 10, 10)))
+        zf.writestr("b.jpg", _jpeg_bytes((10, 200, 10)))
+        zf.writestr("readme.txt", b"hello")
+    arc.with_suffix(".cbz.json").write_text(json.dumps(
+        {"category": "patreon", "id": "777", "title": "Set"}))
+
+    r = importer.import_one(arc)
+    assert r.status == "imported"
+    assert len(r.member_image_ids) == 2
+
+    imgs = importer.session.execute(select(ImageRecord)).scalars().all()
+    assert len(imgs) == 2
+    post = importer.session.execute(select(Post)).scalar_one()
+    provs = importer.session.execute(
+        select(func.count()).select_from(ImageProvenance)
+    ).scalar_one()
+    assert provs == 2  # one Post, both members
+    att = importer.session.execute(select(PostAttachment)).scalar_one()
+    assert att.original_filename == "set.cbz"
+    assert att.post_id == post.id
+
+
+def test_corrupt_archive_still_stored(importer, import_layout):
+    import_root, _ = import_layout
+    arc = import_root / "Bob" / "broken.zip"
+    arc.parent.mkdir(parents=True, exist_ok=True)
+    arc.write_bytes(b"definitely not a zip")
+    r = importer.import_one(arc)
+    assert r.status == "attached"
+    att = importer.session.execute(select(PostAttachment)).scalar_one()
+    assert att.original_filename == "broken.zip"
+    assert importer.session.execute(
+        select(func.count()).select_from(ImageRecord)
+    ).scalar_one() == 0
+
+
+def test_reimport_archive_is_idempotent(importer, import_layout):
+    import_root, _ = import_layout
+    arc = import_root / "Bob" / "set.zip"
+    arc.parent.mkdir(parents=True, exist_ok=True)
+    with zipfile.ZipFile(arc, "w") as zf:
+        zf.writestr("a.jpg", _jpeg_bytes((1, 2, 3)))
+    importer.import_one(arc)
+    importer.import_one(arc)
+    assert importer.session.execute(
+        select(func.count()).select_from(PostAttachment)
+    ).scalar_one() == 1

tests/test_importer_attachments.py

@@ -0,0 +1,61 @@
+"""FC-2d-iii: loose non-media → PostAttachment (no skip)."""
+
+import pytest
+from sqlalchemy import func, select
+
+from backend.app.models import Artist, ImportSettings, PostAttachment
+from backend.app.services.importer import Importer
+from backend.app.services.thumbnailer import Thumbnailer
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture
+def import_layout(tmp_path):
+    import_root = tmp_path / "import"
+    images_root = tmp_path / "images"
+    import_root.mkdir()
+    images_root.mkdir()
+    return import_root, images_root
+
+
+@pytest.fixture
+def importer(db_sync, import_layout):
+    import_root, images_root = import_layout
+    settings = db_sync.execute(
+        select(ImportSettings).where(ImportSettings.id == 1)
+    ).scalar_one()
+    return Importer(
+        session=db_sync, images_root=images_root, import_root=import_root,
+        thumbnailer=Thumbnailer(images_root=images_root), settings=settings,
+    )
+
+
+def test_loose_non_media_becomes_attachment(importer, import_layout):
+    import_root, _ = import_layout
+    f = import_root / "Alice" / "manifesto.pdf"
+    f.parent.mkdir(parents=True, exist_ok=True)
+    f.write_bytes(b"%PDF-1.4 stuff")
+    r = importer.import_one(f)
+    assert r.status == "attached"
+    att = importer.session.execute(select(PostAttachment)).scalar_one()
+    artist = importer.session.execute(
+        select(Artist).where(Artist.slug == "alice")
+    ).scalar_one()
+    assert att.artist_id == artist.id
+    assert att.post_id is None
+    assert att.original_filename == "manifesto.pdf"
+    assert att.ext == ".pdf"
+
+
+def test_json_sidecar_is_not_attached(importer, import_layout):
+    import_root, _ = import_layout
+    j = import_root / "Alice" / "x.json"
+    j.parent.mkdir(parents=True, exist_ok=True)
+    j.write_text("{}")
+    r = importer.import_one(j)
+    assert r.status == "skipped"
+    n = importer.session.execute(
+        select(func.count()).select_from(PostAttachment)
+    ).scalar_one()
+    assert n == 0