← back to Paul Conrad Archive
src/conrad/dedupe.py
232 lines
"""Metadata dedupe. Never deletes: a duplicate gets merged_into=<canonical id> and a 'duplicate_of' link;
uncertain pairs become 'possible_duplicate' rows (exported to data/exports/duplicates.csv for manual review).
Box-range / folder records are never merged into items — items get 'contained_in_candidate' links instead.
Perceptual-hash (image) dedupe is intentionally SKIPPED: the copyright rule forbids downloading Conrad images.
"""
from __future__ import annotations
import re
from collections import defaultdict
from rapidfuzz import fuzz
from .normalize import norm_title
W_DATE, W_TITLE, W_PUB, W_CAPTION, W_SUBJ, W_PEOPLE = 35, 25, 10, 15, 5, 5
AUTO_LINK = 60 # score >= this AND strong title match -> auto-link
REVIEW = 35 # score in [REVIEW, AUTO_LINK) or weak title -> manual review
TITLE_STRONG = 88 # rapidfuzz token_set_ratio
TITLE_SIMILAR = 80
PRIMARY_RANK = {"loc": 0, "wsu": 1, "si": 2, "dpla": 3, "hunt": 4, "syr": 5, "cat": 9}
GENERIC_NORM = {"cartoon", "cartoons", "political cartoon", "political cartoons", "editorial cartoon",
"editorial cartoons", "drawing", "drawings", "cartoon drawing", "cartoon drawings", "untitled",
"untitled drawing", "untitled cartoon", "proof", "proofs", "sketch", "sketches", "no title", "title unknown"}
LOC_PK = re.compile(r"loc\.gov/(?:pictures/)?item/([0-9a-z]+)/?", re.I)
def is_generic_title(title: str | None) -> bool:
"""Cataloger-devised ([bracketed]) or generic titles name a KIND of thing, not a specific cartoon:
a title match on them is not evidence two records are the same work."""
raw = (title or "").strip()
if not raw:
return True
if raw.startswith("[") and raw.rstrip(".").endswith("]"):
return True
t = norm_title(raw)
return not t or t in GENERIC_NORM or len(t.split()) == 1
def _ids(rows) -> set[str]:
"""Normalised identifier tokens for a record's source rows (LOC pk from any loc.gov item URL, plus raw URLs)."""
out = set()
for sid, ident, url in rows:
if sid in ("loc", "seed_loc") and ident:
out.add(f"loc:{ident}")
m = LOC_PK.search(url or "")
if m:
out.add(f"loc:{m[1]}")
elif url:
out.add("url:" + re.sub(r"^https?://(www\.)?", "", url.strip().lower()).rstrip("/"))
return out
def id_relation(a: dict, b: dict) -> str:
"""'agree' (shared identifier), 'conflict' (both carry a LOC pk and they differ) or 'none'."""
ia, ib = a.get("ids") or set(), b.get("ids") or set()
if ia & ib:
return "agree"
la, lb = {x for x in ia if x.startswith("loc:")}, {x for x in ib if x.startswith("loc:")}
if la and lb:
return "conflict"
return "none"
def date_agree(a: dict, b: dict) -> bool:
"""Exact day equal, or identical ranges finer than a whole year (identical bare-year spans do NOT count)."""
if a.get("date_exact") and a.get("date_exact") == b.get("date_exact"):
return True
sa, ea, sb, eb = a.get("date_start"), a.get("date_end"), b.get("date_start"), b.get("date_end")
return bool(sa and ea and (sa, ea) == (sb, eb) and sa[:7] == ea[:7])
def _pub(p: str | None) -> str:
import re
return re.sub(r"^the\s+", "", (p or "").strip().lower())
def score_pair(a: dict, b: dict) -> tuple[float, dict]:
"""a, b: dicts with date_exact, year, title, caption, description, publication, subjects(set), people(set)."""
parts = {}
if a.get("date_exact") and a.get("date_exact") == b.get("date_exact"):
parts["date"] = W_DATE
ta, tb = norm_title(a.get("title")), norm_title(b.get("title"))
tsim = fuzz.token_set_ratio(ta, tb) if ta and tb else 0.0
if tsim >= TITLE_SIMILAR:
parts["title"] = W_TITLE * (tsim / 100.0)
if _pub(a.get("publication")) and _pub(a.get("publication")) == _pub(b.get("publication")):
parts["publication"] = W_PUB
ca = (a.get("caption") or a.get("description") or "").lower()
cb = (b.get("caption") or b.get("description") or "").lower()
if ca and cb and fuzz.token_set_ratio(ca, cb) >= 80:
parts["caption"] = W_CAPTION
if a.get("subjects") and b.get("subjects") and a["subjects"] & b["subjects"]:
parts["subjects"] = W_SUBJ
if a.get("people") and b.get("people") and a["people"] & b["people"]:
parts["people"] = W_PEOPLE
parts["title_sim"] = round(tsim, 1)
return round(sum(v for k, v in parts.items() if k != "title_sim"), 1), parts
def _load_items(conn) -> list[dict]:
rows = conn.execute("""SELECT id, canonical_id, title, caption, description, publication, date_exact, date_start,
date_end, year FROM cartoons WHERE granularity='item'""").fetchall()
subj, ppl = defaultdict(set), defaultdict(set)
for cid, name in conn.execute("SELECT cs.cartoon_id, s.name FROM cartoon_subjects cs JOIN subjects s ON s.id=cs.subject_id"):
subj[cid].add(name.lower())
for cid, name in conn.execute("SELECT cp.cartoon_id, p.name FROM cartoon_people cp JOIN people p ON p.id=cp.person_id"):
ppl[cid].add(name.lower())
ids = defaultdict(list)
for cid, sid, ident, url in conn.execute("SELECT cartoon_id, source_id, identifier, record_url FROM cartoon_sources"):
ids[cid].append((sid, ident, url))
out = []
for r in rows:
d = dict(r)
d["subjects"], d["people"] = subj[r["id"]], ppl[r["id"]]
d["ids"] = _ids(ids[r["id"]])
out.append(d)
return out
def _rank(c: dict) -> tuple:
return (PRIMARY_RANK.get(c["canonical_id"].split(":")[0], 5), c["id"])
def run(conn) -> dict:
conn.execute("DELETE FROM cartoon_links")
conn.execute("UPDATE cartoons SET merged_into=NULL")
items = _load_items(conn)
by_year = defaultdict(list)
for it in items:
if it["year"]:
by_year[it["year"]].append(it)
auto, review = [], []
for year, group in by_year.items():
for i, a in enumerate(group):
for b in group[i + 1:]:
if a["canonical_id"].split(":")[0] == b["canonical_id"].split(":")[0] == "hunt":
continue
s, parts = score_pair(a, b)
if s < REVIEW and parts["title_sim"] < 95:
continue
secondary = "cat" in (a["canonical_id"][:3], b["canonical_id"][:3])
both_titled = bool(norm_title(a["title"]) and norm_title(b["title"]))
if both_titled and parts["title_sim"] < 60 and not secondary:
continue # same day, clearly different cartoons (e.g. Wichita daily sequence)
keep, dup = sorted([a, b], key=_rank)
date_conflict = bool(a["date_exact"] and b["date_exact"] and a["date_exact"] != b["date_exact"])
rel = id_relation(a, b)
title_match = (s >= AUTO_LINK and parts["title_sim"] >= TITLE_STRONG) or \
(parts["title_sim"] >= 95 and not date_conflict)
# a title match on a generic / [cataloger-devised] title is not evidence of identity: it needs
# identifier or (finer-than-year) date agreement. Two different LOC pks are never auto-linked.
generic = is_generic_title(a["title"]) or is_generic_title(b["title"])
if title_match and rel == "conflict":
parts["hold"] = "different LOC identifiers"
title_match = False
elif title_match and generic and not (rel == "agree" or date_agree(a, b)):
parts["hold"] = "generic title: needs identifier or date agreement"
title_match = False
if title_match:
auto.append((dup, keep, s, parts))
else:
review.append((a, b, s, parts))
# identical titles in DIFFERENT years (date conflict / reprint / mis-dated secondary source): review only
seen_pairs = {(min(a["id"], b["id"]), max(a["id"], b["id"])) for a, b, *_ in review} | \
{(min(d["id"], k["id"]), max(d["id"], k["id"])) for d, k, *_ in auto}
by_title = defaultdict(list)
for it in items:
t = norm_title(it["title"])
if len(t) >= 12:
by_title[t].append(it)
for t, group in by_title.items():
for i, a in enumerate(group):
for b in group[i + 1:]:
key = (min(a["id"], b["id"]), max(a["id"], b["id"]))
if a["year"] != b["year"] and key not in seen_pairs:
s, parts = score_pair(a, b)
parts["note"] = "identical title, different year"
review.append((a, b, s, parts))
# apply auto-links (union into the best-ranked canonical)
parent = {}
def find(x):
while parent.get(x, x) != x:
x = parent[x]
return x
for dup, keep, s, parts in sorted(auto, key=lambda t: -t[2]):
rk, rd = find(keep["id"]), find(dup["id"])
if rk == rd:
continue
parent[rd] = rk
conn.execute("INSERT OR REPLACE INTO cartoon_links VALUES (?,?,?,?,?)",
(dup["id"], keep["id"], "duplicate_of", s, str(parts)))
for cid in list(parent):
conn.execute("UPDATE cartoons SET merged_into=? WHERE id=?", (find(cid), cid))
for a, b, s, parts in review:
conn.execute("INSERT OR REPLACE INTO cartoon_links VALUES (?,?,?,?,?)",
(a["id"], b["id"], "possible_duplicate", s, str(parts)))
contained = link_contained(conn)
conn.commit()
from . import db
db.refresh_index(conn)
return {"items": len(items), "auto_linked": len(parent), "possible_duplicates": len(review),
"contained_in_candidates": contained}
def link_contained(conn) -> int:
"""Link item records to the box_range / folder records whose date span fully contains the item's dates."""
boxes = conn.execute("""SELECT MIN(c.id) AS first_id, cs.repository, cs.box, c.date_start, c.date_end, COUNT(*) AS n,
MIN(cs.identifier) AS lo, MAX(cs.identifier) AS hi
FROM cartoons c JOIN cartoon_sources cs ON cs.cartoon_id=c.id AND cs.source_id='huntington'
WHERE c.granularity='box_range' GROUP BY cs.box, c.date_start, c.date_end""").fetchall()
folders = conn.execute("""SELECT c.id AS first_id, cs.repository, cs.folder AS box, c.date_start, c.date_end, 1 AS n,
cs.identifier AS lo, cs.identifier AS hi
FROM cartoons c JOIN cartoon_sources cs ON cs.cartoon_id=c.id AND cs.source_id='seed_syracuse'
WHERE c.granularity='folder' AND c.date_start IS NOT NULL""").fetchall()
items = conn.execute("""SELECT id, date_start, date_end FROM cartoons WHERE granularity='item' AND merged_into IS NULL
AND date_start IS NOT NULL AND date_end IS NOT NULL""").fetchall()
n = 0
for it in items:
for bx in list(boxes) + list(folders):
if bx["date_start"] <= it["date_start"] and it["date_end"] <= bx["date_end"]:
# only month-or-finer items: a bare year would 'fit' many boxes and mean nothing
if it["date_start"][:7] != it["date_end"][:7] and bx["repository"] == "Huntington Library":
continue
detail = (f"{bx['repository']} box {bx['box']} ({bx['lo']}..{bx['hi']}, {bx['n']} slots) spans "
f"{bx['date_start']}..{bx['date_end']}")
conn.execute("INSERT OR REPLACE INTO cartoon_links VALUES (?,?,?,?,?)",
(it["id"], bx["first_id"], "contained_in_candidate", None, detail))
n += 1
return n