diff --git a/backend/app/api/posts.py b/backend/app/api/posts.py index 2823368..49fcf93 100644 --- a/backend/app/api/posts.py +++ b/backend/app/api/posts.py @@ -207,6 +207,8 @@ async def rescan_associations(): be within the window to exist at all); this is the button for a first run over a library that predates the feature.""" async with get_session() as session: - result = await association_rescan(session) + # full=True: the button reaches the whole history, which the hourly + # sweep's 48-hour horizon never does. + result = await association_rescan(session, full=True) await session.commit() return jsonify(result) diff --git a/backend/app/services/post_association_service.py b/backend/app/services/post_association_service.py index f5706c6..2fb0567 100644 --- a/backend/app/services/post_association_service.py +++ b/backend/app/services/post_association_service.py @@ -286,10 +286,28 @@ class PostAssociationService: paths_by_post: dict[int, list[str]] = {} hashes_by_post: dict[int, list[int]] = {} + # An image is counted under the post a reader SEES it on: a Discord + # message absorbed into a drop contributes to the drop, not to itself. + # + # Keyed on `primary_post_id` alone until 2026-09-24, which on the live + # instance meant a drop never had a name or a hash at all — its images + # are owned by its member messages, and the drop only claims them + # through provenance. The identity route therefore never fired there; + # the tests missed it because they attached images to the drop itself, + # which discord_grouping never does. + # + # Counting by the drop also makes the span honest: five wips of one + # piece posted as five messages and grouped into one drop are ONE post + # as far as "how many posts carry this name" is concerned. + owner = Post.__table__.alias("owner") rows = await self.session.execute( select( - ImageRecord.primary_post_id, ImageRecord.path, ImageRecord.phash - ).where( + func.coalesce(owner.c.absorbed_by_post_id, ImageRecord.primary_post_id), + ImageRecord.path, ImageRecord.phash, + ) + .select_from(ImageRecord) + .join(owner, owner.c.id == ImageRecord.primary_post_id) + .where( ImageRecord.artist_id == artist_id, ImageRecord.primary_post_id.is_not(None), ) @@ -333,44 +351,81 @@ class PostAssociationService: self._corpora[artist_id] = corpus return corpus - async def _decided(self, announcement_id: int) -> set[int]: - """Payload posts already proposed for this announcement, in ANY status. + async def _decided(self, announcement_id: int) -> dict[int, PostAssociation]: + """Pairs already recorded for this announcement, keyed by payload. - 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. + Linked and dismissed pairs are never touched again: re-proposing a pair + the operator has already rejected on every subsequent scan is the + single behaviour that makes a review queue get ignored. + + A PENDING pair is different — nobody has decided it — so the caller + re-scores it. Otherwise a pair queued by an older, weaker matcher sits + in the queue forever even once the evidence is conclusive, which is the + chore the operator asked FC not to hand them. """ rows = (await self.session.execute( - select(PostAssociation.payload_post_id) + select(PostAssociation) .where(PostAssociation.announcement_post_id == announcement_id) )).scalars().all() - return set(rows) + return {a.payload_post_id: a for a in rows} async def _candidate_groups( self, announcement: Post, *, window: timedelta, - ) -> list[Post]: - """Synthetic Discord groupings by the SAME artist, inside the window. + ) -> list[tuple[Post, datetime]]: + """Synthetic Discord groupings by the SAME artist with a message inside + the window — each with the time of its message CLOSEST to the teaser. 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. + + Measured on the MESSAGES, not the drop's own date. A drop is dated by + its first message, and since #4390 merges a creator's trickle into one + drop, that can be days before the release the teaser announces — + "Very early Marin" on Sep 7, `MarinaraSauce_base` on Sep 11. Matching + on the drop's date would put every merged trickle outside the window. """ at = _post_time(announcement) - sort_key = func.coalesce(Post.post_date, Post.downloaded_at) - return (await self.session.execute( - select(Post) + member = Post.__table__.alias("member") + member_at = func.coalesce(member.c.post_date, member.c.downloaded_at) + rows = list((await self.session.execute( + select(member.c.absorbed_by_post_id, member_at) .where( + member.c.artist_id == announcement.artist_id, + member.c.absorbed_by_post_id.is_not(None), + member_at >= at - window, + member_at <= at + window, + ) + )).all()) + # The drop's own date counts too — the first message's, so it adds + # nothing for a real drop, but it keeps a drop with no member rows + # (hand-built, or one whose messages were removed) matchable. + own_at = func.coalesce(Post.post_date, Post.downloaded_at) + rows += (await self.session.execute( + select(Post.id, own_at).where( Post.artist_id == announcement.artist_id, Post.synthesized_by == DROP_GROUPER, + own_at >= at - window, + own_at <= at + window, + ) + )).all() + closest: dict[int, datetime] = {} + for group_id, when in rows: + if group_id not in closest or abs(when - at) < abs(closest[group_id] - at): + closest[group_id] = when + if not closest: + return [] + groups = (await self.session.execute( + select(Post).where( + Post.id.in_(list(closest)), + 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() + ranked = sorted(groups, key=lambda g: (abs(closest[g.id] - at), g.id)) + return [(g, closest[g.id]) for g in ranked[:MAX_CANDIDATES]] async def _claimed(self, announcement_id: int, payload_id: int) -> bool: """Is either end of this pair already spoken for by an accepted link? @@ -415,8 +470,9 @@ class PostAssociationService: made = 0 scored: list[tuple[Post, float, dict, float]] = [] - for group in await self._candidate_groups(announcement, window=window): - if group.id in already: + for group, group_at in await self._candidate_groups(announcement, window=window): + prior = already.get(group.id) + if prior is not None and prior.status != "pending": continue named, token = shared_identity( here, @@ -435,7 +491,7 @@ class PostAssociationService: identity = max(named, copied) circumstantial = { "proximity": proximity_signal( - _post_time(group) - _post_time(announcement), window, + group_at - _post_time(announcement), window, ), "declared": declared, "marker": marker_overlap( @@ -494,6 +550,16 @@ class PostAssociationService: status = "linked" if group.id == auto_id else "pending" if status == "linked": linked += 1 + prior = already.get(group.id) + if prior is not None: + # Re-scored in place: a pending pair keeps its row (and id), + # and only an upgrade to linked counts as news. + prior.score = score + prior.signals = signals + if status == "linked": + prior.status = "linked" + prior.linked_by = "fc" + continue self.session.add(PostAssociation( announcement_post_id=announcement.id, payload_post_id=group.id, @@ -573,8 +639,17 @@ class PostAssociationService: return out -async def rescan(session: AsyncSession, *, now: datetime | None = None) -> dict: - """Score every recent non-synthetic post against nearby groupings.""" +async def rescan( + session: AsyncSession, *, now: datetime | None = None, full: bool = False, +) -> dict: + """Score recent non-synthetic posts against nearby groupings. + + `full=True` scores EVERY post by an artist who has Discord drops at all — + the manual button's job, for history that predates the feature or that a + trickle merge (#4390) has just rearranged. It used to share the sweep's + 48-hour horizon, so the button described as "a first run over a library + that predates the feature" could not reach that library. + """ settings = await ImportSettings.load(session) if not settings.discord_link_enabled: return {"enabled": False, "scanned": 0, "proposed": 0, "linked": 0} @@ -586,6 +661,18 @@ async def rescan(session: AsyncSession, *, now: datetime | None = None) -> dict: # 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) + if full: + with_drops = select(Post.artist_id).where( + Post.synthesized_by == DROP_GROUPER + ).distinct() + ids = set((await session.execute( + select(Post.id).where( + Post.synthesized_by.is_(None), + Post.absorbed_by_post_id.is_(None), + Post.artist_id.in_(with_drops), + ) + )).scalars().all()) + return await _score(session, settings, ids, window_hours) ids = set((await session.execute( select(Post.id).where( Post.synthesized_by.is_(None), @@ -605,14 +692,28 @@ async def rescan(session: AsyncSession, *, now: datetime | None = None) -> dict: # That is #4392's third cause, and it is the one that left a measured 0.800 # pair with an empty review queue on the live instance. The other two were # about scoring; this one meant nothing was scored at all. - drop_times = (await session.execute( - select(sort_key).where( + # + # The times are the drops' MESSAGES, not the drops' own dates: a drop that + # grew today by a trickle merge (#4390) is dated by its first stage, days + # earlier, while the teaser sits beside the message that just joined. + recent = ( + select(Post.id).where( Post.synthesized_by == DROP_GROUPER, func.coalesce(Post.last_grew_at, Post.downloaded_at) >= horizon, ) .order_by(func.coalesce(Post.last_grew_at, Post.downloaded_at).desc()) .limit(MAX_RECENT_DROPS) - )).scalars().all() + ) + member = Post.__table__.alias("member") + drop_times = set((await session.execute( + select(func.coalesce(member.c.post_date, member.c.downloaded_at)) + .where(member.c.absorbed_by_post_id.in_(recent)) + )).scalars().all()) + # Plus each drop's own date — its first message's, so a duplicate for a + # real drop, but what a drop with no member rows is placed by. + drop_times |= set((await session.execute( + select(sort_key).where(Post.id.in_(recent)) + )).scalars().all()) # The interval arithmetic is done in Python rather than SQL: a handful of # literal ranges is portable, and `now - INTERVAL` is not. ranges = [ @@ -629,6 +730,12 @@ async def rescan(session: AsyncSession, *, now: datetime | None = None) -> dict: ) )).scalars().all()) + return await _score(session, settings, ids, window_hours) + + +async def _score( + session: AsyncSession, settings: ImportSettings, ids: set[int], window_hours: float, +) -> dict: svc = PostAssociationService(session) proposed = 0 linked = 0 diff --git a/tests/test_post_association.py b/tests/test_post_association.py index ff9a890..2c0a4e3 100644 --- a/tests/test_post_association.py +++ b/tests/test_post_association.py @@ -950,3 +950,156 @@ def test_the_duplicate_threshold_sits_inside_the_measured_gap(): justifies it: 20 bits was the widest true pair, 108 the nearest unrelated one.""" assert 20 < DUPLICATE_MAX_DISTANCE < 108 + + +# --- drops shaped as the grouper actually writes them ---------------------- +# +# Every test above hand-builds a drop that OWNS its images. discord_grouping +# never does that: a drop's images belong to its member messages and the drop +# claims them through provenance. Against that shape the matcher saw no names +# and no hashes at all, so the identity route could not fire on the live +# instance — and nothing here could have noticed. + + +async def _real_drop(db, artist, discord, *, stages): + """Messages, each owning one named image, grouped by the real grouper and + merged by the real trickle pass. `stages` is [(at, name), ...].""" + from backend.app.services.discord_grouping import group_source, merge_trickles + + for i, (at, name) in enumerate(stages): + msg = Post(source_id=discord.id, artist_id=artist.id, + external_post_id=f"{artist.slug}-msg-{i}", post_date=at) + db.add(msg) + await db.flush() + vec = [0.0] * 1152 + vec[i % 1152] = 1.0 + db.add(ImageRecord( + path=f"/images/{artist.slug}/20260901_{1234567890000 + i}_01_{name}.png", + sha256=f"{artist.id:08d}{i:056d}", size_bytes=10, mime="image/png", + width=10, height=10, origin="downloaded", primary_post_id=msg.id, + artist_id=artist.id, siglip_embedding=vec, + )) + await db.flush() + await group_source(db, discord, max_distance=0.10, window_minutes=60) + await merge_trickles(db, discord, gap=timedelta(hours=168), min_images=2, + cooldown=timedelta(hours=24)) + await db.commit() + return (await db.execute( + select(Post).where(Post.source_id == discord.id, + Post.synthesized_by == DROP_GROUPER, + Post.absorbed_by_post_id.is_(None)) + )).scalars().all() + + +@pytest.mark.asyncio +async def test_a_drop_whose_images_belong_to_its_messages_still_links_on_its_name(db): + """The live shape. `0-k` on the teaser and on the drop's MESSAGE, nowhere + else — conclusive, so FC links it without asking.""" + artist, patreon, discord = await _artist_with_channels(db, "liveshape") + now = datetime.now(UTC) - timedelta(days=5) + (drop,) = await _real_drop(db, artist, discord, stages=[(now, "0-k_base")]) + teaser = await _teaser(db, artist, patreon, at=now + timedelta(hours=20), + body="new one", names=["0-k"]) + await db.commit() + + made = await PostAssociationService(db).match_post( + teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW, auto_link=True, + ) + await db.commit() + + assert made == (1, 1) + assoc = (await db.execute(select(PostAssociation))).scalar_one() + assert (assoc.payload_post_id, assoc.signals["identity_token"]) == (drop.id, "0-k") + + +@pytest.mark.asyncio +async def test_a_merged_trickle_is_found_by_its_latest_stage(db): + """A trickle merged by #4390 is dated by its FIRST stage — four days before + the release the teaser announces. Matching on the drop's own date would put + it outside the 24h window.""" + artist, patreon, discord = await _artist_with_channels(db, "trickleshape") + start = datetime.now(UTC) - timedelta(days=10) + (drop,) = await _real_drop(db, artist, discord, stages=[ + (start, "svtt_wip1"), (start + timedelta(days=2), "svtt_wip3"), + (start + timedelta(days=4), "svtt_drench_b"), + ]) + teaser = await _teaser(db, artist, patreon, at=start + timedelta(days=4, hours=1), + body="new one", names=["svtt_teaser"]) + await db.commit() + + made = await PostAssociationService(db).match_post( + teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW, auto_link=True, + ) + await db.commit() + + assert made[0] == 1 + assoc = (await db.execute(select(PostAssociation))).scalar_one() + assert assoc.payload_post_id == drop.id + # Timed against the closest message, an hour away — not the first, four + # days away. + assert assoc.signals["proximity"] > 0.9 + + +@pytest.mark.asyncio +async def test_a_pair_left_pending_is_linked_once_the_evidence_is_conclusive(db): + """Nobody decided a pending pair, so the matcher re-scores it. Otherwise a + pair queued by an older, weaker matcher waits for a click forever.""" + artist, patreon, discord = await _artist_with_channels(db, "pendingshape") + now = datetime.now(UTC) - timedelta(days=5) + (drop,) = await _real_drop(db, artist, discord, stages=[(now, "0-k_base")]) + teaser = await _teaser(db, artist, patreon, at=now + timedelta(hours=2), + body="new one", names=["0-k"]) + db.add(PostAssociation(announcement_post_id=teaser.id, payload_post_id=drop.id, + score=0.61, signals={"proximity": 0.9}, status="pending")) + await db.commit() + + await PostAssociationService(db).match_post( + teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW, auto_link=True, + ) + await db.commit() + + assoc = (await db.execute(select(PostAssociation))).scalar_one() + assert (assoc.status, assoc.linked_by) == ("linked", "fc") + + +@pytest.mark.asyncio +async def test_a_dismissed_pair_is_never_rescored(db): + artist, patreon, discord = await _artist_with_channels(db, "dismissedshape") + now = datetime.now(UTC) - timedelta(days=5) + (drop,) = await _real_drop(db, artist, discord, stages=[(now, "0-k_base")]) + teaser = await _teaser(db, artist, patreon, at=now + timedelta(hours=2), + body="new one", names=["0-k"]) + db.add(PostAssociation(announcement_post_id=teaser.id, payload_post_id=drop.id, + score=0.61, signals={}, status="dismissed")) + await db.commit() + + assert await PostAssociationService(db).match_post( + teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW, auto_link=True, + ) == (0, 0) + assoc = (await db.execute(select(PostAssociation))).scalar_one() + assert assoc.status == "dismissed" + + +@pytest.mark.asyncio +async def test_the_full_rescan_reaches_history_the_sweep_does_not(db): + """The "Scan now" button. It used to share the sweep's 48-hour horizon, so + it could not reach the library it was described as being for.""" + artist, patreon, discord = await _artist_with_channels(db, "historyshape") + long_ago = datetime.now(UTC) - timedelta(days=400) + (drop,) = await _real_drop(db, artist, discord, stages=[(long_ago, "0-k_base")]) + # Authored long ago too. A drop FC wrote TODAY is exactly what the sweep + # follows back to its teaser (#4392's third cause), so it would find this + # pair without the button. + drop.downloaded_at = long_ago + await _teaser(db, artist, patreon, at=long_ago + timedelta(hours=2), + body="new one", names=["0-k"]) + settings = await ImportSettings.load(db) + settings.discord_link_enabled = True + await db.commit() + + assert (await rescan(db))["proposed"] == 0 + await db.commit() + result = await rescan(db, full=True) + await db.commit() + + assert (result["proposed"], result["linked"]) == (1, 1)