diff --git a/backend/app/services/provenance_service.py b/backend/app/services/provenance_service.py new file mode 100644 index 0000000..52f7cd7 --- /dev/null +++ b/backend/app/services/provenance_service.py @@ -0,0 +1,67 @@ +"""Read-only provenance queries. + +Provenance is its own system, intentionally separate from the tag/ML +system (see project_provenance_separation). This service joins +ImageProvenance -> Post/Source/Artist and returns plain dicts. It never +mutates and never imports tag/ML modules. +""" + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from ..models import Artist, ImageProvenance, ImageRecord, Post, Source +from ..utils.html_sanitize import sanitize_post_html + + +def _post_dict(p: Post) -> dict: + return { + "id": p.id, + "external_post_id": p.external_post_id, + "url": p.post_url, + "title": p.post_title, + "date": p.post_date.isoformat() if p.post_date else None, + "description_html": sanitize_post_html(p.description), + "attachment_count": p.attachment_count, + } + + +def _source_dict(s: Source) -> dict: + return {"id": s.id, "platform": s.platform, "url": s.url} + + +def _artist_dict(a: Artist) -> dict: + return {"id": a.id, "name": a.name, "slug": a.slug} + + +class ProvenanceService: + def __init__(self, session: AsyncSession): + self.session = session + + async def for_image(self, image_id: int) -> dict | None: + rec = await self.session.get(ImageRecord, image_id) + if rec is None: + return None + stmt = ( + select(ImageProvenance, Post, Source, Artist) + .join(Post, Post.id == ImageProvenance.post_id) + .join(Source, Source.id == ImageProvenance.source_id) + .join(Artist, Artist.id == Source.artist_id) + .where(ImageProvenance.image_record_id == image_id) + .order_by(ImageProvenance.captured_at.asc(), + ImageProvenance.id.asc()) + ) + rows = (await self.session.execute(stmt)).all() + return { + "image_id": image_id, + "provenance": [ + { + "provenance_id": ip.id, + "captured_at": ip.captured_at.isoformat() + if ip.captured_at else None, + "post": _post_dict(post), + "source": _source_dict(src), + "artist": _artist_dict(art), + } + for ip, post, src, art in rows + ], + } diff --git a/tests/test_provenance_service.py b/tests/test_provenance_service.py new file mode 100644 index 0000000..4f7987e --- /dev/null +++ b/tests/test_provenance_service.py @@ -0,0 +1,131 @@ +from datetime import UTC, datetime + +import pytest + +from backend.app.models import ( + Artist, + ImageProvenance, + ImageRecord, + Post, + Source, +) +from backend.app.services.provenance_service import ProvenanceService + +pytestmark = pytest.mark.integration + + +async def _seed_image(db, sha="a" + "0" * 63) -> ImageRecord: + rec = ImageRecord( + path=f"/images/test/{sha}.jpg", + sha256=sha, + size_bytes=1, mime="image/jpeg", width=1, height=1, + origin="imported_filesystem", integrity_status="unknown", + ) + db.add(rec) + await db.flush() + return rec + + +async def _seed_post(db, *, artist_name, slug, platform, ext_id, + title=None, desc=None, count=None) -> tuple: + artist = Artist(name=artist_name, slug=slug) + db.add(artist) + await db.flush() + source = Source(artist_id=artist.id, platform=platform, + url=f"https://{platform}.test/{slug}") + db.add(source) + await db.flush() + post = Post( + source_id=source.id, external_post_id=ext_id, + post_url=f"https://{platform}.test/p/{ext_id}", + post_title=title, post_date=datetime(2023, 8, 1, tzinfo=UTC), + description=desc, attachment_count=count, + ) + db.add(post) + await db.flush() + return artist, source, post + + +@pytest.mark.asyncio +async def test_for_image_missing_returns_none(db): + svc = ProvenanceService(db) + assert await svc.for_image(999999) is None + + +@pytest.mark.asyncio +async def test_for_image_no_provenance_returns_empty_list(db): + rec = await _seed_image(db) + svc = ProvenanceService(db) + payload = await svc.for_image(rec.id) + assert payload == {"image_id": rec.id, "provenance": []} + + +@pytest.mark.asyncio +async def test_for_image_single_provenance_full_shape(db): + rec = await _seed_image(db) + artist, source, post = await _seed_post( + db, artist_name="Alice", slug="alice", platform="patreon", + ext_id="555", title="Set 1", desc="
hi
", + count=2, + ) + db.add(ImageProvenance(image_record_id=rec.id, post_id=post.id, + source_id=source.id)) + await db.flush() + + svc = ProvenanceService(db) + payload = await svc.for_image(rec.id) + assert payload["image_id"] == rec.id + assert len(payload["provenance"]) == 1 + e = payload["provenance"][0] + assert e["post"]["id"] == post.id + assert e["post"]["external_post_id"] == "555" + assert e["post"]["title"] == "Set 1" + assert e["post"]["attachment_count"] == 2 + assert e["post"]["description_html"] == "hi
" # script removed + assert e["post"]["url"] == "https://patreon.test/p/555" + assert e["post"]["date"].startswith("2023-08-01") + assert e["source"] == {"id": source.id, "platform": "patreon", + "url": source.url} + assert e["artist"] == {"id": artist.id, "name": "Alice", + "slug": "alice"} + assert e["provenance_id"] is not None + assert e["captured_at"] is not None + + +@pytest.mark.asyncio +async def test_for_image_multiple_provenance_peer_ordering(db): + rec = await _seed_image(db) + _, s1, p1 = await _seed_post(db, artist_name="A1", slug="a1", + platform="patreon", ext_id="1") + _, s2, p2 = await _seed_post(db, artist_name="A2", slug="a2", + platform="fanbox", ext_id="2") + ip1 = ImageProvenance(image_record_id=rec.id, post_id=p1.id, + source_id=s1.id) + ip2 = ImageProvenance(image_record_id=rec.id, post_id=p2.id, + source_id=s2.id) + db.add(ip1) + await db.flush() + db.add(ip2) + await db.flush() + + svc = ProvenanceService(db) + payload = await svc.for_image(rec.id) + ids = [e["provenance_id"] for e in payload["provenance"]] + assert ids == sorted(ids) # captured_at, id ascending → insertion order + assert len(payload["provenance"]) == 2 + + +@pytest.mark.asyncio +async def test_for_image_null_post_fields_serialize_null(db): + rec = await _seed_image(db) + _, source, post = await _seed_post( + db, artist_name="Bob", slug="bob", platform="x", ext_id="9", + ) # title/desc/count/post_url default-ish + db.add(ImageProvenance(image_record_id=rec.id, post_id=post.id, + source_id=source.id)) + await db.flush() + svc = ProvenanceService(db) + e = (await svc.for_image(rec.id))["provenance"][0] + assert e["post"]["title"] is None + assert e["post"]["description_html"] is None + assert e["post"]["attachment_count"] is None