CSV & Excel Data Ingest skill for Codex

A good ingest never guesses silently: the file’s encoding, delimiter and schema are declared, every row either loads with the declared types or lands in a quarantine table with a reason, the same file loaded twice produces the same result, and each load records its source, row counts and checksum.

Inspect the file before writing code

python3 scripts/sniff_csv.py data/incoming/partner_2025-10.csv
file -I data/incoming/partner_2025-10.csv        # macOS; GNU: file -i
head -c 600 data/incoming/partner_2025-10.csv | xxd | head   # BOM, CRLF, odd bytes
wc -l data/incoming/partner_2025-10.csv

scripts/sniff_csv.py (standard library only) reports encoding or BOM, line endings, delimiter, header detection, ragged rows with line numbers, and columns that must stay text (leading zeros, scientific-notation IDs, European decimals, null-like tokens). It exits 1 when it finds problems. Run it on every new source and on any file that fails.

Then read AGENTS.md for where raw files land, which library the pipeline uses, and the target (warehouse table, Parquet, Postgres). Ask the data provider, through the user, for a spec if none exists: column names, types, encoding, delimiter, date format, null token.

Decisions

Question	Default	Escape hatch
Encoding	UTF-8 (utf-8-sig if a BOM is present)	cp1252 for Windows exports with é failing as UTF-8; confirm with the source
Initial read	All columns as strings	Typed read only when the file has a trusted, stable spec
Library	Polars or DuckDB for large files; pandas when the codebase uses it	openpyxl for cell-level Excel access
Nulls	Explicit list per source ("", "NULL")	Never rely on library defaults that treat "NA" as null
Bad rows	Quarantine with reason, continue	Fail the whole load when quarantine exceeds a threshold (e.g. 1%)
Re-loads	Replace the partition or batch by file id	Never blind-append

Reading as text, then casting

import pandas as pd

raw = pd.read_csv(
    path,
    dtype=str,
    encoding="utf-8-sig",
    sep=";",
    keep_default_na=False,          # "NA", "null" stay strings until we decide
    na_values=[""],
    on_bad_lines="warn",            # or a callable (engine="python") to capture bad rows
    skipinitialspace=True,
)
raw.columns = raw.columns.str.strip().str.lower().str.replace(r"\W+", "_", regex=True)

import polars as pl

raw = pl.read_csv(path, infer_schema=False, separator=";", encoding="utf8-lossy",
                  null_values=[""], truncate_ragged_lines=False)

infer_schema=False (older Polars: infer_schema_length=0) reads all columns as strings. utf8-lossy replaces bad bytes, so check for the replacement character afterwards; for non-UTF-8 files, transcode first (iconv -f CP1252 -t UTF-8).

-- DuckDB: inspect what it would infer, then load as text and cast deliberately
select * from sniff_csv('data/incoming/partner.csv');
create table stg as
select * from read_csv('data/incoming/partner.csv', all_varchar = true, header = true,
                       delim = ';', store_rejects = true);
select * from reject_errors;     -- rows DuckDB could not parse, with line and reason

Explicit schema and validation

import pandera.pandas as pa   # pandera >= 0.24; older: import pandera as pa

schema = pa.DataFrameSchema(
    {
        "customer_id": pa.Column(str, pa.Check.str_matches(r"^C\d{6}$"), unique=True),
        "zip": pa.Column(str, pa.Check.str_length(5, 5)),
        "amount": pa.Column(float, pa.Check.ge(0)),
        "order_date": pa.Column("datetime64[ns]"),
        "country": pa.Column(str, pa.Check.isin(["DE", "FR", "NL", "US"])),
    },
    strict=True,     # unexpected columns fail: providers add columns without warning
    coerce=False,
)

typed = raw.assign(
    amount=pd.to_numeric(raw["amount"].str.replace(".", "", regex=False).str.replace(",", ".", regex=False),
                         errors="coerce"),
    order_date=pd.to_datetime(raw["order_date"], format="%d.%m.%Y", errors="coerce"),
)
try:
    schema.validate(typed, lazy=True)
