FabledCurator/tests/test_gallery_service.py

from datetime import UTC, datetime, timedelta

import pytest
from sqlalchemy import select

from backend.app.models import ImageRecord, Tag, TagKind
from backend.app.models.tag import image_tag
from backend.app.services.gallery_service import (
    GalleryService,
    decode_cursor,
    encode_cursor,
    image_url,
)

pytestmark = pytest.mark.integration


def test_image_url_percent_encodes_special_chars():
    # A '#' in a post folder ('BLUE#59') would otherwise be parsed as a URL
    # fragment, dropping the rest of the path and 404'ing the original while the
    # hash-named thumbnail still loads. Operator-flagged 2026-06-12.
    url = image_url("/images/dismassd/patreon/2024-04-28_BLUE#59/01 a.jpg")
    assert url == "/images/dismassd/patreon/2024-04-28_BLUE%2359/01%20a.jpg"
    # '/' stays a separator; a plain path is unchanged.
    assert image_url("/images/a/b/c.jpg") == "/images/a/b/c.jpg"


def _now():
    return datetime.now(UTC)


async def _seed_images(db, count: int, sha_prefix: str = "0") -> list[ImageRecord]:
    base = _now()
    records = []
    for i in range(count):
        r = ImageRecord(
            path=f"/images/test/{i}.jpg",
            sha256=f"{sha_prefix}{i:063d}",
            size_bytes=1000,
            mime="image/jpeg",
            width=100,
            height=100,
            origin="imported_filesystem",
            integrity_status="unknown",
        )
        r.created_at = base - timedelta(minutes=i)
        # No post → effective_date == created_at (alembic 0035 denorm).
        r.effective_date = r.created_at
        db.add(r)
        records.append(r)
    await db.flush()
    return records


def test_cursor_roundtrip():
    ts = datetime(2026, 5, 14, 12, 30, 0, tzinfo=UTC)
    encoded = encode_cursor(ts, 42)
    back_ts, back_id = decode_cursor(encoded)
    assert back_ts == ts
    assert back_id == 42


def test_decode_invalid_cursor_raises():
    with pytest.raises(ValueError):
        decode_cursor("not-base64!!!")


@pytest.mark.asyncio
async def test_scroll_returns_newest_first(db):
    await _seed_images(db, 5)
    svc = GalleryService(db)
    page = await svc.scroll(cursor=None, limit=10)
    assert len(page.images) == 5
    assert page.images[0].created_at > page.images[-1].created_at


async def _system_tag(db, name):
    return (await db.execute(
        select(Tag).where(Tag.is_system.is_(True), Tag.name == name)
    )).scalar_one()


@pytest.mark.asyncio
async def test_scroll_hides_presentation_chrome_by_default(db):
    # banner / editor screenshot (presentation system tags) are hidden from the
    # default gallery; wip (also a system tag) is NOT — it's real, in-progress
    # art (milestone 141). Seeded system tags survive the harness TRUNCATE.
    imgs = await _seed_images(db, 3, sha_prefix="p")
    banner = await _system_tag(db, "banner")
    wip = await _system_tag(db, "wip")
    await db.execute(image_tag.insert().values(
        image_record_id=imgs[0].id, tag_id=banner.id, source="manual"))
    await db.execute(image_tag.insert().values(
        image_record_id=imgs[1].id, tag_id=wip.id, source="manual"))
    await db.flush()
    svc = GalleryService(db)

    # Default: the banner image is hidden; the wip image + the plain image stay.
    default_ids = {i.id for i in (await svc.scroll(cursor=None, limit=10)).images}
    assert imgs[0].id not in default_ids      # banner hidden
    assert imgs[1].id in default_ids          # wip visible
    assert imgs[2].id in default_ids          # plain visible

    # include_hidden surfaces the banner image (the Hidden view).
    shown = {i.id for i in (
        await svc.scroll(cursor=None, limit=10, include_hidden=True)
    ).images}
    assert imgs[0].id in shown

    # Filtering explicitly FOR banner shows it (a view exclusively for a
    # presentation tag suppresses the implicit hide).
    only_banner = {i.id for i in (
        await svc.scroll(cursor=None, limit=10, tag_ids=[banner.id])
    ).images}
    assert only_banner == {imgs[0].id}


