Python csv 實戰：DictReader、Dialect 與串流清理資料

Python 學習 - 本文屬於一個選集。

§ 115: 本文

一. 前言：CSV 很簡單，直到它不簡單
#

CSV 看起來只是「每欄用逗號隔開」。

於是第一版程式常常長這樣：

for line in open("orders.csv").read().splitlines():
    order_id, customer, total = line.split(",")

只要欄位裡出現逗號、引號、換行、BOM，這段程式就會開始鬧脾氣。

更麻煩的是，CSV 並沒有一個所有工具都完全照做的單一格式。有人用分號，有人用 tab；有人輸出 UTF-8，有人附帶 BOM；有些欄位還會跨兩個實體行。

Python 標準庫的 csv 模組就是來處理這些邊界的。

它很適合：

讀取外部系統匯出的報表
清理欄位並留下錯誤紀錄
串流處理大檔案
輸出欄位順序固定的匯入檔
寫不依賴 pandas 的小型 ETL 工具

如果你要直接用 SQL 查 CSV，請看 Python DuckDB 實戰；要做大型欄位運算則可看 Python Polars 實戰。

今天專心把最底層的一列資料讀對、洗乾淨、再安全寫回去。

二. 準備：標準庫內建，開檔方式最重要
#

csv 不用安裝。建立練習專案即可：

uv init csv-lab
cd csv-lab

準備 orders.csv：

order_id,customer,total,status
1001,拍拍君,520,paid
1002,拍拍醬,199,pending
1003,chatPTT,0,cancelled

最小讀取程式：

import csv
from pathlib import Path

path = Path("orders.csv")

with path.open("r", encoding="utf-8", newline="") as file:
    reader = csv.reader(file)
    for row in reader:
        print(row)

請把 newline="" 當成固定搭配。

它讓 csv 模組自己處理換行；否則 quoted field 裡的換行可能被錯誤解讀，某些平台寫檔時還可能多出 \r。

三. `csv.reader`：每一列先讀成 list
#

csv.reader() 會依格式解析每筆 record，回傳字串 list：

with Path("orders.csv").open(
    "r", encoding="utf-8", newline=""
) as file:
    reader = csv.reader(file)
    header = next(reader)
    print("欄位：", header)

    for order_id, customer, total, status in reader:
        print(order_id, customer, total, status)

預設不會替你把 "520" 轉成整數 520。格式解析和資料型別驗證是兩件事，最好分開處理。

reader 適合沒有 header，或欄位位置本身就是契約的資料。

只要檔案有 header，拍拍君通常優先用 DictReader，避免欄位順序一改，資料就被放錯位置。

四. `DictReader`：先檢查 Schema 再讀資料
#

csv.DictReader() 用第一列建立 key：

with Path("orders.csv").open(
    "r", encoding="utf-8", newline=""
) as file:
    reader = csv.DictReader(file)

    for row in reader:
        print(row["order_id"], row["customer"], row["total"])

不要等讀到第 5000 列才發現 header 拼錯。先驗證欄位：

REQUIRED_FIELDS = {"order_id", "customer", "total", "status"}


def validate_header(reader: csv.DictReader) -> None:
    actual = set(reader.fieldnames or [])
    missing = REQUIRED_FIELDS - actual
    if missing:
        names = ", ".join(sorted(missing))
        raise ValueError(f"缺少必要欄位：{names}")

每列欄位數量也可能不一致。

DictReader 遇到多餘欄位時，會把它們放進 restkey；缺少欄位則使用 restval：

reader = csv.DictReader(
    file,
    restkey="_extra",
    restval="",
)

因此可以明確擋掉壞資料：

if row.get("_extra"):
    raise ValueError(f"第 {reader.line_num} 行有多餘欄位")

注意 reader.line_num 是讀過的實體行數，不一定等於 record 數；一個 quoted field 可以跨多行。

五. 把一列清理成明確的資料模型
#

真實資料常長這樣：

