"""The unified post card — a teaser shows what it points at (#4402, #4401).

Milestone 388. A Patreon teaser is a POINTER: a cropped, censored fragment
whose job is to say "the full set is in Discord". Until this module the card
rendered the fragment and a text link, and the reader had to make the join FC
had already made.

Operator, 2026-09-24: *"the teaser from the patreon post doesn't show the items
that it's supposed to reference so I'm trying to unify the teaser post with the
content it's meant to draw attention to."*

## A reference, never an absorption

`discord_grouping` folds chat messages into a synthetic post by transferring
ownership (`absorbed_by_post_id`). That is the wrong primitive here, and the
operator said so directly: *"the nested items on the unified post are a
duplicate or reference of existing content. that's why they can show similar
items and not erase or invalidate the way the discord items landed."*

So nothing here writes. The Discord posts keep their own rows, dates and
places in the feed; the teaser's card DISPLAYS them. That is also what makes
reaching back for older variants safe at all: a wrong reference shows one
extra thumbnail in one place, where a wrong regrouping would move content.

## What a teaser references

1. The Discord drops a `linked` PostAssociation joins it to (#4392) — accepted
   by the operator, or linked by FC on a conclusive name match.
2. The rest of that piece's VARIANT FAMILY (#4401): the wips, alts and censor
   passes a creator trickles out under one working name, days or weeks apart.

Families are found by the creator's LEADING working name, not by any shared
token and not by image similarity — see `post_naming.leading_name` for the
measurement, and lesson #4400 for why a whole-image comparison between two
works by one artist cannot separate "same piece" from "same artist".
"""
from __future__ import annotations

from collections import Counter
from collections.abc import Iterable
from dataclasses import dataclass
from datetime import datetime, timedelta

from sqlalchemy import and_, exists, extract, func, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import aliased

from ..models import (
    ImageProvenance,
    ImageRecord,
    ImportSettings,
    Post,
    PostAssociation,
    Source,
)
from ..utils.phash import hamming, hash_bits
from ..utils.text import html_to_plain, truncate_at_word
from .discord_grouping import PLATFORM as DISCORD
from .gallery_service import thumbnail_url
from .post_association_service import DUPLICATE_MAX_DISTANCE
from .post_naming import FAMILY_MAX_POSTS, leading_name, rarity, token_frequencies

# The text each referenced post contributes to the card, per post. The card
# clamps it again; this keeps a long Discord thread from making the feed
# payload the size of the thread.
TEXT_LIMIT = 280


@dataclass(frozen=True)
class Candidate:
    """One image as the family search sees it."""

    image_id: int
    post_id: int
    path: str
    phash: int | None
    at: datetime


def family(
    seed: Iterable[Candidate],
    pool: Iterable[Candidate],
    name_posts: Counter[str],
    hash_posts: Counter[int],
    *,
    anchor: datetime,
    window: timedelta,
    max_posts: int = FAMILY_MAX_POSTS,
) -> list[Candidate]:
    """The images in `pool` that belong to the same piece as `seed`.

    A member shares a seed image's LEADING working name, or is a perceptual
    near-duplicate of one (the same file re-posted), and lies within `window`
    of `anchor` — the teaser's date. Oldest first, so the card reads as the
    trickle it was.

    ONE hop from the seed, never transitive. Every measured family is one hop
    from any of its members, because the members share the name; chaining is
    what lets a family drift from `Year_20k` to whatever `Year_20k_Base`'s
    other tokens happen to touch.

    Both identity routes are rarity-gated against `max_posts`, exactly as the
    matcher gates them. A leading name the creator uses across many posts is a
    character, and a hash on many posts is a banner.
    """
    seed = list(seed)
    names = {
        name for c in seed
        if (name := leading_name(c.path)) is not None
        and rarity(name_posts.get(name, 0), max_posts) > 0
    }
    hashes = [
        c.phash for c in seed
        if c.phash is not None and rarity(hash_posts.get(c.phash, 0), max_posts) > 0
    ]
    taken = {c.image_id for c in seed}

    out: list[Candidate] = []
    for c in pool:
        if c.image_id in taken or abs(c.at - anchor) > window:
            continue
        named = leading_name(c.path) in names
        copied = c.phash is not None and any(
            (d := hamming(c.phash, h)) is not None and d <= DUPLICATE_MAX_DISTANCE
            for h in hashes
        )
        if named or copied:
            out.append(c)
            taken.add(c.image_id)
    return sorted(out, key=lambda c: (c.at, c.image_id))


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


def _text(post: Post) -> str | None:
    plain = html_to_plain(post.description) if post.description else None
    if not plain or not plain.strip():
        return None
    return truncate_at_word(plain.strip(), TEXT_LIMIT)[0]


