diff --git a/backend/app/services/discord_grouping.py b/backend/app/services/discord_grouping.py index 44afda3..b8e68ab 100644 --- a/backend/app/services/discord_grouping.py +++ b/backend/app/services/discord_grouping.py @@ -63,7 +63,7 @@ from collections import Counter from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta -from sqlalchemy import Select, delete, func, select, update +from sqlalchemy import Select, delete, func, select, union, update from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession @@ -126,6 +126,31 @@ def cosine_distance(a, b) -> float: return 1.0 - (dot / (na * nb)) +def _message_images(posts): + """(post_id, image_id) for every image a message carries — owned AND re-posted. + + A message owns an image through `primary_post_id`, but only the FIRST + message imported with a given file does. The same file posted again is a + provenance link, and the backfill runs newest-first, so it is usually the + ORIGINAL message that ends up owning nothing. Reading ownership alone left + 101 of Yellowroom's messages (2018–2020 mostly) ungroupable: every image + they carried also sat in another message. + + `posts` is a list of ids or a select of them; filtering both branches by it + keeps the union to the messages in hand rather than the whole library. + Callers pass MESSAGE posts only — a drop's own provenance rows would read + as images it carries. + """ + owned = select( + ImageRecord.primary_post_id.label("post_id"), ImageRecord.id.label("image_id"), + ).where(ImageRecord.primary_post_id.in_(posts)) + reposted = select( + ImageProvenance.post_id.label("post_id"), + ImageProvenance.image_record_id.label("image_id"), + ).where(ImageProvenance.post_id.in_(posts)) + return union(owned, reposted).subquery() + + def _candidate_stmt(source_id: int, *, not_after: datetime) -> Select: """Ungrouped Discord message-posts, one representative image each, OLDEST FIRST — which is the order `build_groups` requires. @@ -143,13 +168,15 @@ def _candidate_stmt(source_id: int, *, not_after: datetime) -> Select: take the OLDEST candidates instead of the lowest-numbered ones. """ sort_key = func.coalesce(Post.post_date, Post.downloaded_at) + carried = _message_images(select(Post.id).where(Post.source_id == source_id)) inner = ( select( Post.id.label("post_id"), sort_key.label("occurred_at"), ImageRecord.siglip_embedding.label("embedding"), ) - .join(ImageRecord, ImageRecord.primary_post_id == Post.id) + .join(carried, carried.c.post_id == Post.id) + .join(ImageRecord, ImageRecord.id == carried.c.image_id) .where( Post.source_id == source_id, # Never absorb a post FC wrote, and never re-absorb one already @@ -295,9 +322,10 @@ async def _link_member_images( """ if not member_ids: return 0 - image_rows = (await session.execute( - select(ImageRecord.id).where(ImageRecord.primary_post_id.in_(member_ids)) - )).scalars().all() + carried = _message_images(member_ids) + image_rows = sorted(set((await session.execute( + select(carried.c.image_id) + )).scalars().all())) if not image_rows: return 0 await session.execute( @@ -421,9 +449,11 @@ async def _group_seed(session: AsyncSession, post_id: int) -> list[float] | None free to disagree; this way there is one. """ sort_key = func.coalesce(Post.post_date, Post.downloaded_at) + carried = _message_images(select(Post.id).where(Post.absorbed_by_post_id == post_id)) return (await session.execute( select(ImageRecord.siglip_embedding) - .join(Post, ImageRecord.primary_post_id == Post.id) + .join(carried, carried.c.image_id == ImageRecord.id) + .join(Post, Post.id == carried.c.post_id) .where( Post.absorbed_by_post_id == post_id, ImageRecord.siglip_embedding.is_not(None), @@ -725,12 +755,15 @@ async def _load_drops(session: AsyncSession, source: Source) -> list[_Drop]: names: dict[int, set[str]] = {pid: set() for pid in by_id} images: dict[int, list[int]] = {pid: [] for pid in by_id} if owner_of: - for iid, primary, path in (await session.execute( - select(ImageRecord.id, ImageRecord.primary_post_id, ImageRecord.path) - .where(ImageRecord.primary_post_id.in_(list(owner_of))) + carried = _message_images(list(owner_of)) + for iid, message, path in (await session.execute( + select(ImageRecord.id, carried.c.post_id, ImageRecord.path) + .join(carried, carried.c.image_id == ImageRecord.id) .order_by(ImageRecord.id) )).all(): - drop = owner_of[primary] + drop = owner_of[message] + if iid in images[drop]: + continue # one file carried by two of the drop's messages images[drop].append(iid) if (name := leading_name(path)) is not None: names[drop].add(name) @@ -844,7 +877,12 @@ async def merge_trickles( if drop.first_at - earlier.last_at > gap: continue shared = mine & gated(earlier.names) - if shared: + if set(drop.images) & set(earlier.images): + # The creator posted the very same file again — the strongest + # evidence there is, and one nearest-neighbour cannot see: it + # skips the image itself, which is the one they share. + targets[earlier.post.id] = (earlier, "same_image") + elif shared: targets[earlier.post.id] = (earlier, f"name:{min(shared)}") elif drop.nearest & earlier.members or (earlier.nearest or set()) & drop.members: targets[earlier.post.id] = (earlier, "nearest") diff --git a/tests/test_discord_trickles.py b/tests/test_discord_trickles.py index 76b47ad..513bce6 100644 --- a/tests/test_discord_trickles.py +++ b/tests/test_discord_trickles.py @@ -20,6 +20,7 @@ from sqlalchemy import select from backend.app.models import ( Artist, + ImageProvenance, ImageRecord, Post, PostAssociation, @@ -251,3 +252,63 @@ async def test_a_checked_drop_is_not_examined_again(db): for drop in await _drops(db, source): assert drop.synthesis_details["trickle_checked"] is True assert "nearest_message_ids" in drop.synthesis_details + + +# --- the same file posted twice ----------------------------------------------- +# 101 of Yellowroom's messages were never grouped: every image they carried also +# sat in another message, which the newest-first backfill had made its owner. + + +async def _repost(db, artist, source, *, at, of, text=None): + """A message carrying a file another message already owns.""" + n = next(_n) + post = Post(source_id=source.id, artist_id=artist.id, + external_post_id=f"msg-{n}", post_date=at, description=text) + db.add(post) + await db.flush() + image_id = (await db.execute( + select(ImageRecord.id).where(ImageRecord.primary_post_id == of.id) + )).scalar_one() + db.add(ImageProvenance(image_record_id=image_id, post_id=post.id, source_id=source.id)) + await db.flush() + return post + + +@pytest.mark.asyncio +async def test_a_message_that_only_reposts_an_image_is_still_grouped(db): + artist, source = await _seed(db, "repost-artist") + later = await _message(db, artist, source, at=T0, angle=0.0) + original = await _repost(db, artist, source, at=T0 - timedelta(days=400), of=later, + text="first posted here") + await db.commit() + + await group_source(db, source, max_distance=0.10, window_minutes=60) + await db.commit() + + await db.refresh(original) + assert original.absorbed_by_post_id is not None + # Grouping links, it never re-owns: the image keeps its primary post. + owner = (await db.execute( + select(ImageRecord.primary_post_id).join( + ImageProvenance, ImageProvenance.image_record_id == ImageRecord.id, + ).where(ImageProvenance.post_id == original.absorbed_by_post_id) + )).scalar_one() + assert owner == later.id + + +@pytest.mark.asyncio +async def test_the_same_file_posted_a_day_apart_is_one_drop(db): + """The live pair: 32554 (Aug 25) and 32553 (Aug 26) carry one file. + Nearest-neighbour cannot see it — it skips the image itself.""" + artist, source = await _seed(db, "twice-artist") + second = await _message(db, artist, source, at=T0 + timedelta(days=1), angle=0.0) + first = await _repost(db, artist, source, at=T0, of=second) + await db.commit() + + await _group_then_merge(db, source) + + (drop,) = await _drops(db, source) + assert set(drop.synthesis_details["member_post_ids"]) | { + m for r in drop.synthesis_details.get("merged", []) for m in r["message_ids"] + } >= {first.id, second.id} + assert drop.synthesis_details["merged"][0]["route"] == "same_image"