pigway-frame-flow/tests/test_photo_captions.py

from io import BytesIO

import pytest
from PIL import Image

from pigway_frame_flow.providers.photo_names import folder_caption, path_caption
from pigway_frame_flow.providers.photo_sources import PhotoSourceInput
from pigway_frame_flow.renderers.photo_caption import render_captioned_photo


@pytest.mark.parametrize(
    "name,title,day,place",
    [
        ("20141207 婚礼@示例酒店", "婚礼", "2014-12-07", "示例酒店"),
        ("2026-10-05 家庭出游 @ 示例地点", "家庭出游", "2026-10-05", "示例地点"),
        ("2026.10.05_周末＠家", "周末", "2026-10-05", "家"),
        ("2026年10月05日 国庆假期", "国庆假期", "2026-10-05", ""),
        ("20041015 军训合影", "军训合影", "2004-10-15", ""),
        ("20241005 主题@地点@补充说明", "主题", "2024-10-05", "地点@补充说明"),
    ],
)
def test_folder_three_line_caption(name, title, day, place):
    assert folder_caption(name) == {"title": title, "date": day, "place": place}


@pytest.mark.parametrize(
    "name",
    [
        "MobileBackup",
        "20260230 无效日期",
        "20261301 无效月份",
        "20261005123 数字不是日期",
    ],
)
def test_non_date_folders_are_not_curated(name):
    assert folder_caption(name) is None


def test_child_folder_inherits_date_and_location():
    assert path_caption("20141207 婚礼@示例酒店/精选/IMG_001.jpg") == {
        "title": "婚礼",
        "date": "2014-12-07",
        "place": "示例酒店",
    }
    assert path_caption("MobileBackup/20260101.jpg") is None


@pytest.mark.parametrize("profile", ["color", "grayscale"])
def test_native_caption_layout_has_single_watermark_line(service, profile):
    out = BytesIO()
    Image.new("RGB", (1600, 900), "blue").save(out, "JPEG")
    result = render_captioned_photo(
        out.getvalue(),
        folder_caption("20141207 婚礼@示例酒店"),
        service.renderer.font,
        profile=profile,
    )
    with Image.open(BytesIO(result)) as image:
        assert image.size == (1200, 1600)
        assert image.crop((30, 1528, 1170, 1600)).convert("L").getextrema()[0] < 50
        assert image.getpixel((20, 1500))[0] < 230
        assert image.getpixel((20, 1590))[0] > 240
        pixel = image.getpixel((600, 640))
        assert (
            len(set(pixel)) <= 2
            if profile == "grayscale"
            else pixel[2] > pixel[0] + 100
        )


def test_source_filters_unorganized_and_reports_caption(client, service, settings):
    root = settings.data_dir.parent / "photos"
    settings.photo_root = root
    for folder in ["20261005 秋游@金山/精选", "MobileBackup", "20260230 无效日期"]:
        path = root / folder / "photo.jpg"
        path.parent.mkdir(parents=True, exist_ok=True)
        Image.new("RGB", (100, 100), "red").save(path)
    source = PhotoSourceInput(
        name="Curated",
        kind="local",
        location=".",
        organized_only=True,
        folder_caption=True,
        catalog_limit=100000,
    )
    service.photo_sources.save("curated", source)
    r = client.get("/api/v1/photo-sources/curated/images")
    assert r.status_code == 200
    data = r.json()
    assert data["total"] == 1 and data["catalog_limit"] == 100000
    assert data["items"][0]["caption"] == {
        "title": "秋游",
        "date": "2026-10-05",
        "place": "金山",
    }
    accepted = client.post(
        "/api/v1/devices/living-room/upload",
        json={"type": "photo_source", "source_id": "curated"},
    )
    assert accepted.status_code == 202 and service.process_one()
    op = client.get("/api/v1/operations/" + accepted.json()["operation_id"]).json()
    assert op["status"] == "COMPLETED"
    with Image.open(BytesIO(client.get(op["result"]["url"]).content)) as image:
        assert image.size == (1600, 1200)


def test_curated_catalog_includes_photos_beyond_old_ten_thousand_limit(
    service, settings
):
    root = settings.data_dir.parent / "many-photos"
    settings.photo_root = root
    folder = root / "20261005 旅行@示例地点"
    folder.mkdir(parents=True)
    for index in range(10001):
        (folder / f"{index:05}.jpg").write_bytes(b"catalog metadata fixture")
    service.photo_sources.save(
        "all",
        PhotoSourceInput(
            name="All curated",
            kind="local",
            location=".",
            organized_only=True,
            folder_caption=True,
            catalog_limit=100000,
        ),
    )
    assert len(service.photo_sources.catalog("all")) == 10001


def test_oversized_synology_photo_uses_read_only_cached_preview(service, settings):
    root = settings.data_dir.parent / "synology"
    settings.photo_root = root
    settings.max_bytes = 1000
    original = root / "20261005 出游@金山" / "large.jpg"
    original.parent.mkdir(parents=True)
    original.write_bytes(b"x" * 1001)
    preview = original.parent / "@eaDir" / original.name / "SYNOPHOTO_THUMB_XL.jpg"
    preview.parent.mkdir(parents=True)
    Image.new("RGB", (20, 20), "blue").save(preview)
    source = PhotoSourceInput(
        name="Synology",
        kind="local",
        location=".",
        organized_only=True,
        folder_caption=True,
        synology_previews=True,
    )
    service.photo_sources.save("synology", source)
    items = service.photo_sources.catalog("synology")
    assert len(items) == 1 and items[0]["provider_preview"]
    data, name, caption = service.photo_sources.read("synology")
    assert (
        data == preview.read_bytes()
        and name == "large.jpg"
        and caption["place"] == "金山"
    )
    assert original.read_bytes() == b"x" * 1001
    service.photo_sources.save(
        "synology", source.model_copy(update={"synology_previews": False})
    )
    assert service.photo_sources.catalog("synology") == []