155 lines
4.5 KiB
Python
155 lines
4.5 KiB
Python
"""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
|