feat(maintenance): reconcile duplicate posts (gallery-dl→native unify)

An artist first downloaded by gallery-dl gets Post rows keyed by the per-
attachment id; a later native walk keys the SAME real post by the post id. They
never dedup (uq_post_source_external_id is on external_post_id) → duplicate post
rows (cheunart: 943→1109). The real post id is recoverable in-DB from
raw_metadata['post_id'] (both eras store the sidecar there).

reconcile_duplicate_posts (cleanup_service): group posts by (source_id, canonical
post_id = raw_metadata.post_id else external_post_id); for each group >1, keep the
row already keyed by the post id (the format the CURRENT native downloader
produces, so future walks dedup and this can't recur), re-point
ImageRecord.primary_post_id / ImageProvenance / PostAttachment / ExternalLink onto
it conflict-safe (drop the loser's row where the keeper already has the equivalent,
per each table's uniqueness), backfill the keeper's empty date/title/body/raw_meta
from a loser, set external_post_id=post_id + derive post_url, delete losers.
IMAGES ARE NOT TOUCHED (content-addressed/deduped already; operator-confirmed).

Preview/apply share find_duplicate_post_groups (rule 93). API
/api/admin/posts/reconcile-duplicates (dry_run→{groups,posts_to_merge,sample};
apply→{groups,merged,sample}; optional source_id). UI: a second section on
PostMaintenanceCard (preview groups+sample → confirm merge). Tests: merge +
metadata backfill + image move, no-op when unique, provenance-collision dedup.

Design: milestone #73. Forensics: note #917. Out of scope (flagged): cheunart vs
Cheunart case-variant artist dirs/rows.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

tests/test_cleanup_service.py

@@ -5,6 +5,8 @@ side effects use tmp_path. Assertions on mutated rows use COLUMN
 SELECTS per reference_async_coredml_test_assertions — never
 re-read ORM attributes after a service mutates and re-fetches.
 """
+from datetime import UTC, datetime
+
 import pytest
 from sqlalchemy import func, select
 
@@ -767,3 +769,87 @@ def test_purge_gated_previews_no_gated_posts_is_noop(db_sync, tmp_path):
     assert out["matched"] == 0
     assert out["deleted"] == 0
     assert out["gated_posts"] == 0
+
+
+# ---- duplicate-post reconciliation (gallery-dl → native, milestone #73) ----
+
+
+def test_reconcile_merges_attachmentid_and_postid_dupes(db_sync, tmp_path):
+    # gallery-dl row (keyed by attachment id, holds the image + body/date) +
+    # native row (keyed by the real post id, bare) → unified onto ONE post-id-
+    # keyed keeper, image moved over, keeper backfilled from the legacy row.
+    a = _make_artist(db_sync, slug="rc1")
+    img = _make_image(db_sync, artist=a, path=str(tmp_path / "x.jpg"), sha256="c" * 64)
+    legacy = Post(
+        artist_id=a.id, external_post_id="711509",
+        raw_metadata={"post_id": 1923726, "category": "subscribestar"},
+        description="real body", post_title="T",
+        post_date=datetime(2025, 6, 20, tzinfo=UTC),
+    )
+    native = Post(
+        artist_id=a.id, external_post_id="1923726",
+        raw_metadata={"post_id": 1923726, "category": "subscribestar"},
+    )
+    db_sync.add_all([legacy, native])
+    db_sync.flush()
+    img.primary_post_id = legacy.id
+    img_id, native_id, legacy_id = img.id, native.id, legacy.id
+    db_sync.commit()
+
+    prev = cleanup_service.reconcile_duplicate_posts(db_sync, dry_run=True)
+    assert prev["groups"] == 1
+    assert prev["posts_to_merge"] == 1
+    res = cleanup_service.reconcile_duplicate_posts(db_sync, dry_run=False)
+    assert res["merged"] == 1
+
+    rows = db_sync.execute(select(Post.id, Post.external_post_id)).all()
+    assert len(rows) == 1
+    keeper_id, keeper_epid = rows[0]
+    assert keeper_id == native_id  # native (post-id-keyed) row survives
+    assert keeper_epid == "1923726"
+    # image moved to keeper; legacy row gone; keeper backfilled.
+    assert db_sync.execute(
+        select(ImageRecord.primary_post_id).where(ImageRecord.id == img_id)
+    ).scalar_one() == native_id
+    assert db_sync.execute(
+        select(func.count()).select_from(Post).where(Post.id == legacy_id)
+    ).scalar_one() == 0
+    desc, pdate = db_sync.execute(
+        select(Post.description, Post.post_date).where(Post.id == native_id)
+    ).one()
+    assert desc == "real body"
+    assert pdate is not None
+
+
+def test_reconcile_noop_when_posts_unique(db_sync):
+    a = _make_artist(db_sync, slug="rc2")
+    db_sync.add_all([
+        Post(artist_id=a.id, external_post_id="100", raw_metadata={"post_id": 100}),
+        Post(artist_id=a.id, external_post_id="200", raw_metadata={"post_id": 200}),
+    ])
+    db_sync.commit()
+    assert cleanup_service.reconcile_duplicate_posts(db_sync, dry_run=True)["groups"] == 0
+    res = cleanup_service.reconcile_duplicate_posts(db_sync, dry_run=False)
+    assert res["merged"] == 0
+    assert db_sync.execute(select(func.count()).select_from(Post)).scalar_one() == 2
+
+
+def test_reconcile_dedups_provenance_collision(db_sync, tmp_path):
+    # Both dup rows link the SAME image via provenance; merging must DROP the
+    # colliding loser row (unique image_record_id+post_id), not raise.
+    a = _make_artist(db_sync, slug="rc3")
+    img = _make_image(db_sync, artist=a, path=str(tmp_path / "y.jpg"), sha256="d" * 64)
+    legacy = Post(artist_id=a.id, external_post_id="9", raw_metadata={"post_id": 55})
+    native = Post(artist_id=a.id, external_post_id="55", raw_metadata={"post_id": 55})
+    db_sync.add_all([legacy, native])
+    db_sync.flush()
+    db_sync.add(ImageProvenance(image_record_id=img.id, post_id=legacy.id))
+    db_sync.add(ImageProvenance(image_record_id=img.id, post_id=native.id))
+    img_id, native_id = img.id, native.id
+    db_sync.commit()
+
+    cleanup_service.reconcile_duplicate_posts(db_sync, dry_run=False)
+    prov = db_sync.execute(
+        select(ImageProvenance.post_id).where(ImageProvenance.image_record_id == img_id)
+    ).scalars().all()
+    assert prov == [native_id]