01 Deduplication

This notebook reads every export in data/raw/search_exports/, merges duplicate records, and writes data/processed/deduplicated_records.csv. It also writes the identification counts to data/processed/screening_counts.yaml, which 02 Screening extends. Two audit files show what happened. duplicate_groups.csv lists every exported record with the record_id it was merged into and the reason. possible_duplicates.csv lists pairs that look alike but were not merged.

Method

Each export is parsed into title, abstract, authors, year, journal, volume, issue, pages, and DOI. Two records are linked as duplicates in three ways.

• They share a normalized DOI. The DOI is lowercased and stripped of any resolver prefix.
• They share a normalized title and publication year, when the title has at least 20 characters. The title is reduced to lowercase letters and digits after accents are removed.
• bib-dedupe labels them duplicates. It compares authors, title, journal, volume, issue, pages, year, and DOI with fuzzy matching, so it also links records that lack a DOI or differ in punctuation, capitalization, or journal abbreviation. The version used is printed at the end of this page.

Links are transitive. Pairs that bib-dedupe labels “maybe” are not merged. A reviewer settles any pair, merged or not, in data/raw/dedup_decisions.csv, with the columns member_a, member_b, and decision (merge or separate). The member identifiers are in duplicate_groups.csv and possible_duplicates.csv.

From each group, the record with the longest abstract supplies the fields, and any field it lacks is filled from the other records in the group.

The record_id is a hash of the key of the earliest record in the group. Records are ordered by search date, file name, and key. The identifier does not change when a later search adds records or finds the same records again, as long as the earlier exports stay in search_exports/. If a later export links two groups that were separate, the identifier of the earlier group is kept and the other is retired. The notebook lists every identifier that changed since the previous run.

import hashlib
import re
import unicodedata
from importlib.metadata import version
from pathlib import Path

import pandas as pd
import rispy
import yaml

ROOT = next(p for p in [Path.cwd(), *Path.cwd().parents] if (p / "_quarto.yml").exists())
RAW = ROOT / "data" / "raw" / "search_exports"
DECISIONS = ROOT / "data" / "raw" / "dedup_decisions.csv"
OUT = ROOT / "data" / "processed"
OUT.mkdir(parents=True, exist_ok=True)

NAME = re.compile(r"^(?P<database>.+?)_(?P<date>\d{4}-\d{2}-\d{2})(?:_(?P<block>[A-Za-z0-9]+))?$")
files = sorted(p for p in RAW.glob("*") if p.suffix.lower() in {".ris", ".csv"})
have_data = bool(files)
if have_data:
    print(f"{len(files)} export file(s) found.")
else:
    print("No exports found in data/raw/search_exports/. Add exports and render this notebook again.")

No exports found in data/raw/search_exports/. Add exports and render this notebook again.

FIELDS = ["title", "abstract", "authors", "year", "journal", "volume", "number", "pages", "doi"]
CSV_COLUMNS = {
    "title": ["title", "article title", "document title"],
    "abstract": ["abstract"],
    "authors": ["authors", "author full names", "author(s)", "author"],
    "year": ["year", "publication year", "py"],
    "journal": ["journal", "source title", "journal/book", "publication title"],
    "volume": ["volume"],
    "number": ["issue", "number"],
    "pages": ["pages", "page range"],
    "doi": ["doi"],
}
RIS_KEYS = {
    "title": ["primary_title", "title", "translated_title"],
    "abstract": ["abstract", "notes_abstract"],
    "authors": ["authors", "first_authors"],
    "year": ["year", "publication_year", "date"],
    "journal": ["secondary_title", "journal_name", "alternate_title3", "alternate_title2"],
    "volume": ["volume"],
    "number": ["number", "issue"],
    "doi": ["doi"],
}


def first(record, keys):
    for key in keys:
        value = record.get(key)
        if value:
            return "; ".join(map(str, value)) if isinstance(value, list) else str(value)
    return ""


def read_export(path):
    if path.suffix.lower() == ".ris":
        with open(path, encoding="utf-8-sig") as fh:
            entries = rispy.load(fh)
        rows = []
        for r in entries:
            row = {f: first(r, RIS_KEYS[f]) for f in FIELDS if f != "pages"}
            start, end = first(r, ["start_page"]), first(r, ["end_page"])
            row["pages"] = f"{start}-{end}" if start and end else start
            rows.append(row)
        return pd.DataFrame(rows, columns=FIELDS)
    df = pd.read_csv(path, encoding="utf-8-sig", dtype=str).fillna("")
    lower = {c.lower().strip(): c for c in df.columns}

    def column(names):
        col = next((lower[c] for c in names if c in lower), None)
        return df[col] if col else pd.Series("", index=df.index)

    out = pd.DataFrame({f: column(CSV_COLUMNS[f]) for f in FIELDS})
    if (out["pages"] == "").all():
        start, end = column(["page start"]), column(["page end"])
        out["pages"] = start.where(end == "", start + "-" + end)
    return out

