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 (
    Artist,
    ExternalLink,
    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)
    post = importer.session.execute(select(Post)).scalar_one()
    # Filesystem-imported sidecar posts no longer create a synthetic Source
    # (alembic 0030 / nullable post.source_id refactor). The Post is linked
    # to the artist via Post.artist_id; Post.source_id stays NULL until a
    # real subscription for the (artist, platform) gets added.
    assert post.source_id is None
    assert importer.session.execute(
        select(func.count()).select_from(Source)
    ).scalar_one() == 0
    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
    # Denormalized gallery sort key (alembic 0035) tracks the primary post's
    # publish date so /scroll orders off ix_image_record_effective_date.
    assert rec.effective_date == post.post_date


def test_sidecar_source_url_persists_filehash(importer, import_layout):
    """#830 Phase 2: a media sidecar's source_url lands on the ImageRecord as
    source_url + the lowercased CDN filehash (the inline-image join key)."""
    import_root, _ = import_layout
    m = import_root / "Alice" / "b.jpg"
    _split(m, "v")
    _sidecar(m, {
        "category": "patreon", "id": 901, "title": "Body post",
        "source_url": "https://cdn.test/p/0123456789ABCDEF0123456789abcdef/b.jpg",
    })
    r = importer.import_one(m)
    assert r.status == "imported"
    rec = importer.session.get(ImageRecord, r.image_id)
    assert rec.source_url == (
        "https://cdn.test/p/0123456789ABCDEF0123456789abcdef/b.jpg"
    )
    assert rec.source_filehash == "0123456789abcdef0123456789abcdef"


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"
    # No synthetic Source after alembic 0030; both imports still resolve to
    # a single null-source Post (deduped by uq_post_artist_external_id_null_source).
    assert importer.session.execute(
        select(func.count()).select_from(Source)
    ).scalar_one() == 0
    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):
    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()
    # No synthetic Source after alembic 0030; the artist linkage lives on
    # Post.artist_id (NOT NULL FK).
    post = importer.session.execute(select(Post)).scalar_one()
    assert post.artist_id == a.id
    assert post.source_id is None
    assert importer.session.execute(
        select(func.count()).select_from(Source)
    ).scalar_one() == 0


def test_upsert_post_record_creates_media_less_post(importer, import_layout):
    """A post-only sidecar (no media) upserts a Post WITH its body — text posts
    are captured even when they have nothing to download."""
    import_root, _ = import_layout
    artist = Artist(name="Alice", slug="alice")
    importer.session.add(artist)
    importer.session.flush()
    sc = import_root / "Alice" / "_post.json"
    sc.parent.mkdir(parents=True, exist_ok=True)
    sc.write_text(json.dumps({
        "category": "patreon", "id": 777,
        "url": "https://patreon.com/posts/777", "title": "Text only",
        "content": "<p>links below: mega</p>",
        "published_at": "2023-08-01T00:00:00Z",
    }))
    assert importer.upsert_post_record(sc, artist=artist) is True
    post = importer.session.execute(select(Post)).scalar_one()
    assert post.external_post_id == "777"
    assert post.post_title == "Text only"
    assert post.description == "<p>links below: mega</p>"
    assert post.post_url == "https://patreon.com/posts/777"
    assert post.post_date is not None


def test_upsert_post_record_idempotent_no_double(importer, import_layout):
    """Re-running upsert_post_record UPDATES the same Post (keyed on
    external_post_id) — never doubles."""
    import_root, _ = import_layout
    artist = Artist(name="Bob", slug="bob")
    importer.session.add(artist)
    importer.session.flush()
    sc = import_root / "Bob" / "_post.json"
    sc.parent.mkdir(parents=True, exist_ok=True)
    sc.write_text(json.dumps({
        "category": "patreon", "id": 888, "title": "T",
        "content": "<p>body</p>", "url": "https://patreon.com/posts/888",
    }))
    assert importer.upsert_post_record(sc, artist=artist) is True
    assert importer.upsert_post_record(sc, artist=artist) is True
    assert importer.session.execute(
        select(func.count()).select_from(Post)
    ).scalar_one() == 1


def test_external_links_recorded_from_body(importer, import_layout):
    """A mega/gdrive/etc. link in the body is recorded as an external_link row
    (status pending) so it's never silently dropped."""
    import_root, _ = import_layout
    artist = Artist(name="Linker", slug="linker")
    importer.session.add(artist)
    importer.session.flush()
    sc = import_root / "Linker" / "_post.json"
    sc.parent.mkdir(parents=True, exist_ok=True)
    sc.write_text(json.dumps({
        "category": "patreon", "id": 999, "title": "Film",
        "url": "https://patreon.com/posts/999",
        "content": '<p>Get it: <a href="https://mega.nz/file/AbC#key123">Mega</a></p>',
    }))
    assert importer.upsert_post_record(sc, artist=artist) is True
    importer.session.expire_all()  # re-read so the server_default status loads
    links = importer.session.execute(select(ExternalLink)).scalars().all()
    assert len(links) == 1
    assert links[0].host == "mega"
    assert links[0].url == "https://mega.nz/file/AbC#key123"
    assert links[0].label == "Mega"
    assert links[0].status == "pending"


def test_external_links_not_duplicated_on_reimport(importer, import_layout):
    """Re-importing the same body keeps ONE external_link row (insert-missing)."""
    import_root, _ = import_layout
    artist = Artist(name="Linker2", slug="linker2")
    importer.session.add(artist)
    importer.session.flush()
    sc = import_root / "Linker2" / "_post.json"
    sc.parent.mkdir(parents=True, exist_ok=True)
    sc.write_text(json.dumps({
        "category": "patreon", "id": 1000, "title": "Film",
        "url": "https://patreon.com/posts/1000",
        "content": '<a href="https://pixeldrain.com/u/xyz">px</a>',
    }))
    assert importer.upsert_post_record(sc, artist=artist) is True
    assert importer.upsert_post_record(sc, artist=artist) is True
    assert importer.session.execute(
        select(func.count()).select_from(ExternalLink)
    ).scalar_one() == 1