FabledCurator/tests/test_post_association.py

"""Milestone 388 E5: which Patreon post announced which Discord drop.

The failure this step must not have is a WRONG link. Telling the operator that
two different pieces are one is worse than telling them nothing — no link
leaves them where they already were, a wrong one actively misinforms. So most
of what follows pins refusals, and the central one is structural rather than
behavioural: the threshold sits above every single signal weight, which is what
makes "time proximity alone is never sufficient" arithmetic instead of a hope.
"""
from datetime import UTC, datetime, timedelta

import pytest
from sqlalchemy import select

from backend.app.models import (
    Artist,
    ImageRecord,
    ImportSettings,
    Post,
    PostAssociation,
    Source,
)
from backend.app.services.discord_grouping import DROP_GROUPER
from backend.app.services.post_association_service import (
    DECLARED_MENTION,
    WEIGHTS,
    PostAssociationService,
    declared_signal,
    proximity_signal,
    rescan,
    weighted_score,
)
from backend.app.services.post_feed_service import PostFeedService

pytestmark = pytest.mark.integration

DEFAULT_THRESHOLD = 0.60
WINDOW = 24.0


# --- the structural guard -------------------------------------------------


def test_no_single_signal_can_reach_the_threshold():
    """THE load-bearing property of this matcher.

    On a busy day an artist posts several times, so a matcher that could pair
    on proximity alone would turn every busy day into false pairs and the
    review queue would be abandoned. Requiring two signals is what prevents
    that — and it is a fact about the WEIGHTS, not about any code path, so it
    survives every refactor of the scorer.

    If this fails, either a weight grew or the default threshold dropped. Do
    not "fix" it by lowering the assertion; the arithmetic IS the safeguard.
    """
    assert max(WEIGHTS.values()) < DEFAULT_THRESHOLD
    assert sum(WEIGHTS.values()) == pytest.approx(1.0)


def test_perfect_proximity_alone_does_not_propose():
    """The same property, expressed through the scorer."""
    score = weighted_score({"proximity": 1.0, "declared": 0.0})
    assert score < DEFAULT_THRESHOLD


def test_an_explicit_declaration_alone_does_not_propose():
    score = weighted_score({"proximity": 0.0, "declared": 1.0})
    assert score < DEFAULT_THRESHOLD


def test_both_signals_together_do_propose():
    assert weighted_score({"proximity": 1.0, "declared": 1.0}) >= DEFAULT_THRESHOLD


# --- the signals ----------------------------------------------------------


def test_proximity_decays_to_zero_at_the_window_edge():
    window = timedelta(hours=24)
    assert proximity_signal(timedelta(0), window) == 1.0
    assert proximity_signal(timedelta(hours=24), window) == 0.0
    assert proximity_signal(timedelta(hours=48), window) == 0.0
    assert 0.4 < proximity_signal(timedelta(hours=12), window) < 0.6


def test_proximity_is_symmetric_because_either_can_land_first():
    """The teaser usually goes up around the drop, not reliably before it."""
    window = timedelta(hours=24)
    assert proximity_signal(timedelta(hours=-2), window) == proximity_signal(
        timedelta(hours=2), window
    )


def test_an_invite_link_scores_higher_than_a_bare_mention():
    assert declared_signal("Full set on my discord.gg/abc123 now!") == 1.0
    assert 0 < declared_signal("Posted the rest on discord earlier") < 1.0
    assert declared_signal("New piece, hope you like it") == 0.0
    assert declared_signal(None) == 0.0


def test_the_declaration_signal_reads_through_html():
    """Post bodies are HTML; a link inside an anchor tag must still count."""
    assert declared_signal(
        '<p>Full set: <a href="https://discord.gg/xyz">here</a></p>'
    ) > 0


def test_a_word_containing_discord_is_not_a_mention():
    """\"discordant\" is not a declaration. Without the word boundary the
    signal fires on ordinary prose and drags pairs over the threshold."""
    assert declared_signal("a discordant palette, deliberately") == 0.0


# --- end to end -----------------------------------------------------------


async def _artist_with_channels(db, name: str):
    artist = Artist(name=name, slug=name.lower().replace(" ", "-"))
    db.add(artist)
    await db.flush()
    patreon = Source(
        artist_id=artist.id, platform="patreon",
        url=f"https://patreon.com/{name}", enabled=True,
    )
    discord = Source(
        artist_id=artist.id, platform="discord",
        url=f"https://discord.com/channels/1/{name}", enabled=True,
    )
    db.add_all([patreon, discord])
    await db.flush()
    return artist, patreon, discord


