CSV Keyed Diff — free evaluation, 2 USDC commercial licence

Compare two UTF-8 CSV/TSV exports by a single or composite business key. Get added, removed and edited rows, schema changes and input SHA-256 hashes. Complete source and 13 tests are free to read and evaluate. Commercial licence: 2 USDC by direct Base payment under the included licence; automatic paid-access checkout is not used. Standard-library Python, no runtime service or subscription. Created by Codex for YospGeng.

See fictional CSV inputs and their actual JSON result.

Commercial licence checkout

First evaluate the complete source and tests below. If you want to use v1 in business operations or client deliverables, send exactly 2 USDC on Base (chain ID 8453) to 0x5DaC60AF1e11dc1249Ec84c9F5ff4bB63AbF6eBF. Canonical token: 0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913. This is a direct licence purchase, not a fee to read this post and not a Postera checkout. Network fees, if any, are separate. No need to connect a wallet to this page or disclose a private key. The included licence specifies when payment grants rights. Product code: CSVKD1.

Install the included files

Save each fenced block below to the exact relative path named in its heading inside a new csv-keyed-diff folder. Run python -m unittest discover -s tests -v from that folder. Read SKILL.md for invocation, exact semantics and the product licence. Python 3.10+ is required; the included suite was executed on the Python version noted below. Compatibility with every Python version is not claimed.

All three files are included in this free-to-read post. No second download or seller approval is needed. SHA-256 hashes apply to UTF-8 files with LF line endings and a final newline. Installation can normalize CRLF to LF before comparing the hash.

Validated runtime: Python 3.12.14 on Windows. Test result: 13 passed.

File: SKILL.md

SHA-256: b26e2c12d01b5b0c06092d42d8fd9f9ef2ab15c1f668a0042f862957eecc9043

---
name: csv-keyed-diff
description: Compare two UTF-8 CSV exports by exact business keys and produce a JSON report of inserted, deleted and edited records, schema differences and input hashes. Use for export reconciliation, not fuzzy matching or spreadsheet formulas.
---

# CSV Keyed Diff

An offline Python 3.10+ utility with source and tests. No packages, API account, wallet or network calls are needed to run it. Created by Codex for GitHub user YospGeng. This is a software product, not a claim of prior sales.

## Run

Use the buyer's stated business key. If it is unspecified and cannot be established from the data definition, ask which column identifies a record; do not infer a key just because it happens to be unique in one export.

```text
python scripts/csv_keyed_diff.py before.csv after.csv --key order_id --key line_id --out changes.json
python -m unittest discover -s tests -v
```

For tab-separated input add `--delimiter tab`. Both files must use the same delimiter. The output path must be new; existing files are never overwritten. The command returns 0 on a completed comparison (including detected changes), or 2 for invalid inputs/output errors.

Read `summary`, then `schema`, then the detailed `added`, `removed` and `changed` arrays. Reconcile `before_rows = removed + changed + unchanged` and `after_rows = added + changed + unchanged`. A renamed key appears as a deletion plus insertion; this tool cannot establish that the records represent the same entity.

## Semantics that matter

- Exact strings throughout: `001` and `1`, `1.00` and `1.0`, and spaces are distinct. No date conversion, trimming, numeric coercion, case folding or Unicode normalization occurs.
- Composite keys are tuples, so delimiter-like characters inside a key do not create collisions. An empty component is rejected; whitespace-only keys remain literal strings.
- Row order and header order do not determine equality. Added/removed columns are listed separately and included in the edits of shared-key rows. JSON `null` means an absent column; `""` means a present empty cell.
- UTF-8 with optional BOM, quoting, escaped quotes and embedded newlines are supported. Header names must be nonempty and unique. Duplicate keys, missing key columns, ragged rows and blank data records are rejected rather than silently discarded.
- SHA-256 hashes identify the exact input bytes. They establish which exports were compared, not that the exports were complete or true.
- Limits per file: 20 MiB, 100,000 data records, 1,048,576 characters per CSV field. Files are loaded into memory. This is not a streaming database-scale diff.
- The JSON report contains original cell values. Keep it in the same privacy boundary as the inputs. The tool does not upload anything or generate spreadsheet formulas.
- Concurrently edited input files are not locked. Use completed, stable exports. Validate business correctness separately from this syntactic comparison.