@pytest.mark.asyncio
async def test_scroll_sorts_by_earliest_post_date(db):
    """posted_new/posted_old key off earliest_post_date (original publish across
    all of an image's posts), NOT effective_date/created_at — so an old file that
    was first posted recently sorts by that recent post date."""
    imgs = await _seed_images(db, 3)
    now = _now()
    # earliest_post_date order deliberately DIFFERS from created_at order (which
    # _seed_images makes imgs[0] newest → imgs[2] oldest), proving the sort keys
    # off earliest_post_date.
    imgs[0].earliest_post_date = now - timedelta(days=1)     # middle
    imgs[1].earliest_post_date = now - timedelta(days=365)   # oldest post
    imgs[2].earliest_post_date = now - timedelta(hours=1)    # newest post
    await db.flush()

    svc = GalleryService(db)
    newest = await svc.scroll(cursor=None, limit=10, sort="posted_new")
    assert [i.id for i in newest.images] == [imgs[2].id, imgs[0].id, imgs[1].id]
    oldest = await svc.scroll(cursor=None, limit=10, sort="posted_old")
    assert [i.id for i in oldest.images] == [imgs[1].id, imgs[0].id, imgs[2].id]


@pytest.mark.asyncio
async def test_scroll_pagination(db):
    await _seed_images(db, 5)
    svc = GalleryService(db)
    first = await svc.scroll(cursor=None, limit=2)
    assert len(first.images) == 2
    assert first.next_cursor is not None
    second = await svc.scroll(cursor=first.next_cursor, limit=2)
    assert len(second.images) == 2
    assert {i.id for i in first.images}.isdisjoint({i.id for i in second.images})


@pytest.mark.asyncio
async def test_scroll_with_tag_filter(db):
    images = await _seed_images(db, 5)
    tag = Tag(name="filterme", kind=TagKind.general)
    db.add(tag)
    await db.flush()
    await db.execute(
        image_tag.insert().values(
            image_record_id=images[0].id, tag_id=tag.id, source="manual"
        )
    )

    svc = GalleryService(db)
    page = await svc.scroll(cursor=None, limit=10, tag_ids=[tag.id])
    assert len(page.images) == 1
    assert page.images[0].id == images[0].id


@pytest.mark.asyncio
async def test_scroll_fandom_includes_images_via_its_characters(db):
    # A fandom owns a character (Tag.fandom_id). Filtering by the fandom must
    # surface images that carry the CHARACTER but were never tagged with the
    # fandom directly — plus images tagged with the fandom directly. Excluding
    # the fandom is symmetric: it hides both.
    images = await _seed_images(db, 4)
    fandom = Tag(name="Naruto", kind=TagKind.fandom)
    db.add(fandom)
    await db.flush()
    char = Tag(name="Sasuke", kind=TagKind.character, fandom_id=fandom.id)
    other = Tag(name="unrelated", kind=TagKind.general)
    db.add_all([char, other])
    await db.flush()
    await db.execute(image_tag.insert().values([
        {"image_record_id": images[0].id, "tag_id": char.id, "source": "manual"},
        {"image_record_id": images[1].id, "tag_id": fandom.id, "source": "manual"},
        {"image_record_id": images[2].id, "tag_id": other.id, "source": "manual"},
    ]))
    svc = GalleryService(db)
    via = await svc.scroll(cursor=None, limit=10, tag_ids=[fandom.id])
    assert {i.id for i in via.images} == {images[0].id, images[1].id}
    excluded = await svc.scroll(cursor=None, limit=10, tag_exclude=[fandom.id])
    assert {i.id for i in excluded.images} == {images[2].id, images[3].id}


@pytest.mark.asyncio
async def test_timeline_groups_by_month(db):
    await _seed_images(db, 3)
    svc = GalleryService(db)
    buckets = await svc.timeline()
    assert sum(b.count for b in buckets) == 3


@pytest.mark.asyncio
async def test_date_groups_in_page(db):
    await _seed_images(db, 3)
    svc = GalleryService(db)
    page = await svc.scroll(cursor=None, limit=10)
    assert len(page.date_groups) == 1
    y, m, ids = page.date_groups[0]
    assert len(ids) == 3


