Clean up a CSV of mixed dates

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())