Parse a CSV export with inconsistent dates, stray quotes and repeated header rows into clean records, and report what you dropped.

Nobody types all of this in a live round. Write it in three passes: first csv.reader with the three lists and the line-count check; then the date and amount parsers; then the cap on quote spans, which you describe first and code only if time allows. The code below is the finished version, for reference.

“Output is three lists, and the check at the end is that they account for every physical line of the file. The first job is getting from lines to records without letting one bad quote eat the file.”

import csv
import re
from collections.abc import Iterable, Iterator
from dataclasses import dataclass
from datetime import date, datetime
from decimal import Decimal, InvalidOperation
from typing import Any

REQUIRED = ("order_id", "order_date", "amount")
DATE_FORMATS = ("%Y-%m-%d", "%Y/%m/%d", "%d %b %Y", "%b %d, %Y")
SLASH = re.compile(r"(\d{1,2})/(\d{1,2})/\d{4}$")
# Lines one record may span before its quote counts as stray
MAX_SPAN = 5


class RowError(ValueError):
    def __init__(self, code: str, detail: str):
        super().__init__(detail)
        self.code = code


@dataclass
class Chunk:
    # Physical line numbers, 1-based, inclusive
    first: int
    last: int
    # The original text, for the report
    raw: str
    # None when the lines could not be parsed
    fields: list[str] | None
    error: RowError | None = None


class _SpanTooLong(Exception):
    pass


def logical_rows(lines: list[str]) -> Iterator[Chunk]:
    """Group physical lines into records by csv's own rules,
    but cap how many lines one record may span."""
    i = first = 0

    def feed():
        nonlocal i
        while i < len(lines):
            if i - first >= MAX_SPAN:
                raise _SpanTooLong
            i += 1
            yield lines[i - 1]

    while i < len(lines):
        first = i
        try:
            for fields in csv.reader(feed(), strict=True):
                raw = "".join(lines[first:i])
                yield Chunk(first + 1, i, raw, fields)
                first = i
        except (_SpanTooLong, csv.Error) as e:
            if isinstance(e, _SpanTooLong):
                err = RowError(
                    "unbalanced quote",
                    f"quote not closed within {MAX_SPAN} lines")
            else:
                err = RowError("parse error", str(e))
            # Reject only the line the record started on,
            # then parse again from the line after it.
            yield Chunk(first + 1, first + 1, lines[first],
                        None, err)
            i = first + 1

“A stray quote makes csv swallow every line up to the next quote character, and past 131072 characters it raises csv.Error. So I feed it lines myself and give up on a quoted span after MAX_SPAN lines. strict=True turns a malformed quote into a csv.Error instead of a silently merged field. Either way I reject only the line the record started on and parse again from the next one, so a bad quote costs one row, not the rest of the file. A quote in the middle of an unquoted field, like 5" monitor, stays a literal character and parses normally.”

def slash_order(values: Iterable[str]) -> str | None:
    """Settle day/month order for the whole file
    from the dates that prove it."""
    parts = [(int(m[1]), int(m[2]))
             for v in values if (m := SLASH.match(v))]
    # 15/03 fits only day-first; 03/15 only month-first
    day_first = any(a > 12 >= b for a, b in parts)
    month_first = any(b > 12 >= a for a, b in parts)
    if day_first and month_first:
        raise ValueError("the file mixes day-first and "
                         "month-first dates; ask the customer")
    if day_first:
        return "%d/%m/%Y"
    return "%m/%d/%Y" if month_first else None


def parse_date(s: str, slash_fmt: str | None) -> date:
    if SLASH.match(s):
        if slash_fmt is None:
            raise RowError(
                "ambiguous date",
                f"{s!r} could be day-first or month-first")
        formats: tuple[str, ...] = (slash_fmt,)
    else:
        formats = DATE_FORMATS
    for fmt in formats:
        try:
            return datetime.strptime(s, fmt).date()
        except ValueError:
            pass
    raise RowError("bad date", f"unrecognized date {s!r}")


def parse_amount(s: str) -> Decimal:
    # Accounting style: (12.50) is negative
    negative = s.startswith("(") and s.endswith(")")
    num = (s[1:-1] if negative else s)
    num = num.replace("$", "").replace(",", "")
    try:
        d = Decimal(num)
    except InvalidOperation:
        raise RowError("bad amount",
                       f"not a number: {s!r}") from None
    # Decimal accepts "nan" and "Infinity"
    if not d.is_finite():
        raise RowError("bad amount",
                       f"not a finite number: {s!r}")
    return -d if negative else d


