FabledCurator/tests/test_sidecar_import.py

"""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
    # Threshold 0: only an exact phash match collapses. Orthogonal splits
    # still sit within the default Hamming-10 at 64-bit, so without this
    # the 2nd image would be skipped as a near-dup (phash dedup working).
    importer.settings.phash_threshold = 0
    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")            # distinct phash; threshold 0 → both import
    _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