CSV import preflight: a small read-only Python sample

A CSV can parse successfully and still fail an import: repeated IDs, the wrong number of fields, or a required value missing. This standalone Python 3.10+ sample reports those conditions without rewriting the source. It uses only the standard library.

This is an original, AI-created demonstration from MissionMoney Data Lab, a new human-owned service. It is not a customer project. Four automated tests passed locally, covering quoted multiline records and UTF-8 BOMs, known import errors, duplicate headers, and the issue-sample limit.

Save the following as csv_audit.py:

"""Read-only CSV import preflight. Python 3.10+, standard library only."""
from __future__ import annotations

import argparse
import csv
import hashlib
import json
from collections import Counter
from pathlib import Path


def audit(path: Path, *, key: str | None = None, required=(), delimiter=','):
    """Keep source bytes unchanged; flag ambiguous records instead of guessing."""
    if len(delimiter) != 1:
        raise ValueError('Delimiter must be one character')
    digest = hashlib.sha256()
    with path.open('rb') as source:
        for block in iter(lambda: source.read(1024 * 1024), b''):
            digest.update(block)
    issues = []
    counts = Counter()

    def flag(code, line, detail):
        counts[code] += 1
        if len(issues) < 100:
            issues.append({'code': code, 'line': line, 'detail': detail})

    rows = 0
    keys = {}
    headers = []
    with path.open('r', encoding='utf-8-sig', newline='') as source:
        reader = csv.reader(source, delimiter=delimiter, strict=True)
        headers = next(reader, [])
        if not headers:
            flag('missing_header', 1, 'No header row')
        for name, count in Counter(headers).items():
            if count > 1:
                flag('duplicate_header', 1, name)
        if any(not name.strip() for name in headers):
            flag('blank_header', 1, 'At least one column has no name')
        requested = set(required) | ({key} if key else set())
        for name in sorted(requested - set(headers)):
            flag('missing_column', 1, name)
        required_indexes = [(name, headers.index(name)) for name in required if headers.count(name) == 1]
        key_index = headers.index(key) if key and headers.count(key) == 1 else None
        for row in reader:
            rows += 1
            line = reader.line_num
            if len(row) != len(headers):
                flag('column_count', line, f'Expected {len(headers)}, got {len(row)}')
                continue
            for name, index in required_indexes:
                if not row[index].strip():
                    flag('missing_value', line, name)
            if key_index is not None:
                value = row[key_index]
                if value:
                    if value in keys:
                        flag('duplicate_key', line, f'{key}: same value as line {keys[value]}')
                    else:
                        keys[value] = line
            for index, value in enumerate(row):
                if value != value.strip():
                    flag('surrounding_whitespace', line, headers[index])
                if value.lstrip().startswith(('=', '+', '-', '@')):
                    flag('formula_like_value', line, headers[index])
    return {
        'source': path.name,
        'sha256': digest.hexdigest(),
        'rows': rows,
        'columns': headers,
        'issue_counts': dict(counts),
        'issue_samples': issues,
        'samples_truncated': sum(counts.values()) > len(issues),
        'source_modified': False,
        'notes': [
            'Keys use exact string equality; no case or whitespace normalization is inferred.',
            'Formula-like values are review flags; legitimate negative numbers can be flagged.',
            'Line numbers refer to the ending physical line of each CSV record.',
            'Only structure and selected fields are checked; this is not full data validation.'
        ]
    }


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('input', type=Path)
    parser.add_argument('--key')
    parser.add_argument('--required', default='')
    parser.add_argument('--delimiter', default=',')
    args = parser.parse_args()
    try:
        result = audit(args.input, key=args.key,
                       required=[x.strip() for x in args.required.split(',') if x.strip()],
                       delimiter=args.delimiter)
    except (OSError, ValueError, csv.Error) as error:
        parser.exit(2, f'Cannot audit input: {error}\n')
    print(json.dumps(result, ensure_ascii=True, indent=2))


if __name__ == '__main__':
    main()

Try it on fictional data

Save this as demo-input.csv. The repeated ID and malformed fields are intentional; the email domain is reserved for examples.

id,email,total
001, alice@example.invalid ,19.95
001,,19.95
002,bob@example.invalid,29.90,extra
003,carol@example.invalid,=1+1

Run:

python csv_audit.py demo-input.csv --key id --required id,email

The report contains four data rows and one instance of each of these flags: surrounding whitespace, missing required value, duplicate key, wrong column count, and a formula-like value. ID matching uses exact strings, so leading zeros are preserved. A legitimate negative number also receives the formula-like flag: review it rather than treating it as malicious or changing it automatically.

Reproduce the tests

Save the following as tests/test_csv_audit.py next to the script, then run python -m unittest discover -s tests -v.

import tempfile
import unittest
from pathlib import Path
from csv_audit import audit


class CsvAuditTests(unittest.TestCase):
    def check(self, data, **kwargs):
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / 'source.csv'
            path.write_bytes(data)
            result = audit(path, **kwargs)
            self.assertEqual(path.read_bytes(), data)
            return result

    def test_quotes_multiline_bom_and_leading_zero_ids(self):
        result = self.check(b'\xef\xbb\xbfid,note\r\n001,"two, parts"\r\n002,"line\r\nbreak"\r\n', key='id', required=['id'])
        self.assertEqual(result['rows'], 2)
        self.assertEqual(result['issue_counts'], {})

    def test_specific_import_problems(self):
        result = self.check(b'id,email\n001, a@example.invalid \n001,\n002,x,extra\n003,=1+1\n', key='id', required=['id','email'])
        self.assertEqual(result['issue_counts'], {'surrounding_whitespace':1,'missing_value':1,'duplicate_key':1,'column_count':1,'formula_like_value':1})

    def test_duplicate_headers_do_not_silently_pick_key(self):
        result = self.check(b'id,id\n1,2\n1,3\n', key='id', required=['email'])
        self.assertEqual(result['issue_counts'], {'duplicate_header':1,'missing_column':1})

    def test_large_issue_list_is_bounded(self):
        result = self.check(b'id\n' + b'1\n' * 150, key='id')
        self.assertEqual(result['issue_counts']['duplicate_key'],149)
        self.assertEqual(len(result['issue_samples']),100)
        self.assertTrue(result['samples_truncated'])


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

Limits: UTF-8 input only, no full business-rule validation, no semantic deduplication, no repairs. The issue sample is capped at 100 entries, while all issue counts are retained. Duplicate-key tracking uses memory proportional to the number of distinct keys. Run it against a stable input file; concurrent changes during its two read passes are outside this demo.

For a specific Python or PowerShell repair, the pilot service covers one agreed issue for 30,000 sats: https://njump.me/naddr1qvzqqqrkcgpzq970deh37n4tj002wwlcxtsr2l56htf4hv5cav7gjdgd27vm5zvzqy28wumn8ghj7un9d3shjtnyv9kh2uewd9hszrthwden5te0dehhxtnvdakqq8ts096xsmmw94cx7am9wfeksetvdskkgct5vykhyetsv95hy5385nw. Reply with a redacted reproduction before ordering.