用篮球赛事 API 搭建一个自动战报推送机器人-育师

从一个真实痛点说起

做体育社群的运营同学都有过这种经历：比赛结束了半小时，群里才有人喊"谁赢了"；或者熬夜看球时不断刷新网页，只为了确认一个终场比分。这些场景的本质需求是——在正确的时间点，把正确的比赛信息推给正确的人。

手动转发不可持续，爬网页又太脆弱。更合理的做法是接一个稳定的赛事数据 API，自己写一个推送机器人。本文以极数本源篮球赛事 API 为数据源，完整实现一个自动战报机器人。

需求拆解

先把"战报推送"这个模糊需求拆成可执行的技术点：

需求	技术挑战
比赛开始时提醒	状态从`upcoming`→`live`的转换检测
终场后推送战报	状态从`live`→`finished`的转换检测
不重复推送	幂等与状态快照对比
定时检查	调度策略与 API 额度平衡
多渠道分发	推送层抽象

核心洞察是：机器人真正要做的不是"查询比赛"，而是"检测状态变化"。这决定了整个架构应该围绕一个状态机来设计。

数据源特性回顾

接口地址https://v1.apizero.cn/api/basketball，POST 请求，Authorization: Bearer <API Key>鉴权。两个参数：date（北京日期，缺省当日）和limit（1–200，缺省 80）。

返回的matches数组里，每场比赛有一个status字段，取值只有三个：live、upcoming、finished。这就是我们状态机的三个状态。

排序规则是进行中 → 未开赛按时间 → 已结束，这个顺序对推送逻辑很重要：进行中的比赛永远排在最前面。

关键约束：免费方案每天只有 20 次调用。这意味着轮询间隔不能太激进，必须做智能调度。

状态机设计

把每场比赛看作一个状态机，只在状态跃迁时触发推送：

upcoming ──[比赛开始]──> live ──[比赛结束]──> finished │ │ │ └── 推送"即将开赛" └── 推送"实时比分" └── 推送"终场战报"

用代码表达这个跃迁规则：

# state_machine.py from enum import Enum class MatchStatus(str, Enum): UPCOMING = "upcoming" LIVE = "live" FINISHED = "finished" # 只在这些跃迁上触发推送，其余变化忽略 TRANSITIONS = { (MatchStatus.UPCOMING, MatchStatus.LIVE): "game_start", (MatchStatus.LIVE, MatchStatus.FINISHED): "game_end", } # 状态不变但比分变化，触发比分播报 SCORE_UPDATE = "score_update" def detect_event(old_status: str | None, new_status: str, old_score: tuple | None, new_score: tuple | None) -> str | None: """返回事件类型，None 表示无需推送。""" if old_status is None: # 首次见到这场比赛，不推送，只登记 return None if old_status != new_status: key = (MatchStatus(old_status), MatchStatus(new_status)) return TRANSITIONS.get(key) # 状态没变，但进行中的比赛比分变了 if new_status == MatchStatus.LIVE and old_score != new_score: return SCORE_UPDATE return None

为什么首次见到不推送？因为程序启动时会一次性拉取当天所有比赛，如果把它们全当"新比赛"推一遍，用户会被历史数据淹没。首见只登记，变化才推送——这是去重的第一道防线。

数据模型与解析

直接操作原始 dict 会让后续逻辑到处写.get()，先做类型化。

# models.py from dataclasses import dataclass @dataclass(frozen=True) class TeamSnapshot: name: str short: str score: int | None @property def label(self) -> str: return self.short or self.name @dataclass(frozen=True) class MatchSnapshot: match_id: str league_name: str start_at: str status: str home: TeamSnapshot away: TeamSnapshot @property def score_pair(self) -> tuple[int | None, int | None]: return (self.home.score, self.away.score) def headline(self) -> str: h = self.home.score if self.home.score is not None else "-" a = self.away.score if self.away.score is not None else "-" return f"{self.home.label} {h} : {a} {self.away.label}" def winner_text(self) -> str: """终场战报用：判断胜负。""" hs, as_ = self.home.score, self.away.score if hs is None or as_ is None or hs == as_: return "战平" if hs == as_ else "结果待定" return f"{self.home.name if hs > as_ else self.away.name} 获胜" def parse_match(raw: dict) -> MatchSnapshot: def team(obj: dict) -> TeamSnapshot: return TeamSnapshot( name=obj.get("name", ""), short=obj.get("short") or obj.get("name", ""), score=obj.get("score"), ) return MatchSnapshot( match_id=str(raw.get("id", "")), league_name=raw.get("league_name", ""), start_at=raw.get("start_at", ""), status=raw.get("status", ""), home=team(raw.get("home", {})), away=team(raw.get("away", {})), )

