From 937a2bbfd11a4594264fcc7ce648982946b67871 Mon Sep 17 00:00:00 2001 From: kert Date: Sat, 31 Oct 2026 20:27:05 -0400 Subject: [PATCH] feat(prisma): dedup on DOI, PMID, then title and year --- src/cli/prisma.py | 17 ++++ src/prisma/dedup.py | 177 +++++++++++++++++++++++++++++++++++++ tests/prisma/test_dedup.py | 154 ++++++++++++++++++++++++++++++++ 3 files changed, 348 insertions(+) create mode 100644 src/prisma/dedup.py create mode 100644 tests/prisma/test_dedup.py diff --git a/src/cli/prisma.py b/src/cli/prisma.py index 74bc6c8..0bcd5d3 100644 --- a/src/cli/prisma.py +++ b/src/cli/prisma.py @@ -406,6 +406,23 @@ def fetch( typer.echo(f" {k}: {v}") +@app.command(name="dedup") +def dedup_cmd( + db: Path = typer.Option(_DEFAULT_ZOT, "--db"), + hold: bool = typer.Option(True, "--hold/--no-hold"), +) -> None: + """Tag DOI, PMID, and title duplicates. Near-matches stay untagged.""" + from prisma.dedup import mark_duplicates + from zot.db import Db + + def _go() -> int: + with Db(str(db)) as zdb: + return mark_duplicates(zdb) + + tagged = _hold(_go, hold=hold) + typer.echo(f" tagged: {tagged}") + + # ── VPN droplet lifecycle ─────────────────────────────────────── diff --git a/src/prisma/dedup.py b/src/prisma/dedup.py new file mode 100644 index 0000000..678beba --- /dev/null +++ b/src/prisma/dedup.py @@ -0,0 +1,177 @@ +"""Duplicate detection across the LEAD review projects. + +Match order is DOI, then PMID, then normalized title plus year plus the +first author's surname. A near title in the same year stays a borderline +pair and is not merged. +""" + +from __future__ import annotations + +import re +from collections.abc import Sequence +from dataclasses import dataclass +from difflib import SequenceMatcher + +from zot.db import TYPE_MAP, Db + +_YEAR = re.compile(r"(\d{4})") +_SIMILAR = 0.85 +DUPLICATE_TAG = "dedup:duplicate" + + +@dataclass(frozen=True) +class Record: + """One bibliographic item, as dedup sees it.""" + + key: str + doi: str + pmid: str + title: str + year: int | None + first_author: str + + +@dataclass(frozen=True) +class DedupResult: + """Groups of the same work, plus near-miss pairs left unmerged.""" + + groups: tuple[tuple[Record, ...], ...] + borderline: tuple[tuple[Record, Record], ...] + + +def normalize(title: str) -> str: + """Lowercase, drop punctuation, and collapse whitespace.""" + chars = [ch if ch.isalnum() or ch.isspace() else " " for ch in title.casefold()] + return " ".join("".join(chars).split()) + + +def _doi_key(doi: str) -> str: + return doi.strip().casefold().rstrip(".") + + +def _title_key(rec: Record) -> tuple[str, int | None, str] | None: + title = normalize(rec.title) + if not title: + return None + return (title, rec.year, rec.first_author.casefold().strip()) + + +def cluster(records: Sequence[Record]) -> DedupResult: + """Group exact duplicates. Near titles become borderline pairs.""" + parent = list(range(len(records))) + + def find(i: int) -> int: + while parent[i] != i: + parent[i] = parent[parent[i]] + i = parent[i] + return i + + def union(i: int, j: int) -> None: + ri, rj = find(i), find(j) + if ri == rj: + return + if ri < rj: + parent[rj] = ri + else: + parent[ri] = rj + + def merge_on(key_of) -> None: + first: dict[object, int] = {} + for i, rec in enumerate(records): + key = key_of(rec) + if not key: + continue + prior = first.get(key) + if prior is None: + first[key] = i + else: + union(prior, i) + + merge_on(lambda rec: _doi_key(rec.doi)) + merge_on(lambda rec: rec.pmid.strip()) + merge_on(_title_key) + + members: dict[int, list[Record]] = {} + order: list[int] = [] + for i, rec in enumerate(records): + root = find(i) + if root not in members: + members[root] = [] + order.append(root) + members[root].append(rec) + groups = tuple(tuple(members[root]) for root in order) + + singles = [group[0] for group in groups if len(group) == 1] + borderline: list[tuple[Record, Record]] = [] + for i, left in enumerate(singles): + left_title = normalize(left.title) + if not left_title: + continue + for right in singles[i + 1 :]: + if left.year != right.year: + continue + right_title = normalize(right.title) + if not right_title: + continue + ratio = SequenceMatcher(None, left_title, right_title).ratio() + if ratio >= _SIMILAR: + borderline.append((left, right)) + return DedupResult(groups=groups, borderline=tuple(borderline)) + + +def _year(value: str | None) -> int | None: + if not value: + return None + match = _YEAR.search(value) + return int(match.group(1)) if match else None + + +def load_records(db: Db) -> tuple[Record, ...]: + """Bibliographic items, anchors excluded, in item order.""" + rows = db.con.execute( + """ + SELECT i.itemID, i.key + FROM items i + WHERE i.itemTypeID NOT IN (?, ?) + AND NOT EXISTS ( + SELECT 1 FROM itemTags it + JOIN tags t ON t.tagID = it.tagID + WHERE it.itemID = i.itemID AND t.name = 'module:prisma' + ) + ORDER BY i.itemID + """, + (TYPE_MAP["note"], TYPE_MAP["attachment"]), + ).fetchall() + records: list[Record] = [] + for item_id, key in rows: + creators = db.get_creators(item_id) + author = "" + if creators: + author = str(creators[0]["lastName"] or creators[0]["firstName"] or "") + records.append( + Record( + key=str(key), + doi=db.get_field(item_id, "DOI") or "", + pmid=db.get_field(item_id, "PMID") or "", + title=db.get_field(item_id, "title") or "", + year=_year(db.get_field(item_id, "date")), + first_author=author, + ) + ) + return tuple(records) + + +def mark_duplicates(db: Db) -> int: + """Tag every record in a group except the first. Return how many were new.""" + written = 0 + for group in cluster(load_records(db)).groups: + for rec in group[1:]: + item_id = db.find_item_by_key(rec.key) + if item_id is None: + continue + if DUPLICATE_TAG in db.get_tags(item_id): + continue + db.tag_item(item_id, DUPLICATE_TAG) + written += 1 + db.commit() + return written diff --git a/tests/prisma/test_dedup.py b/tests/prisma/test_dedup.py new file mode 100644 index 0000000..67083ee --- /dev/null +++ b/tests/prisma/test_dedup.py @@ -0,0 +1,154 @@ +"""Dedup on DOI, then PMID, then normalized title, year, and author.""" + +from __future__ import annotations + +from typer.testing import CliRunner + +from cli.prisma import app +from prisma.dedup import Record, cluster, mark_duplicates, normalize +from zot.db import TYPE_MAP, Db +from zot.schema import create_db + +runner = CliRunner() + + +def _rec(**kwargs: object) -> Record: + base: dict[str, object] = { + "key": "AAAAAAAA", + "doi": "", + "pmid": "", + "title": "", + "year": 2020, + "first_author": "", + } + base.update(kwargs) + return Record(**base) # type: ignore[arg-type] + + +def test_normalize_lowercases_and_drops_punctuation(): + assert normalize("Risk Adjustment!") == "risk adjustment" + + +def test_exact_doi_ignores_case_and_a_trailing_period(): + first = _rec(key="AAAAAAAA", doi="10.1000/ABC.", title="One", first_author="Kim") + second = _rec(key="BBBBBBBB", doi="10.1000/abc", title="Two", first_author="Lee") + result = cluster([first, second]) + assert len(result.groups) == 1 + assert {rec.key for rec in result.groups[0]} == {"AAAAAAAA", "BBBBBBBB"} + assert result.borderline == () + + +def test_same_pmid_is_one_group(): + first = _rec(key="AAAAAAAA", pmid="12345", title="One") + second = _rec(key="BBBBBBBB", pmid="12345", title="Two") + result = cluster([first, second]) + assert len(result.groups) == 1 + assert {rec.key for rec in result.groups[0]} == {"AAAAAAAA", "BBBBBBBB"} + + +def test_normalized_title_year_and_author_is_one_group(): + first = _rec( + key="AAAAAAAA", title="Risk Adjustment!", year=2020, first_author="Pope" + ) + second = _rec( + key="BBBBBBBB", title="risk adjustment", year=2020, first_author="Pope" + ) + result = cluster([first, second]) + assert len(result.groups) == 1 + assert {rec.key for rec in result.groups[0]} == {"AAAAAAAA", "BBBBBBBB"} + + +def test_near_titles_stay_borderline_and_unmerged(): + first = _rec( + key="AAAAAAAA", + title="High needs frailty index", + year=2024, + first_author="Kim", + ) + second = _rec( + key="BBBBBBBB", + title="High-need frailty index", + year=2024, + first_author="Lee", + ) + result = cluster([first, second]) + assert {tuple(rec.key for rec in group) for group in result.groups} == { + ("AAAAAAAA",), + ("BBBBBBBB",), + } + assert result.borderline == ((first, second),) + + +def test_dissimilar_titles_are_neither_merged_nor_borderline(): + first = _rec(key="AAAAAAAA", title="zzzz unique alpha", first_author="Kim") + second = _rec(key="BBBBBBBB", title="qqqq other beta", first_author="Lee") + result = cluster([first, second]) + assert len(result.groups) == 2 + assert result.borderline == () + + +def _db(tmp_path) -> str: + path = str(tmp_path / "z.sqlite") + con = create_db(path) + con.close() + return path + + +def _add( + db: Db, *, key: str, doi: str, title: str, year: int, author: str, pmid: str = "" +) -> None: + item_id = db.create_item(TYPE_MAP["journalArticle"], key=key) + db.set_fields( + item_id, + {"title": title, "DOI": doi, "date": str(year), "PMID": pmid}, + ) + db.add_creators(item_id, [("", author)]) + db.commit() + + +def _dup_count(db: Db) -> int: + row = db.con.execute( + """ + SELECT count(*) FROM itemTags it + JOIN tags t ON t.tagID = it.tagID + WHERE t.name = 'dedup:duplicate' + """ + ).fetchone() + return int(row[0]) + + +def test_shared_doi_writes_one_duplicate_tag(tmp_path): + with Db(_db(tmp_path)) as db: + _add(db, key="ABCD2345", doi="10.1/ABC.", title="One", year=2020, author="Kim") + _add(db, key="EFGH5678", doi="10.1/abc", title="Two", year=2021, author="Lee") + mark_duplicates(db) + assert _dup_count(db) == 1 + assert "dedup:duplicate" not in db.get_tags(db.find_item_by_key("ABCD2345")) + assert "dedup:duplicate" in db.get_tags(db.find_item_by_key("EFGH5678")) + + +def test_borderline_items_are_not_tagged(tmp_path): + with Db(_db(tmp_path)) as db: + _add( + db, + key="ABCD2345", + doi="", + title="High needs frailty index", + year=2024, + author="Kim", + ) + _add( + db, + key="EFGH5678", + doi="", + title="High-need frailty index", + year=2024, + author="Lee", + ) + mark_duplicates(db) + assert _dup_count(db) == 0 + + +def test_dedup_command_is_registered(): + result = runner.invoke(app, ["dedup", "--help"]) + assert result.exit_code == 0