Catching Column Drift in Recurring CSV Files Before an Automated Run

I work on Locate Next.

If a job reads the same CSV export every day, it will eventually break in the quietest possible way. Someone renames "Owner Name" to "Owner", a county adds a column in the middle, or a vendor swaps the order of two address fields. The job keeps running, maps the wrong column, and sends a batch of bad rows downstream. Nobody notices until the results look strange.

This post shows a small Python pattern for recurring lead files: remember a schema, detect drift, process only new or changed rows, and do a dry run before anything real happens. The examples use property lists, but the idea fits any repeating CSV. Every snippet below was run against small made-up sample files, including a cosmetic header rename, a column added in the middle, short and long rows, a duplicate ID and a missing schema file.

1. Save a schema fingerprint

When a person accepts a file, store its headers in order along with a hash.

import csv, hashlib, json, sqlite3
from pathlib import Path

def read_headers(path):
    with open(path, newline="", encoding="utf-8-sig") as f:
        return next(csv.reader(f))

def normalize(headers):
    return [h.strip().lower() for h in headers]

def fingerprint(headers):
    norm = normalize(headers)
    return hashlib.sha256("\x1f".join(norm).encode()).hexdigest(), norm

def save_schema(headers, store=Path("schema.json")):
    digest, norm = fingerprint(headers)
    store.write_text(json.dumps({"digest": digest, "headers": norm}))

Normalizing case and surrounding spaces avoids false alarms from cosmetic edits. Anything else counts as a change.

2. Detect drift and name it

A plain "schema changed" message is not useful. Say what changed. Compare the relative order of shared columns, so adding one column in the middle does not mark every later column as moved, and catch duplicate headers.

def diff_schema(new_headers, store=Path("schema.json")):
    if not store.exists():
        raise SystemExit("No accepted schema yet. Review the headers, "
                         "then call save_schema to accept them.")
    old = json.loads(store.read_text())
    digest, norm = fingerprint(new_headers)
    dupes = sorted({h for h in norm if norm.count(h) > 1})
    if digest == old["digest"] and not dupes:
        return None
    added = [h for h in norm if h not in old["headers"]]
    removed = [h for h in old["headers"] if h not in norm]
    common_old = [h for h in old["headers"] if h in norm]
    common_new = [h for h in norm if h in old["headers"]]
    moved = [n for n, o in zip(common_new, common_old) if n != o]
    return {"added": added, "removed": removed,
            "moved": moved, "duplicates": dupes}

If no schema has been accepted yet, the check stops and asks a person to accept the current headers with save_schema, rather than treating the first file as trusted automatically.

When diff_schema returns anything, the run stops. It does not try to remap columns on its own. A person looks at the named changes, updates the mapping and accepts the new schema.

drift = diff_schema(read_headers("incoming/2026-10-28.csv"))
if drift:
    raise SystemExit(f"Paused: source columns changed {drift}")

3. Process only new or changed rows

Pick one column that identifies a row, such as a case number or parcel ID. Store a hash of each row's content keyed by that ID. On the next run, skip rows whose hash has not changed.

Read rows with the same header normalization the fingerprint uses, so a cosmetic rename does not send every row to "no ID". Pass id_col in the same normalized form, such as "case number", so the check and the lookup agree.

def read_rows(path):
    with open(path, newline="", encoding="utf-8-sig") as f:
        reader = csv.DictReader(f)
        reader.fieldnames = normalize(reader.fieldnames)
        return list(reader)

def open_seen(read_only=False):
    if read_only:
        if not Path("seen.db").exists():
            return None
        return sqlite3.connect("file:seen.db?mode=ro", uri=True)
    db = sqlite3.connect("seen.db")
    db.execute("CREATE TABLE IF NOT EXISTS seen (id TEXT PRIMARY KEY, h TEXT)")
    return db

def row_hash(row):
    parts = [f"{k}={(row[k] or '').strip()}"
             for k in sorted(k for k in row if k is not None)]
    return hashlib.sha256("\x1f".join(parts).encode()).hexdigest()

def classify(rows, id_col, db):
    out = {"new": [], "changed": [], "unchanged": [],
           "no_id": [], "duplicate": []}
    seen_now = set()
    for r in rows:
        rid = (r.get(id_col) or "").strip()
        if not rid:
            out["no_id"].append(r); continue
        if rid in seen_now:
            out["duplicate"].append(r); continue
        seen_now.add(rid)
        h = row_hash(r)
        prev = None
        if db is not None:
            prev = db.execute("SELECT h FROM seen WHERE id=?", (rid,)).fetchone()
        key = "new" if prev is None else ("unchanged" if prev[0] == h else "changed")
        out[key].append(r)
    return out

The hash trims whitespace in each value, so a trailing space is not a fake change. It sorts keys, so column order does not matter, and it tolerates short rows (missing values) and long rows (extra values with no header). Rows without an ID are reported, not silently processed. A second row with the same ID in one file is held as a duplicate, not processed twice.

After you accept a schema that adds a column, expect every row to read as changed once; decide whether that column should be part of the hash.

4. Dry run first

A dry run reads the file and reports what a real run would do, without calling any external service. It opens seen.db read only, and if the file does not exist yet it treats every row as new without creating it, so a dry run writes nothing.

def dry_run(path, id_col, max_per_run):
    c = classify(read_rows(path), id_col, open_seen(read_only=True))
    work = c["new"] + c["changed"]
    return {
        "would_process": min(len(work), max_per_run),
        "over_limit_waits": max(0, len(work) - max_per_run),
        "skipped_unchanged": len(c["unchanged"]),
        "held_no_id": len(c["no_id"]),
        "held_duplicate": len(c["duplicate"]),
    }

This version only counts rows. In production, attach a reason to every skipped or held row (unchanged, no ID, duplicate, over limit) so the report can be audited. And any change to the job's settings means a fresh dry run before the next real one.

5. Process files in date order and record everything

When a folder collects files over time, process them in date order, using the date in the filename rather than modification time, so a changed row in a later file wins. Copying or re-downloading a file changes its modification time; an ISO date in the name does not.

files = sorted(Path("incoming").glob("*.csv"))  # ISO-dated names sort chronologically

def real_run(path, id_col, max_per_run, send):
    db = open_seen()
    c = classify(read_rows(path), id_col, db)
    for r in (c["new"] + c["changed"])[:max_per_run]:
        send(r)  # your downstream call
        db.execute("INSERT OR REPLACE INTO seen VALUES (?, ?)",
                   (r[id_col].strip(), row_hash(r)))
    db.commit()

Write a row to seen only after its downstream call succeeds, and call db.commit() after each file, so a crash mid-run retries the unfinished rows instead of silently skipping them. Log what happened to each row too. That log is how you answer "why was this row skipped" a month later.

Common mistakes

• Remapping columns automatically by fuzzy header match. It works until "Mail Address" quietly maps to "Property Address".
• Using a row number as the ID. Row numbers change whenever a file is sorted.
• Treating a dry run as optional after a settings change.
• Hashing rows without normalizing whitespace, which turns trailing spaces into fake changes.

How Locate Next uses the same idea

Locate Next is skip tracing software for real estate teams, and its automations follow this shape: a source folder of CSV files read oldest first, one ID column, new rows only, a reuse window (recently searched properties aren't searched again), a per-run search limit and a daily spending cap, and a dry run that shows what a real run would do before anything goes live. When the source file changes its columns, the next run pauses and names them. The product page shows how these settings look in the app: Automated skip tracing for growing lists.

Whatever tool you use, the rule is the same: when the input changes shape, stop and say what changed, then let a person decide.