feat(zot): merge-based URL dedupe engine — union facets, keep-earliest (refs #665)
This commit is contained in:
kert
215
src/zot/merge.py
Normal file
215
src/zot/merge.py
Normal file
@@ -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
|
||||
141
tests/zot/test_merge.py
Normal file
141
tests/zot/test_merge.py
Normal file
@@ -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
|
||||
Reference in New Issue
Block a user