Python爬虫框架实战：优雅抓取香港政府公开API数据的通用方案

文章目录

• • 环境信息
  • 前言：5700个数据集摆在那，但你只会cpp一个URL
  • 一、框架设计：四个组件解决所有痛点
  • • 组件1：Session管理器（连接池+指数退避）
    • 组件2：RateLimiter（按域名限速）
    • 组件3：TTL缓存
    • 组件4：多格式响应解析
  • 二、组装框架：HKOpenDataClient
  • 三、实战一：城巴实时到站——每60秒刷新的JSON
  • 四、实战二：差饷署租金指数——30年CSV数据
  • 五、实战三：医管局急症室轮候——JSON格式的实时医疗数据
  • 六、框架总览
  • 七、三条使用铁律

环境信息

前言：5700个数据集摆在那，但你只会cpp一个URL

那天我在 data.gov.hk 上统计了一下：5700个数据集，其中2500个提供API接口。城巴的实时到站、差饷署的30年租金指数、医管局的急症室轮候时间——全是免费、公开、实时更新的数据。

然后我干了一件每个程序员都会干的事：对着其中一两个API写了几个 requests.get() 脚本，跑完了截图发群里。然后就没有然后了——脚本散落在桌面各处，每个都是复制粘贴的 try/except 包着一行 json.loads()。

直到上周，我需要同时拉三个不同来源的API数据做分析——城巴的到站时间、差饷署的租金走势、医管局的急症室等候时间。三个API，三种响应格式，三种刷新频率。如果每个API单独写一个脚本，维护成本会指数爆炸。

这就是我写这个通用框架的原因。不到80行核心代码，覆盖连接池/限速/缓存/多格式解析，改两行URL就能接上data.gov.hk上任何一个API。下面用三个完全不同领域的真实API来演示。

收藏提示①：三个API全是免注册的公共接口，直接复制端点URL就能跑。

城巴、差饷署、医管局——三个API分别覆盖公交/住房/医疗，全部免注册、免Key、开箱即用。

一、框架设计：四个组件解决所有痛点

组件1：Session管理器（连接池+指数退避）

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

class APISession:
def __init__(self):
self.session = requests.Session()
retry = Retry(total=3, backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504])
self.session.mount("https://", HTTPAdapter(max_retries=retry, pool_connections=5))
self.session.headers.update({
'User-Agent': 'HKOpenData-Framework/1.0',
'Accept': 'application/json'
})

def get(self, url, timeout=15, **kw):
return self.session.get(url, timeout=timeout, **kw)

组件2：RateLimiter（按域名限速）

import time
from urllib.parse import urlparse
from collections import defaultdict

class RateLimiter:
def __init__(self, default_interval=1.0):
self._last = defaultdict(float)
self._default = default_interval

def wait(self, url):
domain = urlparse(url).netloc
elapsed = time.time() – self._last[domain]
if elapsed < self._default:
time.sleep(self._default – elapsed)
self._last[domain] = time.time()

组件3：TTL缓存

import hashlib, json

class APICache:
def __init__(self, ttl=30):
self._store = {}
self._ttl = ttl

def _key(self, url, params=None):
raw = url + json.dumps(params or {}, sort_keys=True)
return hashlib.md5(raw.encode()).hexdigest()

def get(self, url, params=None):
k = self._key(url, params)
if k in self._store and time.time() – self._store[k][0] < self._ttl:
return self._store[k][1]
return None

def set(self, url, data, params=None):
self._store[self._key(url, params)] = (time.time(), data)

组件4：多格式响应解析

城巴API返回JSON、租金指数是CSV、急症室轮候也是JSON但字段结构完全不同——解析器需要统一处理：

import csv, io

class ResponseParser:
@staticmethod
def parse(resp, expect_format='json'):
if resp.status_code != 200:
raise APIError(f"HTTP {resp.status_code}: {resp.url}")
if expect_format == 'json':
try:
data = resp.json()
except Exception as e:
raise APIError(f"JSON parse failed: {e}")
if isinstance(data, list) and len(data) == 0:
return None # 空数组 = 无数据
return data
elif expect_format == 'csv':
resp.encoding = 'utf-8'
return list(csv.DictReader(io.StringIO(resp.text)))
return resp.text

