07 - 實戰自動化範例（資料 Pipeline）
🎯 這篇要回答的問題
前面 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。
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 科、再細分考點，太累。改用「關鍵字規則」自動分： 題幹裡出現某些字，就歸到對應的科 / 考點。
# 「考點 → 關鍵字」的對照表（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：
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）」：
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（斷言）的用途：「我斷定這件事一定為真，不為真就立刻爆給我看。」
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 再一起報。
為什麼這站最關鍵？
自動化 pipeline 的風險是「錯了也不知道，一路傳到上線」。
validate.py就是 「最後一道驗證關卡」——資料正確性的保險。寧可在這裡 fail，也不要讓錯誤資料上線。
🧩 把整條線串起來
每支腳本各做一件事、各自能單獨跑；要全跑時用一個 shell 指令串起來（或寫個總控腳本）：
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 客製，不必改程式 |
你不需要醫學或網頁背景也能照搬這套思路：
任何「一堆原始檔 → 反覆做同樣的加工 → 產出乾淨結果」的工作 （整理發票、批次改檔名、彙整 Excel、抓網頁資料……）， 都能用你在 01~06 學到的基礎語法，組成自己的自動化 pipeline。
🔗 相關筆記
- 02-基本資料型態｜list / dict / comprehension（pipeline 的資料結構基礎）
- 04-函式｜把每個步驟封裝成函式
- 05-常用模組｜檔案讀寫、json、argparse、例外處理
- 06-速查表｜語法快速回查