ヘッダーを失わずに大きなCSVを小さなファイルへ分割する

record_id,team,units,note
R001,Support,12,"starter, pack"
R002,Ops,7,repeat
R003,Support,9,"two
lines"
R004,Sales,5,normal
R005,Ops,11,normal
R006,Sales,6,normal
R007,Support,10,normal

python split_large_csv.py

import csv
from pathlib import Path

SOURCE = Path("sample.csv")
OUTPUT_DIR = Path("outputs") / "sample_parts"
ROWS_PER_FILE = 3


def part_path(number: int) -> Path:
    return OUTPUT_DIR / f"part_{number:04d}.csv"


def split_csv() -> tuple[int, int]:
    if type(ROWS_PER_FILE) is not int or ROWS_PER_FILE < 1:
        raise ValueError("ROWS_PER_FILE must be a positive integer.")

    with SOURCE.open("r", encoding="utf-8-sig", newline="") as source:
        reader = csv.reader(source, strict=True)
        header = next(reader, None)
        if not header or any(not name.strip() for name in header):
            raise ValueError("Missing header or blank column name.")
        if len(set(header)) != len(header):
            raise ValueError("Duplicate column names.")

        OUTPUT_DIR.parent.mkdir(parents=True, exist_ok=True)
        OUTPUT_DIR.mkdir()  # Refuse to reuse an existing output path.
        total = 0
        parts = 0
        target = None
        try:
            for record_no, row in enumerate(reader, start=1):
                if len(row) != len(header):
                    raise ValueError(
                        f"Data record {record_no}: wrong number of fields."
                    )
                if total % ROWS_PER_FILE == 0:
                    if target is not None:
                        target.close()
                    parts += 1
                    target = part_path(parts).open(
                        "x", encoding="utf-8", newline=""
                    )
                    writer = csv.writer(target)
                    writer.writerow(header)
                writer.writerow(row)
                total += 1
        finally:
            if target is not None:
                target.close()

    return total, parts


def verify_csv(expected_rows: int, part_count: int) -> None:
    with SOURCE.open("r", encoding="utf-8-sig", newline="") as source:
        original = csv.reader(source, strict=True)
        header = next(original, None)
        seen = 0
        for number in range(1, part_count + 1):
            with part_path(number).open(
                "r", encoding="utf-8", newline=""
            ) as part:
                rows = csv.reader(part, strict=True)
                if next(rows, None) != header:
                    raise ValueError(f"Header mismatch in part {number}.")
                count = 0
                for row in rows:
                    if row != next(original, None):
                        raise ValueError(f"Data mismatch in part {number}.")
                    count += 1
                required = min(
                    ROWS_PER_FILE,
                    expected_rows - (number - 1) * ROWS_PER_FILE,
                )
                if count != required:
                    raise ValueError(f"Row count mismatch in part {number}.")
                seen += count
        if seen != expected_rows or next(original, None) is not None:
            raise ValueError("Overall row count mismatch.")


if __name__ == "__main__":
    total, parts = split_csv()
    verify_csv(total, parts)
    print(f"Verified {total} data records in {parts} files.")
    print(f"Output folder: {OUTPUT_DIR.as_posix()}")

Verified 7 data records in 3 files.
Output folder: outputs/sample_parts