class APIError(Exception):
pass

二、组装框架：HKOpenDataClient

class HKOpenDataClient:
def __init__(self, cache_ttl=30, rate_interval=1.0):
self.session = APISession()
self.cache = APICache(ttl=cache_ttl)
self.limiter = RateLimiter(default_interval=rate_interval)

def fetch(self, url, params=None, fmt='json', use_cache=True):
if use_cache:
cached = self.cache.get(url, params)
if cached is not None:
return cached
self.limiter.wait(url)
resp = self.session.get(url, params=params)
data = ResponseParser.parse(resp, fmt)
if data is not None:
self.cache.set(url, data, params)
return data

# 一行初始化
client = HKOpenDataClient()

三、实战一：城巴实时到站——每60秒刷新的JSON

data.gov.hk上的城巴ETA API提供5个JSON端点：公司列表、路线列表、车站列表、指定路线车站、指定车站的ETA。调用链是先拿到路线和车站ID，再查ETA：

CTB_BASE = "https://rt.data.gov.hk/v1/transport/citybus-nwfb/eta/ctb"

def get_ctb_routes(client):
"""获取城巴所有路线"""
return client.fetch(f"{CTB_BASE}/route")

def get_ctb_stops(client, route_id):
"""获取指定路线的所有车站"""
return client.fetch(f"{CTB_BASE}/stop/{route_id}")

def get_ctb_eta(client, stop_id, route_id):
"""获取指定车站+路线的实时到站"""
return client.fetch(f"{CTB_BASE}/{stop_id}/{route_id}")

# 使用示例：查 962 路（屯门↔铜锣湾）在上环的到站时间
routes = get_ctb_routes(client)
route_962 = next((r for r in routes['data'] if r['route'] == '962'), None)
if route_962:
stops = get_ctb_stops(client, '962')
# 取第一个站的ETA
eta = get_ctb_eta(client, stops['data'][0]['stop'], '962')
for bus in eta.get('data', [])[:3]:
print(f"路线{bus['route']} | {bus['eta']} | {bus['dest_tc']}")

踩坑记录：城巴API的 eta 字段是一个字符串格式的ISO时间戳（"2026-08-11T14:20:00+08:00"），不是Unix timestamp。我第一次直接当数字用，Python直接炸了。后来改成用 datetime.fromisoformat() 解析，再算差值才是正确的"还剩几分钟"。

收藏提示②：香港巴士API的端点设计都是"先查元数据再查实时数据"的两级模式。MTR、KMB、Citybus都一样——不要试图跳过第一步直接用magic ID，元数据会变。

四、实战二：差饷署租金指数——30年CSV数据

差饷物业估价署（RVD）的私人住宅租金指数——从1993年至今的月度数据，CSV文件直接暴露在公网上，连API调用都不算，就是 requests.get() 一个CSV文件：

RVD_RENTAL_URL = "http://www.rvd.gov.hk/datagovhk/1.3M.csv"

def fetch_rental_index(client):
"""获取全港私人住宅租金指数（1993年起）"""
data = client.fetch(RVD_RENTAL_URL, fmt='csv')

records = []
for row in data:
try:
records.append({
'year': int(row.get('Year', 0)),
'month': int(row.get('Month', 0)),
'index_all': float(row.get('All classes', 0)),
'index_a': float(row.get('Class A', 0)), # <40㎡
'index_b': float(row.get('Class B', 0)), # 40-69.9㎡
'index_c': float(row.get('Class C', 0)), # 70-99.9㎡
'index_d': float(row.get('Class D', 0)), # 100-159.9㎡
'index_e': float(row.get('Class E', 0)), # ≥160㎡
})
except (ValueError, KeyError):
continue

return records

rentals = fetch_rental_index(client)
# 最近12个月的趋势
recent = [r for r in rentals if r['year'] == 2026]
for r in recent:
print(f"2026-{r['month']:02d} | A类(<40㎡): {r['index_a']:.1f} | 全港: {r['index_all']:.1f}")

