FabledCurator/backend/app/services/series_match_service.py

"""FC-6.3 — assisted continuation matcher.

Scores a (post, candidate series) pair from several weighted signals; above the
configured threshold it records a *pending* SeriesSuggestion. Confirm-only — the
operator accepts (post becomes a chapter) or dismisses. No single signal gates;
the score is an additive weighted sum, each signal a 0..1 strength.

The learned "title pattern" isn't persisted — it's derived on the fly from the
post titles already in a series, so it sharpens automatically as more posts are
confirmed into the series.
"""

import re
import time

from sqlalchemy import and_, func, select
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession

from ..models import ImageRecord, Post, Tag, TagKind
from ..models.series_page import SeriesPage
from ..models.series_suggestion import SeriesSuggestion
from ..models.tag import image_tag
from .page_number_parser import parse_page_range
from .series_service import SeriesError, SeriesService

# Additive signal weights (sum to 1.0 → max score 1.0). Kept as constants;
# only the on/off + threshold are operator-tunable (sensitivity is the knob
# that matters; per-signal weights are an over-tune for v1).
WEIGHTS = {"title": 0.40, "artist": 0.20, "pages": 0.25, "tags": 0.15}

_DISTINCTIVE_KINDS = (TagKind.character, TagKind.series, TagKind.fandom)
_MAX_CANDIDATES = 50

# Strip installment markers so titles collapse to their stable stem.
_PAGE_TOKEN = re.compile(r"\b(?:pages?|pgs?|pp\.?)\s*\d+(?:\s*[-–—/]\s*\d+)?", re.I)
_BRACKET_NUM = re.compile(r"[\[(]\s*\d+\s*(?:/\s*\d+)?\s*[\])]")
_TRAILING_NUM = re.compile(r"[\s\-_#]*\d+\s*$")


def normalize_title(title: str | None) -> str:
    if not title:
        return ""
    t = _PAGE_TOKEN.sub("", title)
    t = _BRACKET_NUM.sub("", t)
    t = _TRAILING_NUM.sub("", t)
    return " ".join(t.split()).strip().lower()


def _common_prefix(strings: list[str]) -> str:
    strings = [s for s in strings if s]
    if not strings:
        return ""
    pre = strings[0]
    for s in strings[1:]:
        i = 0
        while i < len(pre) and i < len(s) and pre[i] == s[i]:
            i += 1
        pre = pre[:i]
        if not pre:
            break
    return pre.strip()


def title_signal(series_titles: list[str], post_title: str | None) -> float:
    """Overlap of the post title against the series' title stem (the longest
    common prefix of its known titles, page/installment markers removed)."""
    norm = [normalize_title(t) for t in series_titles]
    norm = [t for t in norm if t]
    pt = normalize_title(post_title)
    if not norm or not pt:
        return 0.0
    stem = _common_prefix(norm)
    if len(stem) < 3:
        stem = max(norm, key=len)
    i = 0
    while i < len(stem) and i < len(pt) and stem[i] == pt[i]:
        i += 1
    return min(1.0, i / max(len(stem), 4))


def pages_signal(series_max_stated_end: int | None, post_start: int | None) -> float:
    """1.0 when the post's first page continues right after the series' last
    stated page; partial for a near-continuation; 0 otherwise."""
    if series_max_stated_end is None or post_start is None:
        return 0.0
    diff = post_start - series_max_stated_end
    if diff == 1:
        return 1.0
    if 2 <= diff <= 3:
        return 0.5
    return 0.0


def tags_signal(shared_distinctive: int) -> float:
    if shared_distinctive <= 0:
        return 0.0
    return min(1.0, shared_distinctive / 3.0)


