2026年8月31日 星期一

Python一下:從網址取得YouBike與開放資料——requests安全入門

《Python一下:從風土資料到智慧生活》第 21 篇

從網址取得YouBike與開放資料——requests安全入門

學會向公開資料服務提出請求、看懂回應,並在網路、格式或欄位改變時安全停止。

CH12 網頁資料requestsYouBike JSON
學習目標
完成本篇後,你能使用 requests.get() 取得JSON,理解HTTP狀態碼、標頭與逾時,使用 raise_for_status() 和例外處理,驗證資料結構,建立節制的重試策略,並保存來源與擷取時間。

一、這次使用哪一份資料?

本篇採用臺北市政府交通局提供的「YouBike2.0臺北市公共自行車即時資訊」作為教學案例。政府資料開放平臺列出的資料格式為JSON,欄位包括站點名稱、行政區、更新時間、可借車輛數、可還空位數及經緯度等,授權方式為政府資料開放授權條款第1版。

即時資料會改變:資料值、端點、欄位名稱、更新頻率及服務條件都可能調整。正式系統應把網址設成設定值、定期查看官方說明頁,並為欄位缺漏與服務中斷準備安全流程。

二、requests不是標準函式庫

requests 是第三方套件。在Thonny或本機環境可安裝;Google Colab通常已具備,但仍應先確認。

python -m pip install requests

Python也提供標準函式庫 urllib.request;本篇先使用介面較直觀的requests,之後再比較兩者。

三、第一個GET請求

import requests

URL = (
    "https://tcgbusfs.blob.core.windows.net/"
    "dotapp/youbike/v2/youbike_immediate.json"
)

response = requests.get(URL, timeout=(3.05, 15))
response.raise_for_status()
data = response.json()

print(type(data))
print("資料筆數:", len(data))

timeout=(3.05, 15) 分別限制建立連線與等待資料的時間。依requests官方文件,若未明確設定timeout,請求預設不會自動逾時;timeout也不是整份下載的總秒數,而是網路操作等待限制。

四、回應物件裡有什麼?

print("狀態碼:", response.status_code)
print("內容類型:", response.headers.get("Content-Type"))
print("回應位元組:", len(response.content))
print("最終網址:", response.url)
狀態碼範圍一般意義程式行動
2xx請求成功仍要驗證格式與欄位
3xx重新導向確認最終網址是否符合預期
4xx請求、權限或資源問題不要盲目重試,先檢查設定
5xx伺服器暫時或內部問題可有限度重試並保留紀錄

五、完整處理網路與JSON錯誤

import requests


def fetch_json(url):
    try:
        response = requests.get(
            url,
            timeout=(3.05, 15),
            headers={"User-Agent": "NFU-Python-USR-Education/1.0"},
        )
        response.raise_for_status()
        return response.json()
    except requests.Timeout:
        print("資料服務逾時,請稍後再試。")
    except requests.HTTPError as error:
        print("HTTP狀態錯誤:", error)
    except requests.ConnectionError:
        print("無法連線,請檢查網路或資料來源。")
    except requests.JSONDecodeError:
        print("回應不是有效JSON,請確認服務格式。")
    except requests.RequestException as error:
        print("其他請求錯誤:", error)
    return None

不要用空白 except: 隱藏問題。錯誤發生時也不要把上一次資料假裝成最新資料;若使用快取,畫面須清楚標示擷取時間與資料可能過期。

六、先確認最外層資料型別

def validate_dataset(data):
    if not isinstance(data, list):
        raise TypeError("預期JSON最外層是串列")
    if not data:
        raise ValueError("資料集是空串列")
    if not all(isinstance(item, dict) for item in data):
        raise TypeError("每一筆站點資料必須是字典")
    return data

response.json() 成功只表示回應可解析為JSON,不代表內容一定符合教學程式預期。

七、欄位名稱也要驗證

REQUIRED_FIELDS = {
    "sno",
    "sna",
    "sarea",
    "available_rent_bikes",
    "available_return_bikes",
    "updateTime",
}


def parse_station(item):
    missing = REQUIRED_FIELDS - item.keys()
    if missing:
        names = "、".join(sorted(missing))
        raise KeyError(f"缺少欄位:{names}")

    return {
        "站點編號": str(item["sno"]),
        "站點名稱": str(item["sna"]).strip(),
        "行政區": str(item["sarea"]).strip(),
        "可借車輛": int(item["available_rent_bikes"]),
        "可還空位": int(item["available_return_bikes"]),
        "更新時間": str(item["updateTime"]),
    }

八、單筆錯誤不拖垮整批資料

def parse_stations(data):
    valid = []
    errors = []

    for index, item in enumerate(data, start=1):
        try:
            station = parse_station(item)
            if station["可借車輛"] < 0:
                raise ValueError("可借車輛不可為負數")
            if station["可還空位"] < 0:
                raise ValueError("可還空位不可為負數")
            valid.append(station)
        except (KeyError, TypeError, ValueError) as error:
            errors.append({"資料序號": index, "原因": str(error)})

    return valid, errors