except pa.errors.SchemaErrors as err:
    failures = err.failure_cases      # column, check, index, failure_case

Without pandera (not installed, no network), write the same checks as boolean masks and collect failing indexes. Install needs network and possibly approval in the sandbox.

Quarantine pattern

bad_mask = typed["amount"].isna() & raw["amount"].ne("")         # present but unparseable
bad_mask |= typed["order_date"].isna()
bad_mask |= ~typed["customer_id"].str.match(r"^C\d{6}$", na=False)

quarantine = raw[bad_mask].assign(_reason="type or format check failed", _file=path.name)
good = typed[~bad_mask]

if len(quarantine) / max(len(raw), 1) > 0.01:
    raise RuntimeError(f"{len(quarantine)} of {len(raw)} rows failed checks; load aborted")

Store quarantined rows with the raw strings, file name, line number and reason so the provider can fix them.

Excel

sheets = pd.read_excel(path, sheet_name=None, dtype=str, engine="openpyxl", header=0)
for name, df in sheets.items():
    print(name, df.shape)

• .xlsx needs openpyxl; .xls needs xlrd; .xlsb needs pyxlsb; Polars read_excel uses the calamine engine by default and is much faster on large sheets.
• Real Excel dates come back as datetimes; dates typed as text stay text. Serial numbers (45567) convert with pd.to_datetime(x, unit="D", origin="1899-12-30").
• Merged header cells, notes above the table and totals rows are common: locate the header row explicitly (header=3, skiprows, usecols="A:H") and drop totals by rule.
• Formulas: openpyxl returns cached values only if the file was saved by Excel (load_workbook(path, data_only=True)); files generated by scripts may have none.

Idempotent loads

- [ ] Land the original file unchanged (object storage or data/raw/) with its sha256
- [ ] Record a load row: file name, sha256, received_at, rows read/loaded/quarantined
- [ ] Skip a file whose sha256 was already loaded successfully
- [ ] Load into staging, validate, then swap or merge into the target in one transaction
- [ ] Re-processing a corrected file replaces that file's rows (delete by _file_id, insert)

Deleting or replacing rows in a production table is destructive; confirm the target and scope with the user before the first run against shared data.

Gotchas

• "NA", "NULL", "None" as legitimate values (country code Namibia, a surname) are nulled by pandas defaults. keep_default_na=False and an explicit list fix it.
• Leading zeros in ZIP codes, phone numbers and account IDs disappear when read as numbers. Keep them as strings.
• Embedded newlines inside quoted fields make wc -l disagree with parsed row counts; that is expected, ragged rows are not.
• A trailing delimiter on every line creates an empty extra column; drop it explicitly.
• Files exported “as CSV” from Excel on European locales use ; and , decimals.
• Date formats 03/04/2025 are ambiguous; get the format from the spec, never infer it from the first rows.
• Large files: stream (pd.read_csv(chunksize=...), Polars scan_csv, DuckDB) instead of loading 10 GB into memory.

Verification

• rows_read == rows_loaded + rows_quarantined, recorded per file.
• Spot-check five loaded rows against the raw file by line number, including one with accents and one with leading zeros.
• Load the same file twice; target row count is unchanged the second time.
• Column sums of key numeric fields match a control total from the provider when one is supplied.

Load report template

**File:** partner_2025-10.csv (sha256 3f9a...c21, 18.2 MB, received 2025-10-02)
**Detected:** cp1252, ';' delimiter, CRLF, header row; transcoded to UTF-8
**Rows:** read 241,330; loaded 240,912; quarantined 418 (amount unparseable 301, bad customer_id 117)
**Target:** staging.partner_orders -> raw.partner_orders (replaced file_id 2025-10)
**Checks:** amount sum 4,812,330.17 matches provider control total; second load idempotent
**Provider follow-up:** quarantine file data/quarantine/partner_2025-10.csv