FabledCurator/tests/test_importer_archive.py

"""FC-2d-iii: archive → import media members + store the archive."""

import io
import json
import zipfile

import pytest
from PIL import Image
from sqlalchemy import func, select

from backend.app.models import (
    ImageProvenance,
    ImageRecord,
    ImportSettings,
    Post,
    PostAttachment,
)
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 _jpeg_bytes(color):
    buf = io.BytesIO()
    Image.new("RGB", (40, 40), color).save(buf, "JPEG")
    return buf.getvalue()


def _split_bytes(orient):
    """Structured half/half image — solid colors phash-collapse (distance
    0); orthogonal splits are distinct only at phash_threshold=0 (see
    reference-phash-test-images)."""
    im = Image.new("L", (256, 256), 0)
    px = im.load()
    for y in range(256):
        for x in range(256):
            if (x / 256 if orient == "v" else y / 256) >= 0.5:
                px[x, y] = 255
    buf = io.BytesIO()
    im.convert("RGB").save(buf, "JPEG")
    return buf.getvalue()


def test_archive_imports_members_and_stores_archive(importer, import_layout):
    import_root, _ = import_layout
    # Distinct structure + threshold 0 so BOTH members import (solid
    # colors would phash-collapse; reference-phash-test-images).
    importer.settings.phash_threshold = 0
    arc = import_root / "Bob" / "set.cbz"
    arc.parent.mkdir(parents=True, exist_ok=True)
    with zipfile.ZipFile(arc, "w") as zf:
        zf.writestr("a.jpg", _split_bytes("v"))
        zf.writestr("b.jpg", _split_bytes("h"))
        zf.writestr("readme.txt", b"hello")
    arc.with_suffix(".cbz.json").write_text(json.dumps(
        {"category": "patreon", "id": "777", "title": "Set"}))

    r = importer.import_one(arc)
    assert r.status == "imported"
    assert len(r.member_image_ids) == 2

    imgs = importer.session.execute(select(ImageRecord)).scalars().all()
    assert len(imgs) == 2
    post = importer.session.execute(select(Post)).scalar_one()
    provs = importer.session.execute(
        select(func.count()).select_from(ImageProvenance)
    ).scalar_one()
    assert provs == 2  # one Post, both members
    att = importer.session.execute(select(PostAttachment)).scalar_one()
    assert att.original_filename == "set.cbz"
    assert att.post_id == post.id


def test_nested_archive_members_imported(importer, import_layout):
    """#718: a 'high-res' pack that wraps a nested archive must still import the
    nested images and link them to the post — not silently drop them (incase's
    per-chapter .rar inside an outer 'High Resolution files' pack)."""
    import_root, _ = import_layout
    importer.settings.phash_threshold = 0  # v/h splits are distinct only at 0
    inner = io.BytesIO()
    with zipfile.ZipFile(inner, "w") as zf:
        zf.writestr("p1.jpg", _split_bytes("v"))
        zf.writestr("p2.jpg", _split_bytes("h"))
    arc = import_root / "Nessie" / "hr.cbz"
    arc.parent.mkdir(parents=True, exist_ok=True)
    with zipfile.ZipFile(arc, "w") as zf:
        zf.writestr("chapter.zip", inner.getvalue())  # nested archive, no direct media
    arc.with_suffix(".cbz.json").write_text(json.dumps(
        {"category": "patreon", "id": "900", "title": "HR Pack"}))

    r = importer.import_one(arc)
    assert r.status == "imported"
    assert len(r.member_image_ids) == 2  # both nested images surfaced

    assert importer.session.execute(
        select(func.count()).select_from(ImageRecord)
    ).scalar_one() == 2
    post = importer.session.execute(select(Post)).scalar_one()
    provs = importer.session.execute(
        select(func.count()).select_from(ImageProvenance)
    ).scalar_one()
    assert provs == 2  # both nested members linked to the one outer post
    # The outer pack is still preserved as an attachment.
    att = importer.session.execute(select(PostAttachment)).scalar_one()
    assert att.original_filename == "hr.cbz"
    assert att.post_id == post.id


def test_archive_all_deduped_is_benign_not_flagged(importer, import_layout):
    """#718 reason-string fix: when every image in an archive already exists
    (cross-posted), the images re-link to the new post and the archive is NOT
    flagged as an unextracted-archive problem (no error set)."""
    import_root, _ = import_layout
    importer.settings.phash_threshold = 0
    pv, ph = _split_bytes("v"), _split_bytes("h")
    first = import_root / "Dup" / "a.cbz"
    first.parent.mkdir(parents=True, exist_ok=True)
    with zipfile.ZipFile(first, "w") as zf:
        zf.writestr("a.jpg", pv)
        zf.writestr("b.jpg", ph)
    first.with_suffix(".cbz.json").write_text(json.dumps(
        {"category": "patreon", "id": "111", "title": "First"}))
    assert importer.import_one(first).status == "imported"

    second = import_root / "Dup" / "b.cbz"  # different post, same two images
    with zipfile.ZipFile(second, "w") as zf:
        zf.writestr("a.jpg", pv)
        zf.writestr("b.jpg", ph)
    second.with_suffix(".cbz.json").write_text(json.dumps(
        {"category": "patreon", "id": "222", "title": "Second"}))

    r = importer.import_one(second)
    assert r.status == "attached"
    assert r.error is None  # benign all-deduped — NOT a flagged problem

    assert importer.session.execute(
        select(func.count()).select_from(ImageRecord)
    ).scalar_one() == 2  # no new records
    provs = importer.session.execute(
        select(func.count()).select_from(ImageProvenance)
    ).scalar_one()
    assert provs == 4  # 2 images × 2 posts (enrich-on-duplicate)


def test_corrupt_archive_still_stored(importer, import_layout):
    import_root, _ = import_layout
    arc = import_root / "Bob" / "broken.zip"
    arc.parent.mkdir(parents=True, exist_ok=True)
    arc.write_bytes(b"definitely not a zip")
    r = importer.import_one(arc)
    assert r.status == "attached"
    att = importer.session.execute(select(PostAttachment)).scalar_one()
    assert att.original_filename == "broken.zip"
    assert importer.session.execute(
        select(func.count()).select_from(ImageRecord)
    ).scalar_one() == 0


def test_reimport_archive_is_idempotent(importer, import_layout):
    import_root, _ = import_layout
    arc = import_root / "Bob" / "set.zip"
    arc.parent.mkdir(parents=True, exist_ok=True)
    with zipfile.ZipFile(arc, "w") as zf:
        zf.writestr("a.jpg", _jpeg_bytes((1, 2, 3)))
    importer.import_one(arc)
    importer.import_one(arc)
    assert importer.session.execute(
        select(func.count()).select_from(PostAttachment)
    ).scalar_one() == 1