"""The announcement matcher: which Patreon post announced which Discord drop.

Milestone 388, step E5.

Two of the operator's artists post a deliberately CROPPED fragment on Patreon
to signal that the real thing has landed in their Discord. This service
proposes those pairs, and proposes only — the operator accepts or dismisses,
following the FC-6.3 series matcher (task 737) rather than linking on its own.

## Why confirm-only is not caution for its own sake

A wrongly-asserted association tells the operator that two different pieces are
one. That is strictly worse than no link at all: no link leaves them exactly
where they already were, a wrong one actively misinforms and then propagates
into whatever reads the association. So the matcher's job is to make a SHORT
list worth reading, not a long list worth trusting.

## Signals, and the one deliberately NOT built

1. **Time proximity.** The Patreon post exists in order to announce the drop,
   so the two are minutes-to-hours apart. Nearly free, and strong.
2. **The post says so.** These announcements routinely name Discord or carry
   an invite link, which is close to a declaration.

3. **Crop-to-source matching is HELD, on the plan's own instruction** — it is
   real work with real false-positive risk, and it is only worth building once
   1 and 2 are shown to be insufficient against the operator's actual artists.
   Nothing here should be read as evidence it is unnecessary; it is deferred,
   and the thing that would justify it is an empty review queue on a pair the
   operator can see with their own eyes.

   Note also that a naive whole-image SigLIP similarity is NOT that signal. A
   cropped teaser and its full version are exactly the pair a whole-image
   comparison handles worst, so adding one as a "bonus" would mostly add noise
   while looking like progress.

## Creator identity comes free, so E4 is not actually a prerequisite

The plan listed E4 (creator identity across the two channels) as a dependency.
It is not one for the pairs that matter today: a Patreon `Source` and a Discord
`Source` the operator has added under the same `Artist` already share
`Post.artist_id`, and the synthetic grouping inherits it (E2). E4 EXTENDS this
to creators whose association FC has to learn rather than being told; it is not
needed to represent an association FC already knows.

## A correction the plan carried, worth restating

`link_extract.py` does exist and does capture off-platform links — but only
file hosts (`SUPPORTED_HOSTS` is mega/gdrive/mediafire/dropbox/pixeldrain).
`host_for()` returns None for a Discord URL, so no `ExternalLink` row is ever
written for one. The declaration signal therefore reads the post body itself
rather than the extracted-links table the plan assumed it could use.
"""

from __future__ import annotations

import logging
import re
from datetime import UTC, datetime, timedelta

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

from ..models import ImportSettings, Post, PostAssociation
from ..utils.text import html_to_plain
from .discord_grouping import DROP_GROUPER

log = logging.getLogger(__name__)

# Additive weights, summing to 1.0. Kept as constants rather than settings —
# the sensitivity knob that matters is the threshold, and per-signal weights
# are an over-tune (same call as series_match_service.WEIGHTS).
#
# THE RELATIONSHIP TO THE THRESHOLD IS THE DESIGN. No single weight may reach
# the default threshold, which is what makes "time proximity alone is never
# enough" arithmetic rather than aspirational: on a busy day an artist posts
# several times, and a matcher that could pair on proximity alone would turn
# every busy day into false pairs. A guard test pins this.
WEIGHTS = {"proximity": 0.55, "declared": 0.45}

# A Discord INVITE in the body is close to a declaration; the bare word is
# weaker but still meaningful, because these posts are short and on-topic.
_INVITE = re.compile(r"discord\.(?:gg|com/invite)/", re.I)
_MENTION = re.compile(r"\bdiscord\b", re.I)

DECLARED_INVITE = 1.0
DECLARED_MENTION = 0.6

MAX_CANDIDATES = 25


def proximity_signal(gap: timedelta, window: timedelta) -> float:
    """1.0 when the two posts are simultaneous, decaying linearly to 0 at the
    window's edge. Linear rather than a step, so a pair an hour outside a
    hand-tuned window degrades instead of vanishing."""
    if window <= timedelta(0):
        return 0.0
    seconds = abs(gap.total_seconds())
    if seconds >= window.total_seconds():
        return 0.0
    return round(1.0 - (seconds / window.total_seconds()), 4)


def declared_signal(description: str | None) -> float:
    """Does the announcement say, in its own body, that this is about Discord?

    The invite is matched against the RAW body and the bare mention against the
    stripped text, which is not fussiness — post bodies are HTML, and these
    creators put the invite in an anchor's `href`. `html_to_plain` discards
    attributes, so stripping first would have thrown away the strongest form of
    the signal and left only whatever the link text happened to say.

    The mention still reads stripped text, so `\bdiscord\b` is matched against
    prose rather than against markup and URLs, where it would fire on any link
    that merely passes through a discord domain.
    """
    if not description:
        return 0.0
    if _INVITE.search(description):
        return DECLARED_INVITE
    if _MENTION.search(html_to_plain(description) or ""):
        return DECLARED_MENTION
    return 0.0


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