def clean(path: str):
    # utf-8-sig drops a byte order mark if there is one
    with open(path, newline="", encoding="utf-8-sig") as f:
        lines = f.readlines()
    chunks = list(logical_rows(lines))
    if not chunks or (head := chunks[0]).fields is None:
        raise ValueError("no readable header")
    body = chunks[1:]
    header = [h.strip().lower() for h in head.fields]
    if missing := set(REQUIRED) - set(header):
        # Stop: this is the wrong file
        raise ValueError(f"header lacks {sorted(missing)}")
    at = header.index("order_date")
    slash_fmt = slash_order(
        c.fields[at].strip() for c in body
        if c.fields and len(c.fields) == len(header))

    records, rejected = [], []
    skipped = [(head.first, head.last, "header")]
    first_seen: dict[str, int] = {}
    for c in body:
        try:
            if c.error or c.fields is None:
                raise c.error or RowError("parse error",
                                          "no fields")
            cells = [x.strip() for x in c.fields]
            if not any(cells):
                skipped.append((c.first, c.last, "blank line"))
                continue
            if [x.lower() for x in cells] == header:
                skipped.append(
                    (c.first, c.last, "repeated header"))
                continue
            if len(cells) != len(header):
                raise RowError(
                    "wrong field count",
                    f"{len(cells)} fields, "
                    f"expected {len(header)}")
            rec: dict[str, Any] = dict(zip(header, cells))
            if empty := [k for k in REQUIRED if not rec[k]]:
                raise RowError("missing field",
                               ", ".join(empty))
            if (oid := rec["order_id"]) in first_seen:
                raise RowError(
                    "duplicate id",
                    f"first seen on line {first_seen[oid]}")
            day = parse_date(rec["order_date"], slash_fmt)
            rec["order_date"] = day.isoformat()
            rec["amount"] = parse_amount(rec["amount"])
            rec["source_lines"] = (c.first, c.last)
            first_seen[oid] = c.first
            records.append(rec)
        except RowError as e:
            rejected.append(
                (c.first, c.last, e.code, str(e), c.raw))

    spans = ([r["source_lines"] for r in records]
             + [r[:2] for r in rejected]
             + [s[:2] for s in skipped])
    covered = sum(b - a + 1 for a, b in spans)
    if covered != len(lines):
        raise RuntimeError(
            f"accounted for {covered} of {len(lines)} lines")
    return records, rejected, skipped

“On dates: slash_order reads the whole column once before any row is parsed. A first part above 12 proves day-first and a second part above 12 proves month-first, but only when the other part could be a month: 31/31/2024 proves nothing, so one typo can’t stop the file; it is rejected later as a bad date. Both proofs at once means the file mixes sources, so I stop and ask. If nothing settles it, each slash date is rejected as ambiguous, and that line in the report is the prompt for the customer to tell us. Two passes are fine here because the file is already in memory.”

“A repeated header in the middle is a sign that someone concatenated exports, and paginated exports overlap, so a second order_id is rejected with the line it first appeared on. If the customer says the later export wins, I keep the later row instead. Decimal accepts nan and Infinity, so amounts must be finite, and (12.50) reads as negative. Stripping commas assumes a US export; a European 1.234,56 would come out wrong, so I ask which decimal separator the source uses rather than guess. A file with Windows-1252 accented letters or curly quotes fails with a decode error when clean reads it, which is what I want: I ask for UTF-8 rather than guess the encoding.”

“The only except catches RowError, which my checks raise with a reason code, so a bug in my code still crashes loudly. The final check counts physical lines, not loop iterations, so a grouping bug that loses lines fails the run instead of shipping a short file. It raises RuntimeError rather than using assert, which python -O strips. The report is what the customer’s data owner reads, and it prints each rejected row as it appeared in their file, so they can find it in the source:”

from collections import Counter


def report(records, rejected, skipped) -> str:
    out = [f"{len(records)} clean, {len(rejected)} rejected, "
           f"{len(skipped)} skipped"]
    codes = Counter(r[2] for r in rejected)
    for code, n in codes.most_common():
        out.append(f"  {n:>5}  {code}")
    for first, last, code, detail, raw in rejected:
        where = (f"line {first}" if first == last
                 else f"lines {first}-{last}")
        out.append(f"{where}: {code} ({detail})")
        out.append(f"    {raw.rstrip()}")
    return "\n".join(out)

“On a small sample export, with a blank line, a repeated header, a duplicate, a nan amount, a slash date and a stray quote, it prints:”

8 clean, 4 rejected, 3 skipped
      1  ambiguous date
      1  duplicate id
      1  bad amount
      1  unbalanced quote
line 3: ambiguous date ('03/04/2026' could be day-first or month-first)
    1002,03/04/2026,75.50,"Lee, Ann"
line 7: duplicate id (first seen on line 4)
    1003,2026-03-05,(12.50),Acme
line 8: bad amount (not a finite number: 'nan')
    1004,2026-03-06,nan,Birch
line 9: unbalanced quote (quote not closed within 5 lines)
    1005,2026-03-06,40.00,"Oak ""West

“The report goes to the person who owns the export, with a short note that asks only what they can answer. To Maria, who runs the order system: ‘Of the 4,212 rows you sent, 4,190 loaded. The other 22 are in the attached report, each with its line number and reason. Two questions only you can answer. 12 dates like 03/04/2026 could be day-first or month-first: which does your system write? And 7 orders appear on both pages of the export: if they differ, should the later one win?’“

“For tests I’d take the worst real rows from the export as fixtures: a BOM, a repeated header in the middle, a duplicate ID, "Acme, Inc." with a quoted comma, an unbalanced quote followed by a long file, a slash date, and an amount of nan. Each asserts the list the row lands in and its reason code. If the file outgrows memory, clean becomes a generator that streams records and writes rejections to a file as it goes. The regrouping only ever looks back MAX_SPAN lines, so it runs over a small buffer, and I’d settle the date order from a sample of the first rows or ask.”