async def _images(db, artist, post, ext, names):
    """Attach named files to a post. The NAME is the point — the working-name
    signal reads it, so a test that cares about identity supplies one."""
    for i, name in enumerate(names):
        db.add(ImageRecord(
            path=f"/images/{artist.id}/{ext}_{i}_{name}.jpg",
            sha256=f"{ext}{i}{name}".ljust(64, "0")[:64],
            size_bytes=10, mime="image/jpeg", width=10, height=10,
            origin="downloaded", primary_post_id=post.id, artist_id=artist.id,
        ))
    await db.flush()


async def _teaser(db, artist, source, *, at, body, ext="teaser", names=None,
                  title="New piece"):
    post = Post(
        source_id=source.id, artist_id=artist.id, external_post_id=ext,
        post_date=at, post_title=title, description=body,
    )
    db.add(post)
    await db.flush()
    await _images(db, artist, post, ext, names or [ext])
    return post


async def _drop(db, artist, source, *, at, ext="fc-drop:1", body=None,
                names=()):
    post = Post(
        source_id=source.id, artist_id=artist.id, external_post_id=ext,
        post_date=at, description=body, synthesized_by=DROP_GROUPER,
        synthesis_details={"message_count": 4, "images_since_surface": 0},
    )
    db.add(post)
    await db.flush()
    if names:
        await _images(db, artist, post, ext.replace(":", "-"), names)
    return post


@pytest.mark.asyncio
async def test_a_teaser_and_its_drop_an_hour_apart_are_proposed(db):
    artist, patreon, discord = await _artist_with_channels(db, "pairartist")
    now = datetime.now(UTC)
    teaser = await _teaser(
        db, artist, patreon, at=now - timedelta(hours=3),
        body="Full set is up on discord.gg/abc now",
    )
    drop = await _drop(db, artist, discord, at=now - timedelta(hours=2))
    await db.commit()

    made = await PostAssociationService(db).match_post(
        teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW,
    )
    await db.commit()
    assert made == 1

    assoc = (await db.execute(select(PostAssociation))).scalar_one()
    assert assoc.announcement_post_id == teaser.id
    assert assoc.payload_post_id == drop.id
    assert assoc.status == "pending", "nothing is linked without the operator"
    # The per-signal breakdown survives, so the proposal stays explicable
    # after the weights or threshold move.
    assert assoc.signals["declared"] == 1.0
    assert assoc.signals["proximity"] > 0.9


@pytest.mark.asyncio
async def test_two_unrelated_posts_the_same_day_are_not_proposed(db):
    """Time proximity alone must not be sufficient, or every busy day becomes
    a false pair. The teaser here says nothing about Discord."""
    artist, patreon, discord = await _artist_with_channels(db, "busyartist")
    now = datetime.now(UTC)
    teaser = await _teaser(
        db, artist, patreon, at=now - timedelta(hours=3),
        body="Just a sketch I liked",
    )
    await _drop(db, artist, discord, at=now - timedelta(hours=2))
    await db.commit()

    made = await PostAssociationService(db).match_post(
        teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW,
    )
    await db.commit()
    assert made == 0
    assert (await db.execute(select(PostAssociation))).scalars().all() == []


@pytest.mark.asyncio
async def test_a_drop_outside_the_window_is_not_proposed(db):
    artist, patreon, discord = await _artist_with_channels(db, "farapartartist")
    now = datetime.now(UTC)
    teaser = await _teaser(
        db, artist, patreon, at=now - timedelta(days=10),
        body="Everything is on discord.gg/abc",
    )
    await _drop(db, artist, discord, at=now)
    await db.commit()

    assert await PostAssociationService(db).match_post(
        teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW,
    ) == 0


@pytest.mark.asyncio
async def test_another_artists_drop_is_never_proposed(db):
    """Same-artist is a HARD filter, not a scored signal: two different
    creators posting minutes apart is a coincidence, not evidence."""
    artist_a, patreon_a, _ = await _artist_with_channels(db, "artista")
    _artist_b, _patreon_b, discord_b = await _artist_with_channels(db, "artistb")
    now = datetime.now(UTC)
    teaser = await _teaser(
        db, artist_a, patreon_a, at=now - timedelta(hours=1),
        body="new drop on discord.gg/abc",
    )
    await _drop(db, _artist_b, discord_b, at=now)
    await db.commit()

    assert await PostAssociationService(db).match_post(
        teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW,
    ) == 0


