用 Python 流式把 CSV 转成 JSONL：保留前导零、多行字段与失败保护

python3 --version
mkdir csv-jsonl-demo
cd csv-jsonl-demo
cat > samples.csv <<'CSV'
sample_id,note
001,中文结果
002,"alpha, beta"
003,"first line
second line"
004,
CSV

"""Stream UTF-8 CSV to JSONL without overwriting output. Python 3.10+, Linux."""
import argparse
import csv
import json
import os
from pathlib import Path
import sys
import tempfile


def convert(source, output):
    with source.open(encoding="utf-8-sig", newline="") as incoming:
        reader = csv.reader(incoming, strict=True)
        header = next(reader, None)
        if not header or any(not name.strip() for name in header):
            raise ValueError("missing or empty column name")
        if len(set(header)) != len(header):
            raise ValueError("duplicate column name")
        count = 0
        with tempfile.TemporaryDirectory(prefix=".csv-jsonl-", dir=output.parent) as work:
            staged = Path(work) / "result.jsonl"
            with staged.open("x", encoding="utf-8", newline="\n") as outgoing:
                for number, row in enumerate(reader, 1):
                    if len(row) != len(header):
                        raise ValueError(
                            f"record {number}, ending at CSV line {reader.line_num}: "
                            f"expected {len(header)} fields, got {len(row)}"
                        )
                    record = dict(zip(header, row))
                    outgoing.write(json.dumps(record, ensure_ascii=False) + "\n")
                    count += 1
            os.link(staged, output)
    return count


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("source", type=Path)
    parser.add_argument("output", type=Path)
    args = parser.parse_args()
    try:
        count = convert(args.source, args.output)
    except (OSError, UnicodeError, csv.Error, ValueError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    print(f"OK: {count} records -> {args.output}")
    return 0


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

python3 csv_to_jsonl.py --help
python3 csv_to_jsonl.py samples.csv samples.jsonl
cat samples.jsonl

{"sample_id": "001", "note": "中文结果"}
{"sample_id": "002", "note": "alpha, beta"}
{"sample_id": "003", "note": "first line\nsecond line"}
{"sample_id": "004", "note": ""}

python3 - <<'PY_CHECK'
import json
from pathlib import Path
payload = Path("samples.jsonl").read_bytes()
assert not payload.startswith(b"\xef\xbb\xbf")
assert payload.endswith(b"\n")
lines = payload.decode("utf-8").splitlines()
assert len(lines) == 4
rows = [json.loads(line) for line in lines]
assert rows[0] == {"sample_id": "001", "note": "中文结果"}
assert rows[1]["note"] == "alpha, beta"
assert rows[2]["note"] == "first line\nsecond line"
assert rows[3]["note"] == ""
print("OK: four objects, strings, UTF-8, and multiline field")
PY_CHECK

python3 - <<'PY_TEST'
from pathlib import Path
import subprocess
import sys
original = Path("samples.jsonl").read_bytes()
command = [sys.executable, "csv_to_jsonl.py"]
again = subprocess.run(command + ["samples.csv", "samples.jsonl"], capture_output=True)
assert again.returncode == 1
assert Path("samples.jsonl").read_bytes() == original
Path("bad.csv").write_text("a,b\n1,2\n3,4,5\n", encoding="utf-8")
bad = subprocess.run(command + ["bad.csv", "bad.jsonl"], capture_output=True, text=True)
assert bad.returncode == 1
assert not Path("bad.jsonl").exists()
assert not list(Path(".").glob(".csv-jsonl-*"))
print(bad.stderr.strip())
print("OK: existing output unchanged; late failure leaves no output")
PY_TEST