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