実践：CSV 整理ツールを作る

name,email
 Alice , ALICE@EXAMPLE.TEST 
Alice duplicate,alice@example.test
Bob,bob@example.test
Missing,

Create clean_contacts.py using Python's standard library only.
Expose clean(source: Path, target: Path) returning kept/duplicate/invalid counts.
Read UTF-8 with optional BOM using csv.DictReader, not string splitting.
Require exactly name,email. Trim both fields and lowercase email.
Keep the first valid row per email; reject empty names and invalid email shapes.
Treat missing/extra columns or malformed CSV as whole-run failures.
Read and validate before creating output. Never change input or overwrite output.
Write UTF-8 with name,email and a newline after each row.
CLI takes source and target paths, reports counts on stderr and exits 0 on success,
1 on data/file errors. Add tests for preservation, duplicates, Unicode, quoted
commas, malformed rows and overwrite refusal. Do not use real contact data.

"""Clean a practice CSV without overwriting either the input or an existing output."""
import argparse
import csv
import json
import re
import sys
from pathlib import Path


def clean(source: Path, target: Path) -> dict[str, int]:
    if source.resolve() == target.resolve():
        raise ValueError("Input and output must differ")
    rows = []
    seen = set()
    counts = {"kept": 0, "duplicate": 0, "invalid": 0}
    with source.open(encoding="utf-8-sig", newline="") as stream:
        reader = csv.DictReader(stream, strict=True)
        if reader.fieldnames != ["name", "email"]:
            raise ValueError("Expected exactly: name,email")
        for row in reader:
            if None in row or any(value is None for value in row.values()):
                raise ValueError("Malformed CSV row")
            name = row["name"].strip()
            email = row["email"].strip().lower()
            # An exercise-level shape check, not proof an address can receive mail.
            if not name or not re.fullmatch(r"[^\s@]+@[^\s@]+\.[^\s@]+", email):
                counts["invalid"] += 1
            elif email in seen:
                counts["duplicate"] += 1
            else:
                seen.add(email)
                rows.append({"name": name, "email": email})
                counts["kept"] += 1
    # Exclusive creation: a rerun never silently replaces an existing result.
    with target.open("x", encoding="utf-8", newline="") as stream:
        writer = csv.DictWriter(stream, fieldnames=["name", "email"], lineterminator="\n")
        writer.writeheader()
        writer.writerows(rows)
    return counts


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("source", type=Path)
    parser.add_argument("target", type=Path)
    args = parser.parse_args()
    try:
        counts = clean(args.source, args.target)
    except (OSError, ValueError, csv.Error) as error:
        print(str(error), file=sys.stderr)
        return 1
    print(json.dumps(counts), file=sys.stderr)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

$csvBefore = (Get-FileHash -LiteralPath contacts.csv -Algorithm SHA256).Hash
py -3 clean_contacts.py contacts.csv cleaned-01.csv
$csvExit = $LASTEXITCODE
$csvExit
Get-Content -LiteralPath cleaned-01.csv
$csvBefore -eq (Get-FileHash -LiteralPath contacts.csv -Algorithm SHA256).Hash

cp contacts.csv contacts-before.csv
python3 clean_contacts.py contacts.csv cleaned-01.csv
csv_exit=$?
echo "$csv_exit"
cat cleaned-01.csv
cmp contacts-before.csv contacts.csv

name,email
Alice,alice@example.test
Bob,bob@example.test

name,email
"林, Alex",alex@example.test

name,email
 ,first@example.test
First valid,FIRST@example.test
Later duplicate,first@example.test

name,email
First valid,first@example.test

import csv
import tempfile
import unittest
from pathlib import Path

from clean_contacts import clean


class CleanupTests(unittest.TestCase):
    def test_contract_and_input_preservation(self):
        original = (Path(__file__).parent / "contacts.csv").read_bytes()
        with tempfile.TemporaryDirectory() as directory:
            source = Path(directory) / "input.csv"
            target = Path(directory) / "output.csv"
            source.write_bytes(original)
            self.assertEqual(clean(source, target), {"kept": 2, "duplicate": 1, "invalid": 1})
            self.assertEqual(source.read_bytes(), original)
            with target.open(encoding="utf-8", newline="") as stream:
                self.assertEqual(list(csv.DictReader(stream)), [{"name": "Alice", "email": "alice@example.test"}, {"name": "Bob", "email": "bob@example.test"}])
            with self.assertRaises(FileExistsError):
                clean(source, target)
            with self.assertRaises(ValueError):
                clean(source, source)

    def test_unicode_and_quoted_comma(self):
        with tempfile.TemporaryDirectory() as directory:
            source, target = Path(directory) / "in.csv", Path(directory) / "out.csv"
            source.write_text('name,email\n"林, Alex",alex@example.test\n', encoding="utf-8")
            self.assertEqual(clean(source, target)["kept"], 1)
            self.assertIn('"林, Alex"', target.read_text(encoding="utf-8"))

    def test_bad_header_and_malformed_rows_create_no_output(self):
        with tempfile.TemporaryDirectory() as directory:
            source, target = Path(directory) / "in.csv", Path(directory) / "out.csv"
            for value in ["email,name\na@b.test,A\n", "name,email\nAlice\n", "name,email\nA,a@b.test,extra\n"]:
                source.write_text(value, encoding="utf-8")
                with self.assertRaises(ValueError):
                    clean(source, target)
                self.assertFalse(target.exists())


if __name__ == "__main__":
    unittest.main()