FabledCurator/backend/app/services/artist_membership_service.py

"""Proposing that a creator FC tracks and a membership it found are the same.

Milestone 388, step E4. An instance of the confirm-only matcher shape
(snippet #3842), and a sibling of `post_association_service`.

## What E4 turned out NOT to need

The step's own first instruction was to verify before building, and the
verification said: not the schema, not the flows. `Source.artist_id` is a plain
FK so many sources per artist already works; `POST /api/sources` already takes
an `artist_id`; the add-source dialog already has an artist autocomplete that
attaches to an EXISTING artist; `SourceService.reassign` already moves a source
between artists WITH post and image re-attribution; and a sweep for
one-source-per-artist assumptions found only `func.count()` calls, which are
the opposite of assuming one.

So the association a Discord source and a Patreon source share is already
expressible today. What was missing is FC OFFERING it.

## Accept adds a SOURCE — it never merges artists

The asymmetry that sets the whole posture: adding a source is trivially undone.
A wrong artist merge silently mixes two creators' work and corrupts tagging,
series and provenance downstream, with nothing left to tell the two apart by.
So the accepted action is "add the missing channel to this artist", and merging
is not offered at all.

## The signals

1. **Name.** The roster's `display_name` and `vanity`, slugified, against the
   artist's `slug`. Graded rather than boolean — an exact match is strong
   evidence, a containment match is a hint.
2. **Declared.** A post already under this artist whose body links to
   `patreon.com/<vanity>` for this exact membership. A creator pointing at
   their own Patreon from their own Discord is close to a statement.

Signal 2 is NOT read from `ExternalLink`, and that correction is worth keeping:
`link_extract.SUPPORTED_HOSTS` is file hosts only (mega/gdrive/mediafire/
dropbox/pixeldrain) and `host_for()` returns None for patreon.com, so no
`ExternalLink` row is ever written for one. The same trap already caught E5 for
Discord invites.

## Weights, and what they make impossible

    name 0.65 · declared 0.35, cut at 0.60

Chosen so the arithmetic encodes the judgement rather than a code path doing it:

* an EXACT name match alone (0.65) proposes — same slug on both sides is
  strong, and requiring corroboration would mean proposing almost nothing;
* a CONTAINMENT name match alone (0.6 * 0.65 = 0.39) does not — "art" inside
  "artgirl" is a coincidence generator, and it needs the declaration;
* the declaration ALONE (0.35) never proposes, at any setting at or above
  0.60 — a creator may link another creator's Patreon, and a link is not a
  claim of identity.

A guard test pins all three against WEIGHTS directly, so they survive a
refactor of the scorer.
"""

from __future__ import annotations

import logging
import re

from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from ..models import (
    Artist,
    ArtistMembershipSuggestion,
    PlatformMembership,
    Post,
    Source,
)
from ..utils.slug import slugify
from ..utils.text import html_to_plain

log = logging.getLogger(__name__)

WEIGHTS = {"name": 0.65, "declared": 0.35}
DEFAULT_THRESHOLD = 0.60

NAME_EXACT = 1.0
# Containment is a hint, not a match: "art" sits inside "artgirl", and slugs
# are short enough that coincidental containment is common.
NAME_CONTAINS = 0.6
# Below this many characters, containment is noise rather than signal — a
# 3-character slug is inside a great many longer ones.
_MIN_CONTAINMENT_LEN = 5

MAX_CANDIDATES = 25


def name_signal(membership: PlatformMembership, artist: Artist) -> float:
    """Graded slug agreement between a membership and an artist.

    Both the display name and the vanity are tried, because creators routinely
    differ between the two ("Team Melon Collie" vs "MelonCollieStudios") and
    either may be the one the operator typed when they created the artist.
    """
    artist_slug = slugify(artist.name or "") if artist.name else ""
    if not artist_slug or artist_slug == "untitled":
        return 0.0
    candidates = {
        slugify(v) for v in (membership.display_name, membership.vanity_or_none())
        if v
    }
    candidates.discard("untitled")
    if not candidates:
        return 0.0
    if artist_slug in candidates:
        return NAME_EXACT
    for c in candidates:
        if len(c) < _MIN_CONTAINMENT_LEN or len(artist_slug) < _MIN_CONTAINMENT_LEN:
            continue
        if c in artist_slug or artist_slug in c:
            return NAME_CONTAINS
    return 0.0


