从一个真实痛点说起

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

手动转发不可持续，爬网页又太脆弱。更合理的做法是接一个稳定的赛事数据 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，天然支持多实例共享状态。

部署建议

  • 容器化：把机器人打成镜像，docker run -e APIZERO_KEY=... bot，环境变量注入密钥。
  • 进程守护：用 systemd 或 Docker 的 restart: always 保证崩溃自愈。
  • 日志结构化：用 JSON 格式输出日志，方便接入 ELK 或 Loki 排查漏推问题。
  • 健康检查：暴露一个 HTTP 端点返回最近一次 tick 的时间戳，超过阈值告警。
  • 密钥管理：绝不要把 Key 写进代码或镜像，用环境变量或密钥管理服务。

合规提醒

官方文档明确说明：赛程与比分来自公开体育数据源，可能有延迟或漏场；CBA 为独立备源，覆盖可能少于 NBA；仅供资讯展示，不构成投注建议。落到机器人设计上有两条红线：

  1. 推送内容里不要出现赔率、盘口、投注引导类文案
  2. 战报要提示数据可能有延迟，避免用户据此做时效性决策

小结

本文把"自动战报推送"这个需求拆解成了一个以状态机为核心的机器人：

  • 用 upcoming → live → finished 的跃迁定义事件，只在变化时推送
  • 快照首见只登记、变化才推送，配合持久化实现重启幂等
  • 自适应轮询间隔 + 时间窗调度，在免费额度内把调用花在刀刃上
  • 推送层用 Protocol 抽象，一套逻辑适配多种群机器人

整个项目不到 300 行代码，但覆盖了状态管理、调度、容错、额度控制这些真正决定生产可用性的细节。把它部署起来，你的社群就再也不会问"谁赢了"了。

# 自动化
Logo
DAMO开发者矩阵

DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐

cover

Microduck：用 Rust 写实时控制栈的开源双足鸭子机器人，这个仓库负责的是部署而不是训练

avatar DAMO开发者矩阵
cover

混凝土表面缺陷识别数据集 | 混凝土缺陷 剥落检测 渗漏水识别 裂缝检测 目标检测9257期

avatar DAMO开发者矩阵
cover

物理AI的核心挑战在于获取符合物理规律的训练数据

avatar DAMO开发者矩阵