From ab0d24eb52c235091c186e7f9a0aa85d92396986 Mon Sep 17 00:00:00 2001
From: rlskoeser <rebecca.s.koeser@princeton.edu>
Date: Tue, 15 Sep 2026 15:36:12 -0400
Subject: [PATCH 1/4] Add script to propagate metadata cleanup to current
 csv/json files

---
 src/corppa/utils/dataset_refine.py | 175 +++++++++++++++++++++++++++++
 1 file changed, 175 insertions(+)
 create mode 100644 src/corppa/utils/dataset_refine.py

diff --git a/src/corppa/utils/dataset_refine.py b/src/corppa/utils/dataset_refine.py
new file mode 100644
index 00000000..06ea86b2
--- /dev/null
+++ b/src/corppa/utils/dataset_refine.py
@@ -0,0 +1,175 @@
+"""
+Utility script for PPA full-text dataset publication prep.
+
+Takes a manually cleaned version of PPA metadata in CSV format, and
+propagates refined author and pub_place fields to CSV and JSON metadata
+in the specified ppa_corpus folder.
+"""
+
+import argparse
+import json
+import shutil
+from pathlib import Path
+
+import polars as pl
+
+
+def refine_metadata(
+    refined_metadata: Path, csv_metadata_path: Path, json_metadata_path: Path
+) -> None:
+    refined_metadata_df = pl.read_csv(refined_metadata)
+    # get the last update time of records in refined metadat
+    refined_last_update = refined_metadata_df["updated"].max()
+    df = pl.read_csv(csv_metadata_path)
+    csv_columns = df.columns  # store columns original order for output file
+    csv_total = df.height
+
+    # join with refined metadata on work id so we can propagate work-specific
+    # resolutions (author names are unambiguous but pub places are not)
+    refined_df = df.join(
+        refined_metadata_df.select("work_id", "author", "pub_place"),
+        on="work_id",
+        suffix="_refined",
+        how="left",  # preserve all original records
+    )
+    # report how many instances of each field will be updated
+    author_changes_df = refined_df.filter(pl.col.author.ne(pl.col.author_refined))
+    num_author_changes = author_changes_df.height
+    uniq_author_changes = (
+        author_changes_df.select("author", "author_refined").unique().height
+    )
+    print(
+        f"Updated author in {num_author_changes:,} works ({uniq_author_changes:,} unique replacements)"
+    )
+
+    pubplace_changes_df = refined_df.filter(
+        pl.col.pub_place.ne(pl.col.pub_place_refined)
+    )
+    num_pubplace_changes = pubplace_changes_df.height
+    uniq_pubplace_changes = pubplace_changes_df.select(
+        "pub_place", "pub_place_refined"
+    ).n_unique()
+    print(
+        f"Updated pub_place in {num_pubplace_changes:,} works ({uniq_pubplace_changes:,} unique replacements)"
+    )
+
+    # then replace the original with the refined fields
+    # (refined metadata preserves unchanged values, so we can just copy them over)
+    refined_df = refined_df.drop("author", "pub_place").rename(
+        {"author_refined": "author", "pub_place_refined": "pub_place"}
+    )
+
+    # identify any records added after the manual cleanup, and update author/pub_place
+    # if there is an unambiguous mapping from the refined set
+    post_refine_records = df.filter(pl.col.added.gt(refined_last_update))
+    if post_refine_records.height:
+        author_lookup = author_changes_df.select("author", "author_refined").unique()
+
+        pub_place_lookup = pubplace_changes_df.select(
+            "pub_place", "pub_place_refined"
+        ).unique()
+        # omit any ambiguous placenames (i.e. Cambridge and Rochester could either be US or UK)
+        pub_place_lookup = pub_place_lookup.filter(
+            ~pub_place_lookup["pub_place"].is_duplicated()
+        )
+        # join and then filter to any that have changes
+        post_refine_records = (
+            post_refine_records.select("work_id", "author", "pub_place")
+            .join(author_lookup, on="author", how="left")
+            .join(pub_place_lookup, on="pub_place", how="left")
+            .filter(
+                # limit to records with at least one refined value
+                ~(pl.col.author_refined.is_null() & pl.col.pub_place_refined.is_null())
+            )
+        )
+
+        # update the refined data with these changes
+        refined_df = (
+            refined_df.join(
+                post_refine_records.select(
+                    "work_id", "author_refined", "pub_place_refined"
+                ),
+                on="work_id",
+                how="left",
+            )
+            .with_columns(
+                # take the refined value if not null, otherwise the previous value
+                author=pl.when(pl.col.author_refined.is_not_null())
+                .then(pl.col.author_refined)
+                .otherwise(pl.col.author),
+                pub_place=pl.when(pl.col.pub_place_refined.is_not_null())
+                .then(pl.col.pub_place_refined)
+                .otherwise(pl.col.pub_place),
+            )
+            .drop("author_refined", "pub_place_refined")
+        )
+
+    # replace refined fields with unrefined and save the file
+    assert refined_df.height == csv_total  # no records lost
+
+    # Path.copy not available in 3.12, so use shutil to make a backup
+    shutil.copy(csv_metadata_path, csv_metadata_path.with_suffix(".csv.bk"))
+    # then replace the original with the refined version
+    refined_df.select(csv_columns).write_csv(csv_metadata_path)
+
+    # load json, update based on csv, write out
+    with json_metadata_path.open() as jsonfile:
+        json_metadata = json.load(jsonfile)
+    refined_json_metadata = []
+    for csv_row, json_row in zip(refined_df.iter_rows(named=True), json_metadata):
+        assert csv_row["work_id"] == json_row["work_id"]
+        if "author" in json_row:
+            json_row["author"] = csv_row["author"]
+        if "pub_place" in json_row:
+            json_row["pub_place"] = csv_row["pub_place"]
+        refined_json_metadata.append(json_row)
+    # no rows lost
+    assert len(json_metadata) == len(refined_json_metadata) == csv_total
+
+    # make a backup and output the refined version
+    shutil.copy(json_metadata_path, json_metadata_path.with_suffix(".json.bk"))
+    with json_metadata_path.open("w") as jsonfile:
+        json.dump(refined_json_metadata, jsonfile, indent=2)
+
+    print(
+        f"Updated metadata has been saved to {csv_metadata_path} and {json_metadata_path}"
+    )
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Propagate PPA metadata standardization to updated metadata files"
+    )
+    parser.add_argument(
+        "refined_metadata",
+        help="PPA metadata file with standardized author/pub_place fields to propagate",
+        type=Path,
+    )
+
+    parser.add_argument(
+        "corpus_dir",
+        help="PPA full-text corpus directory; expected to contain "
+        "work-level metadata files ppa_metadata.csv and ppa_metadata.json",
+        type=Path,
+    )
+
+    args = parser.parse_args()
+    if not args.refined_metadata.is_file():
+        raise SystemExit(f"Input metadata file {args.refined_metadata} does not exist")
+
+    if not args.corpus_dir.is_dir():
+        raise SystemExit(f"PPA corpus path {args.corpus_dir} is not a directory")
+
+    # we expect two formats
+    csv_metadata_path = args.corpus_dir / "ppa_metadata.csv"
+    json_metadata_path = args.corpus_dir / "ppa_metadata.json"
+    if not csv_metadata_path.exists():
+        raise SystemExit(f"PPA metadata file {csv_metadata_path} not found")
+    if not json_metadata_path.exists():
+        raise SystemExit(f"PPA metadata file {json_metadata_path} not found")
+
+    refine_metadata(args.refined_metadata, csv_metadata_path, json_metadata_path)
+
+
+if __name__ == "__main__":
+    main()

