feat(zot): merge-based URL dedupe engine — union facets, keep-earliest (refs #665)

src/zot/merge.py

@@ -0,0 +1,215 @@
+"""Merge-based Zotero dedupe.
+
+Identity = the ``url`` itemData field (only ``http*``/``email:*``
+values). The keeper is the lowest ``itemID`` — the row a deterministic
+``find_item_by_url`` returns, i.e. the copy ``bib.sync`` has been
+updating. Losers are merged facet-by-facet into the keeper, then
+hard-deleted; every decision is logged as JSON before any delete.
+
+Born from the 2026-05-14 incident where a key-matched restore from
+``pre-rekey.bak`` re-inserted ~9k rekeyed rows as URL-twins (issue
+#665; plan: docs/superpowers/plans/2026-08-26-zotero-dedupe.md).
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from collections import defaultdict
+from pathlib import Path
+
+from zot.db import Db, now_iso
+
+URL_FIELD_PREFIXES = ("http", "email:")
+
+
+def _note_hash(text: str) -> str:
+    return hashlib.sha1(" ".join((text or "").split()).encode()).hexdigest()
+
+
+def merge_into(db: Db, keeper_id: int, loser_id: int) -> dict[str, int]:
+    """Union loser's facets into keeper, then delete the loser row.
+
+    Facets unioned: tags, collection memberships, relations, creators
+    (only if keeper has none), and child notes/attachments (reparented
+    to keeper, deduped by note-text hash / attachment path so identical
+    children aren't doubled). Keeper's own EAV fields (title, etc.) are
+    left untouched — keeper wins on every field. Every loser child note
+    or attachment is either reparented onto keeper or deleted as a dup
+    before the loser row itself is deleted, so no orphan rows are left
+    hanging off the loser once ``db.delete_item`` removes it.
+    """
+    out = {
+        "tags_added": 0,
+        "collections_added": 0,
+        "notes_adopted": 0,
+        "attachments_adopted": 0,
+        "relations_moved": 0,
+    }
+    con = db.con
+
+    for (tag_id,) in con.execute(
+        "SELECT tagID FROM itemTags WHERE itemID=?", (loser_id,)
+    ).fetchall():
+        cur = con.execute(
+            "INSERT OR IGNORE INTO itemTags (itemID, tagID, type) VALUES (?,?,0)",
+            (keeper_id, tag_id),
+        )
+        out["tags_added"] += cur.rowcount
+
+    for (coll_id,) in con.execute(
+        "SELECT collectionID FROM collectionItems WHERE itemID=?", (loser_id,)
+    ).fetchall():
+        cur = con.execute(
+            "INSERT OR IGNORE INTO collectionItems (collectionID, itemID) VALUES (?,?)",
+            (coll_id, keeper_id),
+        )
+        out["collections_added"] += cur.rowcount
+
+    keeper_notes = {
+        _note_hash(r[0])
+        for r in con.execute(
+            "SELECT note FROM itemNotes WHERE parentItemID=?", (keeper_id,)
+        )
+    }
+    for note_item, note in con.execute(
+        "SELECT itemID, note FROM itemNotes WHERE parentItemID=?", (loser_id,)
+    ).fetchall():
+        if _note_hash(note) in keeper_notes:
+            db.delete_item(note_item)
+            continue
+        con.execute(
+            "UPDATE itemNotes SET parentItemID=? WHERE itemID=?",
+            (keeper_id, note_item),
+        )
+        out["notes_adopted"] += 1
+
+    keeper_atts = {
+        r[0]
+        for r in con.execute(
+            "SELECT COALESCE(path,'') FROM itemAttachments WHERE parentItemID=?",
+            (keeper_id,),
+        )
+    }
+    for att_item, path in con.execute(
+        "SELECT itemID, COALESCE(path,'') FROM itemAttachments WHERE parentItemID=?",
+        (loser_id,),
+    ).fetchall():
+        if path and path in keeper_atts:
+            db.delete_item(att_item)
+            continue
+        con.execute(
+            "UPDATE itemAttachments SET parentItemID=? WHERE itemID=?",
+            (keeper_id, att_item),
+        )
+        out["attachments_adopted"] += 1
+
+    cur = con.execute(
+        "UPDATE OR IGNORE itemRelations SET itemID=? WHERE itemID=?",
+        (keeper_id, loser_id),
+    )
+    out["relations_moved"] = cur.rowcount
+
+    # Keeper has no creators but loser does → adopt them wholesale.
+    has_keeper = con.execute(
+        "SELECT 1 FROM itemCreators WHERE itemID=? LIMIT 1", (keeper_id,)
+    ).fetchone()
+    if not has_keeper:
+        con.execute(
+            "UPDATE OR IGNORE itemCreators SET itemID=? WHERE itemID=?",
+            (keeper_id, loser_id),
+        )
+
+    db.delete_item(loser_id)
+    return out
+
+
+def find_url_clusters(db: Db) -> list[list[tuple[int, str, int]]]:
+    """Group top-level items sharing an ``http*``/``email:*`` url.
+
+    Excludes notes and attachments (their itemID never sets a ``url``
+    field but they're excluded defensively). Each cluster is sorted by
+    ascending ``itemID`` — the first element is the merge keeper.
+    """
+    rows = db.con.execute(
+        """
+        SELECT i.itemID, i.key, i.itemTypeID, idv.value
+        FROM items i
+        JOIN itemData id ON id.itemID = i.itemID
+        JOIN fields f ON f.fieldID = id.fieldID AND f.fieldName = 'url'
+        JOIN itemDataValues idv ON idv.valueID = id.valueID
+        WHERE i.itemID NOT IN (SELECT itemID FROM deletedItems)
+          AND i.itemID NOT IN (SELECT itemID FROM itemAttachments)
+          AND i.itemID NOT IN (SELECT itemID FROM itemNotes)
+        ORDER BY i.itemID
+        """
+    ).fetchall()
+    by_url: dict[str, list[tuple[int, str, int]]] = defaultdict(list)
+    for item_id, key, type_id, url in rows:
+        if url and url.startswith(URL_FIELD_PREFIXES):
+            by_url[url].append((item_id, key, type_id))
+    return [v for v in by_url.values() if len(v) > 1]
+
+
+def merge_url_duplicates(
+    db_path: str | Path,
+    *,
+    dry_run: bool = True,
+    log_dir: str | Path = ".state/zotero-dedupe",
+) -> dict:
+    """Driver: find URL-duplicate clusters and merge each into its keeper.
+
+    Clusters whose members don't all share the keeper's ``itemTypeID``
+    are skipped (counted in ``skipped_type_mismatch``) rather than
+    merged, since a type mismatch usually means the "duplicate" isn't
+    actually the same record. Writes a JSON decision log to *log_dir*
+    (one entry per cluster) before any delete happens, even in
+    ``dry_run`` mode, so a run can be audited or replayed.
+    """
+    stats = {
+        "clusters": 0,
+        "merged": 0,
+        "skipped_type_mismatch": 0,
+        "tags_added": 0,
+        "collections_added": 0,
+        "notes_adopted": 0,
+        "attachments_adopted": 0,
+        "relations_moved": 0,
+    }
+    log: list[dict] = []
+    log_dir = Path(log_dir)
+    log_dir.mkdir(parents=True, exist_ok=True)
+    run_id = now_iso().replace(" ", "T").replace(":", "")
+
+    with Db(str(db_path)) as db:
+        clusters = find_url_clusters(db)
+        for members in clusters:
+            keeper = members[0]
+            losers = members[1:]
+            if any(m[2] != keeper[2] for m in losers):
+                stats["skipped_type_mismatch"] += 1
+                log.append(
+                    {
+                        "keeper_key": keeper[1],
+                        "action": "skip-type-mismatch",
+                        "loser_keys": [m[1] for m in losers],
+                    }
+                )
+                continue
+            stats["clusters"] += 1
+            entry = {
+                "keeper_key": keeper[1],
+                "loser_keys": [m[1] for m in losers],
+                "action": "dry-run" if dry_run else "merged",
+            }
+            if not dry_run:
+                for m in losers:
+                    for k, v in merge_into(db, keeper[0], m[0]).items():
+                        stats[k] += v
+                    stats["merged"] += 1
+            log.append(entry)
+        if not dry_run:
+            db.commit()
+
+    (log_dir / f"merge-log-{run_id}.json").write_text(json.dumps(log, indent=1))
+    return stats

tests/zot/test_merge.py

@@ -0,0 +1,141 @@
+"""Merge-based dedupe: fidelity facets union into the keeper."""
+
+import json
+from pathlib import Path
+
+import pytest
+
+from zot.db import TYPE_MAP, Db
+from zot.schema import create_db
+
+
+@pytest.fixture
+def dup_db(tmp_path: Path) -> Path:
+    p = tmp_path / "zotero.sqlite"
+    create_db(str(p)).close()
+    with Db(str(p)) as db:
+        keeper = db.create_item(TYPE_MAP["document"], key="AAAAAAAA")
+        loser = db.create_item(TYPE_MAP["document"], key="BBBBBBBB")
+        for iid in (keeper, loser):
+            db.set_field(iid, "url", "https://example.org/x")
+        db.set_field(keeper, "title", "X")
+        db.set_field(loser, "title", "X (stale)")
+        db.sync_tags(keeper, ["shared", "keeper-only"])
+        db.sync_tags(loser, ["shared", "loser-only"])
+        # loser-only child note
+        nid = db.create_item(TYPE_MAP["note"], key="CCCCCCCC")
+        db.con.execute(
+            "INSERT INTO itemNotes (itemID, parentItemID, note, title) VALUES (?,?,?,?)",
+            (nid, loser, "<p>loser note</p>", ""),
+        )
+        # identical note on both — must NOT be doubled
+        for parent, key in ((keeper, "DDDDDDDD"), (loser, "EEEEEEEE")):
+            dn = db.create_item(TYPE_MAP["note"], key=key)
+            db.con.execute(
+                "INSERT INTO itemNotes (itemID, parentItemID, note, title) VALUES (?,?,?,?)",
+                (dn, parent, "<p>same on both</p>", ""),
+            )
+        db.commit()
+    return p
+
+
+def _tags(db: Db, item_id: int) -> set[str]:
+    return {
+        r[0]
+        for r in db.con.execute(
+            "SELECT t.name FROM itemTags it JOIN tags t ON t.tagID=it.tagID"
+            " WHERE it.itemID=?",
+            (item_id,),
+        )
+    }
+
+
+def _notes(db: Db, item_id: int) -> list[str]:
+    return [
+        r[0]
+        for r in db.con.execute(
+            "SELECT note FROM itemNotes WHERE parentItemID=?", (item_id,)
+        )
+    ]
+
+
+class TestMergeInto:
+    def test_unions_tags_and_adopts_missing_notes(self, dup_db):
+        from zot.merge import merge_into
+
+        with Db(str(dup_db)) as db:
+            keeper = db.find_item_by_key("AAAAAAAA")
+            loser = db.find_item_by_key("BBBBBBBB")
+            merge_into(db, keeper, loser)
+            db.commit()
+            assert _tags(db, keeper) >= {"shared", "keeper-only", "loser-only"}
+            notes = _notes(db, keeper)
+            assert "<p>loser note</p>" in notes
+            assert notes.count("<p>same on both</p>") == 1
+            assert db.find_item_by_key("BBBBBBBB") is None
+
+    def test_keeper_fields_win(self, dup_db):
+        from zot.merge import merge_into
+
+        with Db(str(dup_db)) as db:
+            keeper = db.find_item_by_key("AAAAAAAA")
+            loser = db.find_item_by_key("BBBBBBBB")
+            merge_into(db, keeper, loser)
+            row = db.con.execute(
+                "SELECT idv.value FROM itemData id"
+                " JOIN fields f ON f.fieldID=id.fieldID"
+                " JOIN itemDataValues idv ON idv.valueID=id.valueID"
+                " WHERE id.itemID=? AND f.fieldName='title'",
+                (keeper,),
+            ).fetchone()
+            assert row[0] == "X"
+
+
+class TestDriver:
+    def test_dry_run_reports_without_deleting(self, dup_db, tmp_path):
+        from zot.merge import merge_url_duplicates
+
+        stats = merge_url_duplicates(dup_db, dry_run=True, log_dir=tmp_path)
+        assert stats["clusters"] == 1
+        with Db(str(dup_db)) as db:
+            assert db.find_item_by_key("BBBBBBBB") is not None
+        logs = list(tmp_path.glob("merge-log-*.json"))
+        assert logs and json.loads(logs[0].read_text())[0]["keeper_key"] == "AAAAAAAA"
+
+    def test_live_run_merges(self, dup_db, tmp_path):
+        from zot.merge import merge_url_duplicates
+
+        stats = merge_url_duplicates(dup_db, dry_run=False, log_dir=tmp_path)
+        assert stats["merged"] == 1
+        with Db(str(dup_db)) as db:
+            assert db.find_item_by_key("BBBBBBBB") is None
+
+    def test_type_mismatch_skipped(self, tmp_path):
+        from zot.merge import merge_url_duplicates
+
+        p = tmp_path / "z.sqlite"
+        create_db(str(p)).close()
+        with Db(str(p)) as db:
+            a = db.create_item(TYPE_MAP["document"], key="AAAAAAAA")
+            b = db.create_item(TYPE_MAP["journalArticle"], key="BBBBBBBB")
+            for iid in (a, b):
+                db.set_field(iid, "url", "https://example.org/y")
+            db.commit()
+        stats = merge_url_duplicates(p, dry_run=False, log_dir=tmp_path)
+        assert stats["skipped_type_mismatch"] == 1
+        with Db(str(p)) as db:
+            assert db.find_item_by_key("BBBBBBBB") is not None
+
+    def test_junk_urls_ignored(self, tmp_path):
+        from zot.merge import merge_url_duplicates
+
+        p = tmp_path / "z.sqlite"
+        create_db(str(p)).close()
+        with Db(str(p)) as db:
+            a = db.create_item(TYPE_MAP["journalArticle"], key="AAAAAAAA")
+            b = db.create_item(TYPE_MAP["journalArticle"], key="BBBBBBBB")
+            for iid in (a, b):
+                db.set_field(iid, "url", "997-998")
+            db.commit()
+        stats = merge_url_duplicates(p, dry_run=False, log_dir=tmp_path)
+        assert stats["clusters"] == 0