用 Twitch Helix API 检查频道直播状态

写一个命令行工具：给它一份频道清单，它会告诉你哪些频道正在直播、多少人在看、在播什么。工具用 client credentials 方式拿 app access token，再调用 Helix 的 Get Streams；每 100 个频道合成一次请求，遇到 429 会按 Ratelimit-Reset 等待后重试。

先看清成品

pytest 跑了 1 个测试，通过：150 个频道分成 100 + 50 两批，第一次返回 429 后会重试。用官方 Twitch CLI v1.1.24 的 mock-api（本地模拟服务器）完整跑了一遍：拿 token、查询，输出“● marstonkid357 99368 人在看”和“○ twitchdev 未直播”，这些都是 mock 生成的假数据。连真实服务器时我用的是假凭据：token 接口返回 HTTP 400 invalid client，Get Streams 返回 401 Invalid OAuth token。没有用真实的 Client ID 和 Secret 查询过真实的直播数据。

注意：只使用 Twitch 官方 API、EventSub 和官方嵌入播放器，并遵守《Twitch 开发者服务协议》。不要抓取或转录直播流，不要遮挡或去掉播放器上的广告，不要共享账号或 Client Secret，也不要用 VPN 或代理绕过地区限制。

做完之后你会得到什么？

需要准备哪些环境？

官方要点（已和 dev.twitch.tv 核对）

文件结构

35-twitch-A-src/
├── live_check.py
├── channels.txt
└── test_live.py

第 1 步：脚本

live_check.py：

"""查询一组 Twitch 频道当前是否在直播（Helix Get Streams + app access token）。
环境变量：TWITCH_CLIENT_ID、TWITCH_CLIENT_SECRET
可选：TWITCH_AUTH_URL / TWITCH_API_URL（默认指向官方地址；测试时可改成 twitch-cli mock-api）
用法：python live_check.py channels.txt"""
import os, sys, time
import requests

AUTH = os.environ.get("TWITCH_AUTH_URL", "https://id.twitch.tv/oauth2/token")
API = os.environ.get("TWITCH_API_URL", "https://api.twitch.tv/helix")

def app_token(cid, secret):
    r = requests.post(AUTH, data={"client_id": cid, "client_secret": secret, "grant_type": "client_credentials"}, timeout=15)
    if r.status_code != 200:
        raise SystemExit(f"获取 token 失败：HTTP {r.status_code} {r.text[:200]}")
    return r.json()["access_token"]

def get_streams(logins, cid, token):
    """user_login 每次最多 100 个；返回 {login: stream}。只有正在直播的频道会出现在结果里。"""
    live, h = {}, {"Authorization": f"Bearer {token}", "Client-Id": cid}
    for i in range(0, len(logins), 100):
        batch = logins[i:i + 100]
        while True:
            r = requests.get(f"{API}/streams", params=[("user_login", l) for l in batch] + [("first", 100)], headers=h, timeout=15)
            if r.status_code == 429:                                   # 等到 Ratelimit-Reset 再重试
                time.sleep(max(1, int(r.headers.get("Ratelimit-Reset", time.time() + 1)) - int(time.time())))
                continue
            if r.status_code != 200:
                raise SystemExit(f"Get Streams 失败：HTTP {r.status_code} {r.text[:200]}")
            break
        for s in r.json()["data"]:
            live[s["user_login"].lower()] = s
    return live

if __name__ == "__main__":
    logins = [l.strip().lower() for l in open(sys.argv[1], encoding="utf-8") if l.strip() and not l.startswith("#")]
    cid = os.environ["TWITCH_CLIENT_ID"]
    live = get_streams(logins, cid, app_token(cid, os.environ["TWITCH_CLIENT_SECRET"]))
    for l in logins:
        s = live.get(l)
        print(f"● {l}  {s['viewer_count']} 人在看  [{s['game_name']}] {s['title']}" if s else f"○ {l}  未直播")

channels.txt：

# 每行一个频道登录名（login），不区分大小写
marstonkid357
twitchdev

第 2 步：单元测试

test_live.py：

import live_check

class R:
    def __init__(self, code, js=None, headers=None): self.status_code, self._js, self.headers, self.text = code, js, headers or {}, ""
    def json(self): return self._js

def test_batches_and_429(monkeypatch):
    calls = []
    def fake_get(url, params, headers, timeout):
        calls.append([v for k, v in params if k == "user_login"])
        if len(calls) == 1: return R(429, headers={"Ratelimit-Reset": "0"})
        return R(200, {"data": [{"user_login": "U5", "viewer_count": 1, "game_name": "g", "title": "t"}]})
    monkeypatch.setattr(live_check.requests, "get", fake_get)
    monkeypatch.setattr(live_check.time, "sleep", lambda s: None)
    live = live_check.get_streams([f"u{i}" for i in range(150)], "cid", "tok")
    assert [len(c) for c in calls] == [100, 100, 50]           # 第一次 429 后重试同一批
    assert "u5" in live

第 3 步：用 Twitch CLI 的 mock-api 离线运行

twitch mock-api generate          # 生成本地数据库，并打印一组 mock 用的 Client-ID / Secret
twitch mock-api start -p 8765     # 默认端口 8080，如果被占用就换一个
export TWITCH_AUTH_URL=http://localhost:8765/auth/token
export TWITCH_API_URL=http://localhost:8765/mock
export TWITCH_CLIENT_ID=<generate 打印的 Client-ID>
export TWITCH_CLIENT_SECRET=<generate 打印的 Secret>
python live_check.py channels.txt

● marstonkid357  99368 人在看  [Just Making a CLI] Sample stream!
○ twitchdev  未直播

marstonkid357 是 mock 随机生成的用户。换成你自己生成的数据时，请把 channels.txt 里的第一行改成 generate 输出里的用户。连接真实服务器时，去掉两个 URL 变量，填入你自己应用的凭据即可。

检查点：凭据填错时，脚本会立刻停下，并打印接口返回的原文。我用假凭据实测，得到的是：

获取 token 失败：HTTP 400 {"status":400,"message":"invalid client"}

调试示例：多个频道用逗号连在一起

很多 API 习惯用 user_login=a,b 传多个值。我对 mock-api 实测了两种写法：

user_login=marstonkid357,twitchdev            -> {"data":[]}
user_login=marstonkid357&user_login=twitchdev -> {"data":[{"user_login":"marstonkid357", ...}]}

用逗号连接时不会报错，只会返回空列表，看起来就像“大家都没在直播”。文档要求每个值单独写一个参数，所以脚本用 [("user_login", l) for l in batch] 这种元组列表，交给 requests 去编码。

常见问题

模板规则

• Client Secret 只放在环境变量或服务器端，不要放进前端代码，也不要提交到仓库。
• 先用 mock-api 测试，再用真实凭据。
• 定时轮询时，间隔要合理；需要实时通知的话，用 EventSub（见 B 篇）。

项目资源

用一个 token 加一个接口，就能做出实用的“开播检查”。最容易踩的坑不是鉴权，而是参数写法：写错了不会报错，只会悄悄返回空结果。