FabledCurator/backend/app/services/provenance_service.py

"""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,
    PostAttachment,
    Source,
    attachment_download_url,
)
from .post_body import render_post_body


async def _post_dict(session: AsyncSession, p: Post) -> dict:
    """One provenance entry's post.

    NOTE: the key names here deliberately differ from
    `PostFeedService._to_dict` (`url`/`title`/`date` vs
    `post_url`/`post_title`/`post_date`, `attachment_count` vs `attachments`),
    and `description_translated` is the FULL text here where the feed truncates
    it to DESCRIPTION_LIMIT. That divergence is issue #3965's wider half and is
    deliberately NOT addressed here — renaming is a breaking payload change for
    ProvenancePanel with no second reason to spend it.

    What IS fixed here is the body: this used to call `sanitize_post_html`
    alone, so provenance bodies hotlinked the platform CDN for images already
    on disk while the post detail view served local copies. `render_post_body`
    is the whole pipeline, so the two surfaces cannot drift apart again.
    """
    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": await render_post_body(session, p.description, p.artist_id),
        "attachment_count": p.attachment_count,
        # Translation (#143): the English title/description shown by default when
        # a translation exists; the UI toggles to the original. Source lang labels
        # the original.
        "title_translated": p.post_title_translated,
        "description_translated": p.description_translated,
        "translated_source_lang": p.translated_source_lang,
        "translation_override": p.translation_override,
    }


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}


def _attachment_dict(a: PostAttachment) -> dict:
    return {
        "id": a.id,
        "original_filename": a.original_filename,
        "size_bytes": a.size_bytes,
        "ext": a.ext,
        "download_url": attachment_download_url(a.id),
    }


class ProvenanceService:
    def __init__(self, session: AsyncSession):
        self.session = session

    async def _attachments_for_posts(self, post_ids: list[int]) -> list[dict]:
        if not post_ids:
            return []
        rows = (
            await self.session.execute(
                select(PostAttachment)
                .where(PostAttachment.post_id.in_(post_ids))
                .order_by(PostAttachment.id.asc())
            )
        ).scalars().all()
        return [_attachment_dict(a) for a in rows]

    async def _attachment_by_id(self, attachment_id: int) -> list[dict]:
        att = await self.session.get(PostAttachment, attachment_id)
        return [_attachment_dict(att)] if att is not None else []

    async def for_image(self, image_id: int) -> dict | None:
        rec = await self.session.get(ImageRecord, image_id)
        if rec is None:
            return None
        # Artist via Post.artist_id (alembic 0030); Source via LEFT JOIN
        # since both Post.source_id and ImageProvenance.source_id can be
        # NULL for filesystem-imported content. Frontend renders source=
        # null as "filesystem import."
        stmt = (
            select(ImageProvenance, Post, Source, Artist)
            .join(Post, Post.id == ImageProvenance.post_id)
            .join(Artist, Artist.id == Post.artist_id)
            .outerjoin(Source, Source.id == ImageProvenance.source_id)
            .where(ImageProvenance.image_record_id == image_id)
            .order_by(ImageProvenance.captured_at.asc(),
                      ImageProvenance.id.asc())
        )
        rows = (await self.session.execute(stmt)).all()
        post_ids = [ip.post_id for ip, _p, _s, _a in rows]
        # Prefer the EXACT archive this file came out of (milestone #87): if the
        # originating post's provenance row records from_attachment_id, the image
        # was extracted from that one .zip/.rar, so show only it — not the dozens
        # of unrelated archives a "High Resolution Files" bundle post carries.
        from_att_id = next(
            (
                ip.from_attachment_id
                for ip, _p, _s, _a in rows
                if ip.post_id == rec.primary_post_id
                and ip.from_attachment_id is not None
            ),
            None,
        )
        if from_att_id is not None:
            attachments = await self._attachment_by_id(from_att_id)
        else:
            # No recorded containing archive (loose download, or pre-backfill):
            # scope to the originating post only, not every pHash-linked post.
            # primary_post_id is the post this file was actually captured from;
            # fall back to all linked posts when it's unset (older rows /
            # filesystem imports).
            attach_post_ids = (
                [rec.primary_post_id]
                if rec.primary_post_id is not None
                else post_ids
            )
            attachments = await self._attachments_for_posts(attach_post_ids)
        return {
            "image_id": image_id,
            "provenance": [
                {
                    "provenance_id": ip.id,
                    "captured_at": ip.captured_at.isoformat()
                    if ip.captured_at else None,
                    "post": await _post_dict(self.session, post),
                    "source": _source_dict(src) if src is not None else None,
                    "artist": _artist_dict(art),
                }
                for ip, post, src, art in rows
            ],
            "attachments": attachments,
        }

    async def for_post(self, post_id: int) -> dict | None:
        # Same LEFT JOIN to Source — get_post must succeed for a
        # NULL-source post.
        stmt = (
            select(Post, Source, Artist)
            .join(Artist, Artist.id == Post.artist_id)
            .outerjoin(Source, Source.id == Post.source_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": await _post_dict(self.session, post),
            "source": _source_dict(src) if src is not None else None,
            "artist": _artist_dict(art),
            "attachments": await self._attachments_for_posts([post.id]),
        }