85 lines
2.8 KiB
Python
85 lines
2.8 KiB
Python
"""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
|
|
],
|
|
}
|
|
|
|
async def for_post(self, post_id: int) -> dict | None:
|
|
stmt = (
|
|
select(Post, Source, Artist)
|
|
.join(Source, Source.id == Post.source_id)
|
|
.join(Artist, Artist.id == Source.artist_id)
|
|
.where(Post.id == post_id)
|
|
)
|
|
row = (await self.session.execute(stmt)).first()
|
|
if row is None:
|
|
return None
|
|
post, src, art = row
|
|
return {
|
|
"post": _post_dict(post),
|
|
"source": _source_dict(src),
|
|
"artist": _artist_dict(art),
|
|
}
|