Compare Two CSV Reports
Find what changed between two spreadsheet exports. Match records by ID, detect additions, removals, and changed fields, then generate a concise summary and detailed CSV reports.
About this project
You export the same report two days running — a device inventory, a customer list, a compliance scan — and the real question is never “what’s in it?” but “what changed?” Which rows are new, which disappeared, and which ones were quietly edited. Doing that by eye across two spreadsheets is miserable and easy to get wrong. A comparison script does it in a blink and never misses a row.
Why it’s worth building: the CSV Cleaner transformed one file. This project reconciles two datasets and derives new information from them — a genuine step up. It’s the everyday shape of automation in security, IT operations, reporting, and data-quality work: two snapshots in, a clear picture of the delta out. Along the way you’ll meet the trick that makes it fast — matching rows through dictionaries keyed by a unique ID instead of comparing every row against every other.
Build it step by step
We’ll start with the obvious, honest first attempt — then feel exactly why it struggles, and fix it with the one idea that makes comparison scripts fast and readable. First, two files to compare.
Step 0: Two exports to compare
Create these two files next to your script. They’re a tiny device inventory keyed by device_id. Look closely: today.csv gains D-1005, loses D-1003, and edits two others (D-1002’s status, D-1004’s owner). Every case we need to detect is in here.
device_id,status,owner,location
D-1001,active,alice,HQ
D-1002,active,bob,Warehouse
D-1003,retired,carol,HQ
D-1004,active,dan,Remote
Step 0 (continued): today’s export
And the second file, taken a day later.
device_id,status,owner,location
D-1001,active,alice,HQ
D-1002,inactive,bob,Warehouse
D-1004,active,erin,Remote
D-1005,active,frank,HQ
Step 1: Load each file as records
csv.DictReader reads each row into a dictionary keyed by the header, so you write row["status"] instead of counting columns as row[3]. We wrap it in a load_rows function that every later step reuses — the first move toward keeping loading, comparing, and reporting apart.
import csv
def load_rows(path):
"""Read a CSV into a list of dict records, one per row."""
with open(path, newline="", encoding="utf-8") as f:
return list(csv.DictReader(f))
for row in load_rows("today.csv"):
print(row)
{'device_id': 'D-1001', 'status': 'active', 'owner': 'alice', 'location': 'HQ'}
{'device_id': 'D-1002', 'status': 'inactive', 'owner': 'bob', 'location': 'Warehouse'}
{'device_id': 'D-1004', 'status': 'active', 'owner': 'erin', 'location': 'Remote'}
{'device_id': 'D-1005', 'status': 'active', 'owner': 'frank', 'location': 'HQ'}
Step 2: A first, honest attempt — nested loops
How do you find the rows that are new today? The obvious answer: for each of today’s rows, look through all of yesterday’s for a matching device_id; if none matches, it’s an addition. This works, and there’s no shame in writing it first.
old_rows = load_rows("yesterday.csv")
new_rows = load_rows("today.csv")
for new in new_rows:
match = None
for old in old_rows: # scan ALL of yesterday for each row
if old["device_id"] == new["device_id"]:
match = old
break
if match is None:
print("Added:", new["device_id"])
Added: D-1005
Step 3: Why the nested loop won’t scale
The nested version is fine for eight rows and painful for eighty thousand. If each file has n rows, you do roughly n × n comparisons — a real inventory of 50,000 devices is billions of checks. And you’d need a separate nested pass for added, removed, and changed. There’s a better way.
Step 4: Index each file by its ID
Here’s the refactor that changes everything. index_by_id turns a list of rows into a dict keyed by device_id. Now the keys of each dict are just the set of IDs in that file — and set subtraction hands us additions and removals directly, no loops in sight.
def index_by_id(rows, id_column):
"""Key each row by its unique id for instant lookup."""
return {row[id_column]: row for row in rows}
old = index_by_id(old_rows, "device_id")
new = index_by_id(new_rows, "device_id")
added = new.keys() - old.keys() # ids today has that yesterday didn't
removed = old.keys() - new.keys() # ids yesterday had that today doesn't
print("Added:", added)
print("Removed:", removed)
Added: {'D-1005'}
Removed: {'D-1003'}
Step 5: Detect the changed records
Additions and removals are the IDs that appear in only one file. Changes live among the IDs in both — same device, different details. We compare a configurable list of columns so you decide what counts as a change; a row is “changed” if any watched column differs.
COMPARE_COLUMNS = ["status", "owner", "location"] # you choose what matters
changed = []
for device_id in old.keys() & new.keys(): # ids in BOTH files
before, after = old[device_id], new[device_id]
if any(before[c] != after[c] for c in COMPARE_COLUMNS):
changed.append(device_id)
print("Changed:", sorted(changed))
Changed: ['D-1002', 'D-1004']
Step 6: Validate the columns you rely on
The comparison trusts that both files contain device_id and every watched column. If a header is misspelled or an export drops a column, we want a clear message — not a cryptic KeyError three functions deep. One small guard, run against each file before we compare, does it.
def require_columns(rows, columns, source):
"""Stop early with a clear message if a needed column is missing."""
present = set(rows[0].keys()) if rows else set()
missing = [c for c in columns if c not in present]
if missing:
print(f"{source} is missing required column(s): {', '.join(missing)}")
raise SystemExit(1)
needed = ["device_id"] + COMPARE_COLUMNS
require_columns(old_rows, needed, "yesterday.csv")
require_columns(new_rows, needed, "today.csv")
print("Both files have the columns we need.")
Both files have the columns we need.
The finished tool
Everything assembled: load_rows reads, index_by_id keys each file, compare derives the three groups with set arithmetic, and write_reports writes one CSV per category into an output/ folder it creates for you. Notice how each function does one job — loading, comparing, and reporting stay cleanly apart, which is exactly what makes a growing tool easy to change.
"""Compare two CSV exports and report what was added, removed, and changed."""
import csv
import sys
from pathlib import Path
# --- Configure how records are matched and compared -------------------------
ID_COLUMN = "device_id" # the unique key in each row
COMPARE_COLUMNS = ["status", "owner", "location"] # fields watched for changes
def load_rows(path):
"""Read a CSV into a list of dict records, one per row."""
with open(path, newline="", encoding="utf-8") as f:
return list(csv.DictReader(f))
def require_columns(rows, columns, source):
"""Stop early with a clear message if a needed column is missing."""
present = set(rows[0].keys()) if rows else set()
missing = [c for c in columns if c not in present]
if missing:
print(f"{source} is missing required column(s): {', '.join(missing)}")
raise SystemExit(1)
def index_by_id(rows, id_column):
"""Key each row by its unique id. Warn on duplicates (last row wins)."""
by_id = {}
for row in rows:
key = row[id_column]
if key in by_id:
print(f"Warning: duplicate {id_column} {key!r}; keeping the last row.")
by_id[key] = row
return by_id
def compare(old, new, columns):
"""Return (added, removed, changed) records given two id->row dicts."""
added = [new[i] for i in new.keys() - old.keys()] # ids only today has
removed = [old[i] for i in old.keys() - new.keys()] # ids only yesterday had
changed = []
for i in old.keys() & new.keys(): # ids in both files
before, after = old[i], new[i]
if any(before.get(c) != after.get(c) for c in columns):
changed.append(after)
return added, removed, changed
def write_reports(groups, out_dir, fieldnames):
"""Write one CSV per category ({name: rows}) into out_dir."""
out_dir.mkdir(parents=True, exist_ok=True)
for name, rows in groups.items():
with open(out_dir / f"{name}.csv", "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(rows)
def main():
if len(sys.argv) < 3:
print("Usage: python csv_compare.py <yesterday.csv> <today.csv>")
raise SystemExit(1)
old_path, new_path = Path(sys.argv[1]), Path(sys.argv[2])
try:
old_rows = load_rows(old_path)
new_rows = load_rows(new_path)
except FileNotFoundError as err:
print(f"No such file: {err.filename}")
raise SystemExit(1)
needed = [ID_COLUMN] + COMPARE_COLUMNS
require_columns(old_rows, needed, old_path)
require_columns(new_rows, needed, new_path)
old = index_by_id(old_rows, ID_COLUMN)
new = index_by_id(new_rows, ID_COLUMN)
added, removed, changed = compare(old, new, COMPARE_COLUMNS)
out_dir = Path("output")
fieldnames = list(new_rows[0].keys()) if new_rows else needed
write_reports(
{"added": added, "removed": removed, "changed": changed},
out_dir, fieldnames)
print("Comparison complete.\n")
print(f"Added records: {len(added)}")
print(f"Removed records: {len(removed)}")
print(f"Changed records: {len(changed)}\n")
print("Reports written to:")
for name in ("added", "removed", "changed"):
print(f"- output/{name}.csv")
if __name__ == "__main__":
main()
$ python csv_compare.py yesterday.csv today.csv Comparison complete. Added records: 1 Removed records: 1 Changed records: 2 Reports written to: - output/added.csv - output/removed.csv - output/changed.csv
Keep going, make it your own
You’ve built a real reconciliation tool. Each idea below turns it into something you’d trust on production data — and teaches one more technique worth having.
Show which fields changed. A row landing in changed.csv tells you it moved, but not how. Compare the watched columns and record each column: old → new, the detail an auditor actually wants.
def field_changes(before, after, columns):
"""Return {column: (old, new)} for every watched column that differs."""
return {c: (before[c], after[c]) for c in columns if before[c] != after[c]}
diffs = field_changes(old["D-1002"], new["D-1002"], COMPARE_COLUMNS)
print(diffs)
{'status': ('active', 'inactive')}
Make the ID and columns real flags with argparse. Instead of editing constants, run python csv_compare.py old.csv new.csv --id-col sku --columns price,stock — and get a --help for free.
import argparse
parser = argparse.ArgumentParser(description="Compare two CSV exports.")
parser.add_argument("old"); parser.add_argument("new")
parser.add_argument("--id-col", default="device_id")
parser.add_argument("--columns", default="status,owner,location",
help="comma-separated columns to compare")
args = parser.parse_args()
compare_columns = args.columns.split(",")
Write a human-readable summary too. Alongside the CSVs, drop a plain-text summary.txt so someone can skim the result without opening a spreadsheet — the same counts your terminal prints, saved for the record.
from pathlib import Path
lines = [
f"Added: {len(added)}",
f"Removed: {len(removed)}",
f"Changed: {len(changed)}",
]
Path("output/summary.txt").write_text("\n".join(lines), encoding="utf-8")
Report duplicate IDs as their own category. Real exports sometimes list the same device_id twice. Right now the last one silently wins; instead, collect the repeats so you can flag a data-quality problem at the source.
from collections import Counter
ids = [row["device_id"] for row in new_rows]
dupes = [device_id for device_id, n in Counter(ids).items() if n > 1]
if dupes:
print("Duplicate ids in today.csv:", dupes)
Download the files
Grab the two sample exports and the finished tool, then run <code>python csv_compare.py yesterday.csv today.csv</code> and compare your <code>output/</code> files to the summary below.
- yesterday.csv (the “before” export) A small inventory snapshot to diff against.
- today.csv (the “after” export) The same inventory a day later, with an add, a removal, and two changes.
- csv_compare.py (finished tool) The full command-line comparison tool.
Mini exercise (medium)
Write the comparison’s core: find_changes(old, new, columns). Given two dicts keyed by ID ({id: record}), return a sorted list of the IDs present in both whose value differs in any of the given columns. Ignore IDs that are only in one dict — those are additions or removals, not changes.
Give it a shot. Complete the code and press Run; there’s nothing to download or configure.
def find_changes(old, new, columns):
changed = []
# TODO: for each id in BOTH dicts, add it to changed
# if any of the given columns differ between old and new
return sorted(changed)
old = {"A": {"status": "on"}, "B": {"status": "on"}}
new = {"A": {"status": "on"}, "B": {"status": "off"}, "C": {"status": "on"}}
print(find_changes(old, new, ["status"]))Loop over the shared IDs with old.keys() & new.keys(), and keep an ID when any(old[key][c] != new[key][c] for c in columns). Return sorted(...).
def find_changes(old, new, columns):
changed = []
for key in old.keys() & new.keys():
if any(old[key][c] != new[key][c] for c in columns):
changed.append(key)
return sorted(changed)
old = {"A": {"status": "on"}, "B": {"status": "on"}}
new = {"A": {"status": "on"}, "B": {"status": "off"}, "C": {"status": "on"}}
print(find_changes(old, new, ["status"]))