@pytest.mark.asyncio
async def test_an_artist_with_no_discord_source_produces_nothing_and_no_error(db):
    artist = Artist(name="soloartist", slug="soloartist")
    db.add(artist)
    await db.flush()
    patreon = Source(
        artist_id=artist.id, platform="patreon",
        url="https://patreon.com/solo", enabled=True,
    )
    db.add(patreon)
    await db.flush()
    teaser = await _teaser(
        db, artist, patreon, at=datetime.now(UTC),
        body="on discord.gg/abc", ext="solo",
    )
    await db.commit()

    assert await PostAssociationService(db).match_post(
        teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW,
    ) == 0


@pytest.mark.asyncio
async def test_a_synthetic_post_cannot_announce_anything(db):
    """FC wrote it, so it announces nothing — and a grouping proposing itself
    as the teaser for another grouping would be pure noise."""
    artist, _patreon, discord = await _artist_with_channels(db, "noselfannounce")
    now = datetime.now(UTC)
    drop_a = await _drop(db, artist, discord, at=now - timedelta(hours=1), ext="fc-drop:a")
    await _drop(db, artist, discord, at=now, ext="fc-drop:b")
    await db.commit()

    assert await PostAssociationService(db).match_post(
        drop_a.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW,
    ) == 0


@pytest.mark.asyncio
async def test_a_dismissed_pair_is_never_proposed_again(db):
    """The row is what remembers the rejection. Re-proposing a rejected pair on
    every scan is the single behaviour that makes a review queue get ignored."""
    artist, patreon, discord = await _artist_with_channels(db, "dismissartist")
    now = datetime.now(UTC)
    teaser = await _teaser(
        db, artist, patreon, at=now - timedelta(hours=2),
        body="discord.gg/abc has the rest",
    )
    await _drop(db, artist, discord, at=now - timedelta(hours=1))
    await db.commit()

    svc = PostAssociationService(db)
    assert await svc.match_post(
        teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW,
    ) == 1
    await db.commit()

    assoc = (await db.execute(select(PostAssociation))).scalar_one()
    await svc.dismiss(assoc.id)
    await db.commit()

    assert await svc.match_post(
        teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW,
    ) == 0
    await db.commit()
    assert len((await db.execute(select(PostAssociation))).scalars().all()) == 1


@pytest.mark.asyncio
async def test_only_an_accepted_link_reaches_the_post_payload(db):
    """A pending proposal is a question for the review queue, not a claim to
    render beside the artwork."""
    artist, patreon, discord = await _artist_with_channels(db, "payloadartist")
    now = datetime.now(UTC)
    teaser = await _teaser(
        db, artist, patreon, at=now - timedelta(hours=2),
        body="rest is on discord.gg/abc",
    )
    drop = await _drop(db, artist, discord, at=now - timedelta(hours=1))
    await db.commit()

    svc = PostAssociationService(db)
    await svc.match_post(teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW)
    await db.commit()

    feed = PostFeedService(db)
    assert (await feed.get_post(teaser.id))["associations"] == []

    assoc = (await db.execute(select(PostAssociation))).scalar_one()
    await svc.accept(assoc.id)
    await db.commit()

    # Both ends see it, and each sees the OTHER post with its own role.
    teaser_item = await feed.get_post(teaser.id)
    assert teaser_item["associations"] == [
        {"role": "announces", "post_id": drop.id, "id": assoc.id}
    ]
    drop_item = await feed.get_post(drop.id)
    assert drop_item["associations"] == [
        {"role": "announced_by", "post_id": teaser.id, "id": assoc.id}
    ]


@pytest.mark.asyncio
async def test_deleting_the_grouping_takes_its_proposals_with_it(db):
    """E3's reversal path is one DELETE; it must not leave a dangling proposal
    pointing at a post that no longer exists."""
    artist, patreon, discord = await _artist_with_channels(db, "cascadeartist")
    now = datetime.now(UTC)
    teaser = await _teaser(
        db, artist, patreon, at=now - timedelta(hours=2),
        body="discord.gg/abc",
    )
    drop = await _drop(db, artist, discord, at=now - timedelta(hours=1))
    await db.commit()

    await PostAssociationService(db).match_post(
        teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW,
    )
    await db.commit()
    assert len((await db.execute(select(PostAssociation))).scalars().all()) == 1

    await db.delete(drop)
    await db.commit()
    assert (await db.execute(select(PostAssociation))).scalars().all() == []


