---
title: "07 - 實戰自動化範例（資料 Pipeline）"
type: note
specialty: Programming
tags: [python, 07-實戰自動化範例, automation]
---

# 07 - 實戰自動化範例（資料 Pipeline）

← [[06-速查表]] | 回到 → [[00-Index]]

---

## 🎯 這篇要回答的問題

前面 01~06 學了變數、迴圈、函式、檔案讀寫、JSON……但你可能會想：
**「這些基礎語法到底組起來能幹嘛？」**

這篇用一個**真實存在的專案**來示範：把一堆重複、繁瑣、容易出錯的人工工作，
用 Python 變成一條按一個鍵就跑完的**自動化 pipeline**。

---

## 📦 真實案例：考古題刷題網頁的題庫維運

有一個「台灣內科專科考古題刷題網頁」專案（純前端 React 網站，沒有後端）。
它的題庫是一堆 **JSON 檔**，而這些 JSON 不是手打的——而是用一整套 Python 腳本，
從**原始考題 PDF** 自動加工出來的。

這套腳本住在專案的 `tools/` 資料夾，每支各做一件事，串起來就是一條 pipeline：

```
考題 PDF
   │  parse_pdf.py        解析 → 結構化題目
   ▼
questions.<年>.json
   │  extract_figures.py  附圖 PDF 裁成 PNG、回填 image 欄位
   │  classify.py         關鍵字規則 → 分到 11 科
   │  classify_topic.py   再細分「考點」（如 ACS、HF）
   ▼
帶分類的題目
   │  build_index.py      掃全部題目 → 輕量 index.json
   ▼
index.json（前端免載整年題庫就能顯示）
   │  export_questions.py 匯出「還沒寫詳解」的題目餵 AI
   ▼
   │  validate.py         ★ 最後一道驗證關卡：檢查資料正不正確
   ▼
   上線
```

> 重點不在醫學，而在：**這整條線就是基礎語法（檔案讀寫 + JSON + 迴圈 + 函式 + 條件）疊出來的。**
> 下面用簡化過的程式片段，逐站示意。

---

## 1️⃣ 解析：PDF → 結構化資料（`parse_pdf.py`）

原始考題是 PDF，人眼看得懂但程式不行。第一步是把它「讀進來、拆成一題一題」，
變成程式能處理的結構（list of dict），再存成 JSON。

```python
import json

def parse_questions(raw_text):
    """把一大段文字拆成一題一題的 dict（示意）。"""
    questions = []
    # 假設每題以「第N題」開頭，這裡用簡化邏輯示意
    for num, block in enumerate(split_into_blocks(raw_text), start=1):
        q = {
            "id": f"114-{num:03d}",      # f-string 補零 → 114-001
            "year": 114,
            "num": num,
            "question": block["stem"],
            "options": block["options"], # 5 個選項的 list
            "answer": block["answer"],   # 正解索引（0 起）
        }
        questions.append(q)              # 一題一題收進 list
    return questions

# 存成 JSON 檔（ensure_ascii=False 才能正常存中文）
data = parse_questions(raw_text)
with open("questions.114.json", "w", encoding="utf-8") as f:
    json.dump(data, f, ensure_ascii=False, indent=2)
```

用到的基礎：**`for` 迴圈 + `enumerate`**（[[03-控制流程]]）、**dict / list**（[[02-基本資料型態]]）、
**f-string 補零**、**`with open` + json**（[[05-常用模組]]）。

> 不同年份的 PDF 版面不一樣，所以實際專案有 `parse_pdf.py`、`parse_pdf_109.py`、`parse_pdf_old.py`
> 好幾個版本——這就是函式封裝的好處：**同樣的輸出格式，內部解析邏輯各自處理。**

---

## 2️⃣ 分類：用關鍵字規則自動歸類（`classify.py` / `classify_topic.py`）

兩千題要一題一題人工分到 11 科、再細分考點，太累。改用「**關鍵字規則**」自動分：
題幹裡出現某些字，就歸到對應的科 / 考點。