@dataclass
class _Artist:
    """Everything the family search needs about one artist, loaded once."""

    rows: dict[int, tuple]          # image_id -> (post_id, path, phash, sha, mime, thumb)
    posts: dict[int, Post]
    platform: dict[int, str | None]  # post_id -> platform
    name_posts: Counter[str]
    hash_posts: Counter[int]


class PostUnificationService:
    def __init__(self, session: AsyncSession):
        self.session = session
        self._artists: dict[int, _Artist] = {}

    async def _artist(self, artist_id: int) -> _Artist:
        if artist_id in self._artists:
            return self._artists[artist_id]

        posts: dict[int, Post] = {}
        platform: dict[int, str | None] = {}
        for post, plat in (await self.session.execute(
            select(Post, Source.platform)
            .outerjoin(Source, Post.source_id == Source.id)
            .where(Post.artist_id == artist_id)
        )).all():
            posts[post.id] = post
            platform[post.id] = plat

        rows: dict[int, tuple] = {}
        paths_by_post: dict[int, list[str]] = {}
        hashes_by_post: dict[int, set[int]] = {}
        for img_id, post_id, path, phash, sha, mime, thumb in (await self.session.execute(
            select(
                ImageRecord.id, ImageRecord.primary_post_id, ImageRecord.path,
                ImageRecord.phash, ImageRecord.sha256, ImageRecord.mime,
                ImageRecord.thumbnail_path,
            ).where(
                ImageRecord.artist_id == artist_id,
                ImageRecord.primary_post_id.is_not(None),
            )
        )).all():
            bits = hash_bits(phash)
            rows[img_id] = (post_id, path, bits, sha, mime, thumb)
            paths_by_post.setdefault(post_id, []).append(path)
            if bits is not None:
                hashes_by_post.setdefault(post_id, set()).add(bits)

        # Counted over EVERY post the artist has, exactly as the matcher's
        # corpus counts them — a family is judged against the whole library,
        # not against the slice inside the window, or a character name would
        # look rare in any quiet month.
        found = _Artist(
            rows=rows,
            posts=posts,
            platform=platform,
            name_posts=token_frequencies(paths_by_post.values()),
            hash_posts=Counter(h for hs in hashes_by_post.values() for h in hs),
        )
        self._artists[artist_id] = found
        return found

    async def _drop_images(self, drop_ids: list[int]) -> dict[int, list[int]]:
        """drop post id -> its image ids, through provenance as the feed reads them.

        A synthetic drop owns no image outright: its images belong to the
        member messages, and `discord_grouping` gives the drop a provenance row
        for each. The primary_post_id arm keeps any image that has one and no
        row, the same union `PostFeedService._thumbnails_for` takes.
        """
        out: dict[int, list[int]] = {pid: [] for pid in drop_ids}
        if not drop_ids:
            return out
        links = (
            select(
                ImageProvenance.image_record_id.label("image_id"),
                ImageProvenance.post_id.label("post_id"),
            )
            .where(ImageProvenance.post_id.in_(drop_ids))
            .union(
                select(
                    ImageRecord.id.label("image_id"),
                    ImageRecord.primary_post_id.label("post_id"),
                ).where(ImageRecord.primary_post_id.in_(drop_ids))
            )
            .subquery()
        )
        for img_id, pid in (await self.session.execute(
            select(links.c.image_id, links.c.post_id).order_by(links.c.image_id)
        )).all():
            out[pid].append(img_id)
        return out

    async def unified_for(self, posts: Iterable[Post]) -> dict[int, dict]:
        """post id -> the card's reference set, for each post that HAS one.

        Only teasers get one: a post with at least one `linked` association on
        the announcing side. Every other post is absent from the result, and
        the card renders exactly as it did before this module existed.
        """
        teasers = {p.id: p for p in posts if p.synthesized_by is None}
        if not teasers:
            return {}
        links = (await self.session.execute(
            select(PostAssociation)
            .where(
                PostAssociation.status == "linked",
                PostAssociation.announcement_post_id.in_(list(teasers)),
            )
            .order_by(PostAssociation.id)
        )).scalars().all()
        if not links:
            return {}

        settings = await self.session.get(ImportSettings, 1)
        window = timedelta(days=float(
            settings.discord_family_window_days if settings is not None else 60.0
        ))
        drop_images = await self._drop_images(
            sorted({a.payload_post_id for a in links})
        )

        by_teaser: dict[int, list[PostAssociation]] = {}
        for a in links:
            by_teaser.setdefault(a.announcement_post_id, []).append(a)

        out: dict[int, dict] = {}
        for teaser_id, assocs in by_teaser.items():
            teaser = teasers[teaser_id]
            artist = await self._artist(teaser.artist_id)
            out[teaser_id] = self._compose(teaser, assocs, drop_images, artist, window)
        return out

    def _compose(
        self,
        teaser: Post,
        assocs: list[PostAssociation],
        drop_images: dict[int, list[int]],
        artist: _Artist,
        window: timedelta,
    ) -> dict:
        def candidate(img_id: int) -> Candidate | None:
            row = artist.rows.get(img_id)
            if row is None:
                return None
            post_id, path, bits, *_ = row
            post = artist.posts.get(post_id)
            if post is None:
                return None
            return Candidate(img_id, post_id, path, bits, _when(post))

        drop_ids = [a.payload_post_id for a in assocs]
        shown = [i for d in drop_ids for i in drop_images.get(d, [])]
        own = [i for i, row in artist.rows.items() if row[0] == teaser.id]
        seed = [c for i in own + shown if (c := candidate(i)) is not None]

        # Variants come from Discord only. That is where a creator trickles
        # them out, it is the corpus the family rule was measured on, and it
        # keeps one teaser from pulling a DIFFERENT teaser's crop onto its card.
        pool = [
            c for i, row in artist.rows.items()
            if artist.platform.get(row[0]) == DISCORD
            and (c := candidate(i)) is not None
        ]
        variants = family(
            seed, pool, artist.name_posts, artist.hash_posts,
            anchor=_when(teaser), window=window,
        )

        def thumb(img_id: int, post_id: int, role: str) -> dict | None:
            row = artist.rows.get(img_id)
            if row is None:
                return None
            _pid, _path, _bits, sha, mime, tp = row
            return {
                "image_id": img_id,
                "thumbnail_url": thumbnail_url(tp, sha, mime),
                "mime": mime,
                "post_id": post_id,
                "role": role,
            }

        own_ids = set(own)
        thumbnails: list[dict] = []
        seen: set[int] = set(own_ids)
        for drop_id in drop_ids:
            for img_id in drop_images.get(drop_id, []):
                if img_id in seen:
                    continue
                if (t := thumb(img_id, drop_id, "drop")) is not None:
                    thumbnails.append(t)
                    seen.add(img_id)
        for c in variants:
            if c.image_id in seen:
                continue
            if (t := thumb(c.image_id, c.post_id, "variant")) is not None:
                thumbnails.append(t)
                seen.add(c.image_id)

        # The text of every item the card unifies — the operator's *"the
        # unified card should also contain the text for any of the items
        # unified on it"*. A drop's own description already joins its member
        # messages, so a variant's text is its MESSAGE, read off the member
        # post that owns the image. A line said twice (`@everyone 🍈🍈` on
        # every message of a drop) is shown once.
        texts: list[dict] = []
        said: set[str] = set()

        def add_text(post: Post | None, role: str) -> None:
            if post is None:
                return
            text = _text(post)
            if text is None or text in said:
                return
            said.add(text)
            texts.append({
                "post_id": post.id,
                "role": role,
                "date": _when(post).isoformat(),
                "text": text,
            })

        for drop_id in drop_ids:
            add_text(artist.posts.get(drop_id), "drop")
        for post_id in dict.fromkeys(c.post_id for c in variants):
            add_text(artist.posts.get(post_id), "variant")

        return {
            "links": [
                {
                    "association_id": a.id,
                    "post_id": a.payload_post_id,
                    # "fc" | "operator" | None (linked before the column
                    # existed — an operator accept, every one of them).
                    "linked_by": a.linked_by,
                    "token": (a.signals or {}).get("identity_token"),
                }
                for a in assocs
            ],
            "thumbnails": thumbnails,
            "variant_count": sum(1 for t in thumbnails if t["role"] == "variant"),
            "texts": texts,
        }


def fold_clause(fold_hours: float):
    """WHERE clause: this post is NOT a linked drop sitting beside its teaser.

    Operator: *"discord 'posts' land as normal and only hidden from the post
    view they're posted the same day."* Everything else stays — an older
    variant the teaser also references is history, and a reference does not
    remove it from history.

    Built on `post_date`/`downloaded_at`, not the feed's `resurfaced_at`-led
    sort key: whether two posts are the same release is a question about when
    they were published, not about where the feed has since moved one.
    """
    teaser = aliased(Post)
    # The SQL-standard EXTRACT(epoch FROM …), which every Postgres accepts.
    gap = func.abs(extract(
        "epoch",
        func.coalesce(Post.post_date, Post.downloaded_at)
        - func.coalesce(teaser.post_date, teaser.downloaded_at),
    ))
    return ~exists(
        select(PostAssociation.id)
        .join(teaser, teaser.id == PostAssociation.announcement_post_id)
        .where(and_(
            PostAssociation.payload_post_id == Post.id,
            PostAssociation.status == "linked",
            gap <= fold_hours * 3600,
        ))
    )