环境信息

• Python 3.12.13
• macOS / Linux / Windows均可运行
• urllib.request、json、datetime、time为Python标准库，无需安装第三方依赖。
• 数据接口：香港资料一线通 Citybus实时“Next Bus”数据。

我让Codex写过一版“查询香港巴士到站时间”的脚本。第一眼看起来没有什么问题：能请求接口、能解析JSON、能打印ETA。

但真正把它放进定时任务前，我发现了三个不太显眼的坑：站点ID不能凭感觉填写，eta不是“还剩几分钟”，而且接口返回空ETA时不能当成0分钟。代码可以正常运行，不代表结果就可信。

这篇文章用城巴1号线做例子，从路线和方向开始，找到真实站点，再读取ETA。重点不是展示一段“AI生成代码”，而是把第一版代码逐字段对照官方文档，修到它至少知道什么时候应该停下来报警。

1. 先把问题拆成四个接口层

城巴的公开接口不是一个“传路线号就返回到站时间”的单层接口。比较稳妥的调用顺序是：

1. route：确认路线和起讫点；
2. route-stop：根据方向拿到该路线的站点序号和站点ID；
3. stop：补齐站点名称和经纬度；
4. eta：查询某个站点、某条路线的预计抵达时间。

站点ID是6位字符串，例如002403。它不是路线号，也不是可以自己递增出来的序号。路线方向在请求路径中使用inbound或outbound，返回数据里的dir则可能显示成I或O，这两个层次不要混在一起。

2. Codex第一版最容易写错的地方

我给Codex的任务限制在一个小范围内：只使用Python标准库，读取公开JSON接口，输入路线和方向，输出前3个ETA，并在接口失败时返回错误。

这比“帮我写一个完整的香港公交系统”更容易检查。第一版代码通常能完成请求，但人工检查时必须追问三件事：

检查项	不能直接相信的写法	正确处理
站点	把站点ID写死，却没有说明来源	先调用route-stop，从返回结果取站点
时间	用datetime.now()减去没有时区的字符串	用带时区的ISO 8601时间解析
空值	eta为空就显示“0分钟”	跳过该条记录，并保留备注字段

官方资料说明ETA数据每分钟更新。脚本若每秒请求一次，不会让预测更准确，只会增加无意义请求。这里把轮询间隔设为60秒，把静态路线和站点信息当成低频数据处理。

3. 先用route-stop找到真实站点

本文使用的接口基础地址是：

https://rt.data.gov.hk/v1/transport/citybus-nwfb

查询城巴1号线往入境方向的站点：

GET /route-stop/CTB/1/inbound

返回结果里的data是数组，每一项至少包含seq、stop、dir和route。代码不直接假设第一个站点永远不变，而是先排序，再从列表中选择目标站点。

4. 完整Python代码：读取ETA并检查新鲜度

from __future__ import annotations

from datetime import datetime, timezone
from http.client import HTTPException
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
import json
import sys
import time
from typing import Any


BASE_URL = "https://rt.data.gov.hk/v1/transport/citybus-nwfb"
COMPANY = "CTB"
ROUTE = "1"
DIRECTION = "inbound"
WATCH_SECONDS = 60
MAX_DATA_AGE_SECONDS = 180


def fetch_json(path: str, timeout: int = 10) -> dict[str, Any]:
    request = Request(
        BASE_URL + path,
        headers={"User-Agent": "hk-bus-eta-demo/1.0"},
    )
    try:
        with urlopen(request, timeout=timeout) as response:
            payload = json.load(response)
    except HTTPError as exc:
        body = exc.read().decode("utf-8", errors="replace")[:200]
        raise RuntimeError(f"HTTP {exc.code}: {body}") from exc
    except (URLError, TimeoutError, HTTPException) as exc:
        raise RuntimeError(f"request failed: {exc}") from exc

    if not isinstance(payload, dict):
        raise ValueError("API response is not a JSON object")
    return payload


def parse_time(value: str) -> datetime:
    # API返回的是带时区的ISO 8601时间；不要用naive datetime混算。
    return datetime.fromisoformat(value.replace("Z", "+00:00"))


def route_stops(route: str, direction: str) -> list[dict[str, Any]]:
    payload = fetch_json(f"/route-stop/{COMPANY}/{route}/{direction}")
    rows = payload.get("data")
    if not isinstance(rows, list) or not rows:
        raise ValueError("route-stop returned no stops")

    valid_rows = [row for row in rows if isinstance(row, dict) and row.get("stop")]
    return sorted(valid_rows, key=lambda row: int(row.get("seq", 0)))


def read_eta(route: str, stop_id: str) -> list[dict[str, Any]]:
    payload = fetch_json(f"/eta/{COMPANY}/{stop_id}/{route}")
    rows = payload.get("data")
    if not isinstance(rows, list):
        raise ValueError("eta returned an invalid data field")

    now = datetime.now(timezone.utc)
    result = []
    for row in rows:
        if not isinstance(row, dict):
            continue

        eta_text = (row.get("eta") or "").strip()
        if not eta_text:
            # 空ETA表示当前没有可用的预计时间，不等于马上到站。
            continue

        eta_at = parse_time(eta_text)
        minutes = max(0, round((eta_at - now).total_seconds() / 60))
        data_at = row.get("data_timestamp")
        age_seconds = None
        if data_at:
            age_seconds = max(0, int((now - parse_time(data_at)).total_seconds()))

        result.append(
            {
                "eta_seq": row.get("eta_seq"),
                "destination": row.get("dest_sc") or row.get("dest_tc"),
                "eta": eta_at.isoformat(),
                "minutes": minutes,
                "data_age_seconds": age_seconds,
                "remark": row.get("rmk_sc") or row.get("rmk_tc") or "",
            }
        )
    return result


