feat(deep-scan): Importer(deep=) re-derive on sha-match; import_file wires batch.scan_mode

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

tests/test_importer_deep.py

@@ -0,0 +1,99 @@
+"""FC-2d-vi: Importer(deep=True) re-derives on an already-imported row."""
+
+import json
+
+import pytest
+from PIL import Image
+from sqlalchemy import func, select
+
+from backend.app.models import (
+    ImageProvenance,
+    ImageRecord,
+    ImportSettings,
+    Post,
+)
+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
+
+
+def _mk(db_sync, import_layout, *, deep):
+    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,
+        deep=deep,
+    )
+
+
+def _img(path):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    Image.new("RGB", (50, 50), (90, 30, 200)).save(path, "JPEG")
+
+
+def test_deep_rederives_phash_and_provenance(db_sync, import_layout):
+    import_root, _ = import_layout
+    src = import_root / "Mae" / "p.jpg"
+    _img(src)
+
+    # First import (quick). Images get a phash on import; null it +
+    # primary_post_id to simulate a pre-FC-2d-i+ii/v record, then add a
+    # sidecar to simulate provenance that didn't exist at first import.
+    quick = _mk(db_sync, import_layout, deep=False)
+    r1 = quick.import_one(src)
+    assert r1.status == "imported"
+    rec = db_sync.get(ImageRecord, r1.image_id)
+    rec.phash = None
+    rec.primary_post_id = None
+    db_sync.commit()
+    src.with_suffix(".jpg.json").write_text(json.dumps(
+        {"category": "patreon", "id": "9", "title": "P"}))
+
+    deep = _mk(db_sync, import_layout, deep=True)
+    r2 = deep.import_one(src)
+    assert r2.status == "skipped"
+    assert "deep" in (r2.error or "")
+    assert r2.image_id == rec.id
+
+    db_sync.expire_all()
+    rec2 = db_sync.get(ImageRecord, rec.id)
+    assert rec2.phash is not None
+    assert db_sync.execute(
+        select(func.count()).select_from(ImageRecord)
+    ).scalar_one() == 1  # no new record
+    post = db_sync.execute(select(Post)).scalar_one()
+    assert db_sync.execute(
+        select(func.count()).select_from(ImageProvenance)
+        .where(ImageProvenance.image_record_id == rec.id)
+    ).scalar_one() == 1
+    assert rec2.primary_post_id == post.id
+
+    # Idempotent: a second deep pass adds no duplicate provenance.
+    _mk(db_sync, import_layout, deep=True).import_one(src)
+    db_sync.expire_all()
+    assert db_sync.execute(
+        select(func.count()).select_from(ImageProvenance)
+    ).scalar_one() == 1
+
+
+def test_deep_false_is_plain_skip(db_sync, import_layout):
+    import_root, _ = import_layout
+    src = import_root / "Mae" / "q.jpg"
+    _img(src)
+    _mk(db_sync, import_layout, deep=False).import_one(src)
+    r = _mk(db_sync, import_layout, deep=False).import_one(src)
+    assert r.status == "skipped"
+    assert r.error == "sha256 already present"