用 Python 流式处理大型 CSV：逐条检查并统计标签

cat > labels.csv <<'CSV'
sample_id,label
001,1
002,0
"sample,003",unknown
004,1
005,0
006,1
CSV
python3 summarize_labels.py --help
python3 summarize_labels.py labels.csv

{
  "records": 6,
  "label_counts": {
    "0": 2,
    "1": 3,
    "unknown": 1
  }
}

"""Stream a UTF-8 CSV with exactly sample_id,label columns; count 0/1/unknown."""
import argparse
import csv
import json


def summarize(path):
    counts = {"0": 0, "1": 0, "unknown": 0}
    records = 0
    with open(path, encoding="utf-8-sig", newline="") as handle:
        reader = csv.reader(handle, strict=True)
        try:
            header = next(reader, None)
            if header != ["sample_id", "label"]:
                raise ValueError("Expected header: sample_id,label (in that order)")
            for row in reader:
                where = f"Record ending at physical line {reader.line_num}"
                if len(row) != 2:
                    raise ValueError(f"{where}: expected 2 fields, got {len(row)}")
                sample_id, label = row
                if not sample_id or sample_id != sample_id.strip():
                    raise ValueError(f"{where}: sample_id is empty or has outer whitespace")
                if label not in counts:
                    raise ValueError(f"{where}: label must be 0, 1, or unknown")
                counts[label] += 1
                records += 1
        except csv.Error as exc:
            raise ValueError(f"CSV parse error near physical line {reader.line_num}: {exc}") from exc
    return {"records": records, "label_counts": counts}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("input", help="Path to the input CSV")
    args = parser.parse_args()
    try:
        result = summarize(args.input)
    except (OSError, UnicodeError, ValueError) as exc:
        parser.exit(1, f"error: {exc}\n")
    print(json.dumps(result, indent=2))


if __name__ == "__main__":
    main()

cat > bad-labels.csv <<'CSV'
sample_id,label
001,1
002,maybe
CSV
python3 summarize_labels.py bad-labels.csv
echo "exit=$?"

error: Record ending at physical line 3: label must be 0, 1, or unknown
exit=1