bvandeusen
91be9df671
attach_in_place mirrored only the media flow, so gallery-dl-downloaded zips/PDFs/audio bounced back as `skipped+invalid_image`, which download_service counted as an ingest error and flipped runs to status="error" despite N successful image attaches. Lustria patreon event #38998 (21 images + 1 OST zip) went red for exactly this reason. Now attach_in_place dispatches the same way as import_one: archives → _import_archive (extracts media members, captures archive as PostAttachment), non-media → _capture_attachment. Download_service accepts the new `attached` result and treats non-duplicate skips as soft skips, not ingest errors. Also: ShowcaseView always loadInitial() on mount, not just when the store is empty — Pinia persists across navigations and operator wants a fresh shuffle every time the showcase loads.
189 lines
6.3 KiB
Python
189 lines
6.3 KiB
Python
"""attach_in_place — Importer seam used by FC-3c (downloader)."""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from backend.app.models import Artist, ImageRecord, ImportSettings
|
|
from backend.app.services.importer import Importer
|
|
from backend.app.services.thumbnailer import Thumbnailer
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
def _make_jpg(path: Path, color=(255, 0, 0), size=(64, 64)) -> bytes:
|
|
from PIL import Image
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
Image.new("RGB", size, color).save(path, "JPEG")
|
|
return path.read_bytes()
|
|
|
|
|
|
@pytest.fixture
|
|
def importer(db_sync, tmp_path):
|
|
settings = db_sync.execute(
|
|
select(ImportSettings).where(ImportSettings.id == 1)
|
|
).scalar_one()
|
|
images_root = tmp_path / "images"
|
|
images_root.mkdir()
|
|
return Importer(
|
|
session=db_sync,
|
|
images_root=images_root,
|
|
import_root=tmp_path / "ignored",
|
|
thumbnailer=Thumbnailer(images_root=images_root),
|
|
settings=settings,
|
|
)
|
|
|
|
|
|
def test_attach_in_place_imports_happy(importer, db_sync):
|
|
images_root = importer.images_root
|
|
artist = Artist(name="Alice", slug="alice")
|
|
db_sync.add(artist)
|
|
db_sync.flush()
|
|
|
|
target = images_root / "alice" / "patreon" / "post1" / "img.jpg"
|
|
_make_jpg(target)
|
|
|
|
result = importer.attach_in_place(target, artist=artist)
|
|
assert result.status == "imported"
|
|
assert result.image_id is not None
|
|
|
|
rec_path = db_sync.execute(
|
|
select(ImageRecord.path, ImageRecord.origin, ImageRecord.artist_id)
|
|
.where(ImageRecord.id == result.image_id)
|
|
).one()
|
|
assert rec_path.path == str(target)
|
|
assert rec_path.origin == "downloaded"
|
|
assert rec_path.artist_id == artist.id
|
|
|
|
|
|
def test_attach_in_place_with_sidecar(importer, db_sync):
|
|
images_root = importer.images_root
|
|
artist = Artist(name="Bob", slug="bob")
|
|
db_sync.add(artist)
|
|
db_sync.flush()
|
|
|
|
target = images_root / "bob" / "patreon" / "post1" / "img.jpg"
|
|
_make_jpg(target)
|
|
sidecar = target.with_suffix(target.suffix + ".json")
|
|
sidecar.write_text(
|
|
'{"category": "patreon", "id": "12345", "url": "https://patreon.com/post/12345", '
|
|
'"title": "Test Post", "published_at": "2026-05-01T00:00:00+00:00"}'
|
|
)
|
|
|
|
result = importer.attach_in_place(target, artist=artist)
|
|
assert result.status == "imported"
|
|
|
|
primary_post_id = db_sync.execute(
|
|
select(ImageRecord.primary_post_id).where(ImageRecord.id == result.image_id)
|
|
).scalar_one()
|
|
assert primary_post_id is not None # post created from sidecar
|
|
|
|
|
|
def test_attach_in_place_sha256_dedup_does_not_modify_disk(importer, db_sync):
|
|
images_root = importer.images_root
|
|
artist = Artist(name="Cara", slug="cara")
|
|
db_sync.add(artist)
|
|
db_sync.flush()
|
|
|
|
first = images_root / "cara" / "p" / "a.jpg"
|
|
bytes_ = _make_jpg(first)
|
|
importer.attach_in_place(first, artist=artist)
|
|
|
|
second = images_root / "cara" / "p" / "b.jpg"
|
|
second.parent.mkdir(parents=True, exist_ok=True)
|
|
second.write_bytes(bytes_)
|
|
assert second.exists()
|
|
|
|
result = importer.attach_in_place(second, artist=artist)
|
|
assert result.status == "skipped"
|
|
assert result.skip_reason.value == "duplicate_hash"
|
|
# Caller is responsible for deleting the on-disk file; we don't touch it.
|
|
assert second.exists()
|
|
|
|
|
|
def test_attach_in_place_supersede_keeps_file_at_new_path(importer, db_sync):
|
|
"""phash near-dup where the new file is LARGER → existing record
|
|
swaps to the new path; old file deleted."""
|
|
images_root = importer.images_root
|
|
artist = Artist(name="Dee", slug="dee")
|
|
db_sync.add(artist)
|
|
db_sync.flush()
|
|
|
|
smaller = images_root / "dee" / "p" / "small.jpg"
|
|
_make_jpg(smaller, color=(50, 50, 200), size=(64, 64))
|
|
first = importer.attach_in_place(smaller, artist=artist)
|
|
assert first.status == "imported"
|
|
|
|
bigger = images_root / "dee" / "p" / "big.jpg"
|
|
_make_jpg(bigger, color=(50, 50, 200), size=(256, 256))
|
|
|
|
result = importer.attach_in_place(bigger, artist=artist)
|
|
assert result.status == "superseded"
|
|
|
|
new_path = db_sync.execute(
|
|
select(ImageRecord.path).where(ImageRecord.id == first.image_id)
|
|
).scalar_one()
|
|
assert new_path == str(bigger)
|
|
assert not smaller.exists()
|
|
assert bigger.exists()
|
|
|
|
|
|
def test_attach_in_place_phash_smaller_skips_without_modifying_disk(
|
|
importer, db_sync
|
|
):
|
|
images_root = importer.images_root
|
|
artist = Artist(name="Eve", slug="eve")
|
|
db_sync.add(artist)
|
|
db_sync.flush()
|
|
|
|
bigger = images_root / "eve" / "p" / "big.jpg"
|
|
_make_jpg(bigger, color=(0, 200, 0), size=(256, 256))
|
|
importer.attach_in_place(bigger, artist=artist)
|
|
|
|
smaller = images_root / "eve" / "p" / "small.jpg"
|
|
_make_jpg(smaller, color=(0, 200, 0), size=(64, 64))
|
|
|
|
result = importer.attach_in_place(smaller, artist=artist)
|
|
assert result.status == "skipped"
|
|
assert result.skip_reason.value == "duplicate_phash"
|
|
assert smaller.exists()
|
|
|
|
|
|
def test_attach_in_place_invalid_image_returns_skipped(importer):
|
|
images_root = importer.images_root
|
|
bad = images_root / "fred" / "bad.jpg"
|
|
bad.parent.mkdir(parents=True, exist_ok=True)
|
|
bad.write_bytes(b"not a jpeg, just some bytes")
|
|
|
|
result = importer.attach_in_place(bad)
|
|
assert result.status == "skipped"
|
|
assert result.skip_reason.value == "invalid_image"
|
|
|
|
|
|
def test_attach_in_place_non_media_routes_to_attachment(importer, db_sync):
|
|
"""FC-2d-iii dispatch parity for the download path. A non-media file
|
|
(.pdf, .txt etc.) downloaded by gallery-dl must become a PostAttachment,
|
|
not bounce back as `skipped+invalid_image` — the latter flipped
|
|
otherwise-successful runs to status="error" downstream. Operator-flagged
|
|
2026-06-02 (Lustria patreon OST zip)."""
|
|
from backend.app.models import PostAttachment
|
|
|
|
images_root = importer.images_root
|
|
artist = Artist(name="Gus", slug="gus")
|
|
db_sync.add(artist)
|
|
db_sync.flush()
|
|
|
|
txt = images_root / "gus" / "patreon" / "post" / "notes.txt"
|
|
txt.parent.mkdir(parents=True, exist_ok=True)
|
|
txt.write_bytes(b"some non-media payload that the importer should preserve")
|
|
|
|
result = importer.attach_in_place(txt, artist=artist)
|
|
assert result.status == "attached"
|
|
|
|
row = db_sync.execute(
|
|
select(PostAttachment).where(PostAttachment.original_filename == "notes.txt")
|
|
).scalar_one()
|
|
assert row.ext == ".txt"
|
|
assert row.artist_id == artist.id
|