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

适用场景

实时公交到站数据是出行场景的基础组件，常见于以下应用：

• 公交电子站牌：动态显示下趟车到站时间，替代传统静态时刻表；
• 出行助手 App：在路线规划中嵌入具体车次到达预估，让用户掌握候车时间；
• 企业园区通勤系统：查询内部通勤线路当前位置与到站倒计时；
• 智能家居场景：语音查询“下一班 401 路多久到五一广场”。

无论哪种场景，核心流程都是：传入城市与站名 → 获取该站经过的所有线路及每线路即将到站车辆的信息。

接口能力边界

在使用之前需要了解接口的客观约束，避免在设计系统时产生不可行的预期。

• 覆盖范围：支持全国数百个城市的公交数据（具体城市列表以文档为准）。
• 方向支持：可通过direction参数指定查询方向（1=默认正向，2=反向），满足双向候车需求。
• 数据实时性：数据来自公共交通运营方的实时推送或轮询，接口响应中包含updated_at用于判断数据新鲜度。
• 请求限制：QPS 为 10（每秒最多 10 次请求），超出限制将收到 HTTP 429 状态码。
• 协议与格式：仅支持 HTTPS，请求体与响应体均为 JSON。

注意：接口并不提供历史行车轨迹或全路网车辆位置，只返回指定车站的到站预估信息。

请求参数与鉴权

请求地址

POST https://v1.apizero.cn/api/bus-realtime

请求头

请求体（JSON）

示例请求体：

{ "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, json=payload, timeout=self.timeout ) resp.raise_for_status() # 触发 HTTP 错误 data = resp.json() # 业务错误检查 if data.get("code") != 0: raise ValueError(f"API 返回业务错误: {data.get('msg', 'unknown')}") return data

使用示例

# 通过环境变量加载 API Key client = BusRealtimeClient() try: result = client.query(city="长沙", station="五一广场", direction=1) 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 智能提示。可以使用dataclass：

from 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:34，24小时制）。
• lines[].buses[].arrival_timestamp：到站时间的 Unix 毫秒时间戳，用于后端计算倒计时。
• lines[].buses[].status：状态描述，例如 "5站" 表示距离本站还有5站。
• lines[].buses[].stops_remaining：剩余站数（整型）。
• lines[].buses[].travel_minutes：预计还需多少分钟到达本站。

注意：status字段的格式可能随城市不同而变化（如“即将进站”“已过站”），后续工程化处理时建议以travel_minutes和stops_remaining为主要数值依据。

常见错误与调试

调试技巧

1. 开启请求日志：在 curl 中加-v查看完整请求头与握手信息。
2. 检查响应头：X-RateLimit-Remaining和X-RateLimit-Reset（如果存在）可用于跟踪配额。
3. 使用公共测试城市：建议先用“长沙”“北京”等大城市测试，覆盖率高。

工程化注意事项

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(maxsize=128, ttl=30) 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 即可。