用frozen=True的 dataclass 有个好处：快照天然不可变，作为字典的 value 比较时不会出现意外共享引用的问题。

API 客户端

客户端负责鉴权、重试和把原始响应转成快照列表。

# client.py import time import logging from datetime import datetime, timedelta, timezone import requests from models import MatchSnapshot, parse_match logger = logging.getLogger(__name__) BASE_URL = "https://v1.apizero.cn/api/basketball" BEIJING_TZ = timezone(timedelta(hours=8)) class SportsApiError(RuntimeError): pass class SportsClient: def __init__(self, api_key: str, timeout: int = 10): self._session = requests.Session() self._session.headers.update({ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", }) self.timeout = timeout @staticmethod def today_beijing() -> str: return datetime.now(BEIJING_TZ).strftime("%Y-%m-%d") def fetch(self, date: str | None = None, limit: int = 200) -> tuple[list[MatchSnapshot], dict]: if not 1 <= limit <= 200: raise ValueError("limit 范围 1–200") payload = {"date": date or self.today_beijing(), "limit": limit} for attempt in range(1, 4): try: resp = self._session.post(BASE_URL, json=payload, timeout=self.timeout) if resp.status_code == 429: time.sleep(5 * attempt) continue resp.raise_for_status() body = resp.json() if body.get("code") != 0: raise SportsApiError(f"接口错误 code={body.get('code')} msg={body.get('msg')}") data = body["data"] matches = [parse_match(m) for m in data.get("matches", [])] return matches, data.get("summary", {}) except SportsApiError: raise except (requests.ConnectionError, requests.Timeout) as exc: logger.warning("网络异常 attempt=%d: %s", attempt, exc) time.sleep(2 ** attempt) raise SportsApiError("请求失败，已重试 3 次")

fetch返回summary是有意为之——里面的live字段直接告诉我们当前有几场进行中，调度器可以据此决定下次轮询间隔，不必自己再遍历一遍。

推送层抽象

不同渠道（钉钉、企业微信、Telegram、Server 酱）的消息格式差异很大，但机器人只关心"发什么内容"。用一个 Protocol 隔离开。

# notifiers.py from typing import Protocol import requests class Notifier(Protocol): def send(self, title: str, content: str) -> None: ... class WebhookNotifier: """通用 Webhook 推送，适配钉钉/企业微信/飞书等。""" def __init__(self, webhook_url: str, timeout: int = 5): self.webhook_url = webhook_url self.timeout = timeout def send(self, title: str, content: str) -> None: payload = { "msgtype": "markdown", "markdown": {"title": title, "text": f"### {title}\n\n{content}"}, } resp = requests.post(self.webhook_url, json=payload, timeout=self.timeout) resp.raise_for_status() class ConsoleNotifier: """本地调试用，不产生任何网络请求。""" def send(self, title: str, content: str) -> None: print(f"\n{'=' * 40}\n{title}\n{'-' * 40}\n{content}\n{'=' * 40}")

定义Notifier协议而不是抽象基类，是为了让任何有send(title, content)方法的对象都能直接传进来，鸭子类型在 Python 里更自然。

机器人主体

现在把状态机、客户端、推送层组装起来。

