账号检测的原理是拿一个「必须有登录态才返回有效结果」的接口当探针:携带 Cookie 请求 B 站的用户导航接口 /x/web-interface/nav,解析返回的 code 与 isLogin 字段,就能判定当前凭据是否还有效。做这个功能的理由更直接——Cookie 会静默失效,失效后采集任务不会抛异常,只会安静地返回空数据,界面上看起来就像「B 站今天没有热点」。热点监控(AI 热点监控工具)多平台并行采集时,把登录态检测做成独立的前置探针,能把这类数据断档从「几天后才发现」变成「几分钟内告警」。
一、为什么要单独做这个功能
失效不是报错,这是账号类采集特有的坑。带着过期 Cookie 请求,服务端照样返回 HTTP 200,业务码却是未登录,解析逻辑拿到空列表就当成「无新内容」处理完毕。有没有检测环节,差别体现在三处:
| 维度 | 无登录态检测 | 有登录态检测 |
|---|---|---|
| 发现时机 | 人工翻看面板时偶然发现 | 探针失败后数分钟内告警 |
| 表面现象 | 数据条数归零,日志无异常 | 明确指出凭据失效及原因码 |
| 排查成本 | 逐个数据源翻日志、复现请求 | 直接定位到账号维度 |
| 数据完整性 | 断档时段无法追回 | 断档窗口压缩到一轮轮询 |
另一个理由是权限差异。B 站部分接口对匿名请求限制更严,返回字段少、翻页深度受限,登录态下才能拿到完整信息。既然采集链路依赖登录态,就该把它当成一个有明确健康状态的依赖项来管理,而不是假设它一直可用。
需要明确的前提:检测对象是自有账号的登录凭据,用途是确认自家采集任务能否正常工作。整个功能不涉及他人账号、不做批量登录、不绕过平台的安全校验,采集范围限于平台公开可见内容,凭据与数据的存储使用遵守《网络安全法》《数据安全法》《个人信息保护法》。
二、检测原理:三步校验
探针逻辑分三级,从便宜到贵依次执行,避免每次都发网络请求。
2.1 格式预检
Cookie 字符串里缺少关键字段,说明配置本身就错了,不必浪费一次请求。B 站的 Web 登录态核心是 SESSDATA,写操作还需要 bili_jct 作为校验令牌。预检只判断字段是否存在,不解析内容。
2.2 活体探针
用户导航接口是公认的轻量探针:请求体积小、响应快、字段明确,返回里带用户名与用户 ID,可直接确认「当前是哪个账号登录」。
import requests
NAV_API = "https://api.bilibili.com/x/web-interface/nav"
HEADERS = {
"User-Agent": ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36"),
"Referer": "https://www.bilibili.com/",
}
def check_account(cookie: str, timeout: int = 10) -> dict:
"""返回 {'ok': bool, 'code': int, 'user': str, 'reason': str}"""
if "SESSDATA" not in cookie:
return {"ok": False, "code": -1, "user": "",
"reason": "凭据格式不完整,缺少 SESSDATA"}
try:
resp = requests.get(NAV_API, headers={**HEADERS, "Cookie": cookie},
timeout=timeout)
except requests.RequestException as exc:
return {"ok": False, "code": -2, "user": "",
"reason": f"网络异常:{type(exc).__name__}"}
if resp.status_code != 200:
return {"ok": False, "code": resp.status_code, "user": "",
"reason": f"HTTP 状态异常 {resp.status_code}"}
payload = resp.json()
code = payload.get("code", -999)
data = payload.get("data") or {}
if code == 0 and data.get("isLogin") is True:
return {"ok": True, "code": 0, "user": data.get("uname", ""),
"reason": ""}
return {"ok": False, "code": code, "user": "",
"reason": explain(code, payload.get("message", ""))}
2.3 返回码语义判定
同样是「检测失败」,原因不同,处置动作完全不同:
| 业务码 | 含义 | 处置动作 |
|---|---|---|
| 0 且 isLogin 为真 | 登录态正常 | 记录用户名,继续采集 |
| -101 | 账号未登录 | 凭据缺失或已失效,等待更新 |
| -352 | 风控校验未通过 | 凭据过期或触发风控,降速并暂停该源 |
| -412 | 请求被拦截 | 请求头不完整或频率过高,补齐头部并降频 |
| 网络异常 | 探针本身没打通 | 视为不确定状态,不计入失效判定 |
把网络异常单独归类很重要。链路抖动被当成凭据失效,会造成大量误报警,接着就是没人再看告警。
CODE_HINTS = {
-101: "账号未登录,凭据可能已被清除",
-352: "风控校验失败,凭据过期或访问特征异常",
-412: "请求被拦截,检查请求头与访问频率",
}
def explain(code: int, message: str) -> str:
return CODE_HINTS.get(code, message or f"未知业务码 {code}")
三、检测频率与告警抖动
探针不需要高频。凭据有效期通常按天甚至更长计算,把检测挂成独立定时任务,间隔 30 分钟到 1 小时足够。更省的做法是「事件驱动 + 定时兜底」:需要登录态的采集任务连续两轮返回空结果时,主动触发一次探针,同时保留低频定时检测作为兜底。
告警要防抖。连续失败达到阈值才升级为通知,单次失败只记日志:
FAIL_THRESHOLD = 2
_fail_streak = {"bilibili": 0}
def on_check_result(account: str, result: dict, notifier) -> None:
if result["ok"]:
if _fail_streak[account] >= FAIL_THRESHOLD:
notifier.send(f"[恢复] {account} 登录态已恢复,当前账号 {result['user']}")
_fail_streak[account] = 0
return
if result["code"] == -2: # 网络异常不计入失效
return
_fail_streak[account] += 1
if _fail_streak[account] == FAIL_THRESHOLD:
notifier.send(f"[失效] {account} 登录态异常:{result['reason']},"
f"需更新凭据,相关采集任务已降级")
恢复通知同样要发。只发失效不发恢复,运维人员无法确认修复是否生效,会反复手工验证。
四、失效后的处理链路
检测出结果只是起点,关键是让系统在凭据失效期间仍能有序运行。处理顺序建议固定为五步:
- 把账号健康状态写入配置表或缓存,字段包含状态、原因码、检测时间;
- 依赖登录态的采集任务读取该状态,命中失效时直接跳过本轮,不再发无效请求;
- 能匿名获取的数据源切换到匿名模式,接受字段与深度上的缩减,保证有数据可用;
- 前端面板显示该数据源为「降级中」,避免用户把空数据误读为平台无热点;
- 通知负责人更新凭据,更新后立即触发一次探针,通过则清除降级状态并推送恢复消息。
凭据更新可以做成后台页面:调用平台的二维码登录流程拿到新凭据,服务端只做代理与暂存,页面上完成扫码后写入配置。这条链路把「改环境变量再重启服务」变成一次扫码,减少人工失误。
五、凭据保管的几条硬要求
登录凭据等同账号权限,保管方式直接决定风险面。凭据从环境变量或加密配置读取,不写进代码仓库;日志与告警消息里做脱敏,只输出用户名和原因码,不打印凭据内容;数据库存储时加密,并限制读取权限;探针请求维持正常频率与完整请求头,不做异常高频调用;仅使用自有账号,不采集需要授权才能查看的私密内容。
再加一条工程习惯:给探针结果留历史。把每次检测的时间、状态、原因码落表,几周后就能看出凭据的实际有效周期,据此把「被动等失效」改成「到期前主动提醒更新」。
常见问题(FAQ)
Q1:怎么区分凭据失效和网络故障?
看失败发生在哪一层:业务码 -101 或 -352 属凭据问题,连接超时归网络异常。
Q2:检测多久跑一次合适?
定时 30 分钟到 1 小时,另在采集连续返回空结果时事件触发一次探针。
Q3:凭据失效期间采集要停掉吗?
停掉依赖登录态的任务,能匿名取的数据源降级运行,并在面板标注降级状态。