feat(sidecar): importer creates Source/Post/ImageProvenance from sidecars (idempotent, fail-soft)

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

tests/test_sidecar_import.py

@@ -0,0 +1,164 @@
+"""FC-2d-v: sidecar → Source/Post/ImageProvenance (sync importer)."""
+
+import json
+from pathlib import Path
+
+import pytest
+from PIL import Image
+from sqlalchemy import func, select
+
+from backend.app.models import (
+    ImageProvenance,
+    ImageRecord,
+    ImportSettings,
+    Post,
+    Source,
+)
+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 _split(path: Path, orient, size=(256, 256)):
+    """Structured image (half/half) — solid colors phash-collapse, which
+    would route a 2nd image to the supersede path (no sidecar applied)."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    w, h = size
+    im = Image.new("L", size, 0)
+    px = im.load()
+    for y in range(h):
+        for x in range(w):
+            if (x / w if orient == "v" else y / h) >= 0.5:
+                px[x, y] = 255
+    im.convert("RGB").save(path, "JPEG")
+
+
+def _sidecar(media: Path, payload: dict):
+    media.with_suffix(".json").write_text(json.dumps(payload))
+
+
+def test_sidecar_creates_provenance(importer, import_layout):
+    import_root, _ = import_layout
+    m = import_root / "Alice" / "a.jpg"
+    _split(m, "v")
+    _sidecar(m, {
+        "category": "patreon", "id": 555,
+        "url": "https://patreon.com/posts/555", "title": "Set 1",
+        "content": "<p>hi</p>", "page_count": 2,
+        "published_at": "2023-08-01T00:00:00Z",
+    })
+    r = importer.import_one(m)
+    assert r.status == "imported"
+    rec = importer.session.get(ImageRecord, r.image_id)
+    src = importer.session.execute(select(Source)).scalar_one()
+    post = importer.session.execute(select(Post)).scalar_one()
+    assert src.platform == "patreon"
+    assert post.external_post_id == "555"
+    assert post.post_url == "https://patreon.com/posts/555"
+    assert post.post_title == "Set 1"
+    assert post.description == "<p>hi</p>"
+    assert post.attachment_count == 2
+    assert post.post_date is not None
+    assert post.raw_metadata["id"] == 555
+    prov = importer.session.execute(select(ImageProvenance)).scalar_one()
+    assert prov.image_record_id == rec.id and prov.post_id == post.id
+    assert rec.primary_post_id == post.id
+
+
+def test_reimport_same_post_idempotent(importer, import_layout):
+    import_root, _ = import_layout
+    payload = {"category": "patreon", "id": 777, "title": "P"}
+    m1 = import_root / "Bob" / "p1.jpg"
+    _split(m1, "v")
+    _sidecar(m1, payload)
+    importer.import_one(m1)
+    m2 = import_root / "Bob" / "p2.jpg"
+    _split(m2, "h")            # structurally distinct → not a phash dup
+    _sidecar(m2, payload)
+    r2 = importer.import_one(m2)
+    assert r2.status == "imported"
+    assert importer.session.execute(
+        select(func.count()).select_from(Source)
+    ).scalar_one() == 1
+    assert importer.session.execute(
+        select(func.count()).select_from(Post)
+    ).scalar_one() == 1
+    assert importer.session.execute(
+        select(func.count()).select_from(ImageProvenance)
+    ).scalar_one() == 2
+
+
+def test_garbage_sidecar_still_imports(importer, import_layout):
+    import_root, _ = import_layout
+    m = import_root / "Carol" / "c.jpg"
+    _split(m, "v")
+    m.with_suffix(".json").write_text("{ not json")
+    r = importer.import_one(m)
+    assert r.status == "imported"
+    assert importer.session.execute(
+        select(func.count()).select_from(Post)
+    ).scalar_one() == 0
+
+
+def test_no_sidecar_unchanged(importer, import_layout):
+    import_root, _ = import_layout
+    m = import_root / "Dave" / "d.jpg"
+    _split(m, "v")
+    r = importer.import_one(m)
+    assert r.status == "imported"
+    assert importer.session.execute(
+        select(func.count()).select_from(Post)
+    ).scalar_one() == 0
+
+
+def test_no_artist_anywhere_skips_provenance(importer, import_layout):
+    import_root, _ = import_layout
+    m = import_root / "rootfile.jpg"          # no top-level artist folder
+    _split(m, "v")
+    _sidecar(m, {"category": "x", "id": 1})   # no artist key
+    r = importer.import_one(m)
+    assert r.status == "imported"
+    assert importer.session.execute(
+        select(func.count()).select_from(Source)
+    ).scalar_one() == 0
+
+
+def test_sidecar_artist_used_when_no_folder_artist(importer, import_layout):
+    from backend.app.models import Artist
+
+    import_root, _ = import_layout
+    m = import_root / "e.jpg"                 # root → no folder artist
+    _split(m, "v")
+    _sidecar(m, {"category": "pixiv", "id": 9, "artist": "Yuki"})
+    r = importer.import_one(m)
+    assert r.status == "imported"
+    a = importer.session.execute(
+        select(Artist).where(Artist.slug == "yuki")
+    ).scalar_one()
+    src = importer.session.execute(select(Source)).scalar_one()
+    assert src.artist_id == a.id