bvandeusenandClaude Opus 5.5
c2af70e6af
CI and images / lint (push) Successful in 2s
CI and images / extension-version (push) Successful in 2s
CI and images / extension-test (push) Successful in 21s
CI and images / frontend-build (push) Successful in 21s
CI and images / backend-lint-and-test (push) Successful in 45s
CI and images / integration (push) Successful in 2m31s
CI and images / sign-extension (push) Successful in 2s
CI and images / build-agent (push) Successful in 4s
CI and images / build-web (push) Successful in 2m1s
CI and images / smoke-web (push) Successful in 49s
CI and images / promote (push) Successful in 1s
CI and images / refresh-dev (push) Skipped
CI and images / unattended-failure (push) Skipped
Most archive attachments never had their images imported. The re-extract action skipped archives without a post, and nothing kept game builds out of an extraction. - An archive holding executables, libraries, or game-engine data (Ren'Py, Unity, Godot, RPG Maker, Python builds) is kept as an attachment and not extracted. The same holds for one nested inside an art pack. - Re-extract now reaches archives imported from a folder without a post. Their images go to the artist, in a folder named after the archive. - Re-extract passes over archives whose members are already recorded. - Re-extract stages each archive by symlink instead of copying it. - The card says what the action does now. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
308 lines
12 KiB
Python
308 lines
12 KiB
Python
"""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_archive_members_record_containing_archive(importer, import_layout):
|
||
"""Milestone #87: each extracted member's provenance row records the archive
|
||
PostAttachment it came out of, so provenance can show the one archive a file
|
||
lives inside."""
|
||
import_root, _ = import_layout
|
||
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"))
|
||
arc.with_suffix(".cbz.json").write_text(json.dumps(
|
||
{"category": "patreon", "id": "777", "title": "Set"}))
|
||
|
||
importer.import_one(arc)
|
||
att = importer.session.execute(select(PostAttachment)).scalar_one()
|
||
provs = importer.session.execute(select(ImageProvenance)).scalars().all()
|
||
assert len(provs) == 2
|
||
assert all(p.from_attachment_id == att.id for p in provs)
|
||
|
||
|
||
def test_nested_archive_member_records_outer_archive(importer, import_layout):
|
||
"""Nested members link to the OUTER stored archive — the only one persisted
|
||
as a PostAttachment (the inner archive lives in a tempdir)."""
|
||
import_root, _ = import_layout
|
||
importer.settings.phash_threshold = 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())
|
||
arc.with_suffix(".cbz.json").write_text(json.dumps(
|
||
{"category": "patreon", "id": "900", "title": "HR Pack"}))
|
||
|
||
importer.import_one(arc)
|
||
att = importer.session.execute(select(PostAttachment)).scalar_one()
|
||
provs = importer.session.execute(select(ImageProvenance)).scalars().all()
|
||
assert len(provs) == 2
|
||
assert all(p.from_attachment_id == att.id for p in provs)
|
||
|
||
|
||
def test_deduped_archive_member_records_each_posts_archive(importer, import_layout):
|
||
"""A member re-shipped in a second post's archive gets its NEW provenance row
|
||
(for the second post) stamped with the SECOND archive — the UPDATE-of-existing
|
||
path the reextract backfill relies on. Each post's rows point at its own
|
||
archive."""
|
||
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"}))
|
||
importer.import_one(first)
|
||
|
||
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"}))
|
||
importer.import_one(second)
|
||
|
||
posts = {p.external_post_id: p.id for p in
|
||
importer.session.execute(select(Post)).scalars().all()}
|
||
atts = {a.post_id: a.id for a in
|
||
importer.session.execute(select(PostAttachment)).scalars().all()}
|
||
provs = importer.session.execute(select(ImageProvenance)).scalars().all()
|
||
assert len(provs) == 4 # 2 images × 2 posts
|
||
for p in provs:
|
||
assert p.from_attachment_id == atts[p.post_id]
|
||
assert atts[posts["111"]] != atts[posts["222"]]
|
||
|
||
|
||
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
|
||
|
||
|
||
def test_a_game_build_is_kept_whole_and_its_images_left_alone(importer, import_layout):
|
||
import_root, _ = import_layout
|
||
importer.settings.phash_threshold = 0
|
||
arc = import_root / "Bob" / "MyGame-1.0-pc.zip"
|
||
arc.parent.mkdir(parents=True, exist_ok=True)
|
||
with zipfile.ZipFile(arc, "w") as zf:
|
||
zf.writestr("MyGame/lib/py3-windows-x86_64/python.exe", b"MZ")
|
||
zf.writestr("MyGame/game/images/cg01.jpg", _split_bytes("v"))
|
||
zf.writestr("MyGame/game/script.rpyc", b"\x00")
|
||
|
||
r = importer.import_one(arc)
|
||
assert r.status == "attached"
|
||
assert importer.session.execute(select(func.count()).select_from(ImageRecord)).scalar_one() == 0
|
||
att = importer.session.execute(select(PostAttachment)).scalar_one()
|
||
assert att.original_filename == "MyGame-1.0-pc.zip"
|
||
|
||
|
||
def test_a_game_build_inside_an_art_pack_is_skipped_and_the_art_imported(importer, import_layout):
|
||
import_root, _ = import_layout
|
||
importer.settings.phash_threshold = 0
|
||
game = io.BytesIO()
|
||
with zipfile.ZipFile(game, "w") as zf:
|
||
zf.writestr("Game_Data/sharedassets0.assets", b"\x00")
|
||
zf.writestr("Game_Data/sprite.jpg", _split_bytes("h"))
|
||
arc = import_root / "Bob" / "pack.zip"
|
||
arc.parent.mkdir(parents=True, exist_ok=True)
|
||
with zipfile.ZipFile(arc, "w") as zf:
|
||
zf.writestr("art.jpg", _split_bytes("v"))
|
||
zf.writestr("bonus/game.zip", game.getvalue())
|
||
|
||
r = importer.import_one(arc)
|
||
assert r.status == "imported"
|
||
assert len(r.member_image_ids) == 1
|
||
assert importer.session.execute(select(func.count()).select_from(ImageRecord)).scalar_one() == 1
|