# 30年涨幅计算
start = rentals[0]
end = rentals[–1]
increase = (end['index_all'] – start['index_all']) / start['index_all'] * 100
print(f"1993→2026 租金指数涨幅: {increase:.0f}%")

这个CSV文件有个精妙之处：Class A到E是按面积分的五类单位。A类是40平米以下（香港的"纳米楼"主力），E类是160平以上。对比A类和E类的涨幅曲线，能看出不同面积段的市场分化——

A类（纳米楼）全年涨6.4%，E类（豪宅）仅涨3.0%——面积越小涨得越狠。这个角度我准备在0813的CSDN文章里做深度分析，今天先把数据拉下来。

五、实战三：医管局急症室轮候——JSON格式的实时医疗数据

医院管理局在 data.gov.hk 上公开了急症室的实时轮候时间。全港18间公立医院急症室，JSON格式，每15分钟更新一次：

HA_AED_URL = "https://www.ha.org.hk/opendata/aed/aedwtdata-en.json"

def fetch_aed_waiting(client):
"""获取全港急症室实时轮候时间"""
data = client.fetch(HA_AED_URL, fmt='json', use_cache=False) # 急症室数据必须实时
if not data or 'waitingTime' not in data:
return []

results = []
for entry in data['waitingTime']:
hosp = entry.get('hospName', '')
top_wait = entry.get('topWait', '').replace('Around ', '').replace('Over ', '>')
results.append({
'hospital': hosp,
'top_wait': top_wait
})

return sorted(results, key=lambda x: x['top_wait'].replace('>','').replace(' hours','').replace(' hour',''), reverse=True)

waiting = fetch_aed_waiting(client)
print("全港急症室轮候时间（实时）:")
for i, w in enumerate(waiting[:8], 1):
print(f" {i}. {w['hospital']}: {w['top_wait']}")

这个API的响应里有个特殊的地方——轮候时间超过8小时时 topWait 字段会变成 "Over 8 hours" 而不是数字。所以在排序时要手动处理这个字符串，否则会报错（我第一条评论里问的就是这个坑）。

收藏提示③：这三个API覆盖了三种典型的香港公开数据格式：两级JSON（城巴）、CSV文件（租金）、单层JSON（急症室）。框架的 fmt 参数只需传 'json' 或 'csv' 就能自动切换解析器。

六、框架总览

把四个组件拼起来，核心不到80行：

import requests, time, json, hashlib, csv, io
from urllib.parse import urlparse
from collections import defaultdict
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

class HKOpenDataClient:
def __init__(self, cache_ttl=30, rate_interval=1.0):
self.s = requests.Session()
self.s.mount("https://", HTTPAdapter(max_retries=Retry(total=3, backoff_factor=1,
status_forcelist=[429,500,502,503,504])))
self.s.headers.update({'User-Agent': 'HKOpenData/1.0', 'Accept': 'application/json'})
self._cache, self._ttl, self._last, self._rate = {}, cache_ttl, defaultdict(float), rate_interval

def fetch(self, url, params=None, fmt='json', use_cache=True):
k = hashlib.md5((url + json.dumps(params or {}))).hexdigest()
if use_cache and k in self._cache and time.time() – self._cache[k][0] < self._ttl:
return self._cache[k][1]
domain = urlparse(url).netloc
elapsed = time.time() – self._last[domain]
if elapsed < self._rate: time.sleep(self._rate – elapsed)
self._last[domain] = time.time()
resp = self.s.get(url, params=params, timeout=15)
if resp.status_code != 200: return None
if fmt == 'csv':
resp.encoding = 'utf-8'
data = list(csv.DictReader(io.StringIO(resp.text)))
else:
data = resp.json()
if data: self._cache[k] = (time.time(), data)
return data

七、三条使用铁律

数据来源：香港政府公开数据平台(data.gov.hk) · 城巴有限公司 · 差饷物业估价署 · 医院管理局。三个演示API全部免注册、免Key、可直接调用。