《Python一下：從風土資料到智慧生活》第 22 篇
把YouBike JSON變成可分析資料——欄位驗證與快照保存
JSON能解析只是起點；資料還要能被理解、驗證、追蹤版本，才適合統計與比較。
學習目標
完成本篇後，你能理解JSON與Python型別的對應，使用
完成本篇後，你能理解JSON與Python型別的對應，使用
json.loads()、dumps()、load() 與 dump()，驗證欄位與型別、處理欄位別名、建立資料品質報告，並比較兩個YouBike快照的站點變化。一、JSON是資料交換格式
JSON常用於網頁API與裝置資料交換。它是文字格式，不是Python程式碼，也不應使用 eval() 解析。
| JSON | Python | 注意 |
|---|---|---|
| object | dict | 鍵通常是字串 |
| array | list | 元素可有不同型別 |
| string | str | 數字也可能被來源寫成字串 |
| number | int或float | 金融或高精度資料另有考量 |
| true／false | True／False | 大小寫不同 |
| null | None | 缺值不等於0或空字串 |
資料語意原則：
0可能表示真的沒有車，None可能表示未知，缺少欄位可能代表Schema改變；三者不能自動視為相同。若把未知補成0，統計與決策都會被扭曲。二、字串與Python物件互轉
import json
json_text = """
{
"站點": "示範站A",
"可借車輛": 8,
"營運中": true,
"備註": null
}
"""
data = json.loads(json_text)
print(data)
print(type(data["可借車輛"]))
print(data["備註"] is None)
restored = json.dumps(data, ensure_ascii=False, indent=2)
print(restored)loads 的s可理解為string；dumps 產生JSON字串。ensure_ascii=False 讓中文直接顯示，indent=2 方便人工閱讀。
三、檔案與Python物件互轉
import json
from pathlib import Path
path = Path("教學快照.json")
sample = [{"sno": "A001", "sna": "示範站A"}]
with path.open("x", encoding="utf-8") as file:
json.dump(sample, file, ensure_ascii=False, indent=2)
with path.open("r", encoding="utf-8") as file:
loaded = json.load(file)
print(loaded)load、dump 直接處理檔案物件。重要快照使用 x 模式，避免覆蓋歷史資料。
四、不要假設所有來源欄位永遠相同
不同縣市或不同版本可能以 available_rent_bikes、sbi 等名稱表示可借車輛。別名可以相容已知版本，但若同時出現衝突值，應視為錯誤。
FIELD_ALIASES = {
"站點編號": ("sno",),
"站點名稱": ("sna",),
"行政區": ("sarea",),
"可借車輛": ("available_rent_bikes", "sbi"),
"可還空位": ("available_return_bikes", "bemp"),
"更新時間": ("updateTime", "mday"),
}
def read_alias(record, aliases):
found = [
(name, record[name])
for name in aliases
if name in record and record[name] is not None
]
if not found:
raise KeyError(f"找不到欄位：{aliases}")
values = {str(value) for _, value in found}
if len(values) > 1:
raise ValueError(f"別名欄位值衝突：{found}")
return found[0][1]五、建立正規化函式
def parse_nonnegative_int(value, field_name):
if value is None or str(value).strip() == "":
raise ValueError(f"{field_name}是未知值")
number = int(value)
if number < 0:
raise ValueError(f"{field_name}不可為負數")
return number
def normalize_station(record):
station = {
name: read_alias(record, aliases)
for name, aliases in FIELD_ALIASES.items()
}
station["站點編號"] = str(station["站點編號"]).strip()
station["站點名稱"] = str(station["站點名稱"]).strip()
station["行政區"] = str(station["行政區"]).strip()
station["可借車輛"] = parse_nonnegative_int(
station["可借車輛"], "可借車輛"
)
station["可還空位"] = parse_nonnegative_int(
station["可還空位"], "可還空位"
)
station["更新時間"] = str(station["更新時間"]).strip()
if not station["站點編號"] or not station["站點名稱"]:
raise ValueError("站點編號與名稱不可空白")
return station六、Schema版本：告訴未來的程式如何解讀
官方資料有自己的結構，你的「正規化快照」也應明確記錄版本。版本改變時，可保留舊解析器或撰寫轉換程式。
SNAPSHOT_SCHEMA = "nfu-youbike-snapshot/1.0"
def validate_snapshot(snapshot):
if snapshot.get("schema") != SNAPSHOT_SCHEMA:
raise ValueError("不支援的快照Schema版本")
if not isinstance(snapshot.get("stations"), list):
raise TypeError("stations必須是串列")
if not isinstance(snapshot.get("quality"), dict):
raise TypeError("quality必須是字典")
return snapshot七、批次轉換與資料品質報告
from collections import Counter
def normalize_dataset(raw_records):
valid = []
errors = []
station_ids = Counter()
for index, record in enumerate(raw_records, start=1):
try:
if not isinstance(record, dict):
raise TypeError("單筆資料必須是字典")
station = normalize_station(record)
station_ids[station["站點編號"]] += 1
valid.append(station)
except (KeyError, TypeError, ValueError) as error:
errors.append({"序號": index, "原因": str(error)})
duplicate_ids = sorted(
key for key, count in station_ids.items() if count > 1
)
quality = {
"原始筆數": len(raw_records),
"有效筆數": len(valid),
"錯誤筆數": len(errors),
"重複站點編號": duplicate_ids,
}
return valid, errors, quality重複編號不能隨便保留第一筆或最後一筆；應先檢查是否為來源重複、版本衝突或站點資料更新方式造成。
八、計算資料完整率
def completeness_report(records, fields):
total = len(records)
report = {}
for field in fields:
present = sum(
1
for record in records
if field in record
and record[field] is not None
and str(record[field]).strip() != ""
)
report[field] = {
"有值筆數": present,
"完整率": round(present / total, 4) if total else None,
}
return report完整率只表示欄位有值，不代表值正確、即時或合理。資料品質至少還包含唯一性、一致性、時效性與來源可信度。
九、建立可追溯的JSON快照
import json
from datetime import datetime, timezone
from pathlib import Path
def save_snapshot(path, source_url, stations, errors, quality):
snapshot = {
"schema": SNAPSHOT_SCHEMA,
"source": {
"url": source_url,
"fetched_at_utc": datetime.now(
timezone.utc
).isoformat(),
},
"quality": quality,
"errors": errors,
"stations": stations,
}
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("x", encoding="utf-8") as file:
json.dump(snapshot, file, ensure_ascii=False, indent=2)
return path快照不是即時資料：檔案一旦保存就是某個時間點的歷史紀錄。教學或展示時必須呈現擷取時間，不可讓使用者誤以為快照仍是最新車位資訊。
十、讀取快照時再次驗證
import json
from pathlib import Path
def load_snapshot(path):
path = Path(path)
try:
with path.open("r", encoding="utf-8") as file:
snapshot = json.load(file)
except FileNotFoundError:
raise FileNotFoundError(f"找不到快照：{path}")
except json.JSONDecodeError as error:
raise ValueError(
f"JSON格式錯誤，第{error.lineno}行：{error.msg}"
) from error
if not isinstance(snapshot, dict):
raise TypeError("快照最外層必須是字典")
return validate_snapshot(snapshot)十一、比較兩個快照
def index_by_station_id(stations):
result = {}
for station in stations:
station_id = station["站點編號"]
if station_id in result:
raise ValueError(f"重複站點編號：{station_id}")
result[station_id] = station
return result
def compare_snapshots(old_stations, new_stations):
old_map = index_by_station_id(old_stations)
new_map = index_by_station_id(new_stations)
old_ids = old_map.keys()
new_ids = new_map.keys()
added = sorted(new_ids - old_ids)
removed = sorted(old_ids - new_ids)
changed = []
for station_id in sorted(old_ids & new_ids):
old = old_map[station_id]
new = new_map[station_id]
if (
old["可借車輛"] != new["可借車輛"]
or old["可還空位"] != new["可還空位"]
):
changed.append({
"站點編號": station_id,
"舊可借": old["可借車輛"],
"新可借": new["可借車輛"],
"舊可還": old["可還空位"],
"新可還": new["可還空位"],
})
return {"新增": added, "移除": removed, "車位變化": changed}站點「移除」也可能只是新快照缺資料，不一定代表實體站點撤除。解讀變化時要同時看品質報告與來源時間。
十二、產生行政區摘要
from collections import defaultdict
def summarize_by_area(stations):
areas = defaultdict(
lambda: {"站點數": 0, "可借車輛": 0, "可還空位": 0}
)
for station in stations:
area = areas[station["行政區"]]
area["站點數"] += 1
area["可借車輛"] += station["可借車輛"]
area["可還空位"] += station["可還空位"]
return dict(sorted(areas.items()))摘要反映快照內有效資料，不應直接推論長期交通需求。若要進行趨勢分析，必須有規律、合法且時間間隔一致的歷史資料。
十三、常見錯誤檢查表
| 現象 | 原因 | 修正 |
|---|---|---|
| 用eval解析JSON | 把資料誤當程式碼 | 只使用json.loads或json.load |
| 缺值被算成0 | 混淆未知與真實零值 | 保留None或列為錯誤 |
| 欄位改名後全數失敗 | 未管理Schema與別名 | 記錄版本，有限度支援已知別名 |
| 快照被當成即時資料 | 未顯示擷取時間 | 保存來源與UTC時間 |
| 站點消失就判斷撤站 | 忽略資料缺漏 | 先檢查品質報告及後續快照 |
十四、USR實作挑戰
挑戰A｜故障資料集
建立包含null、空字串、負數、重複編號與衝突別名的資料，確認品質報告能逐項揭露。
建立包含null、空字串、負數、重複編號與衝突別名的資料，確認品質報告能逐項揭露。
挑戰B｜兩個時間點
保存兩份教學快照，比較新增、缺少及車位變化；報告中同時顯示兩次擷取時間與錯誤筆數。
保存兩份教學快照，比較新增、缺少及車位變化；報告中同時顯示兩次擷取時間與錯誤筆數。
挑戰C｜轉用養殖API
把Schema改為場域、感測器、數值、單位及時間，明確區分0、None、缺欄與過期資料。
把Schema改為場域、感測器、數值、單位及時間，明確區分0、None、缺欄與過期資料。
十五、與生成式AI協作
請擔任Python JSON與資料品質助教。
請檢查我的YouBike快照程式：
1. 禁止eval，只使用json模組；
2. 明確區分0、None、空字串與缺少欄位；
3. 支援已知欄位別名，但衝突時必須報錯；
4. 保存Schema版本、來源URL與UTC擷取時間；
5. 產生有效、錯誤、重複與完整率報告；
6. 比較快照時先確認站點編號唯一；
7. 不把單次快照推論為長期趨勢。
請提供正常與故障測試資料。本篇小結
JSON解析只回答「文字能否轉成物件」，資料驗證才回答「這些物件能否被正確使用」。加入Schema、來源、時間、品質與版本，才能讓開放資料成為可追溯的分析材料。
JSON解析只回答「文字能否轉成物件」，資料驗證才回答「這些物件能否被正確使用」。加入Schema、來源、時間、品質與版本，才能讓開放資料成為可追溯的分析材料。
沒有留言:
張貼留言