order_id,customer,total,status
1001, 拍拍君 ,520,PAID
1002,拍拍醬,,pending
bad-id,chatPTT,300,paid
1004,,120,unknown

不要把所有 if 都塞在讀檔迴圈。先定義資料模型和清理結果：

from dataclasses import dataclass


@dataclass(frozen=True)
class Order:
    order_id: int
    customer: str
    total: int
    status: str


@dataclass(frozen=True)
class CleanResult:
    order: Order | None
    error: str | None

再集中清理規則：

VALID_STATUSES = {"paid", "pending", "cancelled"}


def clean_order(row: dict[str, str], line: int) -> CleanResult:
    try:
        order_id = int(row["order_id"])
    except (KeyError, ValueError):
        return CleanResult(None, f"line {line}: order_id 不是整數")

    customer = row.get("customer", "").strip()
    if not customer:
        return CleanResult(None, f"line {line}: customer 不可為空")

    try:
        raw_total = row.get("total", "").strip()
        total = int(raw_total) if raw_total else 0
    except ValueError:
        return CleanResult(None, f"line {line}: total 不是整數")

    status = row.get("status", "").strip().lower()
    if status not in VALID_STATUSES:
        return CleanResult(None, f"line {line}: status 不合法: {status}")

    return CleanResult(
        Order(order_id, customer, total, status),
        None,
    )

這樣格式解析、商業規則和流程控制各自有清楚邊界。

六. `DictWriter`：固定欄位順序與引用規則
#

清理完成後，用 csv.DictWriter() 輸出：

FIELDNAMES = ["order_id", "customer", "total", "status"]


def write_order(writer: csv.DictWriter, order: Order) -> None:
    writer.writerow(
        {
            "order_id": order.order_id,
            "customer": order.customer,
            "total": order.total,
            "status": order.status,
        }
    )

建立 writer 時明確指定順序：

with Path("clean.csv").open(
    "w", encoding="utf-8", newline=""
) as file:
    writer = csv.DictWriter(file, fieldnames=FIELDNAMES)
    writer.writeheader()

dict 有未知 key 時，預設 extrasaction="raise" 會直接報錯。這通常比靜默忽略安全。

CSV writer 會自動替含逗號、引號或換行的欄位加上引號：

writer.writerow(
    {
        "order_id": 1005,
        "customer": "拍拍君, Taipei",
        "total": 880,
        "status": "paid",
    }
)

不要先手動加引號；交給 writer 依 quoting 規則處理。

還要記得：writer 會把 None 寫成空字串，這個轉換不可逆。如果空字串和資料庫 NULL 意義不同，請先定義自己的輸出規則。

七. Dialect：CSV 不一定用逗號
#

某些系統用分號：

order_id;customer;total
1001;拍拍君;520

單次讀取可直接指定：

reader = csv.DictReader(file, delimiter=";")

固定合作方的格式則適合命名成 dialect：

csv.register_dialect(
    "partner_export",
    delimiter=";",
    quotechar='"',
    doublequote=True,
    lineterminator="\n",
    strict=True,
)

讀寫兩端共用同一份設定：

reader = csv.DictReader(file, dialect="partner_export")
writer = csv.DictWriter(
    output_file,
    fieldnames=FIELDNAMES,
    dialect="partner_export",
)

strict=True 會在 parser 發現壞格式時拋出 csv.Error，適合不希望「盡量猜」的匯入流程。

來源格式未知時，可以有限度使用 Sniffer：

sample = file.read(4096)
file.seek(0)

dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
reader = csv.DictReader(file, dialect=dialect)

但 Sniffer 是 heuristic，不是格式驗證器。正式管線最好把來源與 dialect 寫進設定，偵測失敗時也要有清楚錯誤。

八. Encoding、BOM 與欄位內換行
#

Excel 匯出的 UTF-8 檔案可能帶 BOM，第一個 key 會變成 "\ufefforder_id"。

固定可能有 BOM 的來源可以用：