```python
# 「考點 → 關鍵字」的對照表（dict）
TOPIC_RULES = {
    "ACS":  ["心肌梗塞", "STEMI", "NSTEMI", "troponin"],
    "HF":   ["心衰竭", "heart failure", "EF", "BNP"],
    "CKD":  ["慢性腎臟病", "eGFR", "蛋白尿"],
}

def classify_topic(question):
    """看題幹+選項裡出現哪些關鍵字，回傳考點。"""
    text = question["question"] + " ".join(question["options"])
    for topic, keywords in TOPIC_RULES.items():
        if any(kw in text for kw in keywords):   # 命中任一關鍵字
            return topic
    return None        # 都沒中 → 先不分類

# 對每一題跑分類，把結果寫回題目
for q in data:
    q["topic"] = classify_topic(q)
```

用到的基礎：**dict 當對照表**、**`for ... in` 遍歷**、**`any()` + `in` 字串包含判斷**（[[06-速查表]]）、
**函式回傳值**（[[04-函式]]）。

> 這是「**啟發式**（heuristic）」——靠規則猜，不保證完美，但能把 90% 的苦工自動做完，剩下人工微調。

---

## 3️⃣ 建索引：濃縮成輕量檔（`build_index.py`）

整年題庫 JSON 很大。網站首頁只需要「題號、年份、科目、考點」就能畫出列表，
不必載入整題內容。所以掃一遍全部題目，**萃取必要欄位**產出一個小小的 `index.json`：

```python
import json, glob

index = []
# glob 找出所有 questions.*.json 檔
for path in glob.glob("questions.*.json"):
    with open(path, encoding="utf-8") as f:
        for q in json.load(f):
            # 只挑前端列表需要的幾個欄位（用 dict comprehension 也行）
            index.append({
                "id": q["id"],
                "year": q["year"],
                "subject": q["subject"],
                "topic": q.get("topic"),   # .get 取不到不會報錯
            })

with open("index.json", "w", encoding="utf-8") as f:
    json.dump(index, f, ensure_ascii=False)

print(f"已建立索引，共 {len(index)} 題")
```

用到的基礎：**`glob` 批次找檔**、**巢狀迴圈**、**`dict.get()` 安全取值**（[[02-基本資料型態]]）、
**`len()` 計數 + f-string 回報**。

> 「大資料只保留必要欄位 → 產生輕量索引」是非常常見的效能手法。

---

## 4️⃣ 匯出：篩出「待辦」項目（`export_questions.py`）

兩千題不可能一次寫完詳解。需要一個工具：**只挑出「還沒寫詳解」的題目**匯出來，
分批餵給 AI 起草。這就是「**篩選（filter）**」：

```python
import json

with open("questions.114.json", encoding="utf-8") as f:
    questions = json.load(f)

with open("explanations.114.json", encoding="utf-8") as f:
    explanations = json.load(f)   # 已有詳解的 id 集合

# 用 list comprehension 篩出「尚無詳解」的題
todo = [q for q in questions if q["id"] not in explanations]

# 還可以再依年/科/題號縮小範圍（argparse 接命令列參數）
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("--subject")     # 例：python export_questions.py --subject 心臟血管
args = parser.parse_args()
if args.subject:
    todo = [q for q in todo if q["subject"] == args.subject]

print(f"待寫詳解：{len(todo)} 題")
```

用到的基礎：**list comprehension 篩選**（[[02-基本資料型態]]）、**`in` 判斷成員**、
**`argparse` 接命令列參數**（讓同一支腳本能用參數客製，不必改程式碼）。

---

## 5️⃣ 驗證：最後一道防線（`validate.py`）★

這是整條 pipeline **最重要**的一站。前面任何一步出小錯（少了選項、答案對不上、圖檔不見），
都可能讓網站出錯、甚至給使用者**錯的答案**。所以最後跑一支 `validate.py`，
把所有「該成立的規則」逐一檢查，**有一條不過就報錯停下**。