def clean_doi(x):
    x = x.strip().lower()
    return re.sub(r"^(https?://(dx\.)?doi\.org/|doi:\s*)", "", x)


def title_key(x):
    x = unicodedata.normalize("NFKD", x).encode("ascii", "ignore").decode().lower()
    return re.sub(r"[^a-z0-9]+", "", x)


def year4(x):
    m = re.search(r"\d{4}", x)
    return m.group(0) if m else ""


MIN_TITLE_CHARS = 20  # shorter titles are linked only by bib-dedupe, which also checks authors and journal

if have_data:
    frames = []
    for p in files:
        m = NAME.match(p.stem)
        if not m:
            raise ValueError(f"{p.name}: name the file <database>_<YYYY-MM-DD>[_<block>], for example scopus_2026-11-02_A.ris")
        df = read_export(p)
        if len(df) and (df["title"].str.strip() == "").all():
            raise ValueError(f"{p.name}: no titles were read. CSV exports need a title column.")
        df["database"] = m["database"]
        df["search_date"] = m["date"]
        df["block"] = m["block"] or ""
        df["file"] = p.name
        df["row"] = range(1, len(df) + 1)
        frames.append(df)
    records = pd.concat(frames, ignore_index=True).fillna("").astype(str)
    if records.empty:
        raise ValueError("The exports contain no records.")
    records["row"] = records["row"].astype(int)
    records["member_id"] = records["file"] + "#" + records["row"].astype(str)
    records["doi"] = records["doi"].map(clean_doi)
    records["year"] = records["year"].map(year4)
    records["tkey"] = records["title"].map(title_key)
    records["abstract_len"] = records["abstract"].str.len()
    long_title = records["tkey"].str.len() >= MIN_TITLE_CHARS
    records["key"] = [
        f"doi:{d}" if d else f"title:{t}:{y}" if ok and y else f"row:{f}:{r}"
        for d, t, y, ok, f, r in zip(records["doi"], records["tkey"], records["year"], long_title, records["file"], records["row"])
    ]
    print(f"{len(records)} exported record(s).")

