用 Python 按 ID 比较两份 CSV：找出新增、删除和字段修改

python3 --version
mkdir csv-diff-demo
cd csv-diff-demo

cat > old.csv <<'CSV'
sample_id,label,note
001,0,stable
002,1,removed
003,1,keep
CSV
cat > new.csv <<'CSV'
sample_id,label,note
003,1,keep
001,1,stable
004,0,added
CSV

"""Compare two CSV snapshots by one unique key. Python 3.9+, standard library."""
import argparse
import csv
import json
from pathlib import Path
import sys


def load_csv(path, key):
    records = {}
    with open(path, encoding="utf-8-sig", newline="") as handle:
        reader = csv.reader(handle, strict=True)
        header = next(reader, None)
        if not header or any(not name or name != name.strip() for name in header):
            raise ValueError(f"{path}: empty or padded column name")
        if len(set(header)) != len(header) or key not in header:
            raise ValueError(f"{path}: columns must be unique and include {key!r}")
        key_index = header.index(key)
        for number, row in enumerate(reader, 1):
            if len(row) != len(header):
                raise ValueError(f"{path}: record {number}: wrong field count")
            identity = row[key_index]
            if not identity or identity != identity.strip():
                raise ValueError(f"{path}: record {number}: empty or padded key")
            if identity in records:
                raise ValueError(f"{path}: record {number}: duplicate key {identity!r}")
            records[identity] = dict(zip(header, row))
    return header, records


def compare(old_path, new_path, key):
    old_header, old = load_csv(old_path, key)
    new_header, new = load_csv(new_path, key)
    if old_header != new_header:
        raise ValueError("headers differ: column names and order must match")
    added = sorted(new.keys() - old.keys())
    removed = sorted(old.keys() - new.keys())
    changed = []
    for identity in sorted(old.keys() & new.keys()):
        fields = {
            name: {"old": old[identity][name], "new": new[identity][name]}
            for name in old_header
            if old[identity][name] != new[identity][name]
        }
        if fields:
            changed.append({"id": identity, "fields": fields})
    return {"key": key, "added": added, "removed": removed, "changed": changed}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("old", type=Path)
    parser.add_argument("new", type=Path)
    parser.add_argument("--key", default="sample_id", help="unique key column")
    args = parser.parse_args()
    try:
        report = compare(args.old, args.new, args.key)
        json.dump(report, sys.stdout, ensure_ascii=True, indent=2)
        sys.stdout.write("\n")
        sys.stdout.flush()
    except (OSError, UnicodeError, ValueError, csv.Error) as exc:
        parser.exit(2, f"error: {exc}\n")
    return int(any(report[name] for name in ("added", "removed", "changed")))


if __name__ == "__main__":
    sys.exit(main())

python3 compare_csv.py --help
if python3 compare_csv.py old.csv new.csv --key sample_id > diff.json; then
  result=0
else
  result=$?
fi
printf 'exit=%s\n' "$result"
cat diff.json

{
  "key": "sample_id",
  "added": ["004"],
  "removed": ["002"],
  "changed": [
    {"id": "001", "fields": {"label": {"old": "0", "new": "1"}}}
  ]
}

import csv
import json
import subprocess
import sys

with open("diff.json", encoding="utf-8") as handle:
    report = json.load(handle)
assert report == {
    "key": "sample_id",
    "added": ["004"],
    "removed": ["002"],
    "changed": [{"id": "001", "fields": {"label": {"old": "0", "new": "1"}}}],
}
with open("old.csv", encoding="utf-8", newline="") as handle:
    rows = list(csv.reader(handle))
with open("reordered.csv", "w", encoding="utf-8-sig", newline="") as handle:
    writer = csv.writer(handle)
    writer.writerow(rows[0])
    writer.writerows(reversed(rows[1:]))
result = subprocess.run(
    [sys.executable, "compare_csv.py", "old.csv", "reordered.csv"],
    text=True, capture_output=True, check=False,
)
assert result.returncode == 0, result.stderr
assert json.loads(result.stdout) == {
    "key": "sample_id", "added": [], "removed": [], "changed": [],
}
with open("duplicate.csv", "w", encoding="utf-8", newline="") as handle:
    csv.writer(handle).writerows([rows[0], rows[1], rows[1]])
result = subprocess.run(
    [sys.executable, "compare_csv.py", "old.csv", "duplicate.csv"],
    text=True, capture_output=True, check=False,
)
assert result.returncode == 2
assert "duplicate key" in result.stderr
assert result.stdout == ""
print("PASS: changes, row order, BOM, and duplicate rejection")

#!/usr/bin/env bash
set -eu
if python3 compare_csv.py old.csv new.csv --key sample_id > diff.json; then
  echo "No record changes"
else
  result=$?
  case "$result" in
    1) echo "Records changed; review diff.json" >&2 ;;
    *) echo "Comparison failed; do not use diff.json" >&2 ;;
  esac
  exit "$result"
fi