@pytest.mark.asyncio
async def test_neighbors(db):
    images = await _seed_images(db, 3)
    svc = GalleryService(db)
    middle = images[1]
    payload = await svc.get_image_with_tags(middle.id)
    assert payload["neighbors"]["prev_id"] == images[0].id
    assert payload["neighbors"]["next_id"] == images[2].id


@pytest.mark.asyncio
async def test_get_image_with_tags_returns_none_for_missing(db):
    svc = GalleryService(db)
    assert await svc.get_image_with_tags(99999) is None


@pytest.mark.asyncio
async def test_get_image_with_tags_includes_integrity_status(db):
    images = await _seed_images(db, 1)
    img = images[0]
    img.integrity_status = "ok"
    await db.flush()
    svc = GalleryService(db)
    payload = await svc.get_image_with_tags(img.id)
    assert payload["integrity_status"] == "ok"


async def _seed_image_with_post(
    db, *, sha: str, image_created_at, post_date, artist_name="test-artist",
    platform="patreon", external_post_id="42",
):
    """Helper: seed an Artist + Source + Post and one ImageRecord whose
    primary_post_id points at that Post. Used for date-coalesce tests."""
    from backend.app.models import Artist, Post, Source
    artist = Artist(name=artist_name, slug=artist_name.lower().replace(" ", "-"))
    db.add(artist)
    await db.flush()
    source = Source(
        artist_id=artist.id, platform=platform,
        url=f"https://www.{platform}.com/{artist.slug}",
    )
    db.add(source)
    await db.flush()
    post = Post(
        source_id=source.id, artist_id=artist.id,
        external_post_id=external_post_id,
        post_title="A Post", post_date=post_date,
    )
    db.add(post)
    await db.flush()
    img = ImageRecord(
        path=f"/images/test/{sha[:8]}.jpg",
        sha256=sha, size_bytes=1000, mime="image/jpeg",
        width=100, height=100,
        origin="imported_filesystem", integrity_status="unknown",
        primary_post_id=post.id,
    )
    img.created_at = image_created_at
    # Mirror the importer's denorm (alembic 0035): a linked post with a date
    # sets effective_date to that date, else it falls back to created_at.
    img.effective_date = post_date or image_created_at
    db.add(img)
    await db.flush()
    return img, post


@pytest.mark.asyncio
async def test_scroll_sorts_by_post_date_when_available(db):
    """Operator-flagged 2026-05-25: ~57k IR images all imported in the
    same week sort by image.created_at and pile up in one month bucket.
    Once primary_post_id is wired (via tag_apply phase 4), the gallery
    should sort by Post.post_date instead, spreading them across the
    actual publish years."""
    base_import = _now()
    # Image A: imported NOW, but post was made 2 years ago.
    img_a, _ = await _seed_image_with_post(
        db, sha="a" * 64,
        image_created_at=base_import,
        post_date=base_import - timedelta(days=730),
        artist_name="Aria", external_post_id="A-1",
    )
    # Image B: imported NOW (1 min later), post made YESTERDAY.
    img_b, _ = await _seed_image_with_post(
        db, sha="b" * 64,
        image_created_at=base_import - timedelta(minutes=1),
        post_date=base_import - timedelta(days=1),
        artist_name="Bea", external_post_id="B-1",
    )
    # Image C: filesystem-imported, no primary_post_id, created 5 days ago.
    img_c = ImageRecord(
        path="/images/test/c.jpg", sha256="c" * 64,
        size_bytes=1000, mime="image/jpeg",
        width=100, height=100,
        origin="imported_filesystem", integrity_status="unknown",
    )
    img_c.created_at = base_import - timedelta(days=5)
    img_c.effective_date = img_c.created_at  # no post → tracks created_at
    db.add(img_c)
    await db.flush()

    svc = GalleryService(db)
    page = await svc.scroll(cursor=None, limit=10)
    # Effective-date order: B (yesterday) > C (5 days ago) > A (2 years ago)
    assert [i.id for i in page.images] == [img_b.id, img_c.id, img_a.id]

    # API exposes both fields explicitly so the UI can show "Posted X / Imported Y".
    a_payload = next(i for i in page.images if i.id == img_a.id)
    assert a_payload.posted_at is not None
    assert a_payload.posted_at < a_payload.created_at
    c_payload = next(i for i in page.images if i.id == img_c.id)
    assert c_payload.posted_at is None
    assert c_payload.effective_date == c_payload.created_at