def _post_time(post: Post) -> datetime:
    return post.post_date or post.downloaded_at


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

    async def _decided(self, announcement_id: int) -> set[int]:
        """Payload posts already proposed for this announcement, in ANY status.

        Dismissed pairs are included deliberately: re-proposing a pair the
        operator has already rejected on every subsequent scan is the single
        behaviour that makes a review queue get ignored.
        """
        rows = (await self.session.execute(
            select(PostAssociation.payload_post_id)
            .where(PostAssociation.announcement_post_id == announcement_id)
        )).scalars().all()
        return set(rows)

    async def _candidate_groups(
        self, announcement: Post, *, window: timedelta,
    ) -> list[Post]:
        """Synthetic Discord groupings by the SAME artist, inside the window.

        Same-artist is the identity signal and it is free (see the module
        docstring on E4). It is also a hard filter rather than a scored one:
        two different creators posting minutes apart is a coincidence, not
        evidence, and letting it score at all would mean a busy hour across the
        library could out-vote everything else.
        """
        at = _post_time(announcement)
        sort_key = func.coalesce(Post.post_date, Post.downloaded_at)
        return (await self.session.execute(
            select(Post)
            .where(
                Post.artist_id == announcement.artist_id,
                Post.synthesized_by == DROP_GROUPER,
                Post.id != announcement.id,
                sort_key >= at - window,
                sort_key <= at + window,
            )
            .order_by(sort_key)
            .limit(MAX_CANDIDATES)
        )).scalars().all()

    async def match_post(
        self, announcement_id: int, *, threshold: float, window_hours: float,
    ) -> int:
        """Score one announcement against nearby groupings. Returns proposals made."""
        announcement = await self.session.get(Post, announcement_id)
        if announcement is None or announcement.synthesized_by is not None:
            # A synthetic post cannot announce anything — FC wrote it.
            return 0

        window = timedelta(hours=window_hours)
        declared = declared_signal(announcement.description)
        already = await self._decided(announcement_id)

        made = 0
        for group in await self._candidate_groups(announcement, window=window):
            if group.id in already:
                continue
            signals = {
                "proximity": proximity_signal(
                    _post_time(group) - _post_time(announcement), window,
                ),
                "declared": declared,
            }
            score = weighted_score(signals)
            if score < threshold:
                continue
            self.session.add(PostAssociation(
                announcement_post_id=announcement.id,
                payload_post_id=group.id,
                score=score,
                signals=signals,
                status="pending",
            ))
            made += 1
        return made

    async def list_pending(self) -> list[dict]:
        rows = (await self.session.execute(
            select(PostAssociation)
            .where(PostAssociation.status == "pending")
            .order_by(PostAssociation.score.desc(), PostAssociation.id.desc())
        )).scalars().all()
        return [
            {
                "id": a.id,
                "announcement_post_id": a.announcement_post_id,
                "payload_post_id": a.payload_post_id,
                "score": a.score,
                "signals": a.signals,
            }
            for a in rows
        ]

    async def accept(self, association_id: int) -> dict | None:
        a = await self.session.get(PostAssociation, association_id)
        if a is None:
            return None
        a.status = "linked"
        return {"id": a.id, "status": a.status}

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

    async def linked_for(self, post_ids: list[int]) -> dict[int, list[dict]]:
        """Accepted links touching these posts, keyed by post id, BOTH ways.

        A post is either end of the relationship, and each end wants the other
        one: the teaser wants "the full set is over here", the grouping wants
        "this is what announced me". One query, both directions.
        """
        if not post_ids:
            return {}
        rows = (await self.session.execute(
            select(PostAssociation).where(
                PostAssociation.status == "linked",
                or_(
                    PostAssociation.announcement_post_id.in_(post_ids),
                    PostAssociation.payload_post_id.in_(post_ids),
                ),
            )
        )).scalars().all()
        out: dict[int, list[dict]] = {}
        for a in rows:
            if a.announcement_post_id in post_ids:
                out.setdefault(a.announcement_post_id, []).append(
                    {"role": "announces", "post_id": a.payload_post_id, "id": a.id}
                )
            if a.payload_post_id in post_ids:
                out.setdefault(a.payload_post_id, []).append(
                    {"role": "announced_by", "post_id": a.announcement_post_id, "id": a.id}
                )
        return out


async def rescan(session: AsyncSession, *, now: datetime | None = None) -> dict:
    """Score every recent non-synthetic post against nearby groupings."""
    settings = await ImportSettings.load(session)
    if not settings.discord_link_enabled:
        return {"enabled": False, "scanned": 0, "proposed": 0}

    now = now or datetime.now(UTC)
    window_hours = float(settings.discord_link_window_hours)
    # Only look at announcements that could still have a partner in range —
    # a full-library rescan is the manual button's job, not the sweep's.
    horizon = now - timedelta(hours=window_hours * 2)
    sort_key = func.coalesce(Post.post_date, Post.downloaded_at)
    ids = (await session.execute(
        select(Post.id).where(
            Post.synthesized_by.is_(None),
            Post.absorbed_by_post_id.is_(None),
            sort_key >= horizon,
        )
    )).scalars().all()

    svc = PostAssociationService(session)
    proposed = 0
    for pid in ids:
        proposed += await svc.match_post(
            pid,
            threshold=float(settings.discord_link_threshold),
            window_hours=window_hours,
        )
    log.info(
        "discord announcement matcher: scanned %d post(s), proposed %d pair(s)",
        len(ids), proposed,
    )
    return {"enabled": True, "scanned": len(ids), "proposed": proposed}