def run_once() -> None:
    stops = route_stops(ROUTE, DIRECTION)
    target = stops[0]
    stop_id = str(target["stop"]).zfill(6)
    stop_info = fetch_json(f"/stop/{stop_id}").get("data") or {}
    rows = read_eta(ROUTE, stop_id)

    print(
        f"route={ROUTE} direction={DIRECTION} "
        f"stop={stop_id} name={stop_info.get('name_sc', '')}"
    )
    if not rows:
        print("ETA unavailable: no non-empty eta field returned")
        return

    for row in rows:
        age = row["data_age_seconds"]
        freshness = ""
        if age is not None and age > MAX_DATA_AGE_SECONDS:
            freshness = f" [STALE data_age={age}s]"
        print(
            f"#{row['eta_seq']} {row['destination']} "
            f"in about {row['minutes']} min, eta={row['eta']}"
            f"{freshness} {row['remark']}"
        )


def main() -> None:
    watch = "--watch" in sys.argv
    while True:
        try:
            run_once()
        except (OSError, ValueError, RuntimeError) as exc:
            print(f"ERROR: {exc}", file=sys.stderr)
        if not watch:
            return
        time.sleep(WATCH_SECONDS)


if __name__ == "__main__":
    main()

这段代码有意没有把“ETA为空”转换成一个数字，也没有用系统本地时间直接相减。因为香港接口返回的是带+08:00的时间戳，程序应该让时间对象保持时区信息，避免部署到其他时区的机器后出现偏差。

这段代码可以直接复用，收藏备用：换掉ROUTE和DIRECTION即可查询其他城巴路线；如果要固定某一站，可以把target = stops[0]改为按站点ID筛选，并在筛选失败时显式报错。

5. 运行方式和输出边界

保存为bus_eta.py后，单次查询：

python3 bus_eta.py

每分钟轮询一次：

python3 bus_eta.py --watch

接口返回内容会随着时间、交通状况和运营安排变化，下面只展示输出结构，不把某一次查询的分钟数当成固定结果：

route=1 direction=inbound stop=002403 name=跑马地 (上)
#1 中环 (港澳码头) in about <N> min, eta=<ISO-8601 timestamp>

如果看到STALE，说明接口里的data_timestamp距离本机时间已经超过阈值；这时不应该继续把旧ETA当成实时结果。生产任务还应把异常写入日志，并设置请求超时、重试上限和429退避，而不是无限循环请求。

6. 这次实际修掉的三个坑

6.1 路线号不等于站点ID

路线1只是线路，站点ID来自route-stop接口。站点顺序还和方向有关，同一条路线的inbound和outbound不能混用。

6.2 ETA是预测时间，不是服务承诺

官方App说明ETA会根据车辆位置和实际交通状况持续变化。脚本计算出来的“还剩几分钟”只是客户端基于当前时间的一次换算，不能把它写成绝对准点承诺。

6.3 Codex能写出请求代码，但不会替你确认数据含义

模型很容易根据变量名猜出eta是整数分钟，也可能忽略data_timestamp。接口文档、真实响应和异常分支仍然要由工程师逐项核对。AI减少的是重复输入，不是字段语义的责任。

7. 如果要扩展到KMB或长期监控

本文先聚焦城巴，是因为它的路线、站点、路线-站点和ETA接口层次清晰。KMB/龙运也提供实时到站JSON API，但路径和字段组织不同，不能只替换域名；扩展时应重新对照对应的数据字典。

长期运行还需要补三件事：

1. 用缓存保存路线和站点元数据，避免每次轮询都重复请求；
2. 为429、5xx、超时设置有限重试和指数退避；
3. 将data_timestamp、请求耗时和空ETA数量写入结构化日志，才能区分“没有车”和“接口没有新数据”。

总结

这次真正值得记录的不是“Codex几秒钟写出一个脚本”，而是生成代码之后的校验过程：先用官方接口确认数据关系，再用真实字段修正时间处理和异常分支，最后根据ETA更新频率限制轮询。

如果把这类脚本放进定时任务，最低限度也要守住三条边界：站点ID必须来自路线-站点接口，ETA必须按带时区时间解析，空值和旧数据不能伪装成实时结果。

如果这篇对你有用，收藏+点赞。下次换成KMB或绿巴数据时，最容易踩到的坑会是接口路径和字段定义不一样，而不是Python语法本身。

参考资料

1. DATA.GOV.HK：Citybus实时“Next Bus”到站数据
2. Citybus：实时到站数据字典
3. Citybus：API Specifications

数据说明：本文只使用公开接口做技术演示，不承诺ETA与实际到站完全一致；接口字段、更新频率和运营安排以数据提供方最新文档为准。