AIでCSVデータのクリーニング規則を作り、変更セルをすべて確認する

row_id,name,email,department,hours
1, alice kim ,ALICE.KIM@example.com, sales ,8
2,Bob Lee,bob.lee@example.com,Engineering,7.5
3,Carol Park, carol.park@example.com ,SALES, 8.0
4,Dan Choi,,engineering,NA
5,Eve Han,eve.han@example.com, Sales,9

import csv
import sys
from decimal import Decimal, InvalidOperation
from pathlib import Path

INPUT = Path("sample_people.csv")
OUTPUT_DIR = Path("outputs")
CLEANED = OUTPUT_DIR / "sample_people_cleaned.csv"
CHANGES = OUTPUT_DIR / "changes.csv"

EXPECTED_FIELDS = ["row_id", "name", "email", "department", "hours"]
DEPARTMENT_MAP = {
    "sales": "Sales",
    "engineering": "Engineering",
}


def clean_hours(value):
    trimmed = value.strip()
    if trimmed == "NA":
        return "", "NA converted to blank"
    if trimmed == "":
        return "", "trim whitespace"

    try:
        number = Decimal(trimmed)
    except InvalidOperation:
        return trimmed, "trim whitespace"

    if number == number.to_integral():
        normalized = str(number.quantize(Decimal("1")))
    else:
        normalized = format(number.normalize(), "f")
    return normalized, "normalize numeric notation"


def clean_cell(column, value):
    if column == "row_id":
        return value, "preserve row_id"

    if column == "name":
        return value.strip(), "trim name whitespace"

    if column == "email":
        trimmed = value.strip()
        return trimmed.lower() if trimmed else "", "trim and lowercase email"

    if column == "department":
        trimmed = value.strip()
        normalized = DEPARTMENT_MAP.get(trimmed.lower(), trimmed)
        return normalized, "trim and normalize known department"

    if column == "hours":
        return clean_hours(value)

    return value, "no rule"


def main():
    if not INPUT.is_file():
        sys.exit(f"Input file not found: {INPUT}")

    if OUTPUT_DIR.exists():
        sys.exit(f"Stop: {OUTPUT_DIR} already exists. Review or rename it first.")

    OUTPUT_DIR.mkdir()

    cleaned_rows = []
    changes = []

    with INPUT.open("r", encoding="utf-8", newline="") as source:
        reader = csv.DictReader(source)
        if reader.fieldnames != EXPECTED_FIELDS:
            sys.exit(
                f"Unexpected columns. Expected {EXPECTED_FIELDS}, got {reader.fieldnames}"
            )

        for row in reader:
            cleaned = {}
            for column in EXPECTED_FIELDS:
                before = row[column]
                after, rule = clean_cell(column, before)
                cleaned[column] = after

                if before != after:
                    changes.append(
                        {
                            "row_id": row["row_id"],
                            "column": column,
                            "before": before,
                            "after": after,
                            "rule": rule,
                        }
                    )

            cleaned_rows.append(cleaned)

    with CLEANED.open("x", encoding="utf-8", newline="") as target:
        writer = csv.DictWriter(target, fieldnames=EXPECTED_FIELDS)
        writer.writeheader()
        writer.writerows(cleaned_rows)

    with CHANGES.open("x", encoding="utf-8", newline="") as target:
        writer = csv.DictWriter(
            target,
            fieldnames=["row_id", "column", "before", "after", "rule"],
        )
        writer.writeheader()
        writer.writerows(changes)

    print(f"Rows written: {len(cleaned_rows)}")
    print(f"Changed cells: {len(changes)}")
    print(f"Cleaned copy: {CLEANED}")
    print(f"Change log: {CHANGES}")


if __name__ == "__main__":
    main()