CSV内の混在した日付形式を標準化し、解析できなかった値を一覧化する

record_id,event_date,note
R001,2026-09-01,ISO format
R002,09/02/2026,US slash format
R003,2026/09/03,year first with slashes
R004,04-09-2026,day first with hyphens
R005,2026.09.05,dot separated
R006,20260906,compact numeric
R007,2026-02-30,invalid calendar date
R008,Sep 7 2026,unsupported text format

import csv
from datetime import datetime
from pathlib import Path

SOURCE = Path("mixed_dates.csv")
OUTPUT_DIR = Path("outputs") / "date_cleanup_result"
CLEANED = OUTPUT_DIR / "cleaned_dates.csv"
UNPARSED = OUTPUT_DIR / "unparsed_dates.csv"
DATE_COLUMN = "event_date"

FORMATS = (
    "%Y-%m-%d",
    "%m/%d/%Y",
    "%Y/%m/%d",
    "%d-%m-%Y",
    "%Y.%m.%d",
    "%Y%m%d",
)


def parse_date(value: str) -> str | None:
    text = value.strip()
    for format_string in FORMATS:
        try:
            parsed = datetime.strptime(text, format_string)
        except ValueError:
            continue
        return parsed.strftime("%Y-%m-%d")
    return None


def main() -> None:
    if not SOURCE.is_file():
        raise FileNotFoundError(f"Source CSV not found: {SOURCE}")

    OUTPUT_DIR.parent.mkdir(parents=True, exist_ok=True)
    OUTPUT_DIR.mkdir()  # Stop rather than reuse an existing result folder.

    with SOURCE.open("r", encoding="utf-8-sig", newline="") as source:
        reader = csv.DictReader(source)
        if reader.fieldnames is None:
            raise ValueError("CSV has no header row.")
        if DATE_COLUMN not in reader.fieldnames:
            raise ValueError(f"Missing required column: {DATE_COLUMN}")
        if "standardized_date" in reader.fieldnames or "date_status" in reader.fieldnames:
            raise ValueError("Output column name already exists in source CSV.")

        cleaned_fields = [*reader.fieldnames, "standardized_date", "date_status"]
        review_fields = ["row_number", *reader.fieldnames]

        with CLEANED.open("x", encoding="utf-8", newline="") as cleaned_stream, \
             UNPARSED.open("x", encoding="utf-8", newline="") as review_stream:
            cleaned_writer = csv.DictWriter(cleaned_stream, fieldnames=cleaned_fields)
            review_writer = csv.DictWriter(review_stream, fieldnames=review_fields)
            cleaned_writer.writeheader()
            review_writer.writeheader()

            total = 0
            parsed_count = 0
            unparsed_count = 0

            for row_number, row in enumerate(reader, start=2):
                total += 1
                original_value = row[DATE_COLUMN]
                standardized = parse_date(original_value)

                output_row = dict(row)
                if standardized is None:
                    output_row["standardized_date"] = ""
                    output_row["date_status"] = "UNPARSED"
                    review_writer.writerow({"row_number": row_number, **row})
                    unparsed_count += 1
                else:
                    output_row["standardized_date"] = standardized
                    output_row["date_status"] = "PARSED"
                    parsed_count += 1

                cleaned_writer.writerow(output_row)

    if parsed_count + unparsed_count != total:
        raise RuntimeError("Record counts do not balance.")

    print(f"Processed {total} records.")
    print(f"Parsed: {parsed_count}; unparsed: {unparsed_count}.")
    print(f"Cleaned CSV: {CLEANED.as_posix()}")
    print(f"Review CSV: {UNPARSED.as_posix()}")


if __name__ == "__main__":
    main()

python date_format_cleanup.py

Processed 8 records.
Parsed: 6; unparsed: 2.
Cleaned CSV: outputs/date_cleanup_result/cleaned_dates.csv
Review CSV: outputs/date_cleanup_result/unparsed_dates.csv