if have_data:
    import bib_dedupe.bib_dedupe as bd

    position = {m: i for i, m in enumerate(records["member_id"])}
    pairs = {}  # (i, j) -> the reasons the two records are linked

    def link(i, j, why):
        if i != j:
            pairs.setdefault((min(i, j), max(i, j)), set()).add(why)

    # 1. Exact keys
    first_with = {}
    for i, r in enumerate(records.itertuples()):
        keys = []
        if r.doi:
            keys.append(("doi", r.doi))
        if len(r.tkey) >= MIN_TITLE_CHARS and r.year:
            keys.append(("title-year", r.tkey, r.year))
        for k in keys:
            if k in first_with:
                link(first_with[k], i, k[0])
            else:
                first_with[k] = i

    # 2. bib-dedupe. One process, so the result does not depend on the number of processors.
    maybe_pairs = []
    if len(records) > 1:
        bd_input = pd.DataFrame({
            "ID": [str(i) for i in range(len(records))],
            "ENTRYTYPE": "article",
            "title": records["title"],
            "author": records["authors"].str.replace("; ", " and ", regex=False),
            "year": records["year"],
            "journal": records["journal"],
            "volume": records["volume"],
            "number": records["number"],
            "pages": records["pages"],
            "doi": records["doi"],
            "abstract": records["abstract"],
            "search_set": "",
        })
        prepared = bd.prep(bd_input, verbosity_level=0, cpu=1)
        blocked = bd.block(prepared, verbosity_level=0, cpu=1)
        if len(blocked):
            matched = bd.match(blocked, verbosity_level=0, cpu=1)
            for r in matched.itertuples():
                if r.duplicate_label == "duplicate":
                    link(int(r.ID_1), int(r.ID_2), "bib-dedupe")
                elif r.duplicate_label == "maybe":
                    maybe_pairs.append((min(int(r.ID_1), int(r.ID_2)), max(int(r.ID_1), int(r.ID_2))))

    # 3. Reviewer decisions
    decided = {}
    if DECISIONS.exists():
        dec = pd.read_csv(DECISIONS, dtype=str).fillna("")
        dec.columns = [c.strip().lower() for c in dec.columns]
        absent = [c for c in ("member_a", "member_b", "decision") if c not in dec.columns]
        if absent:
            raise ValueError("dedup_decisions.csv is missing column(s): " + ", ".join(absent))
        dec = dec.apply(lambda s: s.str.strip())
        dec["decision"] = dec["decision"].str.lower()
        errs = []
        for n, r in enumerate(dec.itertuples(), start=2):
            if r.member_a not in position or r.member_b not in position:
                errs.append(f"line {n}: member id not found in the exports")
            elif r.member_a == r.member_b:
                errs.append(f"line {n}: member_a and member_b are the same record")
            elif r.decision not in ("merge", "separate"):
                errs.append(f"line {n}: decision must be merge or separate")
        if errs:
            raise ValueError("dedup_decisions.csv has problems:\n- " + "\n- ".join(errs))
        for r in dec.itertuples():
            i, j = position[r.member_a], position[r.member_b]
            decided[(min(i, j), max(i, j))] = r.decision
            if r.decision == "merge":
                link(i, j, "manual")
            else:
                pairs.pop((min(i, j), max(i, j)), None)

    # Group the linked records
    parent = list(range(len(records)))

    def find(i):
        while parent[i] != i:
            parent[i] = parent[parent[i]]
            i = parent[i]
        return i

    for i, j in pairs:
        parent[find(i)] = find(j)
    records["group"] = [find(i) for i in range(len(records))]

    # A group needs a look when records are joined only by fuzzy matching or a reviewer decision
    exact_parent = list(range(len(records)))

    def exact_find(i):
        while exact_parent[i] != i:
            exact_parent[i] = exact_parent[exact_parent[i]]
            i = exact_parent[i]
        return i

    for (i, j), reasons in pairs.items():
        if reasons & {"doi", "title-year"}:
            exact_parent[exact_find(i)] = exact_find(j)
    records["exact_group"] = [exact_find(i) for i in range(len(records))]

    still_joined = [k for k, d in decided.items() if d == "separate" and find(k[0]) == find(k[1])]
    if still_joined:
        names = [f"{records['member_id'][i]} and {records['member_id'][j]}" for i, j in still_joined]
        raise ValueError("These records are marked separate in dedup_decisions.csv but other links still join them: " + "; ".join(names))

    why = {}
    for (i, j), reasons in pairs.items():
        why.setdefault(find(i), set()).update(reasons)
    records["matched_by"] = records["group"].map(lambda g: "; ".join(sorted(why.get(g, ()))))
    print(f"{len(pairs)} link(s) between records, {records['group'].nunique()} group(s).")

def pick(g, best, col):
    return best[col] or next((v for v in g[col] if v), "")


if have_data:
    ordered = records.sort_values(["search_date", "file", "key", "row"], kind="stable")
    rows, members, used = [], [], set()
    for _, g in ordered.groupby("group", sort=False):
        earliest = g.iloc[0]
        basis = earliest["key"] if earliest["key"] not in used else f"{earliest['key']}|{earliest['member_id']}"
        used.add(earliest["key"])
        rid = "R" + hashlib.sha1(basis.encode()).hexdigest()[:10]
        best = g.sort_values("abstract_len", ascending=False, kind="stable").iloc[0]
        rows.append({
            "record_id": rid,
            "title": pick(g, best, "title"),
            "abstract": pick(g, best, "abstract"),
            "authors": pick(g, best, "authors"),
            "year": pick(g, best, "year"),
            "journal": pick(g, best, "journal"),
            "doi": next((d for d in g["doi"] if d), ""),
            "databases": "; ".join(sorted(set(g["database"]))),
            "blocks": "; ".join(sorted(b for b in set(g["block"]) if b)),
            "first_seen": g["search_date"].min(),
            "n_exports": len(g),
        })
        members.append(g.assign(record_id=rid, needs_review="yes" if g["exact_group"].nunique() > 1 else ""))
    dedup = pd.DataFrame(rows).sort_values("record_id").reset_index(drop=True)
    membership = (
        pd.concat(members)[["record_id", "member_id", "database", "block", "search_date", "doi", "year", "title", "matched_by", "needs_review"]]
        .sort_values(["record_id", "search_date", "member_id"])
        .reset_index(drop=True)
    )

    if dedup["record_id"].duplicated().any():
        raise ValueError("Two groups received the same record_id.")
    if len(membership) != len(records) or membership["member_id"].duplicated().any():
        raise ValueError("Some exported records were lost or counted twice while grouping.")
    if int(dedup["n_exports"].sum()) != len(records):
        raise ValueError("The group sizes do not add up to the number of exported records.")

    undecided = [
        (i, j) for i, j in sorted(set(maybe_pairs))
        if find(i) != find(j) and (i, j) not in decided
    ]
    pick_cols = ["member_id", "database", "year", "doi", "title"]
    left = records.loc[[i for i, _ in undecided], pick_cols].add_suffix("_a").reset_index(drop=True)
    right = records.loc[[j for _, j in undecided], pick_cols].add_suffix("_b").reset_index(drop=True)
    possible = pd.concat([left, right], axis=1).rename(columns={"member_id_a": "member_a", "member_id_b": "member_b"})
    print(f"{len(dedup)} unique record(s). {len(possible)} possible duplicate pair(s) not merged.")

