文章目录环境信息前言5700个数据集摆在那但你只会cpp一个URL一、框架设计四个组件解决所有痛点组件1Session管理器连接池指数退避组件2RateLimiter按域名限速组件3TTL缓存组件4多格式响应解析二、组装框架HKOpenDataClient三、实战一城巴实时到站——每60秒刷新的JSON四、实战二差饷署租金指数——30年CSV数据五、实战三医管局急症室轮候——JSON格式的实时医疗数据六、框架总览七、三条使用铁律环境信息项目版本/说明Python3.10requests2.31数据源data.gov.hk香港政府公开数据平台演示API城巴实时到站 / 差饷署租金指数 / 医管局急症室轮候认证三个API全部免注册、免Key前言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、开箱即用。一、框架设计四个组件解决所有痛点组件1Session管理器连接池指数退避importrequestsfromrequests.adaptersimportHTTPAdapterfromurllib3.util.retryimportRetryclassAPISession:def__init__(self):self.sessionrequests.Session()retryRetry(total3,backoff_factor1,status_forcelist[429,500,502,503,504])self.session.mount(https://,HTTPAdapter(max_retriesretry,pool_connections5))self.session.headers.update({User-Agent:HKOpenData-Framework/1.0,Accept:application/json})defget(self,url,timeout15,**kw):returnself.session.get(url,timeouttimeout,**kw)组件2RateLimiter按域名限速importtimefromurllib.parseimporturlparsefromcollectionsimportdefaultdictclassRateLimiter:def__init__(self,default_interval1.0):self._lastdefaultdict(float)self._defaultdefault_intervaldefwait(self,url):domainurlparse(url).netloc elapsedtime.time()-self._last[domain]ifelapsedself._default:time.sleep(self._default-elapsed)self._last[domain]time.time()组件3TTL缓存importhashlib,jsonclassAPICache:def__init__(self,ttl30):self._store{}self._ttlttldef_key(self,url,paramsNone):rawurljson.dumps(paramsor{},sort_keysTrue)returnhashlib.md5(raw.encode()).hexdigest()defget(self,url,paramsNone):kself._key(url,params)ifkinself._storeandtime.time()-self._store[k][0]self._ttl:returnself._store[k][1]returnNonedefset(self,url,data,paramsNone):self._store[self._key(url,params)](time.time(),data)组件4多格式响应解析城巴API返回JSON、租金指数是CSV、急症室轮候也是JSON但字段结构完全不同——解析器需要统一处理importcsv,ioclassResponseParser:staticmethoddefparse(resp,expect_formatjson):ifresp.status_code!200:raiseAPIError(fHTTP{resp.status_code}:{resp.url})ifexpect_formatjson:try:dataresp.json()exceptExceptionase:raiseAPIError(fJSON parse failed:{e})ifisinstance(data,list)andlen(data)0:returnNone# 空数组 无数据returndataelifexpect_formatcsv:resp.encodingutf-8returnlist(csv.DictReader(io.StringIO(resp.text)))returnresp.textclassAPIError(Exception):pass二、组装框架HKOpenDataClientclassHKOpenDataClient:def__init__(self,cache_ttl30,rate_interval1.0):self.sessionAPISession()self.cacheAPICache(ttlcache_ttl)self.limiterRateLimiter(default_intervalrate_interval)deffetch(self,url,paramsNone,fmtjson,use_cacheTrue):ifuse_cache:cachedself.cache.get(url,params)ifcachedisnotNone:returncached self.limiter.wait(url)respself.session.get(url,paramsparams)dataResponseParser.parse(resp,fmt)ifdataisnotNone:self.cache.set(url,data,params)returndata# 一行初始化clientHKOpenDataClient()三、实战一城巴实时到站——每60秒刷新的JSONdata.gov.hk上的城巴ETA API提供5个JSON端点公司列表、路线列表、车站列表、指定路线车站、指定车站的ETA。调用链是先拿到路线和车站ID再查ETACTB_BASEhttps://rt.data.gov.hk/v1/transport/citybus-nwfb/eta/ctbdefget_ctb_routes(client):获取城巴所有路线returnclient.fetch(f{CTB_BASE}/route)defget_ctb_stops(client,route_id):获取指定路线的所有车站returnclient.fetch(f{CTB_BASE}/stop/{route_id})defget_ctb_eta(client,stop_id,route_id):获取指定车站路线的实时到站returnclient.fetch(f{CTB_BASE}/{stop_id}/{route_id})# 使用示例查 962 路屯门↔铜锣湾在上环的到站时间routesget_ctb_routes(client)route_962next((rforrinroutes[data]ifr[route]962),None)ifroute_962:stopsget_ctb_stops(client,962)# 取第一个站的ETAetaget_ctb_eta(client,stops[data][0][stop],962)forbusineta.get(data,[])[:3]:print(f路线{bus[route]}|{bus[eta]}|{bus[dest_tc]})踩坑记录城巴API的eta字段是一个字符串格式的ISO时间戳2026-08-11T14:20:0008:00不是Unix timestamp。我第一次直接当数字用Python直接炸了。后来改成用datetime.fromisoformat()解析再算差值才是正确的还剩几分钟。收藏提示②香港巴士API的端点设计都是先查元数据再查实时数据的两级模式。MTR、KMB、Citybus都一样——不要试图跳过第一步直接用magic ID元数据会变。四、实战二差饷署租金指数——30年CSV数据差饷物业估价署RVD的私人住宅租金指数——从1993年至今的月度数据CSV文件直接暴露在公网上连API调用都不算就是requests.get()一个CSV文件RVD_RENTAL_URLhttp://www.rvd.gov.hk/datagovhk/1.3M.csvdeffetch_rental_index(client):获取全港私人住宅租金指数1993年起dataclient.fetch(RVD_RENTAL_URL,fmtcsv)records[]forrowindata: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):continuereturnrecords rentalsfetch_rental_index(client)# 最近12个月的趋势recent[rforrinrentalsifr[year]2026]forrinrecent:print(f2026-{r[month]:02d}| A类(40㎡):{r[index_a]:.1f}| 全港:{r[index_all]:.1f})# 30年涨幅计算startrentals[0]endrentals[-1]increase(end[index_all]-start[index_all])/start[index_all]*100print(f1993→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_URLhttps://www.ha.org.hk/opendata/aed/aedwtdata-en.jsondeffetch_aed_waiting(client):获取全港急症室实时轮候时间dataclient.fetch(HA_AED_URL,fmtjson,use_cacheFalse)# 急症室数据必须实时ifnotdataorwaitingTimenotindata:return[]results[]forentryindata[waitingTime]:hospentry.get(hospName,)top_waitentry.get(topWait,).replace(Around ,).replace(Over ,)results.append({hospital:hosp,top_wait:top_wait})returnsorted(results,keylambdax:x[top_wait].replace(,).replace( hours,).replace( hour,),reverseTrue)waitingfetch_aed_waiting(client)print(全港急症室轮候时间实时:)fori,winenumerate(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行importrequests,time,json,hashlib,csv,iofromurllib.parseimporturlparsefromcollectionsimportdefaultdictfromrequests.adaptersimportHTTPAdapterfromurllib3.util.retryimportRetryclassHKOpenDataClient:def__init__(self,cache_ttl30,rate_interval1.0):self.srequests.Session()self.s.mount(https://,HTTPAdapter(max_retriesRetry(total3,backoff_factor1,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_intervaldeffetch(self,url,paramsNone,fmtjson,use_cacheTrue):khashlib.md5((urljson.dumps(paramsor{}))).hexdigest()ifuse_cacheandkinself._cacheandtime.time()-self._cache[k][0]self._ttl:returnself._cache[k][1]domainurlparse(url).netloc elapsedtime.time()-self._last[domain]ifelapsedself._rate:time.sleep(self._rate-elapsed)self._last[domain]time.time()respself.s.get(url,paramsparams,timeout15)ifresp.status_code!200:returnNoneiffmtcsv:resp.encodingutf-8datalist(csv.DictReader(io.StringIO(resp.text)))else:dataresp.json()ifdata:self._cache[k](time.time(),data)returndata七、三条使用铁律尊重刷新周期。城巴每1分钟、急症室每15分钟、租金每月——设缓存TTL不要短于刷新周期的一半。否则你拿到的只是重复数据服务器还会记恨你。先查元数据再查实时数据。城巴/九巴/港铁的API都是两级结构——先拿路线列表、再拿车站列表、最后查ETA。跳过前两步直接写死ID可能在你的机器上跑得通但换个车站就崩。空数据≠错误。香港政府API在极端情况巴士末班车已过、急症室无等待数据会返回空数组[]而不是404。你的代码必须兼容两种响应——有数据正常处理空数组也要优雅跳过。数据来源香港政府公开数据平台(data.gov.hk) · 城巴有限公司 · 差饷物业估价署 · 医院管理局。三个演示API全部免注册、免Key、可直接调用。