用 Python 按记录拆分大型 CSV：重复表头、保留多行字段与成功标记

python3 --version
mkdir csv-split-demo
cd csv-split-demo
python3 - <<'PY_DATA'
import csv

with open("input.csv", "w", encoding="utf-8-sig", newline="") as handle:
    writer = csv.writer(handle)
    writer.writerow(["id", "note"])
    writer.writerows([
        ["001", "plain"],
        ["002", "comma, inside"],
        ["003", "first line\nsecond line"],
        ["004", 'say "hello"'],
        ["005", "最后一条"],
    ])
PY_DATA

"""Split a UTF-8 CSV by data records, repeating its header. Python 3.9+."""
import argparse
import csv
import json
from pathlib import Path
import sys


def positive_int(value):
    number = int(value)
    if number < 1:
        raise argparse.ArgumentTypeError("must be greater than zero")
    return number


def split_csv(source, output, rows_per_part):
    records = chunks = 0
    handle = None
    with source.open(encoding="utf-8-sig", newline="") as src:
        reader = csv.reader(src, strict=True)
        try:
            header = next(reader, None)
            if not header or any(not name.strip() for name in header):
                raise ValueError("missing header or empty column name")
            if len(set(header)) != len(header):
                raise ValueError("duplicate column names")
            output.mkdir()  # Existing paths are rejected, including empty dirs.
            for row in reader:
                if len(row) != len(header):
                    raise ValueError(
                        f"physical line {reader.line_num}: "
                        f"expected {len(header)} fields, got {len(row)}"
                    )
                if records % rows_per_part == 0:
                    if handle is not None:
                        handle.close()
                    chunks += 1
                    handle = (output / f"part-{chunks:08d}.csv").open(
                        "x", encoding="utf-8", newline=""
                    )
                    writer = csv.writer(handle, lineterminator="\n")
                    writer.writerow(header)
                writer.writerow(row)
                records += 1
        except csv.Error as exc:
            raise ValueError(
                f"CSV error near physical line {reader.line_num}: {exc}"
            ) from exc
        finally:
            if handle is not None:
                handle.close()
    # Only create the marker after input and all part files close successfully.
    with (output / "_SUCCESS").open("x", encoding="utf-8"):
        pass
    return {"records": records, "chunks": chunks, "output": str(output)}


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("input", type=Path)
    parser.add_argument("output", type=Path, help="new directory; parent must exist")
    parser.add_argument("--rows", type=positive_int, default=100000,
                        help="data records per part (default: 100000)")
    args = parser.parse_args()
    try:
        report = split_csv(args.input, args.output, args.rows)
    except (OSError, UnicodeError, ValueError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    print(json.dumps(report, ensure_ascii=False))
    return 0


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

python3 split_csv.py --help
python3 split_csv.py input.csv parts --rows 2

{"records": 5, "chunks": 3, "output": "parts"}

parts/
  part-00000001.csv
  part-00000002.csv
  part-00000003.csv
  _SUCCESS

python3 - <<'PY_CHECK'
import csv
from pathlib import Path

with open("input.csv", encoding="utf-8-sig", newline="") as handle:
    original = list(csv.reader(handle))
assert Path("parts/_SUCCESS").is_file()
files = sorted(Path("parts").glob("part-*.csv"))
assert len(files) == 3
combined = []
sizes = []
for file in files:
    with file.open(encoding="utf-8", newline="") as handle:
        rows = list(csv.reader(handle))
    assert rows[0] == original[0], file
    sizes.append(len(rows) - 1)
    combined.extend(rows[1:])
assert sizes == [2, 2, 1], sizes
assert combined == original[1:]
assert combined[2][1] == "first line\nsecond line"
print("OK: headers, order, 2/2/1 records, multiline field")
PY_CHECK

python3 - <<'PY_FAILURE'
from pathlib import Path
import subprocess
import sys

Path("bad.csv").write_text("id,note\n001,ok\n002,ok\n003\n", encoding="utf-8")
result = subprocess.run(
    [sys.executable, "split_csv.py", "bad.csv", "bad-parts", "--rows", "2"],
    capture_output=True, text=True, check=False,
)
assert result.returncode == 1, result
assert result.stdout == "" and "expected 2 fields, got 1" in result.stderr
assert Path("bad-parts/part-00000001.csv").is_file()
assert not Path("bad-parts/_SUCCESS").exists()
print("OK: late failure returns 1 and leaves no success marker")
PY_FAILURE

if python3 split_csv.py input.csv parts-next --rows 100000; then
  test -f parts-next/_SUCCESS && printf 'Ready for the next step\n'
else
  printf 'Split failed; do not consume partial files\n' >&2
fi