@pytest.mark.asyncio
async def test_the_rescan_is_a_no_op_when_the_switch_is_off(db):
    settings = await ImportSettings.load(db)
    settings.discord_link_enabled = False
    await db.commit()
    assert await rescan(db) == {"enabled": False, "scanned": 0, "proposed": 0}


# --- the identity route ----------------------------------------------------
#
# Milestone 388. Everything below was calibrated against the operator's real
# library (artist 8: 520 images, 15 same-artist pairs that share a name, no
# false positives) rather than invented, so the numbers in these docstrings are
# measurements.


@pytest.mark.asyncio
async def test_a_shared_working_name_proposes_a_pair_time_cannot_reach(db):
    """The reason identity is a route and not a fourth weight.

    These two are 20 hours apart and the teaser says nothing about Discord, so
    every circumstantial signal is near zero — proximity scores 0.17 and the
    weighted total 0.075, a long way under the bar. What links them is that
    the creator exported both from one file and the name came along.

    Measured equivalents on the live instance: `ConnFront` ↔ `ConnFront` and
    `Thicc Tomboy scene2` ↔ `Thicc_Tomboy_scene2`, the latter 23.8 hours apart
    at a proximity of 0.005.
    """
    artist, patreon, discord = await _artist_with_channels(db, "namedartist")
    now = datetime.now(UTC)
    teaser = await _teaser(
        db, artist, patreon, at=now - timedelta(hours=20),
        body="a little preview", names=["ConnFront"],
    )
    drop = await _drop(db, artist, discord, at=now, names=["01_ConnFront"])
    await db.commit()

    made = await PostAssociationService(db).match_post(
        teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW,
    )
    await db.commit()

    assert made == 1
    assoc = (await db.execute(select(PostAssociation))).scalar_one()
    assert assoc.payload_post_id == drop.id
    assert assoc.signals["identity"] == 1.0
    assert weighted_score(assoc.signals) < DEFAULT_THRESHOLD, (
        "the circumstantial bundle alone must NOT reach the bar here — if it "
        "does, this test has stopped proving what it claims to"
    )


@pytest.mark.asyncio
async def test_the_proposal_says_which_name_it_matched_on(db):
    """A review queue that cannot explain itself is one the operator learns to
    click through without reading."""
    artist, patreon, discord = await _artist_with_channels(db, "explainartist")
    now = datetime.now(UTC)
    teaser = await _teaser(
        db, artist, patreon, at=now - timedelta(hours=20),
        body="a little preview", names=["LoisLaneTB2"],
    )
    await _drop(db, artist, discord, at=now, names=["01_LoisLaneTB2"])
    await db.commit()

    await PostAssociationService(db).match_post(
        teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW,
    )
    await db.commit()

    assoc = (await db.execute(select(PostAssociation))).scalar_one()
    assert assoc.signals["identity_token"] == "loislanetb2"


@pytest.mark.asyncio
async def test_a_name_the_creator_reuses_everywhere_links_nothing(db):
    """THE false-positive guard for this route.

    A character name is not an identity. Measured on artist 8: `anya` is on
    images in 8 posts, `riju` 9, `bea` 16 — ungated, every Anya post would
    match every Anya drop, and the matcher would confidently assert that a
    year of unrelated pieces are all the same piece.
    """
    artist, patreon, discord = await _artist_with_channels(db, "habitartist")
    now = datetime.now(UTC)
    teaser = await _teaser(
        db, artist, patreon, at=now - timedelta(hours=20),
        body="a little preview", ext="t0", names=["Anya"],
    )
    await _drop(db, artist, discord, at=now, names=["01_Anya"])
    # The same name across enough of this artist's OTHER posts to make it a
    # habit. Nothing about the pair above changes; only its context does.
    for i in range(7):
        await _teaser(
            db, artist, patreon, at=now - timedelta(days=30 + i),
            body="older", ext=f"other{i}", names=["Anya"],
        )
    await db.commit()

    made = await PostAssociationService(db).match_post(
        teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW,
    )
    await db.commit()

    assert made == 0
    assert (await db.execute(select(PostAssociation))).scalars().all() == []


