DE Day 18 公開資料實戰:政府開放資料平台
執行需求:CPU 可跑。今天進入採集篇的第三篇,把 Day 17 的禮貌爬蟲套件接到「中華民國政府資料開放平臺」(data.gov.tw)的真實目錄。這個平臺收錄了上萬個政府機關產出的開放資料集,依「政府資料開放授權條款第 1 版」釋出。我們會用一支可重現的腳本,把「找資料 → 看 metadata → 確認授權 → 下載 CSV/JSON → 落進 DuckDB」的完整流程做出來。版本基準仍是 2025 年 11 月:Python 3.13、httpx 0.28、DuckDB 1.4、Polars 1.33。
引言
Day 15 與 Day 16 給了爬蟲與動態瀏覽器兩個工具,Day 17 加上了禮貌與工程紀律。今天把它們用在「真實世界的公開資料」上:政府資料開放平臺。這個平臺由國家發展委員會管理,從 2012 年推動至今已累積上萬個資料集,涵蓋交通、氣象、財稅、戶政、警政、統計、環境等領域。對資料工程師來說,這是「真實、可商用、有版本」的來源,比 Kaggle 的練習資料更具實務價值。
今天的目標很具體:用一份可重現的 Python 腳本,去平臺找一份「每日會更新」的資料集(例如空氣品質監測、即時雨量、停水公告),把它每日拉到本地、落地成 Parquet、寫進 DuckDB,並用一段 SQL 確認資料完整性。我們也會談到一個資料工程師必修的倫理議題:「政府資料開放授權條款第 1 版」允許你做什麼、不允許你做什麼、與個資法的界線在哪裡。這不是法律諮詢,但至少把幾個常見誤解講清楚。
在動手前,先說明一個重要前提:本系列的所有範例都「不鎖定單一資料集」。原因是政府平臺的資料集 ID、欄位名稱、檔案 URL 都可能改版,鎖定特定 ID 會讓範例在某天突然失效。所以今天我們教的是「方法」:用平臺的搜尋 API 或瀏覽器查詢頁,找到資料集描述,讀 metadata,確認授權,最後用通用方法下載。你讀完之後可以把方法套到任何資料集。
政府資料開放平臺的結構與授權
先認識平臺的結構。data.gov.tw 主要由四個部分組成:首頁的搜尋、目錄頁、資料集頁、檔案下載頁。「資料集(dataset)」是多個檔案的集合,例如「空氣品質監測資料」就包含每月、每日、每小時三種檔案。「檔案(resource)」是實際可下載的 CSV、JSON、XML。當我們說「抓一份資料」時,多半是指「找到某個資料集 → 從它的檔案清單挑一個檔 → 下載」。
授權方面,平臺預設採「政府資料開放授權條款第 1 版」。這個授權的精神是「使用者可自由重製、散布、公開傳輸、修改、編輯著作(含翻譯),且不需標示著作人」。聽起來很自由,但有兩個但書必須注意:第一,授權條款要求「使用時仍須標示資料來源出處」,常見寫法是「資料來源:政府資料開放平臺 data.gov.tw」;第二,當資料集中包含「第三人著作」時,使用者必須另行取得該第三人的授權。例如資料集裡包含某家公司提供的營業統計,這部分仍受原公司著作權保護,使用上要再確認。
另外兩個實務上常碰到的議題是「個資」與「持續提供」。個資法對政府單位同樣有約束,平臺上的資料集在釋出前通常已經過「去識別化」處理,但若資料中包含間接可識別個人的欄位(如「年齡 + 居住地 + 罕見疾病」),仍可能受個資法規範。處理上建議:碰到含個人欄位的資料集時,先停下來看 metadata 裡的「更新日期」與「備註」,若 metadata 提到「依個資法 X 條釋出」就要特別小心。持續提供方面,政府資料不像 API 有 99.9% 的 SLA,平臺的檔案 URL、欄位命名、甚至整個資料集都可能改版或下架。我們在 Day 20 設計增量載入、在 Day 38 設計資料合約時,都要把「來源可能改變」這件事考慮進去。
完整實作:找資料、確認授權、下載、落地
今天的實作分四階段。第一階段用平臺的瀏覽器頁面(或搜尋 API)找一份有「每日更新」標註的資料集;第二階段檢查 robots.txt 與 metadata;第三階段用昨天的禮貌套件下載檔案;第四階段寫進 DuckDB。為了讓範例可在離線環境重現,我們以「中華民國政府資料開放平臺」這個「目標」為例,把每一階段的關鍵程式碼寫成獨立函式,並用一份本地 mock metadata 模擬抓回來的結果。
第一步:先安裝套件與建立工作目錄:
cd de-journey
mkdir -p data/government logs
uv pip install httpx==0.28.1 polars==1.33.0 duckdb==1.4.1
這段指令與 Day 17 相同,加上 Polars 1.33 做落地轉換。DuckDB 1.4 在 2025 年 11 月是穩定版本,read_json_auto 對政府平臺常見的 JSON 結構相容性最好。data/government/ 專門放政府資料的原始檔,與 Day 1 的 data/ 平行,便於日後備份與稽核。
第二步:寫一個「資料集搜尋器」,把平臺的查詢頁用昨天的禮貌方法抓回來:
"""pipelines/day18_search.py:用 httpx 查政府資料開放平臺的目錄頁。"""
import asyncio
import httpx
from urllib.parse import urlencode
from bs4 import BeautifulSoup
UA = "HaoBot/1.0 (+https://blog.hao-code.com/bots)"
SEARCH_URL = "https://data.gov.tw/datasets/search"
async def search_datasets(keyword: str, page: int = 1) -> list[dict]:
params = {"q": keyword, "page": page}
async with httpx.AsyncClient(headers={"User-Agent": UA}, timeout=20.0) as client:
r = await client.get(f"{SEARCH_URL}?{urlencode(params)}")
r.raise_for_status()
soup = BeautifulSoup(r.text, "html.parser")
items = []
for a in soup.select("a.dataset-item"):
title_el = a.select_one(".dataset-title")
desc_el = a.select_one(".dataset-description")
items.append({
"title": title_el.get_text(strip=True) if title_el else "",
"url": a.get("href", ""),
"description": desc_el.get_text(strip=True) if desc_el else "",
})
return items
if __name__ == "__main__":
results = asyncio.run(search_datasets("空氣品質"))
for it in results[:5]:
print(f"{it['title']}({it['url']})")
print(f" {it['description'][:80]}")
這段程式用 httpx 送查詢,再用 BeautifulSoup 解析 HTML。政府平臺的查詢頁是伺服器端渲染(不是 SPA),所以不需要 Playwright。實際查詢時,把 "空氣品質" 換成你想找的關鍵字;常見的選擇有「空氣品質監測」、「停水公告」、「即時雨量」、「YouBike 站點」、「國道車流」等,每個都是「每日更新」的典型範例。實際輸出會依當下平臺內容而定,示範關鍵字可能找到 10 到 30 筆。這個查詢頁本身可能被改版,所以程式碼重點是「方法」而非「特定 CSS selector」。
第三步:拿到資料集清單後,針對單一資料集把它的 metadata 與檔案下載 URL 抽出來:
"""pipelines/day18_metadata.py:從資料集頁抽出 metadata 與檔案下載 URL。"""
import asyncio
import httpx
from bs4 import BeautifulSoup
UA = "HaoBot/1.0 (+https://blog.hao-code.com/bots)"
async def fetch_dataset(dataset_url: str) -> dict:
async with httpx.AsyncClient(headers={"User-Agent": UA}, timeout=20.0) as client:
r = await client.get(dataset_url)
r.raise_for_status()
soup = BeautifulSoup(r.text, "html.parser")
meta = {"url": dataset_url, "files": [], "license": ""}
for tr in soup.select("table.dataset-files tbody tr"):
cells = [td.get_text(strip=True) for td in tr.select("td")]
if len(cells) >= 2:
link = tr.select_one("a")
meta["files"].append({
"name": cells[0],
"format": cells[1] if len(cells) > 1 else "",
"url": link.get("href", "") if link else "",
})
lic_el = soup.select_one(".dataset-license, .license-info")
if lic_el:
meta["license"] = lic_el.get_text(strip=True)
return meta
if __name__ == "__main__":
sample = "https://data.gov.tw/datasets/xxxxx"
m = asyncio.run(fetch_dataset(sample))
print(f"授權:{m['license']}")
for f in m["files"]:
print(f" - {f['name']}({f['format']})→ {f['url']}")
這段重點在兩個地方:第一,「檔案下載 URL」藏在 table.dataset-files 的 <a> 標籤,這是平臺一致的版面;第二,「授權」放在 .dataset-license,政府資料預設會出現「政府資料開放授權條款第 1 版」字樣。拿到檔案 URL 後,下一步就是把檔案抓回來。注意:實際抓到的 dataset_url 要從第二步的搜尋結果挑一個真實存在的;網址格式像 https://data.gov.tw/datasets/<id>,ID 是平臺內部編號。
第四步:用昨天的禮貌套件把檔案下載到本地,並用 DuckDB 直接讀進來:
"""pipelines/day18_land.py:把政府資料下載並落進 DuckDB。"""
import asyncio
from pathlib import Path
from datetime import datetime
import duckdb
from pipelines._etiquette import PoliteSession
from pipelines._robots import fetch_robots
UA = "HaoBot/1.0 (+https://blog.hao-code.com/bots)"
DATA_ROOT = Path("data/government")
async def download_file(url: str, dest: Path) -> Path:
DATA_ROOT.mkdir(parents=True, exist_ok=True)
async with PoliteSession(UA, qps=1.0) as session:
import httpx
async with httpx.AsyncClient() as client:
r = await session.get(client, url)
r.raise_for_status()
dest.write_bytes(r.content)
return dest
def land_to_duckdb(file_path: Path, table_name: str) -> int:
con = duckdb.connect("warehouse/de-journey.duckdb")
con.execute(f"CREATE SCHEMA IF NOT EXISTS raw")
if file_path.suffix.lower() == ".csv":
con.execute(
f"CREATE OR REPLACE TABLE raw.{table_name} AS "
f"SELECT *, CURRENT_TIMESTAMP AS _loaded_at "
f"FROM read_csv_auto('{file_path.as_posix()}', sample_size=-1)"
)
elif file_path.suffix.lower() == ".json":
con.execute(
f"CREATE OR REPLACE TABLE raw.{table_name} AS "
f"SELECT *, CURRENT_TIMESTAMP AS _loaded_at "
f"FROM read_json_auto('{file_path.as_posix()}', format='auto')"
)
n = con.execute(f"SELECT COUNT(*) FROM raw.{table_name}").fetchone()[0]
return n
if __name__ == "__main__":
sample_url = "https://data.gov.tw/api/v2/rest/dataset/xxxxx"
robots = fetch_robots("https://data.gov.tw", UA)
print(f"data.gov.tw Crawl-delay: {robots.crawl_delay} 秒")
if not robots.can_fetch(sample_url):
print("robots.txt 禁止,跳過")
else:
dest = DATA_ROOT / f"{datetime.now():%Y%m%d}_{Path(sample_url).name}"
path = asyncio.run(download_file(sample_url, dest))
n = land_to_duckdb(path, "day18_demo")
print(f"落地 {n} 筆到 raw.day18_demo")
這段把昨天的 PoliteSession 與 fetch_robots 整個套進來:先用 robots 確認能抓,再用節流 session 下載,最後用 DuckDB 的 read_csv_auto 與 read_json_auto 直接讀進來。sample_size=-1 告訴 DuckDB 全檔掃描而不是抽樣,這對小檔(< 100 MB)很實用。注意 DuckDB 的 read_csv_auto 對編碼有自動判斷,UTF-8 與 BIG5 都能讀;如果碰到奇怪的編碼錯誤,請用 encoding='utf-8' 明確指定(Day 13 會展開清洗章節)。
第五步:建立一份「稽核 SQL」,把這次抓取結果摘要一下:
"""pipelines/day18_audit_sql.py:抓完之後跑 SQL 做稽核摘要。"""
import duckdb
con = duckdb.connect("warehouse/de-journey.duckdb")
print("=== 抓取摘要 ===")
summary = con.execute("""
SELECT
COUNT(*) AS n_rows,
COUNT(DISTINCT _loaded_at) AS n_loads,
MIN(_loaded_at) AS first_load,
MAX(_loaded_at) AS last_load
FROM raw.day18_demo
""").fetchone()
print(f"筆數:{summary[0]},載入次數:{summary[1]}")
print(f"首次載入:{summary[2]},最近載入:{summary[3]}")
print("=== 欄位型別 ===")
cols = con.execute("DESCRIBE raw.day18_demo").fetchall()
for name, dtype, _ in cols:
print(f" {name:30s} {dtype}")
print("=== 授權標示 ===")
print('資料來源:政府資料開放平臺 https://data.gov.tw/(政府資料開放授權條款第 1 版)')
這段是稽核的最小可用版本:筆數、載入次數、首次與最近載入時間,用來判斷抓取是否成功、是否有重複。DESCRIBE 列出每個欄位的型別,方便日後做清洗時對應正確的轉換函式。最後一行把授權標示印出來,這是「政府資料開放授權條款第 1 版」要求的最基本標示。實際筆數會依資料集而定,政府平臺的常見 CSV 多落在數千到數萬筆。
第七步:用一份 Python 程式把前六步組合成一支「端到端抓取」流程,從搜尋到落地的所有步驟一次跑完:
"""pipelines/day18_e2e.py:把搜尋 + metadata + 下載 + 落地串成一支。"""
import asyncio
from pathlib import Path
import duckdb
from pipelines.day18_search import search_datasets
from pipelines.day18_metadata import fetch_dataset
from pipelines.day18_land import download_file, land_to_duckdb
UA = "HaoBot/1.0 (+https://blog.hao-code.com/bots)"
async def run(keyword: str, file_format: str = "CSV") -> int:
candidates = await search_datasets(keyword, page=1)
print(f"找到 {len(candidates)} 個候選資料集")
for c in candidates[:3]:
meta = await fetch_dataset(c["url"])
chosen = next((f for f in meta["files"] if f["format"].upper() == file_format.upper()), None)
if not chosen:
print(f" 跳過 {c['title']}(無 {file_format} 檔)")
continue
dest = Path("data/government") / Path(chosen["url"]).name
await download_file(chosen["url"], dest)
table = "day18_" + Path(chosen["url"]).stem.replace("-", "_")[:30]
n = land_to_duckdb(dest, table)
print(f" {c['title']} → {n} 筆")
return n
return 0
if __name__ == "__main__":
n = asyncio.run(run("空氣品質"))
print(f"完成,共落地 {n} 筆")
這段把前面六個獨立函式組合成一支端到端腳本:先搜尋、再讀 metadata、挑 CSV 檔、下載、落地。這是「採集管線」的標準樣貌,Day 30 的端到端章節會用更完整的方式重構。把這支程式包進排程(Day 21)後,它就會每天自動找關鍵字、抓資料、寫進 DuckDB。實際輸出筆數依當天平臺的檔案而定,示範關鍵字「空氣品質」常見落在數千到數萬筆。
第六步:把 DuckDB 結果轉 Parquet 永久落地,並用一段 SQL 示範最基本的探索:
"""pipelines/day18_explore.py:把 DuckDB 表轉 Parquet 並做簡單探索。"""
import duckdb
import polars as pl
con = duckdb.connect("warehouse/de-journey.duckdb")
con.execute("""
COPY (SELECT * FROM raw.day18_demo) TO 'warehouse/day18_demo.parquet'
(FORMAT PARQUET, COMPRESSION 'snappy')
""")
df = pl.read_parquet("warehouse/day18_demo.parquet")
print(f"Parquet 筆數:{df.height},欄位數:{df.width}")
print(df.head(3))
nulls = df.null_count().row(0)
print(f"空值欄位數:{sum(1 for n in nulls if n > 0)}")
這段做三件事:第一,用 DuckDB 的 COPY ... TO ... (FORMAT PARQUET) 把 DuckDB 表寫成 Parquet,壓縮用 snappy(Day 9 學過的標準選擇);第二,用 Polars 讀 Parquet 並印出前三列;第三,統計有空值的欄位數。Parquet 是 Day 9 與後續管線(Day 19 到 Day 26)的標準落地格式,比 CSV 小且保留型別。實際筆數與空值分布會依資料集而定。
常見錯誤與踩雷
第一個雷是「沒看授權就把資料商用」。即使 metadata 寫「政府資料開放授權條款第 1 版」,仍要檢查資料集是否包含「第三人著作」。對應排查方向:進到資料集頁面,捲到最下方,看有沒有「本資料集部分內容由 OO 公司提供」之類標示;若有的話,這部分需另尋授權。另一個檢查方式是看 metadata 裡的「聯絡人」欄位,必要時直接寫信確認。
第二個雷是「把政府資料當 API 用」。政府平臺的檔案 URL 可能改版、下架、暫時 404;資料 schema 也可能調整欄位名稱或加新欄。對應排查方向:把 metadata 的「更新日期」與「檔案 URL」一起記到資料表(例如 _meta.json),每次抓取前先確認 URL 還能下載;如果 URL 失效,用「資料集頁」當備案入口;schema 變動則靠 Day 14 學過的品質檢查與 Day 38 的資料合約來擋。
第三個雷是「用 requests 抓但網站是 SPA」。早期政府平臺大多是伺服器端渲染,但部分新上線的查詢頁與儀表板是 SPA。對應排查方向:用 curl 或 httpx.get() 看 HTML 是否有資料;沒有就改用 Day 16 的 Playwright。但記得禮貌套件的 robots 檢查仍然要做,別因為切到瀏覽器就忘了禮儀。
第四個雷是「抓下來的編碼是 BIG5 不是 UTF-8」。早期政府檔案常用 BIG5,UTF-8 開頭讀會出現亂碼。對應排查方向:先看檔案前幾個位元組判斷(UTF-8 通常開頭是 EF BB BF 或沒有 BOM);用 DuckDB 讀時加 encoding='big5' 或先用 Python 轉成 UTF-8 再讀。Day 13 的清洗章節會展開編碼判斷的方法。
第五個雷是「忽略 robots.txt 結果」。data.gov.tw 在 2025 年的 robots.txt 不禁止 /datasets/ 路徑,但這個設定可能改。對應排查方向:每次跑之前先讀 robots.txt;若設定變嚴格(例如禁止整個 /api/),要立刻通知負責人評估要不要改用其他來源(例如機關自己的開放資料專區)。
效能與實務提醒
政府資料檔案大小差異極大。從數 KB(如公告文字)到數 GB(如即時空氣品質的全國觀測站資料)都有。對應做法:小檔(< 10 MB)直接全檔讀進 DuckDB;中檔(10 MB 到 1 GB)用 Polars 的 lazy 讀取,先過濾再分析;大檔(> 1 GB)考慮先用 DuckDB 的 read_csv_auto 邊讀邊篩選,或乾脆把原始檔轉成 Parquet 後再分析。
另一個常見取捨是「每日全量重抓 vs 增量更新」。全量重抓簡單但浪費頻寬與時間;增量更新節省資源但要設計「水位標記」(Day 20 會完整展開)。實務建議:剛開始一週先全量重抓,觀察檔案大小、更新頻率、欄位變動幅度,再決定要不要改增量。一個簡單的判斷是「每日新資料佔比 < 10%」就值得改增量。
排程與通知是另一個實務重點。每天凌晨三點跑抓取、跑完把摘要寫到 Slack,這是常見的設計。但政府資料不像 API 有嚴格 SLA,若某天平臺當機,你的管線也會跟著失敗。對應做法:抓取失敗時不要直接中斷整條管線,而是把失敗狀態寫進 raw._status 表,通知負責人;隔天再試一次,並在 metadata 裡加註「昨日資料缺漏」。Day 22 的失敗處理篇章會展開這個模式。
最後,政府資料的可信度差異很大。有些資料經過嚴格審核(如戶政、財稅),有些只是各機關自己產出未經統一品質管理。資料工程師的責任之一是「評估資料品質」:拿到資料後先看空值率、欄位一致性、更新頻率是否符合 metadata 描述;如果差太多,要回頭與資料提供機關確認。把這套評估寫進 README 與交接文件,是「管線活得比作者久」的關鍵之一。
小結
今天把 Day 17 的禮貌套件接到「中華民國政府資料開放平臺」的真實目錄,建立了「找資料 → 看 metadata → 確認授權 → 下載 → 落地 → 稽核」的完整流程。我們示範了四種關鍵動作:搜尋資料集、讀 metadata、下載檔案、DuckDB 落地,並把授權標示與個資法界線講清楚。版本仍是 2025 年 11 月主流,授權依「政府資料開放授權條款第 1 版」,可在合法範圍內自由重製、改作與散布,但仍須標示來源。
把今天的關鍵詞整理進筆記本:data.gov.tw、政府資料開放授權條款第 1 版、metadata、Crawl-delay、read_csv_auto、read_json_auto、Parquet、snappy。這些詞在 Day 19(API 分頁)、Day 20(增量載入)、Day 38(資料合約)會反覆出現。把 pipelines/day18_search.py、pipelines/day18_metadata.py、pipelines/day18_land.py 都存起來,明天 Day 19 會在這個基礎上加上 API 分頁與增量抓取。
結語
今天我們把爬蟲從「會動」變成「能處理真實世界的開放資料」。但這只是開始——真實世界的 API 跟範例 API 差很多:分頁、限流、版本控管、欄位變動,每一個都是坑。明天 Day 19 我們會把 Day 18 的「找資料 → 下載」流程擴充成「分頁 → 重試 → 增量」,用一個「指標查詢 API」做示範,介紹 cursor-based pagination 與 offset-based pagination 的差別,以及如何設計「水位標記」避免重複抓。
明天,我們會把禮貌套件與 DuckDB 結合,寫一支可以每天自動抓數萬筆 API 資料的管線。
延伸資源
- 中華民國政府資料開放平臺(2025):
https://data.gov.tw/,依「政府資料開放授權條款第 1 版」可商用、可改作。 - 政府資料開放授權條款第 1 版(2025):
https://data.gov.tw/license,本系列所有政府資料來源都依此授權。 - DuckDB CSV 讀取官方文件(2025,1.4 版):
https://duckdb.org/docs/data/csv/reading.html,read_csv_auto與sample_size的標準說明。 - DuckDB JSON 讀取官方文件(2025,1.4 版):
https://duckdb.org/docs/data/json/reading.html,read_json_auto的格式選項。 - 國家發展委員會 開放資料專區(2025):
https://data.gov.tw/,政策與規範的權威來源。
留言
張貼留言