Clean a CSV of mixed date formats with Python

import argparse
import csv
import io
import re
import sys
from pathlib import Path

from parseapi import ParseAPI, ParseAPIError

from datetime import date as CalendarDate

DESCRIPTION = "Clean a date CSV using one declared numeric date order."
REQUIRED = ["date"]
COLUMNS = ["parse_date", "parse_date_order", "parse_status", "parse_error", "parse_http_status"]
SAMPLE = """id,date,note
001,2026-03-04,Year first
002,"March 4, 2026",Month in words
003,03/04/2026,Choose the source's date order
004,13/04/2026,Same source order applies
005,2026-02-30,Impossible calendar date
006,,Missing date
"""

# example:read
def read_rows(stream):
    reader = csv.reader(stream, strict=True)
    fields = next(reader, None)
    if (not fields or not set(REQUIRED).issubset(fields)
            or any(not field.strip() for field in fields)
            or len(fields) != len(set(fields))):
        raise ValueError("Use distinct, nonempty headers including " + ", ".join(REQUIRED) + ".")
    if set(fields) & set(COLUMNS):
        raise ValueError("The input already contains output columns. Use the original CSV.")
    rows = []
    for cells in reader:
        if len(cells) != len(fields):
            raise ValueError("Every row must have the same number of cells as the header.")
        rows.append(dict(zip(fields, cells)))
        if len(rows) > 200:
            raise ValueError("Use 1 to 200 data rows for this example.")
    if not rows:
        raise ValueError("The CSV needs at least one data row.")
    return fields, rows
# /example:read


def outcome(status, error="", http_status="", **values):
    return {**dict.fromkeys(COLUMNS, ""), "parse_status": status,
            "parse_error": error, "parse_http_status": http_status, **values}


def write_rows(stream, fields, rows):
    writer = csv.DictWriter(stream, fieldnames=fields + COLUMNS)
    writer.writeheader()
    writer.writerows(rows)

# example:interpret
def interpretation(value, date_order):
    # A single choice governs all year-last numeric rows, including 13/04/2026.
    if re.fullmatch(r"[0-9]{1,2}([/.-])[0-9]{1,2}\1[0-9]{4}", value):
        return (None, "date_order_required") if date_order == "review" else (date_order, None)
    year_first = re.fullmatch(r"[0-9]{4}([/.-])[0-9]{1,2}\1[0-9]{1,2}", value)
    month_words = re.fullmatch(
        r"(?:[A-Za-z]+\.? [0-9]{1,2}(?:st|nd|rd|th)?,? [0-9]{4}|"
        r"[0-9]{1,2}(?:st|nd|rd|th)?[ -][A-Za-z]+\.?,?[ -][0-9]{4})", value)
    if year_first or month_words:
        return None, None
    # Compact dates, two-digit years, and timestamps need an explicit cleanup rule.
    return None, "unsupported_date_format"
# /example:interpret


# example:result
def parsed(data, value, order):
    if not isinstance(data, dict) or type(data.get("valid")) is not bool:
        return outcome("error", "unexpected_response")
    if not data["valid"]:
        if data.get("date") != value:
            return outcome("error", "unexpected_response")
        return outcome("invalid_date", http_status=200, parse_date_order=order or "explicit")
    iso = data.get("date")
    try:
        if not isinstance(iso, str) or not re.fullmatch(r"[0-9]{4}-[0-9]{2}-[0-9]{2}", iso):
            raise ValueError()
        CalendarDate.fromisoformat(iso)
    except ValueError:
        return outcome("error", "unexpected_response")
    return outcome("parsed", http_status=200, parse_date=iso, parse_date_order=order or "explicit")
# /example:result


def enrich(rows, parse, date_order="review"):
    if date_order not in ("review", "mdy", "dmy"):
        raise ValueError("Choose review, mdy, or dmy for the date order.")
    results, cache, stopped = [], {}, False
    for row in rows:
        value = " ".join(row["date"].split())
        order, issue = interpretation(value, date_order)
        identity = (value, order)
        if not value:
            result = outcome("missing_date")
        elif len(value) > 64:
            result = outcome("unsupported_date_format")
        elif issue:
            result = outcome(issue)
        elif identity in cache:
            result = cache[identity]
        elif stopped:
            result = outcome("not_attempted")
        else:
            try:
                # example:request
                result = parsed(parse.date(value, format=order), value, order)
                # /example:request
            except ParseAPIError as error:
                result = outcome("error", error.code, error.status)
                stopped = error.status in (401, 403, 429)
            except Exception:
                result = outcome("error", "request_failed")
            cache[identity] = result
        results.append({**row, **result})
    return results

def main():
    parser = argparse.ArgumentParser(description=DESCRIPTION)
    parser.add_argument("input", nargs="?", help="UTF-8 CSV filename, or omit to run the sample")
    parser.add_argument("--output", help="New CSV filename, or omit for standard output")
    parser.add_argument("--date-order", choices=("review", "mdy", "dmy"), default="review")
    args = parser.parse_args()
    try:

        if args.output and (Path(args.output).exists() or Path(args.output).is_symlink()):
            raise ValueError("The output path already exists. Choose a new filename.")
        if args.input:
            path = Path(args.input)
            if path.stat().st_size > 65536:
                raise ValueError("Use a CSV no larger than 64 KiB.")
            with path.open(encoding="utf-8-sig", newline="") as stream:
                fields, rows = read_rows(stream)
        else:
            fields, rows = read_rows(io.StringIO(SAMPLE))
        # Reads PARSEAPI_KEY. Reuse one client without automatic retries.
        with ParseAPI(timeout=10.0, retries=0) as parse:
            results = enrich(rows, parse, args.date_order)
        if args.output:
            # Exclusive creation protects against a file appearing during the lookups.
            with Path(args.output).open("x", encoding="utf-8", newline="") as stream:
                write_rows(stream, fields, results)
        else:
            write_rows(sys.stdout, fields, results)
    except (ValueError, OSError, csv.Error) as error:
        print(f"Could not complete: {error}", file=sys.stderr)
        return 1
    review = sum(row["parse_status"] != "parsed" for row in results)
    print(f"{len(results)} rows written, {review} need review.", file=sys.stderr)
    return 2 if review else 0


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