def declared_signal(body: str | None, vanity: str | None) -> float:
    """Does this post body point at THIS membership's Patreon page?

    Matched against the RAW body, not the stripped text: these links live in an
    anchor's `href`, and `html_to_plain` discards attributes — the same trap
    that caught E5's invite detection. The stripped text is checked too, for
    bodies that paste the URL as plain text.
    """
    if not body or not vanity:
        return 0.0
    pattern = re.compile(
        r"patreon\.com/(?:c/|cw/|checkout/)?" + re.escape(vanity) + r"\b", re.I
    )
    if pattern.search(body):
        return 1.0
    return 1.0 if pattern.search(html_to_plain(body) or "") else 0.0


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


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

    async def _decided(self, membership_id: int) -> set[int]:
        """Artists already proposed for this membership, in ANY status.

        Dismissed included: the row is what remembers the rejection, and
        re-proposing a rejected pair every scan is what makes a queue ignored.
        """
        rows = (await self.session.execute(
            select(ArtistMembershipSuggestion.artist_id).where(
                ArtistMembershipSuggestion.platform_membership_id == membership_id
            )
        )).scalars().all()
        return set(rows)

    async def _candidate_artists(self, membership: PlatformMembership) -> list[Artist]:
        """Artists that have SOME source but none for this membership's platform.

        A hard filter, not a scored signal. An artist FC already tracks on this
        platform needs no suggestion — the link exists — and an artist with no
        sources at all is not a creator FC is following through another channel,
        which is the whole case this step is about.
        """
        # `select(...).exists()` rather than a bare `exists().where(...)`: the
        # latter has no FROM to correlate against and does not reliably render.
        has_any = select(Source.id).where(Source.artist_id == Artist.id).exists()
        has_this = (
            select(Source.id)
            .where(
                Source.artist_id == Artist.id,
                Source.platform == membership.platform,
            )
            .exists()
        )
        return (await self.session.execute(
            select(Artist).where(has_any, ~has_this).limit(MAX_CANDIDATES)
        )).scalars().all()

    async def _declared_for(self, artist_id: int, vanity: str | None) -> float:
        if not vanity:
            return 0.0
        # Bounded scan: the newest posts are where a creator's current links
        # live, and an unbounded body scan per (artist, membership) pair would
        # be the expensive part of this sweep.
        bodies = (await self.session.execute(
            select(Post.description)
            .where(Post.artist_id == artist_id, Post.description.is_not(None))
            .order_by(func.coalesce(Post.post_date, Post.downloaded_at).desc())
            .limit(50)
        )).scalars().all()
        for body in bodies:
            if declared_signal(body, vanity) > 0:
                return 1.0
        return 0.0

    async def match_membership(
        self, membership_id: int, *, threshold: float = DEFAULT_THRESHOLD,
    ) -> int:
        membership = await self.session.get(PlatformMembership, membership_id)
        if membership is None:
            return 0
        already = await self._decided(membership_id)

        made = 0
        for artist in await self._candidate_artists(membership):
            if artist.id in already:
                continue
            signals = {
                "name": name_signal(membership, artist),
                "declared": await self._declared_for(
                    artist.id, membership.vanity_or_none()
                ),
            }
            score = weighted_score(signals)
            if score < threshold:
                continue
            self.session.add(ArtistMembershipSuggestion(
                platform_membership_id=membership.id,
                artist_id=artist.id,
                score=score,
                signals=signals,
                status="pending",
            ))
            made += 1
        return made

    async def list_pending(self) -> list[dict]:
        rows = (await self.session.execute(
            select(ArtistMembershipSuggestion, PlatformMembership, Artist)
            .join(
                PlatformMembership,
                PlatformMembership.id
                == ArtistMembershipSuggestion.platform_membership_id,
            )
            .join(Artist, Artist.id == ArtistMembershipSuggestion.artist_id)
            .where(ArtistMembershipSuggestion.status == "pending")
            .order_by(
                ArtistMembershipSuggestion.score.desc(),
                ArtistMembershipSuggestion.id.desc(),
            )
        )).all()
        return [
            {
                "id": s.id,
                "score": s.score,
                "signals": s.signals,
                "artist": {"id": a.id, "name": a.name, "slug": a.slug},
                "membership": {
                    "id": m.id,
                    "platform": m.platform,
                    "display_name": m.display_name,
                    "url": m.url,
                },
            }
            for s, m, a in rows
        ]

    async def accept(self, suggestion_id: int) -> dict | None:
        """Add the missing channel to the artist. NEVER merges two artists.

        Returns the created source's id, or `already_linked` when a source for
        that platform appeared between the proposal and the click — which is
        not an error, it is the operator having done it by hand.
        """
        s = await self.session.get(ArtistMembershipSuggestion, suggestion_id)
        if s is None:
            return None
        membership = await self.session.get(PlatformMembership, s.platform_membership_id)
        if membership is None or not membership.url:
            return None

        existing = (await self.session.execute(
            select(Source.id).where(
                Source.artist_id == s.artist_id,
                Source.platform == membership.platform,
            )
        )).scalars().first()
        if existing is not None:
            s.status = "linked"
            return {"id": s.id, "status": s.status, "already_linked": existing}

        # Through SourceService, NOT a bare Source() insert. It carries the
        # platform/URL validation, the duplicate check and the #693
        # backfill-arming that a hand-added source gets — building a second,
        # quieter way to create a source is how the two drift until one of them
        # is subtly broken (rule 28: repurpose the existing surface).
        from .source_service import DuplicateSourceError, SourceService

        try:
            record = await SourceService(self.session).create(
                artist_id=s.artist_id,
                platform=membership.platform,
                url=membership.url,
            )
        except DuplicateSourceError as exc:
            # The same URL already exists for this artist — the operator got
            # there first by a different route. Not an error.
            s.status = "linked"
            return {"id": s.id, "status": s.status, "already_linked": exc.existing_id}
        s.status = "linked"
        return {"id": s.id, "status": s.status, "source_id": record.id}

    async def dismiss(self, suggestion_id: int) -> dict | None:
        s = await self.session.get(ArtistMembershipSuggestion, suggestion_id)
        if s is None:
            return None
        # Kept, not deleted — the row is what remembers the rejection.
        s.status = "dismissed"
        return {"id": s.id, "status": s.status}


async def rescan(session: AsyncSession, *, threshold: float = DEFAULT_THRESHOLD) -> dict:
    """Offer every known membership to the artists FC already tracks."""
    ids = (await session.execute(select(PlatformMembership.id))).scalars().all()
    svc = ArtistMembershipService(session)
    proposed = 0
    for mid in ids:
        proposed += await svc.match_membership(mid, threshold=threshold)
    log.info(
        "artist/membership matcher: scanned %d membership(s), proposed %d pair(s)",
        len(ids), proposed,
    )
    return {"scanned": len(ids), "proposed": proposed}