@pytest.mark.asyncio
async def test_timeline_buckets_use_post_date_when_available(db):
    """Timeline group-by must follow the same effective_date rule so the
    UI's year/month navigation surfaces publish-date buckets, not the
    single FC-scan bucket all migrated images share."""
    base = datetime(2026, 6, 15, 12, 0, tzinfo=UTC)
    await _seed_image_with_post(
        db, sha="1" * 64,
        image_created_at=base,
        post_date=datetime(2024, 3, 10, tzinfo=UTC),
        artist_name="Carl", external_post_id="C-1",
    )
    await _seed_image_with_post(
        db, sha="2" * 64,
        image_created_at=base,
        post_date=datetime(2024, 3, 11, tzinfo=UTC),
        artist_name="Dee", external_post_id="D-1",
    )
    await _seed_image_with_post(
        db, sha="3" * 64,
        image_created_at=base,
        post_date=datetime(2025, 9, 1, tzinfo=UTC),
        artist_name="Eli", external_post_id="E-1",
    )
    svc = GalleryService(db)
    buckets = await svc.timeline()
    bucket_keys = {(b.year, b.month, b.count) for b in buckets}
    # Two posts in 2024-03, one in 2025-09 — even though all imported in 2026-06.
    assert (2024, 3, 2) in bucket_keys
    assert (2025, 9, 1) in bucket_keys
    # The FC-import bucket should NOT appear since all 3 images have post_date.
    assert not any(b.year == 2026 and b.month == 6 for b in buckets)


@pytest.mark.asyncio
async def test_get_image_with_tags_includes_posted_at_when_present(db):
    base = _now()
    img, _ = await _seed_image_with_post(
        db, sha="f" * 64,
        image_created_at=base,
        post_date=base - timedelta(days=365),
        artist_name="Fred", external_post_id="F-1",
    )
    svc = GalleryService(db)
    payload = await svc.get_image_with_tags(img.id)
    assert payload["posted_at"] is not None
    # Image's own created_at is still surfaced separately.
    assert payload["created_at"] != payload["posted_at"]


@pytest.mark.asyncio
async def test_scroll_multi_tag_and(db):
    images = await _seed_images(db, 5)
    a = Tag(name="aa", kind=TagKind.general)
    b = Tag(name="bb", kind=TagKind.general)
    db.add_all([a, b])
    await db.flush()
    # images[0] carries BOTH tags; images[1] carries only a.
    await db.execute(image_tag.insert().values([
        {"image_record_id": images[0].id, "tag_id": a.id, "source": "manual"},
        {"image_record_id": images[0].id, "tag_id": b.id, "source": "manual"},
        {"image_record_id": images[1].id, "tag_id": a.id, "source": "manual"},
    ]))
    svc = GalleryService(db)
    both = await svc.scroll(cursor=None, limit=10, tag_ids=[a.id, b.id])
    assert [i.id for i in both.images] == [images[0].id]  # AND: only the both-tagged
    just_a = await svc.scroll(cursor=None, limit=10, tag_ids=[a.id])
    assert {i.id for i in just_a.images} == {images[0].id, images[1].id}


@pytest.mark.asyncio
async def test_scroll_tag_or_group(db):
    # #6: one OR-group — match images carrying AT LEAST ONE of the group's tags.
    images = await _seed_images(db, 3)
    a = Tag(name="ora", kind=TagKind.general)
    b = Tag(name="orb", kind=TagKind.general)
    db.add_all([a, b])
    await db.flush()
    await db.execute(image_tag.insert().values([
        {"image_record_id": images[0].id, "tag_id": a.id, "source": "manual"},
        {"image_record_id": images[1].id, "tag_id": b.id, "source": "manual"},
        # images[2] carries neither.
    ]))
    svc = GalleryService(db)
    page = await svc.scroll(cursor=None, limit=10, tag_or_groups=[[a.id, b.id]])
    assert {i.id for i in page.images} == {images[0].id, images[1].id}