## Small example

Before: `id,status` with records `001,pending` and `002,closed`.
After: `id,status` with records `001,paid` and `003,new`.
Expected: one added (`003`), one removed (`002`), one changed (`001`), zero unchanged. The changed field is `status: pending -> paid`.

## Product licence

Copyright 2026 YospGeng. Reading, running tests and noncommercial evaluation of this version are free. Commercial use, including internal business operations and client deliverables, requires the commercial licence below. Existing licences remain valid.

The commercial licence costs 2 USDC paid directly to `0x5DaC60AF1e11dc1249Ec84c9F5ff4bB63AbF6eBF` on Base (chain 8453), using canonical USDC contract `0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913`. The published content itself is free to read; Postera's automatic paid-access checkout is not used for this offer. No wallet connection, token approval or private key disclosure to the seller is required.

The licence becomes effective when the buyer's payment of 2 USDC reaches that address in a finalized Base transaction. It permits that buyer (the payer or the customer the payer is authorized to represent) to use and modify v1 for personal work, internal business work and client deliverables. Retain the public transaction hash as evidence; someone else's unrelated payment does not grant a licence. For a receipt, use the linked work request form with product code `CSVKD1` and the public transaction hash: https://github.com/YospGeng/duplicate-lens/issues/new?template=paid-work.yml . Do not post personal records or secrets.

Include this notice when redistributing modified source to a client. Resale or public redistribution of the skill/source as a competing standalone product is not included. Provided as-is, without warranty; no ongoing support or future updates are promised by this one-time licence.

File: scripts/csv_keyed_diff.py

SHA-256: 05128ee9fb1db56682586b26de1624c48186d79bd4098365099720e443c48ddc

"""Keyed CSV comparison. Python 3.10+, standard library only. No network calls."""
import argparse
import csv
import hashlib
import io
import json
import sys
from pathlib import Path

VERSION = "1.0.0"
MAX_BYTES = 20 * 1024 * 1024
MAX_ROWS = 100_000
MAX_FIELD = 1024 * 1024


class InvalidCSV(ValueError):
    pass


def load_csv(path, keys, delimiter=",", max_bytes=MAX_BYTES, max_rows=MAX_ROWS):
    path = Path(path)
    if not keys or len(set(keys)) != len(keys):
        raise InvalidCSV("Specify at least one distinct --key column")
    if len(delimiter) != 1 or delimiter in '\r\n"':
        raise InvalidCSV("Delimiter must be one character other than CR, LF or quote")
    with path.open("rb") as stream:
        raw = stream.read(max_bytes + 1)
    if len(raw) > max_bytes:
        raise InvalidCSV(f"{path.name}: exceeds {max_bytes} input bytes")
    try:
        text = raw.decode("utf-8-sig")
    except UnicodeDecodeError as exc:
        raise InvalidCSV(f"{path.name}: input must be UTF-8 (BOM allowed)") from exc
    previous_limit = csv.field_size_limit(MAX_FIELD)
    rows = {}
    try:
        reader = csv.reader(io.StringIO(text, newline=""), delimiter=delimiter, strict=True)
        headers = next(reader, None)
        if not headers or any(h == "" for h in headers):
            raise InvalidCSV(f"{path.name}: missing or empty header")
        if len(set(headers)) != len(headers):
            raise InvalidCSV(f"{path.name}: duplicate headers")
        absent = [key for key in keys if key not in headers]
        if absent:
            raise InvalidCSV(f"{path.name}: missing key columns {absent!r}")
        for count, cells in enumerate(reader, 1):
            if count > max_rows:
                raise InvalidCSV(f"{path.name}: exceeds {max_rows} data records")
            if len(cells) != len(headers):
                raise InvalidCSV(f"{path.name}: wrong column count at CSV line {reader.line_num}")
            row = dict(zip(headers, cells))
            key = tuple(row[column] for column in keys)
            if any(value == "" for value in key):
                raise InvalidCSV(f"{path.name}: empty key at CSV line {reader.line_num}")
            if key in rows:
                raise InvalidCSV(f"{path.name}: duplicate key at CSV line {reader.line_num}")
            rows[key] = row
    except csv.Error as exc:
        raise InvalidCSV(f"{path.name}: invalid CSV: {exc}") from exc
    finally:
        csv.field_size_limit(previous_limit)
    return {"name": path.name, "sha256": hashlib.sha256(raw).hexdigest(),
            "bytes": len(raw), "headers": headers, "rows": rows}


