从 curl 到工程封装：实时公交到站接口集成实践

适用场景实时公交到站数据是出行场景的基础组件常见于以下应用公交电子站牌动态显示下趟车到站时间替代传统静态时刻表出行助手 App在路线规划中嵌入具体车次到达预估让用户掌握候车时间企业园区通勤系统查询内部通勤线路当前位置与到站倒计时智能家居场景语音查询“下一班 401 路多久到五一广场”。无论哪种场景核心流程都是传入城市与站名 → 获取该站经过的所有线路及每线路即将到站车辆的信息。接口能力边界在使用之前需要了解接口的客观约束避免在设计系统时产生不可行的预期。覆盖范围支持全国数百个城市的公交数据具体城市列表以文档为准。方向支持可通过direction参数指定查询方向1默认正向2反向满足双向候车需求。数据实时性数据来自公共交通运营方的实时推送或轮询接口响应中包含updated_at用于判断数据新鲜度。请求限制QPS 为 10每秒最多 10 次请求超出限制将收到 HTTP 429 状态码。协议与格式仅支持 HTTPS请求体与响应体均为 JSON。注意接口并不提供历史行车轨迹或全路网车辆位置只返回指定车站的到站预估信息。请求参数与鉴权请求地址POST https://v1.apizero.cn/api/bus-realtime请求头参数名必需类型说明Content-Type是string固定为application/jsonX-API-Key是stringAPI 密钥通过开发者控制台获取请求体JSON字段必需类型说明示例city是string城市名支持中文长沙station是string站点名或关键词兼容别名line五一广场direction否number方向1默认2反方向1示例请求体{ city: 长沙, station: 五一广场, direction: 1 }curl 直接调用curl 是最直接的接口调试方式。以下示例假设你已经将 API Key 保存在环境变量APIZERO_API_KEY中curl -sS \ -X POST \ -H X-API-Key: $APIZERO_API_KEY \ -H Content-Type: application/json \ -d {city: 长沙, station: 五一广场, direction: 1} \ https://v1.apizero.cn/api/bus-realtime | jq .参数说明-sS静默模式避免输出进度信息保留错误输出-X POST明确指定请求方法| jq .对输出结果做 JSON 格式化需安装jq。如果不需要管道美化去掉| jq .即可直接查看原始 JSON 响应。工程化封装以 Python 为例直接使用 curl 适合临时测试在工程项目中通常需要封装成可复用函数统一管理 API Key、错误处理和超时。以下是一个完整的 Python 封装示例。环境准备pip install requests封装类import os import time import requests from typing import Optional, Dict, Any class BusRealtimeClient: 实时公交到站查询客户端 BASE_URL https://v1.apizero.cn/api/bus-realtime def __init__(self, api_key: Optional[str] None, timeout: int 5): self.api_key api_key or os.environ[APIZERO_API_KEY] self.timeout timeout self.session requests.Session() self.session.headers.update({ X-API-Key: self.api_key, Content-Type: application/json }) def query(self, city: str, station: str, direction: int 1) - Dict[str, Any]: 查询指定车站的到站信息 Args: city: 城市名 station: 站名 direction: 方向1默认2反方向 Returns: dict: 响应 JSON Raises: requests.RequestException: 网络或鉴权失败 ValueError: 参数不合法或服务端返回错误 payload { city: city, station: station, direction: direction } resp self.session.post( self.BASE_URL, jsonpayload, timeoutself.timeout ) resp.raise_for_status() # 触发 HTTP 错误 data resp.json() # 业务错误检查 if data.get(code) ! 0: raise ValueError(fAPI 返回业务错误: {data.get(msg, unknown)}) return data使用示例# 通过环境变量加载 API Key client BusRealtimeClient() try: result client.query(city长沙, station五一广场, direction1) print(f查询成功共 {result[data][line_count]} 条线路) for line in result[data][lines]: print(f线路 {line[line]} 方向 {line[terminal]}票价 {line[price]} 元) for bus in line[buses]: print(f 车牌 {bus[bus_id]}剩余 {bus[stops_remaining]} 站预计 {bus[travel_minutes]} 分钟) except Exception as e: print(f查询失败: {e})响应数据模型建议在工程中进一步定义数据类便于类型检查和 IDE 智能提示。可以使用dataclassfrom dataclasses import dataclass, field from typing import List dataclass class BusInfo: bus_id: str arrival_time: str arrival_timestamp: int status: str stops_remaining: int travel_minutes: int dataclass class LineInfo: line: str price: str terminal: str bus_count: int buses: List[BusInfo] dataclass class BusRealtimeResponse: code: int msg: str request_id: str city: str station: str direction: int line_count: int lines: List[LineInfo] updated_at: str响应字段解读当code为 0 时data字段包含完整的公交到站信息。字段结构如下{ code: 0, msg: 成功, request_id: a1b2c3d4, data: { city: 长沙, station: 五一广场, direction: 1, line_count: 2, lines: [ { line: 401路, price: 2, terminal: 汽车西站, bus_count: 1, buses: [ { bus_id: 湘A02882D, arrival_time: 2026-07-01 12:34, arrival_timestamp: 1751344440000, status: 5站, stops_remaining: 5, travel_minutes: 6 } ] } ], updated_at: 2026-07-01 12:30:00 } }关键字段说明code/msg业务状态码。0 表示成功其他表示错误如参数缺失、城市不支持。request_id每次请求的唯一标识便于排查问题时定位。data.updated_at数据最新更新时间服务端缓存刷新的时刻。line_count该站通过的总线路数。lines[].line线路名称如 401路。lines[].price票价字符串格式“2”代表2元。lines[].terminal该线路终点站名。lines[].bus_count当前即将到站的车辆总数。lines[].buses[].bus_id车牌号。lines[].buses[].arrival_time预计到站时间形如2026-07-01 12:3424小时制。lines[].buses[].arrival_timestamp到站时间的 Unix 毫秒时间戳用于后端计算倒计时。lines[].buses[].status状态描述例如 5站 表示距离本站还有5站。lines[].buses[].stops_remaining剩余站数整型。lines[].buses[].travel_minutes预计还需多少分钟到达本站。注意status字段的格式可能随城市不同而变化如“即将进站”“已过站”后续工程化处理时建议以travel_minutes和stops_remaining为主要数值依据。常见错误与调试错误现象可能原因排查方式HTTP 401API Key 缺失或无效检查环境变量APIZERO_API_KEY是否正确设置确认 Key 在控制台未过期。HTTP 400请求参数格式错误或缺少必填字段确认city和station是否提供direction是否为数字。HTTP 429请求超过速率限制QPS 10增加本地限流如令牌桶等待1秒后重试。code ! 0业务层面错误如城市不支持、站点不存在检查msg字段内容确认城市名称是否完全匹配如“长沙”而非“长沙市”。网络超时服务端响应过慢或本地网络问题增加超时时间默认建议 5s检查是否在公司内网或需要代理。调试技巧开启请求日志在 curl 中加-v查看完整请求头与握手信息。检查响应头X-RateLimit-Remaining和X-RateLimit-Reset如果存在可用于跟踪配额。使用公共测试城市建议先用“长沙”“北京”等大城市测试覆盖率高。工程化注意事项1. 密钥管理绝不将 API Key 硬编码到代码仓库中。应通过环境变量、配置中心或密钥管理服务如 Vault注入。示例中的os.environ[APIZERO_API_KEY]是基础做法生产环境可考虑读取.env文件并加入.gitignore。2. 限流与重试接口 QPS 为 10单客户端应自我节流。可以在客户端中实现简单的速率限制from threading import Lock import time class RateLimiter: def __init__(self, max_per_second): self.max_per_second max_per_second self.lock Lock() self.last_called time.time() self.calls [] def acquire(self): with self.lock: now time.time() # 移除1秒前的记录 self.calls [t for t in self.calls if t now - 1] if len(self.calls) self.max_per_second: sleep_time self.calls[0] 1 - now if sleep_time 0: time.sleep(sleep_time) self.calls.append(time.time())对非业务错误如 HTTP 429、502实现指数退避重试最多3次。3. 缓存策略实时公交数据的有效窗口通常在 30-60 秒。如果同一城市的同一站点被频繁查询如轮询刷新建议在客户端层面做短期缓存import cachetools.func cachetools.func.ttl_cache(maxsize128, ttl30) def query_cached(city, station, direction): return client.query(city, station, direction)缓存 TTL 建议 15-30 秒既减少重复请求又不至于让用户看到明显过时的数据。别忘了清除缓存当用户手动“刷新”时直接绕过缓存调用原始请求。4. 监控与告警记录每次请求的延迟、状态码、错误类型到日志系统如 ELK。对业务错误城市不识别、站点不存在设置告警阈值可能意味着前端输入不合法或数据源变动。利用request_id在出问题时快速关联日志。5. 并发安全如果使用同一个BusRealtimeClient实例处理多个请求注意requests.Session是线程安全的但限流器需要加锁如上例。或者使用requests_futures异步发送但限流逻辑仍需同步控制。6. 环境差异开发/测试/生产环境使用不同的 API Key且通过环境变量区分。接口地址在测试阶段可以使用 Mock 服务如 WireMock进行模拟。参考文档官方文档首页https://apizero.cn/aidocs/bus-realtime原始 Markdown 文档https://apizero.cn/aidocs/bus-realtime/raw.md演示与调试可使用上述 curl 命令直接测试替换 API Key 即可。