with Path("orders.csv").open(
    "r", encoding="utf-8-sig", newline=""
) as file:
    reader = csv.DictReader(file)

utf-8-sig 讀取時會吃掉 BOM；沒有 BOM 的 UTF-8 也能正常讀。

至於 Big5、CP950、Shift-JIS，不要靠猜。來源系統應該提供 encoding 契約，匯入工具也應讓設定可以調整。

欄位裡合法包含逗號和換行：

order_id,note
1001,"第一行備註：鍵盤, 滑鼠
第二行備註：請一起包裝"

這就是不能先 splitlines() 的原因。對 parser 而言，上面是 header 加一筆資料，不是三筆資料。

九. 串流清理：記憶體保持固定
#

DictReader 本身就是 iterator，不必先 list(reader)。

可以邊讀、邊驗證、邊寫：

def clean_csv(
    input_path: Path,
    output_path: Path,
    error_path: Path,
) -> tuple[int, int]:
    clean_count = 0
    errors: list[str] = []

    with input_path.open(
        "r", encoding="utf-8-sig", newline=""
    ) as source, output_path.open(
        "w", encoding="utf-8", newline=""
    ) as target:
        reader = csv.DictReader(source, restkey="_extra", restval="")
        validate_header(reader)

        writer = csv.DictWriter(target, fieldnames=FIELDNAMES)
        writer.writeheader()

        try:
            for row in reader:
                if row.get("_extra"):
                    errors.append(f"line {reader.line_num}: 多餘欄位")
                    continue

                result = clean_order(row, reader.line_num)
                if result.order is None:
                    errors.append(result.error or "unknown error")
                    continue

                write_order(writer, result.order)
                clean_count += 1
        except csv.Error as exc:
            errors.append(f"line {reader.line_num}: CSV 格式錯誤: {exc}")

    error_path.write_text("\n".join(errors), encoding="utf-8")
    return clean_count, len(errors)

資料列不會全部留在記憶體；只有錯誤訊息會累積。

如果錯誤也可能有數百萬筆，就把錯誤檔一起打開，發現錯誤時立即 write()。

十. 做成可重跑的 CLI
#

加入 argparse：

import argparse


def main() -> None:
    parser = argparse.ArgumentParser(
        description="清理訂單 CSV，輸出合法資料與錯誤紀錄",
    )
    parser.add_argument("input", type=Path)
    parser.add_argument("--output", type=Path, default=Path("clean.csv"))
    parser.add_argument("--errors", type=Path, default=Path("errors.txt"))
    args = parser.parse_args()

    clean_count, error_count = clean_csv(
        args.input,
        args.output,
        args.errors,
    )
    print(f"clean rows: {clean_count}")
    print(f"errors: {error_count}")


if __name__ == "__main__":
    main()

執行：

python clean_orders.py messy-orders.csv \
  --output clean-orders.csv \
  --errors errors.txt

一個實用的小工具至少應具備：

輸入、輸出路徑可設定
欄位契約和型別規則集中管理
壞資料不會悄悄混進結果
錯誤帶有可追查的行號
同一份輸入可以重跑並得到一致結果

十一. 用暫存目錄測試完整流程
#

最值得測的是「讀入到輸出」的邊界：

from tempfile import TemporaryDirectory


def smoke_test() -> None:
    raw = (
        "order_id,customer,total,status\n"
        "1001,拍拍君,520,PAID\n"
        "bad-id,拍拍醬,300,paid\n"
    )

    with TemporaryDirectory() as tmp:
        root = Path(tmp)
        source = root / "orders.csv"
        target = root / "clean.csv"
        errors = root / "errors.txt"
        source.write_text(raw, encoding="utf-8-sig")

        clean_count, error_count = clean_csv(source, target, errors)

        assert clean_count == 1
        assert error_count == 1
        assert "拍拍君" in target.read_text(encoding="utf-8")
        assert "order_id 不是整數" in errors.read_text(encoding="utf-8")