@pytest.mark.asyncio
async def test_scroll_tag_or_groups_anded(db):
    # Two OR-groups AND together: image must satisfy (a OR b) AND (c).
    images = await _seed_images(db, 3)
    a = Tag(name="aaa", kind=TagKind.general)
    b = Tag(name="bbb", kind=TagKind.general)
    c = Tag(name="ccc", kind=TagKind.general)
    db.add_all([a, b, c])
    await db.flush()
    await db.execute(image_tag.insert().values([
        {"image_record_id": images[0].id, "tag_id": a.id, "source": "manual"},
        {"image_record_id": images[0].id, "tag_id": c.id, "source": "manual"},
        {"image_record_id": images[1].id, "tag_id": b.id, "source": "manual"},
        {"image_record_id": images[2].id, "tag_id": a.id, "source": "manual"},
    ]))
    svc = GalleryService(db)
    page = await svc.scroll(
        cursor=None, limit=10, tag_or_groups=[[a.id, b.id], [c.id]],
    )
    # Only images[0] has both an (a-or-b) tag AND c.
    assert [i.id for i in page.images] == [images[0].id]


@pytest.mark.asyncio
async def test_scroll_tag_exclude(db):
    # #6: exclude — drop images carrying ANY excluded tag.
    images = await _seed_images(db, 3)
    x = Tag(name="xxx", kind=TagKind.general)
    db.add(x)
    await db.flush()
    await db.execute(image_tag.insert().values(
        image_record_id=images[0].id, tag_id=x.id, source="manual",
    ))
    svc = GalleryService(db)
    page = await svc.scroll(cursor=None, limit=10, tag_exclude=[x.id])
    assert {i.id for i in page.images} == {images[1].id, images[2].id}


@pytest.mark.asyncio
async def test_scroll_or_and_exclude_compose(db):
    # The structured model composes: (a OR b) AND NOT x.
    images = await _seed_images(db, 3)
    a = Tag(name="ca", kind=TagKind.general)
    b = Tag(name="cb", kind=TagKind.general)
    x = Tag(name="cx", kind=TagKind.general)
    db.add_all([a, b, x])
    await db.flush()
    await db.execute(image_tag.insert().values([
        {"image_record_id": images[0].id, "tag_id": a.id, "source": "manual"},
        {"image_record_id": images[0].id, "tag_id": x.id, "source": "manual"},
        {"image_record_id": images[1].id, "tag_id": b.id, "source": "manual"},
    ]))
    svc = GalleryService(db)
    page = await svc.scroll(
        cursor=None, limit=10, tag_or_groups=[[a.id, b.id]], tag_exclude=[x.id],
    )
    # images[0] matches the OR-group but is excluded by x; images[1] survives.
    assert [i.id for i in page.images] == [images[1].id]


@pytest.mark.asyncio
async def test_require_single_filter_rejects_post_id_with_or_or_exclude(db):
    svc = GalleryService(db)
    with pytest.raises(ValueError, match="post_id cannot be combined"):
        await svc.scroll(cursor=None, limit=10, post_id=5, tag_or_groups=[[1]])
    with pytest.raises(ValueError, match="post_id cannot be combined"):
        await svc.scroll(cursor=None, limit=10, post_id=5, tag_exclude=[1])


@pytest.mark.asyncio
async def test_scroll_media_filter(db):
    imgs = await _seed_images(db, 2)  # image/jpeg
    vid = ImageRecord(
        path="/images/test/v.mp4", sha256="v" * 64, size_bytes=1,
        mime="video/mp4", width=1, height=1,
        origin="imported_filesystem", integrity_status="unknown",
    )
    vid.created_at = _now()
    vid.effective_date = vid.created_at
    db.add(vid)
    await db.flush()
    svc = GalleryService(db)
    vids = await svc.scroll(cursor=None, limit=10, media_type="video")
    assert [i.id for i in vids.images] == [vid.id]
    pics = await svc.scroll(cursor=None, limit=10, media_type="image")
    assert {i.id for i in pics.images} == {i.id for i in imgs}


@pytest.mark.asyncio
async def test_scroll_sort_oldest_reverses_order(db):
    images = await _seed_images(db, 4)  # images[0] newest ... images[3] oldest
    svc = GalleryService(db)
    newest = await svc.scroll(cursor=None, limit=10)
    oldest = await svc.scroll(cursor=None, limit=10, sort="oldest")
    newest_ids = [i.id for i in newest.images]
    assert newest_ids[0] == images[0].id
    assert [i.id for i in oldest.images] == list(reversed(newest_ids))