def weighted_score(signals: dict) -> float:
    return round(sum(WEIGHTS[k] * signals.get(k, 0.0) for k in WEIGHTS), 4)


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

    async def _post_image_ids(self, post_id: int) -> list[int]:
        rows = (
            await self.session.execute(
                select(ImageRecord.id).where(ImageRecord.primary_post_id == post_id)
            )
        ).scalars().all()
        return list(rows)

    async def _series_image_ids(self, series_tag_id: int) -> set[int]:
        rows = (
            await self.session.execute(
                select(SeriesPage.image_id).where(
                    SeriesPage.series_tag_id == series_tag_id
                )
            )
        ).scalars().all()
        return set(rows)

    async def _series_post_titles(self, series_tag_id: int) -> list[str]:
        rows = (
            await self.session.execute(
                select(Post.post_title)
                .select_from(SeriesPage)
                .join(ImageRecord, ImageRecord.id == SeriesPage.image_id)
                .join(Post, Post.id == ImageRecord.primary_post_id)
                .where(
                    and_(
                        SeriesPage.series_tag_id == series_tag_id,
                        Post.post_title.isnot(None),
                    )
                )
                .distinct()
            )
        ).scalars().all()
        return [t for t in rows if t]

    async def _series_max_stated_end(self, series_tag_id: int) -> int | None:
        # Highest printed page across the series' pages (FC-6.x: stated pages
        # live on series_page now; chapters are cosmetic dividers).
        return await self.session.scalar(
            select(func.max(SeriesPage.stated_page)).where(
                SeriesPage.series_tag_id == series_tag_id
            )
        )

    async def _distinctive_tags(self, image_ids: list[int] | set[int]) -> set[int]:
        ids = list(image_ids)
        if not ids:
            return set()
        rows = (
            await self.session.execute(
                select(image_tag.c.tag_id)
                .join(Tag, Tag.id == image_tag.c.tag_id)
                .where(
                    and_(
                        image_tag.c.image_record_id.in_(ids),
                        Tag.kind.in_(_DISTINCTIVE_KINDS),
                    )
                )
                .distinct()
            )
        ).scalars().all()
        return set(rows)

    async def _candidate_series(self, artist_id: int) -> list[int]:
        """Series that already contain a page by this artist — cross-artist
        series are rare, so same-artist is the candidate bound."""
        rows = (
            await self.session.execute(
                select(SeriesPage.series_tag_id)
                .join(ImageRecord, ImageRecord.id == SeriesPage.image_id)
                .where(ImageRecord.artist_id == artist_id)
                .distinct()
            )
        ).scalars().all()
        return list(rows)

    async def _decided_series(self, post_id: int) -> set[int]:
        rows = (
            await self.session.execute(
                select(SeriesSuggestion.series_tag_id).where(
                    and_(
                        SeriesSuggestion.post_id == post_id,
                        SeriesSuggestion.status.in_(["added", "dismissed"]),
                    )
                )
            )
        ).scalars().all()
        return set(rows)

    async def match_post(self, post_id: int, *, threshold: float) -> int:
        """Score a post against its artist's series; upsert pending suggestions
        for those at/above threshold. Returns the number written."""
        post = await self.session.get(Post, post_id)
        if post is None or post.artist_id is None:
            return 0
        post_images = await self._post_image_ids(post_id)
        if not post_images:
            return 0
        post_image_set = set(post_images)
        rng = parse_page_range(f"{post.post_title or ''} {post.description or ''}")
        post_start = rng[0] if rng else None
        post_dtags = await self._distinctive_tags(post_images)

        decided = await self._decided_series(post_id)
        candidates = await self._candidate_series(post.artist_id)
        made = 0
        for sid in candidates[:_MAX_CANDIDATES]:
            if sid in decided:
                continue
            s_images = await self._series_image_ids(sid)
            if post_image_set & s_images:
                continue  # the post is already (partly) in this series
            signals = {
                "title": title_signal(
                    await self._series_post_titles(sid), post.post_title
                ),
                "artist": 1.0,  # candidates are same-artist by construction
                "pages": pages_signal(
                    await self._series_max_stated_end(sid), post_start
                ),
                "tags": tags_signal(
                    len(post_dtags & await self._distinctive_tags(s_images))
                ),
            }
            score = weighted_score(signals)
            if score < threshold:
                continue
            await self.session.execute(
                pg_insert(SeriesSuggestion)
                .values(
                    post_id=post_id, series_tag_id=sid,
                    score=score, signals=signals, status="pending",
                )
                .on_conflict_do_update(
                    constraint="uq_series_suggestion_post_series",
                    set_={"score": score, "signals": signals, "status": "pending"},
                    where=SeriesSuggestion.status == "pending",
                )
            )
            made += 1
        return made

    # ---- queue ops --------------------------------------------------------

    async def list_pending(self) -> list[dict]:
        rows = (
            await self.session.execute(
                select(
                    SeriesSuggestion.id,
                    SeriesSuggestion.post_id,
                    SeriesSuggestion.series_tag_id,
                    SeriesSuggestion.score,
                    SeriesSuggestion.signals,
                    Post.post_title,
                    Post.external_post_id,
                    Tag.name.label("series_name"),
                )
                .join(Post, Post.id == SeriesSuggestion.post_id)
                .join(Tag, Tag.id == SeriesSuggestion.series_tag_id)
                .where(SeriesSuggestion.status == "pending")
                .order_by(SeriesSuggestion.score.desc())
            )
        ).all()
        return [
            {
                "id": r.id,
                "post_id": r.post_id,
                "series_tag_id": r.series_tag_id,
                "series_name": r.series_name,
                "post_title": r.post_title or f"Post {r.external_post_id}",
                "score": r.score,
                "signals": r.signals or {},
            }
            for r in rows
        ]

    async def accept(self, suggestion_id: int) -> dict:
        s = await self.session.get(SeriesSuggestion, suggestion_id)
        if s is None:
            raise SeriesError(f"suggestion {suggestion_id} not found")
        if s.status != "pending":
            raise SeriesError(f"suggestion {suggestion_id} is already {s.status}")
        out = await SeriesService(self.session).add_post(
            s.series_tag_id, s.post_id
        )
        s.status = "added"
        self.session.add(s)
        return out

    async def dismiss(self, suggestion_id: int) -> None:
        s = await self.session.get(SeriesSuggestion, suggestion_id)
        if s is None:
            raise SeriesError(f"suggestion {suggestion_id} not found")
        if s.status == "pending":
            s.status = "dismissed"
            self.session.add(s)

    async def rescan(
        self, *, threshold: float, time_budget_seconds: float | None = None,
        after_post_id: int = 0,
    ) -> dict:
        """Score every post (id > after_post_id) against its artist's series,
        time-boxed + resumable like the other long maintenance sweeps."""
        post_ids = (
            await self.session.execute(
                select(Post.id)
                .where(Post.id > after_post_id)
                .order_by(Post.id.asc())
            )
        ).scalars().all()
        summary = {
            "scanned": 0, "suggested": 0,
            "partial": False, "resume_after_id": after_post_id,
        }
        start = time.monotonic()
        for pid in post_ids:
            summary["scanned"] += 1
            summary["resume_after_id"] = pid
            summary["suggested"] += await self.match_post(pid, threshold=threshold)
            await self.session.commit()  # commit per post so progress survives
            if (
                time_budget_seconds is not None
                and time.monotonic() - start >= time_budget_seconds
            ):
                summary["partial"] = True
                break
        else:
            summary["partial"] = False
        return summary