Audit Rejected CSV Rows and Reconcile Every Input Record

A clean CSV is not enough if you cannot explain which rows were left out. This small import audit gives every parsed data record an identity, sends it to either an accepted file or a rejected file, and checks that the two outputs account for the whole input.

Use it for a small spreadsheet export whose rules you control. It uses Python 3.10 or newer and the standard library, with no package installation. The course lesson on CSV reading and conversion covers the general module workflow. Here the result is a reconciled audit trail that someone can use to correct an export.

Agree on the export contract first

This example expects a comma-separated UTF-8 file, optionally with a UTF-8 byte-order mark, and these exact headers in this order: sku,quantity,unit_price. It trims surrounding field spaces, but it does not guess missing values or interpret locale-specific currency formatting.

• SKU: one capital ASCII letter followed by three digits.
• Quantity: an integer from 1 to 9999, with no leading zeros.
• Unit price: a nonnegative value below one million, using a decimal point and exactly two decimal places. No currency sign or grouping comma.
• Blank records and wrong field counts: record them as rejected rather than silently dropping them.
• Repeated SKUs: allow them. Two line items can legitimately use the same SKU; source record numbers distinguish them.

Export currency cells as plain decimal values. A quoted "₱1,250.00" is valid CSV syntax, but it violates this import contract. That row belongs in the rejection report until the source is corrected. A semicolon-delimited or differently named export fails the header check; adapt the agreed contract rather than guessing its meaning.

Prepare a file with known failures

In a new practice folder, save this as export.csv, keeping the blank line between D404 and E505. There are seven data records, including that blank record. A final newline terminates the last record; it does not add another one.

sku,quantity,unit_price
A101,2,95.00
B202,3,35.50
C303,,12.00
D404,2,"₱1,250.00"

E505,1,10.00,extra
F606,1,4.25

Expect three accepted records and four rejected records. The accepted value should be 300.75: A101 contributes 190.00, B202 contributes 106.50 and F606 contributes 4.25.

Run one complete audit

Save this as audit_csv.py. The script keeps this small file in memory so malformed CSV can stop the run before any output directory is created. The source file is opened only for reading.

import argparse
import csv
from decimal import Decimal
import json
from pathlib import Path
import re


def validate(row):
    if not row or not any(cell.strip() for cell in row):
        return None, "blank record"
    if len(row) != 3:
        return None, "expected 3 fields"
    sku, quantity, price = (cell.strip() for cell in row)
    if not re.fullmatch(r"[A-Z][0-9]{3}", sku):
        return None, "sku must be one capital letter and 3 digits"
    if not re.fullmatch(r"[1-9][0-9]{0,3}", quantity):
        return None, "quantity must be an integer from 1 to 9999"
    if not re.fullmatch(r"[0-9]{1,6}\.[0-9]{2}", price):
        return None, "unit_price must be 0.00 to 999999.99, without currency or grouping"
    value = Decimal(price)
    return [sku, int(quantity), f"{value:.2f}", f"{int(quantity) * value:.2f}"], None


def audit(source, destination):
    accepted, rejected = [], []
    with source.open(encoding="utf-8-sig", newline="") as stream:
        reader = csv.reader(stream, strict=True)
        header = next(reader, None)
        if header != ["sku", "quantity", "unit_price"]:
            raise ValueError("Expected exactly: sku,quantity,unit_price")
        previous_end = reader.line_num
        for record, row in enumerate(reader, start=1):
            start, end = previous_end + 1, reader.line_num
            previous_end = end
            clean, reason = validate(row)
            if reason:
                rejected.append([record, start, end, reason, json.dumps(row, ensure_ascii=False)])
            else:
                accepted.append([record, *clean])

    # Create outputs only after the entire CSV has parsed successfully.
    # A fresh directory prevents overwriting an earlier audit.
    destination.mkdir(exist_ok=False)
    with (destination / "accepted.csv").open("w", encoding="utf-8", newline="") as stream:
        writer = csv.writer(stream)
        writer.writerow(["source_record", "sku", "quantity", "unit_price", "line_total"])
        writer.writerows(accepted)
    with (destination / "rejected.csv").open("w", encoding="utf-8", newline="") as stream:
        writer = csv.writer(stream)
        writer.writerow(["source_record", "line_start", "line_end", "reason", "raw_fields_json"])
        writer.writerows(rejected)

    # Re-read outputs before reporting a completed audit.
    with (destination / "accepted.csv").open(encoding="utf-8", newline="") as stream:
        saved_ok = list(csv.DictReader(stream))
    with (destination / "rejected.csv").open(encoding="utf-8", newline="") as stream:
        saved_bad = list(csv.DictReader(stream))
    count = len(accepted) + len(rejected)
    recorded = sorted(int(row["source_record"]) for row in saved_ok + saved_bad)
    assert recorded == list(range(1, count + 1)), "Missing or duplicated source record"
    assert len(saved_ok) == len(accepted) and len(saved_bad) == len(rejected)
    total = sum((Decimal(row["line_total"]) for row in saved_ok), Decimal("0.00"))
    print(f"Input records: {count}; accepted: {len(saved_ok)}; rejected: {len(saved_bad)}")
    print(f"Accepted value: {total:.2f}")
    print("PASS: every parsed data record appears in exactly one output")


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("source", type=Path)
    parser.add_argument("destination", type=Path)
    args = parser.parse_args()
    audit(args.source, args.destination)

python audit_csv.py export.csv audit-01

Input records: 7; accepted: 3; rejected: 4
Accepted value: 300.75
PASS: every parsed data record appears in exactly one output

Run from the folder containing the script and export. The destination must be new. To repeat the same audit, use audit-02; the script refuses to overwrite audit-01.

Reconcile the outputs before using them

Open audit-01/accepted.csv in a text editor first. The source records should be 1, 2 and 7. The rejected source records should be 3, 4, 5 and 6, with these first-failure reasons:

The re-read check compares the combined source record IDs against the full sequence, not just a count. That catches a duplicated output record replacing a missing one. Decimal keeps the example's two-decimal arithmetic exact. The accepted total covers only accepted rows; it is not a claim about the value of rejected rows with unknown quantities or prices.

The rejection file stores decoded fields as JSON inside a CSV field, so commas and embedded newlines remain inspectable. It does not preserve the original quote spelling or bytes; keep the original export as the evidence. Treat rejected text as untrusted data and inspect it in a text editor rather than executing spreadsheet formulas.

Handle failures without disguising them

• Header or decoding failure: correct the agreed export format. No completed audit exists.
• CSV parser error: an unclosed quoted field, for example, stops the run before outputs are created. The row reconciliation guarantee applies to successfully parsed files.
• Existing destination: choose a fresh audit folder; preserve the earlier run for comparison.
• Disk or permission error while writing: treat any partial folder as incomplete. Use a new destination after correcting the cause, and require the final PASS line.
• Several problems in one row: this script reports the first reason. Correct it and re-audit to expose the next one.

Try a multiline rejected record

Replace the D404 row with a quoted SKU containing a line break, such as "D404 on one physical line and note",2,1250.00 on the next. Will the input record count become eight?

Answer: no. It remains seven parsed data records, with three accepted and four rejected. The multiline SKU fails its rule, and its rejection now spans two physical lines. reader.line_num tracks the end of that record; the saved start/end range avoids calling a record number a spreadsheet line number. Later physical line positions shift by one while their record IDs stay stable.

Reference: Python csv reader and line-number behavior.