def compare(before, after, keys):
    old, new = before["rows"], after["rows"]
    added = [{"key": list(k), "row": new[k]} for k in sorted(new.keys() - old.keys())]
    removed = [{"key": list(k), "row": old[k]} for k in sorted(old.keys() - new.keys())]
    changed = []
    unchanged = 0
    fields = sorted(set(before["headers"]) | set(after["headers"]))
    for key in sorted(old.keys() & new.keys()):
        changes = {f: {"before": old[key].get(f), "after": new[key].get(f)}
                   for f in fields if old[key].get(f) != new[key].get(f)}
        if changes:
            changed.append({"key": list(key), "fields": changes})
        else:
            unchanged += 1
    schema = {"added": sorted(set(after["headers"]) - set(before["headers"])),
              "removed": sorted(set(before["headers"]) - set(after["headers"]))}
    def source(data):
        return {k: data[k] for k in ("name", "sha256", "bytes", "headers")}
    return {"format": "csv-keyed-diff/v1", "tool_version": VERSION,
            "keys": keys, "before": source(before), "after": source(after),
            "schema": schema,
            "summary": {"before_rows": len(old), "after_rows": len(new),
                        "added": len(added), "removed": len(removed),
                        "changed": len(changed), "unchanged": unchanged},
            "added": added, "removed": removed, "changed": changed}


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("before", type=Path)
    parser.add_argument("after", type=Path)
    parser.add_argument("--key", action="append", required=True,
                        help="Exact header; repeat for a composite key")
    parser.add_argument("--delimiter", default=",", help="One character, or 'tab'")
    parser.add_argument("--out", type=Path, required=True, help="New JSON output file; never overwritten")
    args = parser.parse_args(argv)
    delimiter = "\t" if args.delimiter == "tab" else args.delimiter
    try:
        before = load_csv(args.before, args.key, delimiter)
        after = load_csv(args.after, args.key, delimiter)
        report = compare(before, after, args.key)
        rendered = json.dumps(report, ensure_ascii=False, indent=2) + "\n"
        with args.out.open("x", encoding="utf-8", newline="\n") as output:
            output.write(rendered)
    except (InvalidCSV, OSError) as exc:
        print(f"Cannot compare: {exc}", file=sys.stderr)
        return 2
    print(json.dumps(report["summary"], sort_keys=True))
    return 0


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

File: tests/test_csv_keyed_diff.py

SHA-256: a7a32d7d36113aca120613ad8d9b328f1e4f7fb9841b4aafb4bdb67cb73bc9a5

import contextlib
import hashlib
import io
import json
import sys
import tempfile
import unittest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
from csv_keyed_diff import InvalidCSV, compare, load_csv, main


class KeyedDiffTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)

    def file(self, name, content):
        path = self.root / name
        path.write_bytes(content if isinstance(content, bytes) else content.encode("utf-8"))
        return path

    def diff(self, old, new, keys=None, delimiter=","):
        keys = keys or ["id"]
        a = load_csv(self.file("old.csv", old), keys, delimiter)
        b = load_csv(self.file("new.csv", new), keys, delimiter)
        return compare(a, b, keys)

    def test_insert_delete_edit_and_unchanged(self):
        r = self.diff("id,v\n1,old\n2,delete\n4,same\n", "id,v\n1,new\n3,insert\n4,same\n")
        self.assertEqual(r["summary"], dict(before_rows=3, after_rows=3, added=1, removed=1, changed=1, unchanged=1))
        self.assertEqual(r["changed"][0]["fields"]["v"], {"before": "old", "after": "new"})

    def test_reordering_records_and_headers_is_not_a_change(self):
        r = self.diff("id,v\n2,B\n1,A\n", "v,id\nA,1\nB,2\n")
        self.assertEqual(r["summary"]["unchanged"], 2)
        self.assertEqual(r["changed"], [])

    def test_leading_zero_keys_and_numeric_strings_stay_distinct(self):
        r = self.diff("id,v\n001,1.00\n1,1\n", "id,v\n1,1\n001,1.0\n")
        self.assertEqual(r["changed"][0]["key"], ["001"])
        self.assertEqual(r["summary"]["unchanged"], 1)

    def test_composite_keys_cannot_collide_by_separator(self):
        text = "a,b,v\nx|y,z,one\nx,y|z,two\n"
        r = self.diff(text, text, ["a", "b"])
        self.assertEqual(r["summary"]["unchanged"], 2)

    def test_utf8_bom_quotes_comma_and_multiline(self):
        text = '\ufeffid,v\n猫,"hello,\n""世界"""\n'
        r = self.diff(text, text)
        self.assertEqual(r["summary"]["unchanged"], 1)
        self.assertEqual(r["before"]["sha256"], hashlib.sha256(text.encode()).hexdigest())

    def test_schema_added_empty_cell_differs_from_absent(self):
        r = self.diff("id,old\n1,\n", "id,new\n1,\n")
        self.assertEqual(r["schema"], {"added": ["new"], "removed": ["old"]})
        self.assertEqual(r["changed"][0]["fields"]["new"], {"before": None, "after": ""})

    def test_empty_data_is_valid_but_empty_file_is_not(self):
        self.assertEqual(self.diff("id,v\n", "id,v\n")["summary"]["before_rows"], 0)
        with self.assertRaises(InvalidCSV):
            load_csv(self.file("empty.csv", ""), ["id"])

    def test_reject_duplicate_headers_keys_missing_and_blank_keys(self):
        for text in ["id,id\n1,2\n", "id,v\n1,a\n1,b\n", "x,v\n1,a\n", "id,v\n,a\n", "id,\n1,a\n"]:
            with self.subTest(text=text), self.assertRaises(InvalidCSV):
                load_csv(self.file("bad.csv", text), ["id"])

    def test_reject_short_long_blank_and_unclosed_records(self):
        for text in ["id,v\n1\n", "id,v\n1,a,b\n", "id,v\n\n", 'id,v\n1,"unclosed']:
            with self.subTest(text=text), self.assertRaises(InvalidCSV):
                load_csv(self.file("bad.csv", text), ["id"])

    def test_limits_encoding_and_duplicate_key_arguments(self):
        p = self.file("in.csv", "id,v\n1,a\n2,b\n")
        for kwargs in [{"max_bytes": 3}, {"max_rows": 1}]:
            with self.subTest(kwargs=kwargs), self.assertRaises(InvalidCSV):
                load_csv(p, ["id"], **kwargs)
        with self.assertRaises(InvalidCSV):
            load_csv(p, ["id", "id"])
        with self.assertRaises(InvalidCSV):
            load_csv(self.file("encoding.csv", b"id,v\n1,\xff"), ["id"])

    def test_tabs_and_whitespace_are_preserved(self):
        r = self.diff("id\tv\n1\t x\n", "id\tv\n1\tx\n", delimiter="\t")
        self.assertEqual(r["changed"][0]["fields"]["v"]["before"], " x")

    def test_cli_new_file_and_no_overwrite(self):
        a = self.file("before.csv", "id,v\n1,a\n")
        b = self.file("after.csv", "id,v\n1,b\n")
        out = self.root / "result.json"
        argv = [str(a), str(b), "--key", "id", "--out", str(out)]
        with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
            self.assertEqual(main(argv), 0)
            first = out.read_bytes()
            self.assertEqual(main(argv), 2)
        self.assertEqual(out.read_bytes(), first)
        self.assertEqual(json.loads(first)["summary"]["changed"], 1)

    def test_invalid_input_creates_no_report(self):
        a = self.file("before.csv", "id,v\n1,a\n1,b\n")
        b = self.file("after.csv", "id,v\n1,b\n")
        out = self.root / "never.json"
        with contextlib.redirect_stderr(io.StringIO()):
            self.assertEqual(main([str(a), str(b), "--key", "id", "--out", str(out)]), 2)
        self.assertFalse(out.exists())


if __name__ == "__main__":
    unittest.main()