這正是 `assert`（斷言）的用途：**「我斷定這件事一定為真，不為真就立刻爆給我看。」**

```python
import json, os

with open("questions.114.json", encoding="utf-8") as f:
    questions = json.load(f)

LETTERS = ["A", "B", "C", "D", "E"]
errors = []

for q in questions:
    qid = q["id"]

    # 規則 1：每題一定要有這些欄位
    for field in ["id", "question", "options", "answer", "answerLetter"]:
        if field not in q:
            errors.append(f"{qid} 缺欄位 {field}")

    # 規則 2：選項數必須是 5
    if len(q["options"]) != 5:
        errors.append(f"{qid} 選項數 = {len(q['options'])}，應為 5")

    # 規則 3：answer（索引）要和 answerLetter（字母）一致
    if LETTERS[q["answer"]] != q["answerLetter"]:
        errors.append(f"{qid} answer 與 answerLetter 不一致")

    # 規則 4：若標了有附圖，圖檔必須真的存在
    if q.get("needsImage") and not os.path.exists(q["image"]):
        errors.append(f"{qid} 找不到圖檔 {q['image']}")

# 有任何錯就印出來、讓程式以「失敗」結束
if errors:
    for e in errors:
        print("❌", e)
    raise SystemExit(f"驗證失敗，共 {len(errors)} 個問題")

print("✅ 全部通過驗證")
```

實際的 `validate.py` 還會檢查：科目名稱合法、題號連續、index 與題庫同步、詳解覆蓋率……
原則都一樣：**把「人工很容易看漏」的檢查，寫成程式一次跑完。**

用到的基礎：**`for` 巢狀檢查、`if` 條件、`len()`、`in`、`os.path.exists`、收集錯誤到 list 再一起報**。

> [!important] 為什麼這站最關鍵？
> 自動化 pipeline 的風險是「錯了也不知道，一路傳到上線」。`validate.py` 就是
> **「最後一道驗證關卡」**——資料正確性的保險。寧可在這裡 fail，也不要讓錯誤資料上線。

---

## 🧩 把整條線串起來

每支腳本各做一件事、各自能單獨跑；要全跑時用一個 shell 指令串起來（或寫個總控腳本）：

```bash
python tools/parse_pdf.py --year 114        # 解析
python tools/extract_figures.py --year 114  # 裁圖回填
python tools/classify.py --year 114         # 分科
python tools/classify_topic.py --year 114   # 分考點
python tools/build_index.py                 # 建索引
python tools/validate.py                    # ★ 驗證
```

---

## 💡 從這個案例學到的觀念

| 觀念 | 在這個案例裡的體現 |
|------|---------------------|
| **資料 pipeline** | 解析 → 分類 → 建索引 → 匯出 → 驗證，前一站的輸出是後一站的輸入 |
| **單一職責** | 每支腳本只做一件事，好懂、好測、好重用 |
| **自動化重複工** | 兩千題的轉檔/分類/檢查，人工要做到瘋掉，程式幾秒跑完 |
| **可重跑（idempotent）** | 改了原始檔重跑就更新，不怕手殘 |
| **驗證是保險** | `assert` / 規則檢查擋下錯誤資料，不讓它上線 |
| **參數化** | `argparse` 讓同一支腳本能用 `--year`/`--subject` 客製，不必改程式 |

> [!tip] 你不需要醫學或網頁背景也能照搬這套思路：
> 任何「**一堆原始檔 → 反覆做同樣的加工 → 產出乾淨結果**」的工作
> （整理發票、批次改檔名、彙整 Excel、抓網頁資料……），
> 都能用你在 01~06 學到的基礎語法，組成自己的自動化 pipeline。

---

## 🔗 相關筆記

- [[02-基本資料型態]]｜list / dict / comprehension（pipeline 的資料結構基礎）
- [[04-函式]]｜把每個步驟封裝成函式
- [[05-常用模組]]｜檔案讀寫、json、argparse、例外處理
- [[06-速查表]]｜語法快速回查

---

← [[06-速查表]] | 回到 → [[00-Index]]