From 3ffbc1ba76ffc3afa29bb9126d9aa07bd8ab31fc Mon Sep 17 00:00:00 2001
From: rlskoeser <rebecca.s.koeser@princeton.edu>
Date: Tue, 15 Sep 2026 15:47:58 -0400
Subject: [PATCH 2/4] Clean up and fixes for unmatched records, ambiguous
 lookups

- keep original values when refined data does not have the record
- handle nulls when counting changes
- guard against ambiguous author lookup
- document that json is known to match csv data

Assisted-by: OpenCode:BigPickle (glm-5.1)
---
 src/corppa/utils/dataset_refine.py | 66 ++++++++++++++++++++++--------
 1 file changed, 48 insertions(+), 18 deletions(-)

diff --git a/src/corppa/utils/dataset_refine.py b/src/corppa/utils/dataset_refine.py
index 06ea86b2..30b9b10c 100644
--- a/src/corppa/utils/dataset_refine.py
+++ b/src/corppa/utils/dataset_refine.py
@@ -18,12 +18,22 @@ def refine_metadata(
     refined_metadata: Path, csv_metadata_path: Path, json_metadata_path: Path
 ) -> None:
     refined_metadata_df = pl.read_csv(refined_metadata)
-    # get the last update time of records in refined metadat
-    refined_last_update = refined_metadata_df["updated"].max()
+    # get the last update time of records in refined metadata
+    # (drop_nulls: a null/empty updated column means there is no cutoff to
+    # derive, and the post-refine block below is safely skipped)
+    refined_last_update = refined_metadata_df["updated"].drop_nulls().max()
     df = pl.read_csv(csv_metadata_path)
     csv_columns = df.columns  # store columns original order for output file
     csv_total = df.height
 