九、篩選可借車站點

def find_available_stations(stations, area=None, minimum=1):
    result = []
    for station in stations:
        if area is not None and station["行政區"] != area:
            continue
        if station["可借車輛"] >= minimum:
            result.append(station)

    return sorted(
        result,
        key=lambda item: item["可借車輛"],
        reverse=True,
    )

這只反映資料擷取當下的回應,不保證使用者抵達時仍有車。導覽或交通建議應顯示更新時間,避免把即時資料誤寫成長期事實。

十、把來源與時間一起保存

import json
from datetime import datetime, timezone
from pathlib import Path


def save_snapshot(path, source_url, stations, errors):
    snapshot = {
        "source_url": source_url,
        "fetched_at_utc": datetime.now(
            timezone.utc
        ).isoformat(),
        "valid_count": len(stations),
        "error_count": len(errors),
        "stations": stations,
        "errors": errors,
    }

    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

使用 x 避免覆蓋快照。檔案名稱可加入時間戳;日後分析時才知道資料是哪一刻取得、有效與錯誤各多少筆。

十一、有限度重試GET請求

GET通常是冪等操作,可對連線錯誤、429或部分5xx狀態有限度重試;不要無限迴圈,也不要對所有4xx重試。

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


def build_session():
    retry = Retry(
        total=3,
        connect=3,
        read=2,
        backoff_factor=0.5,
        status_forcelist=(429, 500, 502, 503, 504),
        allowed_methods=frozenset({"GET"}),
        respect_retry_after_header=True,
    )
    adapter = HTTPAdapter(max_retries=retry)
    session = requests.Session()
    session.mount("https://", adapter)
    return session

urllib3官方文件說明,是否針對特定狀態碼重試由 status_forcelist 與允許方法共同決定。若服務提供 Retry-After,應尊重伺服器要求。

十二、限制下載大小

來源即使可信也可能異常回傳超大內容。串流下載時可設定上限:

import requests


def download_limited(url, max_bytes=10_000_000):
    chunks = []
    total = 0

    with requests.get(
        url, timeout=(3.05, 15), stream=True
    ) as response:
        response.raise_for_status()
        for chunk in response.iter_content(chunk_size=64 * 1024):
            if not chunk:
                continue
            total += len(chunk)
            if total > max_bytes:
                raise ValueError("回應超過允許大小")
            chunks.append(chunk)

    return b"".join(chunks)

十三、整合主程式

def main():
    data = fetch_json(URL)
    if data is None:
        return

    try:
        validate_dataset(data)
    except (TypeError, ValueError) as error:
        print("資料集結構錯誤:", error)
        return

    stations, errors = parse_stations(data)
    print("有效站點:", len(stations))
    print("錯誤資料:", len(errors))

    available = find_available_stations(
        stations, minimum=5
    )
    for station in available[:10]:
        print(
            station["站點名稱"],
            station["可借車輛"],
            station["更新時間"],
        )


if __name__ == "__main__":
    main()

十四、來源網址不可任由使用者輸入

伺服器端程式若直接下載使用者提供的任意網址,可能造成SSRF,連到內部管理服務或雲端中繼資料。教學程式應使用固定官方端點或嚴格允許清單;不要關閉TLS憑證驗證,也不要在網址或日誌中放API金鑰。

十五、常見錯誤檢查表

現象原因修正
程式一直等待沒有設定timeout明確設定連線與讀取逾時
404頁面被當JSON未先檢查HTTP狀態呼叫raise_for_status()
json()成功後仍KeyError欄位改變或缺漏驗證最外層及必要欄位
資料服務壓力過大頻繁輪詢或無限重試快取、退避、有限次數並遵守條款
舊資料看似即時未保存或顯示擷取時間一併保存來源與UTC時間

十六、USR實作挑戰

挑戰A|YouBike資料品質
下載一次官方快照,統計有效筆數、錯誤筆數、行政區數量與更新時間範圍,不進行高頻輪詢。
挑戰B|離線教學
保存一份符合授權的教學快照,讓沒有網路時仍可練習解析;畫面清楚標示「非即時資料」。
挑戰C|移植水井場域
把站點欄位驗證概念改寫為智慧養殖感測API,設計來源、時間、有效筆數、錯誤筆數與過期警示。

十七、與生成式AI協作

請擔任Python requests與開放資料安全助教。
請檢查我的JSON下載程式:
1. 使用固定HTTPS官方端點;
2. 必須設定連線與讀取timeout;
3. 先raise_for_status()再解析JSON;
4. 驗證最外層型別及必要欄位;
5. 單筆錯誤不能拖垮整批;
6. 重試只限GET及特定暫時性錯誤;
7. 保存來源、擷取時間與資料品質;
8. 不關閉TLS驗證,不接受任意使用者網址。
請提供測試案例與失敗時的安全行為。
本篇小結
下載JSON只需幾行,可靠使用卻需要逾時、狀態碼、格式、欄位、重試、大小與來源治理。網路資料永遠是不穩定的外部輸入,先驗證再使用。

沒有留言:

張貼留言