# bot.py import time import logging from datetime import datetime, timedelta, timezone from client import SportsClient from models import MatchSnapshot from state_machine import detect_event from notifiers import Notifier, ConsoleNotifier logger = logging.getLogger(__name__) BEIJING_TZ = timezone(timedelta(hours=8)) class ScoreBot: def __init__(self, client: SportsClient, notifier: Notifier, idle_interval: int = 600, live_interval: int = 60): self.client = client self.notifier = notifier self.idle_interval = idle_interval self.live_interval = live_interval # match_id -> MatchSnapshot，记录上一轮看到的状态 self._snapshot: dict[str, MatchSnapshot] = {} self._seen_date: str | None = None def _reset_if_new_day(self, today: str) -> None: """跨天后清空快照，避免旧数据干扰。""" if self._seen_date != today: if self._seen_date is not None: logger.info("跨天，清空快照 %s -> %s", self._seen_date, today) self._snapshot.clear() self._seen_date = today def _handle(self, match: MatchSnapshot) -> None: old = self._snapshot.get(match.match_id) event = detect_event( old_status=old.status if old else None, new_status=match.status, old_score=old.score_pair if old else None, new_score=match.score_pair, ) if event == "game_start": self.notifier.send( f"🏀 {match.league_name} 开赛", f"**{match.headline()}**\n\n开赛时间：{match.start_at}", ) elif event == "score_update": self.notifier.send( f"⚡ 比分更新 · {match.league_name}", f"**{match.headline()}**", ) elif event == "game_end": self.notifier.send( f"🏆 终场战报 · {match.league_name}", f"**{match.headline()}**\n\n{match.winner_text()}\n开赛时间：{match.start_at}", ) # 无论是否推送，都要更新快照 self._snapshot[match.match_id] = match def tick(self) -> int: """执行一轮检查，返回当前进行中的场次数。""" today = self.client.today_beijing() self._reset_if_new_day(today) matches, summary = self.client.fetch(limit=200) for m in matches: self._handle(m) live = summary.get("live", 0) logger.info("本轮共 %d 场，进行中 %d 场", len(matches), live) return live def run_forever(self) -> None: while True: try: live = self.tick() wait = self.live_interval if live > 0 else self.idle_interval except Exception as exc: logger.exception("本轮出错：%s", exc) wait = self.idle_interval time.sleep(wait) if __name__ == "__main__": import os logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") client = SportsClient(os.environ["APIZERO_KEY"]) notifier = ConsoleNotifier() # 换成 WebhookNotifier("https://...") 即可接入群机器人 ScoreBot(client, notifier, idle_interval=600, live_interval=60).run_forever()

几个设计细节值得展开：

快照永远更新：即使这一轮不推送，也要把最新状态存下来。否则下一轮会比较到过期的旧状态，导致漏推或重复推。
跨天清空：_reset_if_new_day处理的是服务器长期运行的场景。北京时间过零点后，date参数变了，昨天的快照必须丢弃。
异常不中断循环：run_forever里 catch 住所有异常，只影响这一轮，机器人不会因为一次网络抖动就退出。
额度自适应：有直播时 60 秒一轮，空闲时 10 分钟一轮。按免费方案每天 20 次算，空闲时段基本不消耗额度，把调用留给真正的比赛时段。

调度策略与额度控制

免费方案每天 20 次、QPS 40 是硬约束。上文的策略在 NBA 常规赛日（一天可能有 5–10 场比赛集中在 3 小时内）依然可能超支，需要更精细的方案：

策略	说明	适用场景
自适应间隔	有直播才高频，空闲降频	通用
时间窗调度	只在已知开赛时段前后密集检查	赛程固定
本地缓存	`finished`状态缓存 24h 不再查	历史回看
多实例错峰	不同机器人错开轮询时间	多群分发

一个实用的时间窗实现思路：从start_at解析出开赛时间，只在「开赛前 5 分钟到赛后 30 分钟」这个窗口内高频轮询，其余时间完全休眠。

def in_active_window(match: MatchSnapshot, now: datetime, pre_min: int = 5, post_min: int = 30) -> bool: """判断当前是否处于该比赛需要关注的窗口内。""" try: start = datetime.strptime(match.start_at, "%Y-%m-%d %H:%M:%S").replace(tzinfo=BEIJING_TZ) except ValueError: return True # 解析失败就保守处理，保持关注 return (start - timedelta(minutes=pre_min)) <= now <= (start + timedelta(minutes=post_min))

把这个函数接到调度器上，就能在没比赛时把轮询彻底停掉，额度利用率能提升数倍。

幂等与容灾

推送机器人最怕两种情况：重复推送和漏推。上面的状态机解决了运行期的重复，但进程重启会丢失内存快照，导致重启后把已结束的比赛重新推一遍。

解决办法是把快照持久化到磁盘或 Redis：

import json from pathlib import Path def save_snapshot(path: str, snapshot: dict[str, MatchSnapshot], seen_date: str) -> None: data = { "seen_date": seen_date, "matches": { mid: { "status": m.status, "home_score": m.home.score, "away_score": m.away.score, } for mid, m in snapshot.items() }, } Path(path).write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") def load_snapshot(path: str) -> tuple[dict[str, dict], str | None]: p = Path(path) if not p.exists(): return {}, None data = json.loads(p.read_text(encoding="utf-8")) return data.get("matches", {}), data.get("seen_date")

启动时加载，每轮 tick 后写回。生产环境直接换成 Redis 的HSET，天然支持多实例共享状态。