@pytest.mark.asyncio
async def test_a_name_below_the_floor_is_recorded_but_carries_nothing(db):
    """Identity may override, never dilute. Below the floor it is real
    evidence that is not strong enough to assert sameness on its own, so it is
    written down for the operator and moves no score.

    The measured cost of the floor sitting at 0.75: artist 8's `680lc` and
    `cnni18x` pairs each span four posts and are genuine, and neither will
    propose on this signal alone.
    """
    artist, patreon, discord = await _artist_with_channels(db, "floorartist")
    now = datetime.now(UTC)
    teaser = await _teaser(
        db, artist, patreon, at=now - timedelta(hours=20),
        body="a little preview", ext="t0", names=["cnni18x"],
    )
    await _drop(db, artist, discord, at=now, names=["01_cnni18x"])
    for i in range(2):
        await _teaser(
            db, artist, patreon, at=now - timedelta(days=30 + i),
            body="older", ext=f"other{i}", names=["cnni18x"],
        )
    await db.commit()

    svc = PostAssociationService(db)
    made = await svc.match_post(
        teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW,
    )
    await db.commit()

    assert made == 0
    corpus = await svc._corpus(artist.id)
    assert corpus.token_posts["cnni18x"] == 4, "four posts carry the name"


# --- the marker, and the vocabulary ----------------------------------------


@pytest.mark.asyncio
async def test_a_marker_the_creator_uses_once_lifts_a_pair_over_the_line(db):
    """The operator's own tie-back: `🍈🍈` in the Patreon title and
    `@everyone 🍈 🍈` in the Discord message. No vocabulary list would predict
    it, and no filename carries it.

    Measured here: 0.446 on proximity alone, 0.646 once the marker counts —
    the bar is 0.60. The mirror case is the guard below it.
    """
    artist, patreon, discord = await _artist_with_channels(db, "markerartist")
    now = datetime.now(UTC)
    teaser = await _teaser(
        db, artist, patreon, at=now - timedelta(minutes=12),
        body="it's up", title="Anya -- \U0001F348\U0001F348",
    )
    await _drop(db, artist, discord, at=now, body="@everyone \U0001F348 \U0001F348")
    await db.commit()

    made = await PostAssociationService(db).match_post(
        teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW,
    )
    await db.commit()

    assert made == 1
    assoc = (await db.execute(select(PostAssociation))).scalar_one()
    assert assoc.signals["marker"] == 1.0


@pytest.mark.asyncio
async def test_a_marker_the_creator_uses_constantly_lifts_nothing(db):
    """The same arrangement, with the marker turned into punctuation.

    This is a real measured proposal, not a hypothetical: 💦 is in 13 of artist
    8's 300 posts and it was the DECIDING term for a pair that proximity alone
    scored 0.441. A habitual marker riding along with proximity is just
    proximity wearing a hat, which the threshold sits above 0.45 to prevent.
    """
    artist, patreon, discord = await _artist_with_channels(db, "punctartist")
    now = datetime.now(UTC)
    teaser = await _teaser(
        db, artist, patreon, at=now - timedelta(minutes=12),
        body="it's up", title="Drizzle \U0001F4A6", ext="t0",
    )
    await _drop(db, artist, discord, at=now, body="@everyone \U0001F4A6")
    for i in range(5):
        await _teaser(
            db, artist, patreon, at=now - timedelta(days=30 + i),
            body="\U0001F4A6", title=f"older {i}", ext=f"other{i}",
        )
    await db.commit()

    made = await PostAssociationService(db).match_post(
        teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW,
    )
    await db.commit()

    assert made == 0


def test_the_creators_own_phrasing_counts_as_a_declaration():
    """`discord` is not how these creators actually write once the audience
    already knows where the server is. Measured on artist 8: 21 of 42 posts
    say `discord`, 7 say `the server`, and it is the RECENT ones that say the
    latter — so a vocabulary list written from old posts silently stops
    matching the posts that still need it."""
    assert declared_signal("<p>Full res is on the server</p>") == DECLARED_MENTION
    assert declared_signal("<p>up on our server now</p>") == DECLARED_MENTION
    assert declared_signal("<p>posted to my server</p>") == DECLARED_MENTION


def test_an_unrelated_server_is_still_not_a_declaration():
    """The widened vocabulary must not widen into ordinary prose."""
    assert declared_signal("<p>the servers were down all morning</p>") == 0.0