Python CSV 编码排障：识别 BOM，严格转换 GB18030 和 UTF-16

python3 - <<'PYTHON'
from pathlib import Path
text = 'sample_id,note\r\n001,中文\r\n002,"comma,inside"\r\n003,"line1\r\nline2"\r\n'
for name, encoding in (
    ("source-gb.csv", "gb18030"),
    ("source-bom.csv", "utf-8-sig"),
    ("source-utf16.csv", "utf-16"),
):
    with Path(name).open("xb") as handle:
        handle.write(text.encode(encoding))
PYTHON

python3 - <<'PYTHON'
from pathlib import Path
for name in ("source-gb.csv", "source-bom.csv", "source-utf16.csv"):
    with Path(name).open("rb") as handle:
        prefix = handle.read(4)
    print(name, prefix.hex(" "))
PYTHON

"""Convert a small text/CSV file to UTF-8 without overwriting any file."""
import argparse
import codecs
from pathlib import Path
import sys

MAX_BYTES = 20 * 1024 * 1024


def bom_name(data):
    # UTF-32 must precede UTF-16: their prefixes can overlap.
    for mark, name in (
        (codecs.BOM_UTF32_LE, "utf-32-le"),
        (codecs.BOM_UTF32_BE, "utf-32-be"),
        (codecs.BOM_UTF8, "utf-8-sig"),
        (codecs.BOM_UTF16_LE, "utf-16-le"),
        (codecs.BOM_UTF16_BE, "utf-16-be"),
    ):
        if data.startswith(mark):
            return name
    return None


def convert(source, destination, encoding):
    with source.open("rb") as handle:
        data = handle.read(MAX_BYTES + 1)
    if len(data) > MAX_BYTES:
        raise ValueError("input exceeds the 20 MiB limit")
    bom = bom_name(data)
    allowed = {
        "utf-8-sig": {None, "utf-8-sig"},
        "gb18030": {None},
        "utf-16": {"utf-16-le", "utf-16-be"},
    }
    if bom not in allowed[encoding]:
        raise ValueError(f"BOM {bom!r} conflicts with {encoding}")
    text = data.decode(encoding, errors="strict")
    result = text.encode("utf-8", errors="strict")
    # Input validation finishes before exclusive output creation.
    with destination.open("xb") as handle:
        handle.write(result)
    return len(data), len(result)


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("input", type=Path)
    parser.add_argument("output", type=Path)
    parser.add_argument("--from-encoding", required=True,
                        choices=("utf-8-sig", "gb18030", "utf-16"))
    args = parser.parse_args()
    try:
        before, after = convert(args.input, args.output, args.from_encoding)
    except (OSError, UnicodeError, ValueError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    print(f"OK: input_bytes={before} output_bytes={after}")
    return 0


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

python3 --version
python3 convert_csv_encoding.py --help

python3 convert_csv_encoding.py source-gb.csv clean-gb.csv --from-encoding gb18030
python3 convert_csv_encoding.py source-bom.csv clean-bom.csv --from-encoding utf-8-sig
python3 convert_csv_encoding.py source-utf16.csv clean-utf16.csv --from-encoding utf-16

python3 - <<'PYTHON'
import codecs
import csv
from pathlib import Path

expected = [
    ["sample_id", "note"],
    ["001", "中文"],
    ["002", "comma,inside"],
    ["003", "line1\r\nline2"],
]
outputs = [Path(name).read_bytes() for name in
           ("clean-gb.csv", "clean-bom.csv", "clean-utf16.csv")]
assert outputs[0] == outputs[1] == outputs[2]
assert not outputs[0].startswith(codecs.BOM_UTF8)
for name in ("clean-gb.csv", "clean-bom.csv", "clean-utf16.csv"):
    with open(name, encoding="utf-8", newline="") as handle:
        assert list(csv.reader(handle, strict=True)) == expected
print("PASS: identical UTF-8 bytes; IDs, quotes and embedded CRLF preserved")
PYTHON

PASS: identical UTF-8 bytes; IDs, quotes and embedded CRLF preserved

python3 convert_csv_encoding.py source-gb.csv wrong.csv --from-encoding utf-8-sig
printf 'exit=%s\n' "$?"
python3 -c 'from pathlib import Path; assert not Path("wrong.csv").exists()'

python3 convert_csv_encoding.py source-bom.csv clean-bom.csv --from-encoding utf-8-sig
printf 'exit=%s\n' "$?"

python3 convert_csv_encoding.py source-gb.csv ready.csv --from-encoding gb18030 &&
  python3 -c 'import csv; f=open("ready.csv", encoding="utf-8", newline=""); print(next(csv.reader(f))); f.close()'