"""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_chapter import SeriesChapter 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: return await self.session.scalar( select(func.max(SeriesChapter.stated_page_end)).where( SeriesChapter.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_as_chapter( 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