if have_data:
    # Identifiers that changed since the previous run
    changed = pd.DataFrame(columns=["previous_id", "record_id", "previous_id_still_used"])
    previous_members = OUT / "duplicate_groups.csv"
    if previous_members.exists():
        before = pd.read_csv(previous_members, dtype=str).fillna("")[["member_id", "record_id"]]
        both = before.rename(columns={"record_id": "previous_id"}).merge(membership[["member_id", "record_id"]], on="member_id")
        changed = both[both["previous_id"] != both["record_id"]][["previous_id", "record_id"]].drop_duplicates().sort_values("previous_id")
        changed["previous_id_still_used"] = changed["previous_id"].isin(set(membership["record_id"])).map({True: "yes", False: "no"})

    dedup.to_csv(OUT / "deduplicated_records.csv", index=False)
    membership.to_csv(previous_members, index=False)
    possible.to_csv(OUT / "possible_duplicates.csv", index=False)
    print("Wrote deduplicated_records.csv, duplicate_groups.csv, and possible_duplicates.csv in data/processed/")

    if len(changed):
        print(f"WARNING: {len(changed)} change(s) of record_id since the previous run, because groups were joined or split.")
        print("Where previous_id_still_used is no, screening decisions and charted data that use the previous_id must be updated. Where it is yes, check that they still apply.")
        display(changed)

Records per search

if have_data:
    by_export = (
        records.groupby(["database", "block", "search_date"], as_index=False)
        .size()
        .rename(columns={"size": "records"})
    )
    display(by_export)
    total, unique = len(records), len(dedup)
    print(f"Records identified: {total}")
    print(f"Duplicate records removed: {total - unique}")
    print(f"Unique records: {unique}")

if have_data:
    path = OUT / "screening_counts.yaml"
    previous = ((yaml.safe_load(path.read_text()) or {}) if path.exists() else {}).get("identification", {})
    is_other = by_export["database"].str.startswith("other-")

    def entry(r, strip=False):
        name = r.database.removeprefix("other-") if strip else r.database
        return {"name": name, "block": r.block or None, "search_date": r.search_date, "n": int(r.records)}

    # The record set may have changed, so the later sections are dropped. Run notebook 02 again to rebuild them.
    counts = {
        "identification": {
            "databases": [entry(r) for r in by_export[~is_other].itertuples()],
            "other_methods": [entry(r, strip=True) for r in by_export[is_other].itertuples()],
            "duplicates_removed": int(total - unique),
            "marked_ineligible_by_automation": previous.get("marked_ineligible_by_automation", 0),
            "removed_other_reasons": previous.get("removed_other_reasons", 0),
            "deduplication": f"bib-dedupe {version('bib-dedupe')}, with exact DOI and title-year keys",
        }
    }
    path.write_text(yaml.safe_dump(counts, sort_keys=False))
    print("Wrote data/processed/screening_counts.yaml")

Merged records, for a spot check

Groups that would fall apart without fuzzy matching or a reviewer decision are the ones to check, and they are listed in full. Groups held together by a DOI or an exact title and year are counted by the keys they share. Every group is in duplicate_groups.csv, with needs_review marked.

if have_data:
    sizes = membership.groupby("record_id")["member_id"].transform("size")
    merged = membership[sizes > 1]
    summary = merged.drop_duplicates("record_id")["matched_by"].value_counts().rename_axis("Linked by").reset_index(name="groups")
    if summary.empty:
        print("No records were merged.")
    else:
        display(summary)
        fuzzy = merged[merged["needs_review"] == "yes"]
        if len(fuzzy):
            print(f"{fuzzy['record_id'].nunique()} group(s) are joined only by fuzzy matching or a reviewer decision:")
            with pd.option_context("display.max_rows", None, "display.max_colwidth", 90):
                display(fuzzy[["record_id", "matched_by", "database", "year", "title"]])
    if len(possible):
        print(f"{len(possible)} possible duplicate pair(s) were not merged. Review possible_duplicates.csv and record the decisions in data/raw/dedup_decisions.csv.")
    print(f"Matching used bib-dedupe {version('bib-dedupe')}.")