+    # each work_id must resolve to at most one refined record, otherwise the
+    # join below would inflate the record count
+    if refined_metadata_df["work_id"].is_duplicated().any():
+        raise ValueError(
+            "Refined metadata contains duplicate work_id values; each work "
+            "must appear at most once"
+        )
+
     # join with refined metadata on work id so we can propagate work-specific
     # resolutions (author names are unambiguous but pub places are not)
     refined_df = df.join(
@@ -32,8 +42,14 @@ def refine_metadata(
         suffix="_refined",
         how="left",  # preserve all original records
     )
-    # report how many instances of each field will be updated
-    author_changes_df = refined_df.filter(pl.col.author.ne(pl.col.author_refined))
+    # report how many instances of each field will be updated. A change is
+    # counted only when the refined column carries a value that differs from
+    # the original (the original is filled so comparisons with null are safe);
+    # an empty refined value means "leave the original alone".
+    author_changes_df = refined_df.filter(
+        pl.col("author_refined").is_not_null()
+        & pl.col("author_refined").ne(pl.col("author").fill_null(""))
+    )
     num_author_changes = author_changes_df.height
     uniq_author_changes = (
         author_changes_df.select("author", "author_refined").unique().height
@@ -43,7 +59,8 @@ def refine_metadata(
     )
 
     pubplace_changes_df = refined_df.filter(
-        pl.col.pub_place.ne(pl.col.pub_place_refined)
+        pl.col("pub_place_refined").is_not_null()
+        & pl.col("pub_place_refined").ne(pl.col("pub_place").fill_null(""))
     )
     num_pubplace_changes = pubplace_changes_df.height
     uniq_pubplace_changes = pubplace_changes_df.select(
@@ -53,17 +70,24 @@ def refine_metadata(
         f"Updated pub_place in {num_pubplace_changes:,} works ({uniq_pubplace_changes:,} unique replacements)"
     )
 
-    # then replace the original with the refined fields
-    # (refined metadata preserves unchanged values, so we can just copy them over)
-    refined_df = refined_df.drop("author", "pub_place").rename(
-        {"author_refined": "author", "pub_place_refined": "pub_place"}
-    )
+    # then replace the original with the refined fields, keeping the original
+    # value whenever the refined set has none for a record (work_ids absent from
+    # the refined set, or freshly-added records). This prevents the join from
+    # silently wiping author/pub_place to null.
+    refined_df = refined_df.with_columns(
+        author=pl.coalesce("author_refined", "author"),
+        pub_place=pl.coalesce("pub_place_refined", "pub_place"),
+    ).drop("author_refined", "pub_place_refined")
 
     # identify any records added after the manual cleanup, and update author/pub_place
     # if there is an unambiguous mapping from the refined set
-    post_refine_records = df.filter(pl.col.added.gt(refined_last_update))
+    post_refine_records = df.filter(pl.col("added").gt(refined_last_update))
     if post_refine_records.height:
         author_lookup = author_changes_df.select("author", "author_refined").unique()
+        # omit any ambiguous authors (a single original author resolving to more
+        # than one refined value cannot be applied automatically, and would make
+        # the join below emit duplicate rows per work_id)
+        author_lookup = author_lookup.filter(~author_lookup["author"].is_duplicated())
 
         pub_place_lookup = pubplace_changes_df.select(
             "pub_place", "pub_place_refined"
@@ -79,7 +103,10 @@ def refine_metadata(
             .join(pub_place_lookup, on="pub_place", how="left")
             .filter(
                 # limit to records with at least one refined value
-                ~(pl.col.author_refined.is_null() & pl.col.pub_place_refined.is_null())
+                ~(
+                    pl.col("author_refined").is_null()
+                    & pl.col("pub_place_refined").is_null()
+                )
             )
         )
 
@@ -94,12 +121,12 @@ def refine_metadata(
             )
             .with_columns(
                 # take the refined value if not null, otherwise the previous value
-                author=pl.when(pl.col.author_refined.is_not_null())
-                .then(pl.col.author_refined)
-                .otherwise(pl.col.author),
-                pub_place=pl.when(pl.col.pub_place_refined.is_not_null())
-                .then(pl.col.pub_place_refined)
-                .otherwise(pl.col.pub_place),
+                author=pl.when(pl.col("author_refined").is_not_null())
+                .then(pl.col("author_refined"))
+                .otherwise(pl.col("author")),
+                pub_place=pl.when(pl.col("pub_place_refined").is_not_null())
+                .then(pl.col("pub_place_refined"))
+                .otherwise(pl.col("pub_place")),
             )
             .drop("author_refined", "pub_place_refined")
         )
@@ -113,6 +140,9 @@ def refine_metadata(
     refined_df.select(csv_columns).write_csv(csv_metadata_path)
 
     # load json, update based on csv, write out
+    # NOTE: the CSV and JSON are known to be equivalent and in the same order
+    # (both are generated from the same dataset), so the rows are paired
+    # positionally; the per-row assert below guards against drift.
     with json_metadata_path.open() as jsonfile:
         json_metadata = json.load(jsonfile)
     refined_json_metadata = []

From 708ebe8de6160563ae8d87b194d4dd167ae900cc Mon Sep 17 00:00:00 2001
From: rlskoeser <rebecca.s.koeser@princeton.edu>
Date: Tue, 15 Sep 2026 15:54:46 -0400
Subject: [PATCH 3/4] Remove unnecessary checks, clean up comments

---
 src/corppa/utils/dataset_refine.py | 51 +++++++++++-------------------
 1 file changed, 18 insertions(+), 33 deletions(-)

diff --git a/src/corppa/utils/dataset_refine.py b/src/corppa/utils/dataset_refine.py
index 30b9b10c..f65e24d6 100644
--- a/src/corppa/utils/dataset_refine.py
+++ b/src/corppa/utils/dataset_refine.py
@@ -18,22 +18,12 @@ def refine_metadata(
     refined_metadata: Path, csv_metadata_path: Path, json_metadata_path: Path
 ) -> None:
     refined_metadata_df = pl.read_csv(refined_metadata)
-    # get the last update time of records in refined metadata
-    # (drop_nulls: a null/empty updated column means there is no cutoff to
-    # derive, and the post-refine block below is safely skipped)
-    refined_last_update = refined_metadata_df["updated"].drop_nulls().max()
+    # get the last update time of records in refined metadata (all records have added/updated time)
+    refined_last_update = refined_metadata_df["updated"].max()
     df = pl.read_csv(csv_metadata_path)
     csv_columns = df.columns  # store columns original order for output file
     csv_total = df.height
 
-    # each work_id must resolve to at most one refined record, otherwise the
-    # join below would inflate the record count
-    if refined_metadata_df["work_id"].is_duplicated().any():
-        raise ValueError(
-            "Refined metadata contains duplicate work_id values; each work "
-            "must appear at most once"
-        )
-
     # join with refined metadata on work id so we can propagate work-specific
     # resolutions (author names are unambiguous but pub places are not)
     refined_df = df.join(
@@ -47,8 +37,8 @@ def refine_metadata(
     # the original (the original is filled so comparisons with null are safe);
     # an empty refined value means "leave the original alone".
     author_changes_df = refined_df.filter(
-        pl.col("author_refined").is_not_null()
-        & pl.col("author_refined").ne(pl.col("author").fill_null(""))
+        pl.col.author_refined.is_not_null()
+        & pl.col.author_refined.ne(pl.col.author.fill_null(""))
     )
     num_author_changes = author_changes_df.height
     uniq_author_changes = (
@@ -59,8 +49,8 @@ def refine_metadata(
     )
 
     pubplace_changes_df = refined_df.filter(
-        pl.col("pub_place_refined").is_not_null()
-        & pl.col("pub_place_refined").ne(pl.col("pub_place").fill_null(""))
+        pl.col.pub_place_refined.is_not_null()
+        & pl.col.pub_place_refined.ne(pl.col.pub_place.fill_null(""))
     )
     num_pubplace_changes = pubplace_changes_df.height
     uniq_pubplace_changes = pubplace_changes_df.select(
@@ -70,10 +60,8 @@ def refine_metadata(
         f"Updated pub_place in {num_pubplace_changes:,} works ({uniq_pubplace_changes:,} unique replacements)"
     )
 
-    # then replace the original with the refined fields, keeping the original
-    # value whenever the refined set has none for a record (work_ids absent from
-    # the refined set, or freshly-added records). This prevents the join from
-    # silently wiping author/pub_place to null.
+    # then replace the original with the refined fields, but preserve the original
+    # value whenever the refined set is absent
     refined_df = refined_df.with_columns(
         author=pl.coalesce("author_refined", "author"),
         pub_place=pl.coalesce("pub_place_refined", "pub_place"),
@@ -81,7 +69,7 @@ def refine_metadata(
 
     # identify any records added after the manual cleanup, and update author/pub_place
     # if there is an unambiguous mapping from the refined set
-    post_refine_records = df.filter(pl.col("added").gt(refined_last_update))
+    post_refine_records = df.filter(pl.col.added.gt(refined_last_update))
     if post_refine_records.height:
         author_lookup = author_changes_df.select("author", "author_refined").unique()
         # omit any ambiguous authors (a single original author resolving to more
@@ -103,10 +91,7 @@ def refine_metadata(
             .join(pub_place_lookup, on="pub_place", how="left")
             .filter(
                 # limit to records with at least one refined value
-                ~(
-                    pl.col("author_refined").is_null()
-                    & pl.col("pub_place_refined").is_null()
-                )
+                ~(pl.col.author_refined.is_null() & pl.col.pub_place_refined.is_null())
             )
         )
 
@@ -121,12 +106,12 @@ def refine_metadata(
             )
             .with_columns(
                 # take the refined value if not null, otherwise the previous value
-                author=pl.when(pl.col("author_refined").is_not_null())
-                .then(pl.col("author_refined"))
-                .otherwise(pl.col("author")),
-                pub_place=pl.when(pl.col("pub_place_refined").is_not_null())
-                .then(pl.col("pub_place_refined"))
-                .otherwise(pl.col("pub_place")),
+                author=pl.when(pl.col.author_refined.is_not_null())
+                .then(pl.col.author_refined)
+                .otherwise(pl.col.author),
+                pub_place=pl.when(pl.col.pub_place_refined.is_not_null())
+                .then(pl.col.pub_place_refined)
+                .otherwise(pl.col.pub_place),
             )
             .drop("author_refined", "pub_place_refined")
         )
@@ -141,8 +126,8 @@ def refine_metadata(
 
     # load json, update based on csv, write out
     # NOTE: the CSV and JSON are known to be equivalent and in the same order
-    # (both are generated from the same dataset), so the rows are paired
-    # positionally; the per-row assert below guards against drift.
+    # (both are generated from the same data), so rows are paired in sequence
+    # the per-row assert is a check to catch any mismatches
     with json_metadata_path.open() as jsonfile:
         json_metadata = json.load(jsonfile)
     refined_json_metadata = []

From 8ab7a537adea25903a1025bcb96345bf7689152e Mon Sep 17 00:00:00 2001
From: rlskoeser <rebecca.s.koeser@princeton.edu>
Date: Tue, 15 Sep 2026 16:07:11 -0400
Subject: [PATCH 4/4] Add unit tests for data refine script

Assisted-by: OpenCode:BigPickle (glm-5.1)
---
 tests/test_utils/test_dataset_refine.py | 401 ++++++++++++++++++++++++
 1 file changed, 401 insertions(+)
 create mode 100644 tests/test_utils/test_dataset_refine.py

diff --git a/tests/test_utils/test_dataset_refine.py b/tests/test_utils/test_dataset_refine.py
new file mode 100644
index 00000000..d3eaf9df
--- /dev/null
+++ b/tests/test_utils/test_dataset_refine.py
@@ -0,0 +1,401 @@
+# Copyright (c) 2024-2026, Center for Digital Humanities, Princeton University
+# SPDX-License-Identifier: Apache-2.0
+
+import json
+from pathlib import Path
+
+import polars as pl
+import pytest
+
+import corppa.utils.dataset_refine as dataset_refine
+
+
+def _make_corpus(tmp_path: Path, rows, refined_rows) -> tuple[Path, Path]:
+    """Write ppa_metadata.csv/json + refined.csv; returns (corpus_dir, refined_csv)."""
+    corpus_dir = tmp_path / "corpus"
+    corpus_dir.mkdir()
+    pl.DataFrame(rows).write_csv(corpus_dir / "ppa_metadata.csv")
+    refined_path = tmp_path / "refined.csv"
+    pl.DataFrame(refined_rows).write_csv(refined_path)
+    with (corpus_dir / "ppa_metadata.json").open("w") as f:
+        json.dump(rows, f, indent=2)
+    return corpus_dir, refined_path
+
+
+def _csv_columns(corpus_dir: Path) -> dict[str, list]:
+    return pl.read_csv(corpus_dir / "ppa_metadata.csv").to_dict(as_series=False)
+
+
+def _json_rows(corpus_dir: Path) -> list[dict]:
+    with (corpus_dir / "ppa_metadata.json").open() as f:
+        return json.load(f)
+
+
+def test_propagates_refined_author_and_pub_place(tmp_path):
+    rows = [
+        {
+            "work_id": "w1",
+            "author": "Alice A.",
+            "pub_place": "Cambridge",
+            "added": "2024-01-01",
+        },
+        {
+            "work_id": "w2",
+            "author": "Bob B.",
+            "pub_place": "Oxford",
+            "added": "2024-01-01",
+        },
+    ]
+    refined_rows = [
+        {
+            "work_id": "w1",
+            "author": "Alice Adams",
+            "pub_place": "Cambridge (UK)",
+            "updated": "2024-02-01",
+        },
+        {
+            "work_id": "w2",
+            "author": "Bob Brown",
+            "pub_place": "Oxford",
+            "updated": "2024-02-01",
+        },
+    ]
+    corpus_dir, refined = _make_corpus(tmp_path, rows, refined_rows)
+
+    dataset_refine.refine_metadata(
+        refined, corpus_dir / "ppa_metadata.csv", corpus_dir / "ppa_metadata.json"
+    )
+
+    out = _csv_columns(corpus_dir)
+    assert out["author"] == ["Alice Adams", "Bob Brown"]
+    assert out["pub_place"] == ["Cambridge (UK)", "Oxford"]
+    js = _json_rows(corpus_dir)
+    assert js[0]["author"] == "Alice Adams"
+    assert js[0]["pub_place"] == "Cambridge (UK)"
+    assert "added" in out  # extra columns are preserved
+
+
+def test_unmatched_record_keeps_original_values(tmp_path):
+    # a work absent from the refined set must not have its author/pub_place
+    # wiped to null
+    rows = [
+        {
+            "work_id": "w1",
+            "author": "Alice A.",
+            "pub_place": "Cambridge",
+            "added": "2024-01-01",
+        },
+        {
+            "work_id": "w2",
+            "author": "Bob B.",
+            "pub_place": "Oxford",
+            "added": "2024-01-01",
+        },
+    ]
+    refined_rows = [
+        {
+            "work_id": "w1",
+            "author": "Alice Adams",
+            "pub_place": "Cambridge (UK)",
+            "updated": "2024-02-01",
+        },
+    ]
+    corpus_dir, refined = _make_corpus(tmp_path, rows, refined_rows)
+
+    dataset_refine.refine_metadata(
+        refined, corpus_dir / "ppa_metadata.csv", corpus_dir / "ppa_metadata.json"
+    )
+
+    out = _csv_columns(corpus_dir)
+    assert out["author"] == ["Alice Adams", "Bob B."]
+    assert out["pub_place"] == ["Cambridge (UK)", "Oxford"]
+    js = _json_rows(corpus_dir)
+    assert js[1]["author"] == "Bob B."
+
+
+def test_fills_null_original_author_from_refined(tmp_path):
+    rows = [
+        {"work_id": "w1", "author": None, "pub_place": "Boston", "added": "2024-01-01"}
+    ]
+    refined_rows = [
+        {
+            "work_id": "w1",
+            "author": "Anonymous",
+            "pub_place": "Boston",
+            "updated": "2024-02-01",
+        }
+    ]
+    corpus_dir, refined = _make_corpus(tmp_path, rows, refined_rows)
+
+    dataset_refine.refine_metadata(
+        refined, corpus_dir / "ppa_metadata.csv", corpus_dir / "ppa_metadata.json"
+    )
+
+    assert _csv_columns(corpus_dir)["author"] == ["Anonymous"]
+
+
+def test_post_refine_records_updated_via_unambiguous_lookup(tmp_path):
+    rows = [
+        {
+            "work_id": "w1",
+            "author": "Doe",
+            "pub_place": "Cambridge",
+            "added": "2024-01-01",
+        },
+        {
+            "work_id": "w2",
+            "author": "Doe",
+            "pub_place": "Cambridge",
+            "added": "2024-07-01",
+        },
+    ]
+    refined_rows = [
+        {
+            "work_id": "w1",
+            "author": "John Doe",
+            "pub_place": "Cambridge (UK)",
+            "updated": "2024-06-01",
+        },
+    ]
+    corpus_dir, refined = _make_corpus(tmp_path, rows, refined_rows)
+
+    dataset_refine.refine_metadata(
+        refined, corpus_dir / "ppa_metadata.csv", corpus_dir / "ppa_metadata.json"
+    )
+
+    out = _csv_columns(corpus_dir)
+    assert out["author"] == ["John Doe", "John Doe"]
+    assert out["pub_place"] == ["Cambridge (UK)", "Cambridge (UK)"]
+
+
+def test_ambiguous_author_lookup_is_not_applied(tmp_path):
+    # "Doe" resolves to two different refined authors, so a post-refine record
+    # with author "Doe" cannot be updated automatically and keeps its value
+    rows = [
+        {"work_id": "w1", "author": "Doe", "pub_place": "X", "added": "2024-01-01"},
+        {"work_id": "w2", "author": "Doe", "pub_place": "Y", "added": "2024-01-01"},
+        {"work_id": "w3", "author": "Doe", "pub_place": "Z", "added": "2024-07-01"},
+    ]
+    refined_rows = [
+        {
+            "work_id": "w1",
+            "author": "John Doe",
+            "pub_place": "X",
+            "updated": "2024-06-01",
+        },
+        {
+            "work_id": "w2",
+            "author": "Jane Doe",
+            "pub_place": "Y",
+            "updated": "2024-06-01",
+        },
+    ]
+    corpus_dir, refined = _make_corpus(tmp_path, rows, refined_rows)
+
+    dataset_refine.refine_metadata(
+        refined, corpus_dir / "ppa_metadata.csv", corpus_dir / "ppa_metadata.json"
+    )
+
+    out = _csv_columns(corpus_dir)
+    assert out["author"] == ["John Doe", "Jane Doe", "Doe"]
+    # no rows were lost or duplicated by the ambiguous lookup
+    assert len(out["author"]) == 3
+
+
+def test_ambiguous_pub_place_lookup_is_not_applied(tmp_path):
+    # "Cambridge" resolves to two distinct refined places; a post-refine record
+    # with pub_place "Cambridge" keeps its original value
+    rows = [
+        {
+            "work_id": "w1",
+            "author": "A",
+            "pub_place": "Cambridge",
+            "added": "2024-01-01",
+        },
+        {
+            "work_id": "w2",
+            "author": "B",
+            "pub_place": "Cambridge",
+            "added": "2024-01-01",
+        },
+        {
+            "work_id": "w3",
+            "author": "C",
+            "pub_place": "Cambridge",
+            "added": "2024-07-01",
+        },
+    ]
+    refined_rows = [
+        {
+            "work_id": "w1",
+            "author": "A",
+            "pub_place": "Cambridge (US)",
+            "updated": "2024-06-01",
+        },
+        {
+            "work_id": "w2",
+            "author": "B",
+            "pub_place": "Cambridge (UK)",
+            "updated": "2024-06-01",
+        },
+    ]
+    corpus_dir, refined = _make_corpus(tmp_path, rows, refined_rows)
+
+    dataset_refine.refine_metadata(
+        refined, corpus_dir / "ppa_metadata.csv", corpus_dir / "ppa_metadata.json"
+    )
+
+    out = _csv_columns(corpus_dir)
+    assert out["pub_place"] == ["Cambridge (US)", "Cambridge (UK)", "Cambridge"]
+    assert len(out["pub_place"]) == 3
+
+
+def test_duplicate_work_id_in_refined_raises(tmp_path):
+    # a duplicated work_id in the refined set inflates the join and trips the
+    # no-records-lost assertion
+    rows = [{"work_id": "w1", "author": "A", "pub_place": "X", "added": "2024-01-01"}]
+    refined_rows = [
+        {"work_id": "w1", "author": "B", "pub_place": "X", "updated": "2024-02-01"},
+        {"work_id": "w1", "author": "C", "pub_place": "X", "updated": "2024-02-01"},
+    ]
+    corpus_dir, refined = _make_corpus(tmp_path, rows, refined_rows)
+
+    with pytest.raises(AssertionError):
+        dataset_refine.refine_metadata(
+            refined, corpus_dir / "ppa_metadata.csv", corpus_dir / "ppa_metadata.json"
+        )
+
+
+def test_backup_files_created(tmp_path):
+    rows = [{"work_id": "w1", "author": "A", "pub_place": "X", "added": "2024-01-01"}]
+    refined_rows = [
+        {"work_id": "w1", "author": "B", "pub_place": "Y", "updated": "2024-02-01"}
+    ]
+    corpus_dir, refined = _make_corpus(tmp_path, rows, refined_rows)
+
+    dataset_refine.refine_metadata(
+        refined, corpus_dir / "ppa_metadata.csv", corpus_dir / "ppa_metadata.json"
+    )
+
+    assert (corpus_dir / "ppa_metadata.csv.bk").is_file()
+    assert (corpus_dir / "ppa_metadata.json.bk").is_file()
+    # backups carry the original, unrefined values
+    assert pl.read_csv(corpus_dir / "ppa_metadata.csv.bk")["author"].to_list() == ["A"]
+
+
+def test_column_order_and_row_count_preserved(tmp_path):
+    rows = [
+        {
+            "work_id": "w1",
+            "title": "T1",
+            "author": "A",
+            "pub_place": "X",
+            "added": "2024-01-01",
+        },
+        {
+            "work_id": "w2",
+            "title": "T2",
+            "author": "B",
+            "pub_place": "Y",
+            "added": "2024-01-01",
+        },
+    ]
+    refined_rows = [
+        {"work_id": "w1", "author": "C", "pub_place": "Z", "updated": "2024-02-01"}
+    ]
+    corpus_dir, refined = _make_corpus(tmp_path, rows, refined_rows)
+    csv_path = corpus_dir / "ppa_metadata.csv"
+    original_columns = pl.read_csv(csv_path).columns
+
+    dataset_refine.refine_metadata(refined, csv_path, corpus_dir / "ppa_metadata.json")
+
+    out_df = pl.read_csv(csv_path)
+    assert out_df.columns == original_columns
+    assert out_df.height == 2
+
+
+def test_json_row_without_author_pub_place_keys_unchanged(tmp_path):
+    rows = [
+        {"work_id": "w1", "author": "A", "pub_place": "X", "added": "2024-01-01"},
+        {"work_id": "w2", "title": "no author field", "added": "2024-01-01"},
+    ]
+    refined_rows = [
+        {"work_id": "w1", "author": "B", "pub_place": "Y", "updated": "2024-02-01"}
+    ]
+    corpus_dir, refined = _make_corpus(tmp_path, rows, refined_rows)
+
+    dataset_refine.refine_metadata(
+        refined, corpus_dir / "ppa_metadata.csv", corpus_dir / "ppa_metadata.json"
+    )
+
+    js = _json_rows(corpus_dir)
+    assert js[0]["author"] == "B"
+    # the second json row has neither key, so it stays untouched even though its
+    # csv counterpart's author was (coalesced)...
+    assert "author" not in js[1]
+    assert js[1]["title"] == "no author field"
+
+
+def test_json_order_mismatch_raises(tmp_path):
+    rows = [
+        {"work_id": "w1", "author": "A", "pub_place": "X", "added": "2024-01-01"},
+        {"work_id": "w2", "author": "B", "pub_place": "Y", "added": "2024-01-01"},
+    ]
+    refined_rows = [
+        {"work_id": "w1", "author": "C", "pub_place": "Z", "updated": "2024-02-01"}
+    ]
+    corpus_dir, refined = _make_corpus(tmp_path, rows, refined_rows)
+    # shuffle the json rows so they no longer match the csv order
+    with (corpus_dir / "ppa_metadata.json").open("w") as f:
+        json.dump([rows[1], rows[0]], f, indent=2)
+
+    with pytest.raises(AssertionError):
+        dataset_refine.refine_metadata(
+            refined, corpus_dir / "ppa_metadata.csv", corpus_dir / "ppa_metadata.json"
+        )
+
+
+# --- main() CLI ---
+
+
+def test_main_missing_refined_file_exits(tmp_path, monkeypatch):
+    corpus_dir = tmp_path / "corpus"
+    corpus_dir.mkdir()
+    monkeypatch.setattr(
+        "sys.argv",
+        ["dataset_refine.py", str(tmp_path / "nope.csv"), str(corpus_dir)],
+    )
+
+    with pytest.raises(SystemExit, match="Input metadata file"):
+        dataset_refine.main()
+
+
+def test_main_missing_csv_exits(tmp_path, monkeypatch):
+    corpus_dir = tmp_path / "corpus"
+    corpus_dir.mkdir()
+    refined = tmp_path / "refined.csv"
+    refined.write_text("work_id,author,pub_place,updated\n")
+    with (corpus_dir / "ppa_metadata.json").open("w") as f:
+        json.dump([], f)  # only json present; csv is missing
+    monkeypatch.setattr(
+        "sys.argv", ["dataset_refine.py", str(refined), str(corpus_dir)]
+    )
+
+    with pytest.raises(SystemExit, match="ppa_metadata.csv not found"):
+        dataset_refine.main()
+
+
+def test_main_full_run_exits_zero(tmp_path, monkeypatch, capsys):
+    rows = [{"work_id": "w1", "author": "A", "pub_place": "X", "added": "2024-01-01"}]
+    refined_rows = [
+        {"work_id": "w1", "author": "B", "pub_place": "Y", "updated": "2024-02-01"}
+    ]
+    corpus_dir, refined = _make_corpus(tmp_path, rows, refined_rows)
+    monkeypatch.setattr(
+        "sys.argv", ["dataset_refine.py", str(refined), str(corpus_dir)]
+    )
+
+    dataset_refine.main()
+
+    assert "Updated author in 1 works" in capsys.readouterr().out
+    assert _csv_columns(corpus_dir)["author"] == ["B"]