尧图建网站 尧图建网站 YAOTU WEB BUILD 免费咨询
ARTICLE DETAIL

资讯详情

深耕网站建设与建站编程的一线实战洞察。

Python打造可扩展的B站自动化机器人:从Cookie管理到定时任务调度实战

Python打造可扩展的B站自动化机器人:从Cookie管理到定时任务调度实战 很多人第一次听到robotbilibili这个词第一反应是这不就是一个挂着 B 站 API 的签到脚本吗如果你抱着这个预期去读代码会发现它远不止一个脚本。它本质上是一套把 B 站账号重复操作拆解成可调度任务的自动化框架登录态管理、请求客户端、任务抽象、定时调度、异常处理和日志监控每一层都可以独立替换和扩展。先说一个明确判断robotbilibili这类项目的核心价值不是和平台风控赛跑也不是帮你实现那些灰色操作而是把确定性的重复劳动交给代码让人把时间放在内容、策略和异常处理上。只有先理解这个边界后面的代码写起来才有意义。这篇文章我会从真实开发痛点出发先讲清楚 B 站自动化机器人要解决什么问题再给出一套完整的 Python 工程示例涵盖Cookie登录态管理、请求客户端封装、任务抽象、定时调度、运行验证和常见问题排查。读完以后你不只是会复制脚本而是能自己设计一套可维护、可灰度、可回滚的 B 站自动化任务系统。1. 这篇文章真正要解决的问题如果你是 B 站内容创作者、账号运营者或者只是把 B 站账号当测试对象的技术爱好者一定经历过下面这些场景每天早上要登录后台看一遍数据记录播放、点赞、评论变化。视频发布后要在固定时间点检查评论、回复私信。直播前要发动态预告直播后要整理数据。多个账号需要重复执行相同的维护操作。手动操作最大的问题不是累而是不可沉淀。今天手动做了一遍明天还要再做一遍后天依然如此。操作步骤没有存档数据没有留痕偶尔漏了一次也很难追溯。更有意思的是这些操作绝大多数是确定性的给定登录态调一个接口写一条数据。既然是确定性的事就应该用代码表达。robotbilibili要解决的就是这个矛盾如何把 B 站账号的日常维护动作变成一组可编排、可调度、可监控的任务。读这篇文章之前我希望你带着三个问题登录态怎么管理才能既不频繁失效又不把 Cookie 泄露到代码仓库里任务之间如何抽象才能让“发评论”和“查数据”这样完全不同的动作共用一套调度和执行框架定时任务跑到一半抛异常了怎么才能及时发现而不是等到第二天数据报表出来才发现这三个问题正是robotbilibili一类项目在设计时最需要想清楚的部分。下面我会用完整代码逐一回答。2. robotbilibili 的核心概念与整体架构在写出第一行代码之前先建立一套统一的概念模型。这个模型不依赖 B 站具体接口任何平台的自动化项目都可以套用。2.1 四个核心抽象Session会话一次登录成功后建立的持久化状态。在 B 站场景里Session 的核心就是 Cookie 和请求头。Session 管理得好不好直接决定自动化任务的稳定性。Action原子操作一次具体的 API 调用比如查询视频列表、读取用户信息、发送一条动态。Action 是任务的最小执行单元。Task任务由若干个 Action 按业务逻辑组合而成比如“每日数据巡检任务”可以拆成“获取用户信息 - 获取视频列表 - 获取每条视频的播放量 - 写入本地数据库”。Scheduler调度器负责在指定时间或周期性触发 Task。调度器需要支持 cron 表达式、任务去重、失败重试和日志输出。2.2 与手工操作和浏览器自动化的对比很多人会问既然有requests直接调接口为什么还要用浏览器自动化Playwright、Selenium这里有一个选择依据维度手工操作API 自动化浏览器自动化执行速度慢依赖人快毫秒级中等需要启动浏览器稳定性不稳定易漏操作高只要接口不变受页面结构和网络影响登录态维护人工维护Cookie/Token需要处理登录态持久化维护成本无代码成本接口变更时需要改代码页面 DOM 变更时需要改选择器适合场景临时、低频操作批量、定时、高频复杂交互、强验证场景robotbilibili走的是 API 自动化路线。它的核心优势是轻量和稳定一个 Python 进程、一个请求客户端、一张定时任务表就能跑完整的自动化流程。2.3 整体目录结构一个可扩展的robotbilibili工程建议用下面的目录组织代码robotbilibili/ ├── main.py # 程序入口 ├── requirements.txt # 依赖清单 ├── .env # 环境变量不要提交到 Git ├── robotbilibili/ │ ├── __init__.py │ ├── client.py # 请求客户端封装 │ ├── login.py # 登录态获取 │ ├── scheduler.py # 任务调度器 │ ├── config.py # 配置读取 │ ├── api/ │ │ └── user.py # 用户相关接口 │ └── tasks/ │ ├── base.py # 任务基类 │ └── daily.py # 具体任务实现 └── logs/ └── robot.log # 运行日志这个结构的设计原则是入口只做组装业务逻辑放在任务层底层能力放在 client 和 api 层。后续无论新增任务还是更换接口改动都能收敛在局部。3. 环境准备与基础依赖为了让示例可复现本文统一使用以下环境。如果你是新手建议严格按照这个顺序操作。3.1 操作系统与 Python 版本Ubuntu 20.04 / macOS / Windows WSL2 都可以。Python 3.9 及以上推荐 Python 3.10 或 3.11。检查 Python 版本python3 --version如果本机 Python 版本较低建议先安装pyenv或使用系统包管理器升级。3.2 创建虚拟环境mkdir robotbilibili cd robotbilibili python3 -m venv venv source venv/bin/activateWindows 用户执行venv\Scripts\activate3.3 安装依赖创建requirements.txtrequests2.28.0 APScheduler3.10.0 python-dotenv1.0.0安装pip install -r requirements.txt这里解释一下三个依赖的用途requests处理 HTTP 请求是 API 自动化的基础。APScheduler高级定时任务调度库支持 cron 表达式、任务持久化和错过任务的补偿。python-dotenv从.env文件读取配置避免把 Cookie 等敏感信息写死在代码里。3.4 准备登录态B 站自动化项目最核心的一步是拿到有效的 Cookie。有两种方式手动从浏览器复制 Cookie登录 bilibili.com 后按 F12 打开开发者工具在 Network 面板中找到任意 API 请求复制请求头里的 Cookie。集成扫码登录让用户用 B 站 App 扫码程序自动换取 Cookie。第一种方式最快但 Cookie 有有效期过期后需要重新复制不适合长期运行的定时任务。第二种方式用户体验更好也是推荐方案下一节会给出实现思路。4. 登录与会话管理机器人稳定运行的地基4.1 Cookie 与 Session 的关系HTTP 协议本身是无状态的服务器不知道两次请求是不是同一个用户。B 站通过 Set-Cookie 在浏览器里写入一串身份凭证后续请求带上这串凭证服务器就能识别用户身份。在robotbilibili里我们用requests.Session()维持一个本地会话对象。Session 会自动保存服务器返回的 Cookie并在后续请求中自动携带。相比每次请求都手动拼 CookieSession 是更规范的做法。4.2 扫码登录的原理与示例扫码登录本质上是一个“轮询”过程客户端向 B 站服务端申请一个二维码链接和唯一的qrcode_key。用户用 B 站 App 扫描二维码并确认。客户端每隔 1-2 秒轮询一次查询扫码状态。当状态返回成功时服务端会下发登录 Cookie客户端保存并复用。下面给出一个轮询流程示例。注意接口地址和参数可能随版本调整你需要以 B 站官方接口文档或开源项目的最新实现为准。# robotbilibili/login.py import time import requests def get_qrcode(): 申请登录二维码返回 qrcode_key 和二维码内容 URL。 resp requests.post( https://passport.bilibili.com/x/passport-login/web/qrcode/generate, timeout10, ) data resp.json() if data.get(code) ! 0: raise RuntimeError(f获取二维码失败: {data}) return data[data] def poll_qrcode(qrcode_key: str): 轮询扫码结果成功则返回 Cookie 列表。 while True: time.sleep(2) resp requests.post( https://passport.bilibili.com/x/passport-login/web/qrcode/poll, data{qrcode_key: qrcode_key}, timeout10, ) data resp.json().get(data, {}) status_code data.get(code) if status_code 0: return data.get(cookie_info, {}).get(cookies, []) if status_code 86038: raise RuntimeError(二维码已失效请重新获取) if status_code 86090: print(已扫码等待确认...) elif status_code 86101: print(等待扫码...) def login_by_qrcode() - str: 执行扫码登录返回可用的 Cookie 字符串。 qr get_qrcode() print(请在 B 站 App 中扫描二维码: , qr.get(url)) cookies poll_qrcode(qr.get(qrcode_key)) cookie_parts [] for c in cookies: cookie_parts.append(f{c[name]}{c[value]}) return ; .join(cookie_parts)这段代码的关键点在于轮询状态机的处理。86101表示等待扫码86090表示已经扫码等待确认86038表示二维码过期只有code 0才意味着登录成功并返回 Cookie。生产环境中你不会希望每次启动程序都重新扫码所以拿到 Cookie 后应该把它写入环境变量或本地凭据文件cat .env EOF BILI_COOKIE你的Cookie字符串 EOF4.3 统一请求客户端封装requests.Session固然好用但真正进入工程化之后还需要统一处理请求头、超时、异常重试和请求间隔。否则每个任务自己写一套 HTTP 逻辑代码会迅速腐烂。# robotbilibili/client.py import time import requests class BiliClient: Bilibili API 请求客户端统一管理会话和请求策略。 def __init__(self, cookie: str, max_retries: int 3): self.session requests.Session() self.session.headers.update({ User-Agent: ( Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 ), Referer: https://www.bilibili.com/, Origin: https://www.bilibili.com, Cookie: cookie, }) self.max_retries max_retries def request(self, method: str, url: str, **kwargs): 带重试和简单频率控制的请求方法。 for attempt in range(1, self.max_retries 1): try: kwargs.setdefault(timeout, 10) response self.session.request(method, url, **kwargs) response.raise_for_status() return response except requests.RequestException as exc: if attempt self.max_retries: raise print(f[client] 请求失败第 {attempt} 次重试: {exc}) time.sleep(attempt * 2) def get(self, url: str, **kwargs): return self.request(GET, url, **kwargs) def post(self, url: str, **kwargs): return self.request(POST, url, **kwargs)这段封装带来了三个好处统一了请求头避免每个接口单独设置User-Agent。增加了重试机制网络抖动时自动重试而不是直接抛异常中止整个任务链。后续要加统一限流、日志、埋点只需要改request方法所有任务都生效。5. 核心功能模块与完整代码实现进入本文最重要的部分。我会从最底层的 API 封装开始逐步搭建一个可以运行的最小robotbilibili工程。5.1 获取登录用户信息获取当前登录用户信息是很多自动化任务的第一步用来校验 Cookie 是否有效也用来确认当前操作的是哪个账号。# robotbilibili/api/user.py from robotbilibili.client import BiliClient def get_my_info(client: BiliClient) - dict: 获取当前登录用户的基本信息。 注意接口地址和返回结构以 B 站实际接口为准这里仅作示例。 resp client.get(https://api.bilibili.com/x/web-interface/nav) json_data resp.json() if json_data.get(code) ! 0: raise RuntimeError(f获取用户信息失败: {json_data}) data json_data.get(data, {}) return { uid: data.get(mid), name: data.get(uname), level: data.get(level_info, {}).get(current_level), }如果 Cookie 失效B 站通常会返回一个非 0 的业务 code。实现中判断code ! 0并抛异常就是为了让上层调度器能捕获这个异常并触发告警。5.2 定义任务抽象基类一个健壮的任务框架最重要的一层抽象是任务接口。所有具体任务都实现同一个run方法调度器不关心任务内部逻辑只负责在正确的时间调用它。# robotbilibili/tasks/base.py import traceback from abc import ABC, abstractmethod from robotbilibili.client import BiliClient class BaseTask(ABC): 所有自动化任务的基类。 name: str base_task description: str abstractmethod def execute(self, client: BiliClient) - dict: 执行任务逻辑返回结构化结果。 raise NotImplementedError def run(self, client: BiliClient) - dict: 统一的执行入口包含异常兜底。 print(f[task] 开始执行 {self.name}) try: result self.execute(client) result.setdefault(success, True) result.setdefault(task, self.name) print(f[task] {self.name} 执行完成: {result}) return result except Exception as exc: print(f[task] {self.name} 执行失败: {exc}) print(traceback.format_exc()) return { success: False, task: self.name, error: str(exc), }BaseTask把异常处理公共化。具体任务只需要实现execute方法不需要关心调度器怎么调它也不需要对上层调用方暴露异常细节。5.3 实现一个具体任务用户信息巡检下面实现一个最简单的巡检任务它调用前面写的get_my_info并把结果返回给调度器。# robotbilibili/tasks/user_inspect.py from robotbilibili.api.user import get_my_info from robotbilibili.client import BiliClient from robotbilibili.tasks.base import BaseTask class UserInspectTask(BaseTask): name user_inspect description 获取当前登录用户信息校验登录态是否有效 def execute(self, client: BiliClient) - dict: info get_my_info(client) return {info: info}你可能会觉得这个任务太简单。没错本文的目的是先把框架跑通。真实项目里的“视频数据统计”“每周动态发布”任务只需要在execute里组合多个 API 调用即可框架本身不需要改动。5.4 组装定时调度器一次性的自动化脚本价值有限真正有用的是按计划自动执行。这里使用 APScheduler 的BlockingScheduler配合 cron 触发方式。# robotbilibili/scheduler.py from apscheduler.schedulers.blocking import BlockingScheduler from robotbilibili.client import BiliClient from robotbilibili.tasks.base import BaseTask def build_scheduler(client: BiliClient, tasks: list[BaseTask]): 把任务列表注册到调度器。 scheduler BlockingScheduler(timezoneAsia/Shanghai) for task in tasks: scheduler.add_job( task.run, interval, minutestask.run_interval_minutes, args[client], idtask.name, replace_existingTrue, max_instances1, coalesceTrue, ) return scheduler为了让调度器能拿到任务的运行间隔我们需要在任务类上增加一个统一字段。更新一下base.pyclass BaseTask(ABC): name: str base_task description: str run_interval_minutes: int 60然后让UserInspectTask指定运行频率class UserInspectTask(BaseTask): name user_inspect description 获取当前登录用户信息校验登录态是否有效 run_interval_minutes 305.5 主入口程序main.py是程序的唯一入口职责是读取配置、初始化客户端、注册任务、启动调度器。# main.py import os from dotenv import load_dotenv from robotbilibili.client import BiliClient from robotbilibili.scheduler import build_scheduler from robotbilibili.tasks.user_inspect import UserInspectTask load_dotenv() def main(): cookie os.getenv(BILI_COOKIE) if not cookie: raise ValueError(未找到 BILI_COOKIE 环境变量请先配置登录 Cookie) client BiliClient(cookie) tasks [ UserInspectTask(), ] scheduler build_scheduler(client, tasks) print([main] 调度器启动等待任务执行...) scheduler.start() if __name__ __main__: main()到这里一个最小的robotbilibili工程就跑通了启动后每 30 分钟执行一次用户信息巡检Cookie 有效则记录下当前账号信息Cookie 失效则打印异常。虽然功能很简单但整个架构已经具备扩展性。6. 定时任务与生产化调度6.1 为什么选择 APSchedulerwhile True sleep也能实现定时效果但工程上远远不够程序重启后任务状态丢失任务执行超时会阻塞下一次调度多个任务并发时没有任务去重机制。APScheduler 解决了这些问题。能力whilesleepAPSchedulercron 触发需要手动计算时间原生支持任务重叠控制无max_instances错过的任务补执行无coalesce任务持久化无支持 SQLAlchemyJobStore进程外管理无支持后台调度器6.2 错过任务的补偿策略max_instances1表示同一个任务在同一时间只能有一个实例在运行避免上一次还没执行完下一次又触发导致请求风暴。coalesceTrue表示如果系统停机错过多个调度点恢复后只补执行最近一次避免任务堆叠。这两个参数在生产环境非常重要。假设你的机器人凌晨跑数据统计任务电脑休眠了 3 个小时醒来后如果不加coalesce调度器可能连续触发 6 次任务瞬间打爆接口加上之后只执行一次。6.3 日志接入print只能用在开发阶段。生产环境需要把运行日志写入文件并记录结构化信息。Python 标准库的logging就够用。# robotbilibili/logger.py import logging from logging.handlers import RotatingFileHandler def setup_logger(name: str robotbilibili, log_file: str logs/robot.log): logger logging.getLogger(name) logger.setLevel(logging.INFO) file_handler RotatingFileHandler( log_file, maxBytes10 * 1024 * 1024, backupCount3, encodingutf-8 ) console_handler logging.StreamHandler() formatter logging.Formatter( %(asctime)s - %(name)s - %(levelname)s - %(message)s ) file_handler.setFormatter(formatter) console_handler.setFormatter(formatter) logger.addHandler(file_handler) logger.addHandler(console_handler) return logger6.4 失败重试与告警网络请求失败和业务 code 不等于 0 是两类不同的问题。前者可以重试后者通常不需要重试因为参数或权限出了问题重试只会浪费请求。在BiliClient.request中我们只对requests.RequestException做了重试这是合理的。业务 code 错误应该在api层通过抛异常终止任务然后在任务层记录日志并触发告警。生产环境建议接入企业微信机器人、钉钉机器人或邮件通知。最简单的做法是当任务的返回结果中successFalse时调用一个send_alert函数。def send_alert(task_name: str, error: str): 发送告警实际可替换为企业微信/钉钉/邮件。 print(f[alert] 任务 {task_name} 失败: {error})7. 运行结果与效果验证7.1 启动程序python main.py如果一切正常你会看到类似下面的输出[main] 调度器启动等待任务执行... [task] 开始执行 user_inspect [task] user_inspect 执行完成: {info: {uid: 12345678, name: 测试账号, level: 6}, success: True, task: user_inspect}这里的uid、name、level是自动从接口返回值中解析出来的说明整个调用链是通的。7.2 如何判断成功一个自动化任务是否成功不应该只凭“没有异常”来判断而要看业务结果。拿用户信息巡检来说成功的标准有两个HTTP 请求返回 200。业务 code 为 0且能解析出用户信息。如果请求 200 但业务 code 返回 -101通常表示未登录这个任务实际上已经失败了。所以在get_my_info里判断code ! 0并抛异常是非常关键的一步。7.3 模拟 Cookie 失效的验证方法把.env里的BILI_COOKIE改成一段乱码重新运行程序[task] user_inspect 执行失败: 获取用户信息失败: {code: -101, message: 账号未登录} [task] 执行失败: 获取用户信息失败: ...正常情况下调度器不会因为一个任务失败而退出。它会等待下一个调度周期继续执行。这就是把任务异常捕获放进BaseTask.run的价值任务失败不影响调度器存活。8. 常见问题与排查思路在 B 站自动化项目里最容易踩坑的不是代码逻辑而是登录态、接口变更和请求频率。下面整理了一张排查表建议收藏以备后用。问题现象可能原因排查方式解决方案接口返回 code-101Cookie 无效或过期先用浏览器打开 bilibili.com看是否已登录重新扫码登录并更新 Cookie接口返回 code-412请求频率过高或请求头异常查看请求日志统计同一接口调用频率增加请求间隔避免高频任务重叠定时任务不触发调度器没启动或时区设置错误检查 main.py 是否执行到 scheduler.start()明确指定 timezoneAsia/Shanghai任务执行到一半中断网络超时或接口响应过慢查看日志中的 timeout 异常调整 BiliClient 的 timeout 参数多次任务同时提交重复操作上一次任务没执行完就触发了下一次查看日志中是否有并发调用设置 max_instances1coalesceTrue.env 配置生效不了python-dotenv 没有找到 .env 文件检查文件路径和 load_dotenv() 调用位置在 main.py 顶部调用 load_dotenv()依赖版本冲突项目依赖与系统包冲突使用 pip freeze 检查依赖树使用虚拟环境锁定 requirements.txt 版本其中-412是最需要重视的。它往往意味着你的请求被风控策略盯上了。遇到这种情况正确做法不是加大并发去对抗而是降低频率、增加随机延迟、或者停止攻击性操作。robotbilibili的设计目标从来不是对抗风控而是做一个“礼貌”的自动化客户端。9. 使用边界与工程最佳实践技术本身是中性的但自动化脚本的使用场景必须设置边界。下面这些经验是我建议任何接触 B 站自动化的开发者都遵守的。9.1 合规底线第一认真阅读并遵守 B 站用户协议、开发者协议和 API 使用规范。只对你有权管理的账号进行操作不做任何影响平台正常秩序的事情比如批量注册、刷播放、刷评论、绕过验证码、攻击接口等。第二自动化只用来替代人工重复操作不能用来放大规模。人工一天发 10 条评论脚本也不应该变成一天发 10 万条。这不是能力问题而是合规问题。9.2 敏感信息管理Cookie 等同于账号的临时密码。把 Cookie 写死在代码里、提交到 GitHub、上传到公开博客都属于安全事故。正确做法使用.env文件存储并加入.gitignore。敏感文件加密存储或使用环境变量注入。定期更换 Cookie避免长期有效凭据泄露。# .gitignore .env logs/ __pycache__/ venv/9.3 请求频率与随机化B 站是商业平台接口有明确的负载压力。机器人的请求应该像人一样“有礼貌”默认请求间隔不低于 1 秒。批量任务之间增加统计抖动比如 1.5 秒到 3 秒的随机延迟。避免在整点时间并发跑大量任务。import random import time def polite_delay(min_seconds: float 1.0, max_seconds: float 3.0): time.sleep(random.uniform(min_seconds, max_seconds))9.4 接口版本兼容B 站接口会不定期调整。不要让所有代码直接依赖原始接口路径而是封装到api层。这样接口变更时只需要修改一个文件而不需要全局替换。建议在api层增加一个简单的版本标记# robotbilibili/api/user.py API_VERSION web-interface def get_my_info(client: BiliClient) - dict: url fhttps://api.bilibili.com/{API_VERSION}/nav ...9.5 日志与可观测性一个没有日志的自动化项目出问题时基本没法排查。至少需要做到每次请求记录接口路径、耗时、HTTP 状态码。每次任务记录开始时间、结束时间、结果摘要。日志按天或按大小轮转避免磁盘耗尽。任务失败时要有告警不能只靠人看日志。9.6 灰度发布与回滚如果机器人会执行写操作比如发动态、发评论建议先在小号上验证再逐步扩大到正式账号。每次变更代码后先在测试环境跑一个周期确认无误后再更新生产任务。# 先跑一次单次任务不启动调度器 python -c from dotenv import load_dotenv load_dotenv() import os from robotbilibili.client import BiliClient from robotbilibili.tasks.user_inspect import UserInspectTask client BiliClient(os.getenv(BILI_COOKIE)) print(UserInspectTask().run(client)) 这一步叫做“单次冒烟测试”它能让你在不等待调度周期的情况下快速验证代码是否可用。10. 总结与后续扩展方向robotbilibili教会我们的不只是一个 B 站脚本怎么写而是一套自动化任务系统怎么设计。梳理一下全文的核心要点登录态是整个自动化的地基Cookie 管理不好一切功能都白搭。请求客户端要统一封装把重试、超时、请求头集中处理才能支撑多任务场景。任务抽象是扩展性的关键把execute和run分离业务代码和调度代码就能各自演进。APScheduler 的max_instances、coalesce、时区设置是生产环境定时任务必须注意的细节。日志、告警、灰度、合规边界决定了这个项目能跑多久、跑得多稳。接下来如果你想继续深入可以往这几个方向做扩展增加 Web 管理端用 FastAPI 暴露一个管理界面在线查看任务状态、手动触发任务、更新 Cookie。接入消息队列用 Redis Celery 替代单机调度器把任务分发到多台机器执行适合账号规模较大的场景。数据持久化把接口返回的数据写入 SQLite 或 MySQL积累历史数据后做趋势分析。多平台扩展把BaseTask抽象复制一份新增抖音、小红书等平台的客户端实现整体架构可以直接复用.最后提醒一句自动化脚本不是越复杂越好而是越可控越好。先把用户信息巡检这个最小环节跑稳定再逐步叠加功能。希望这篇文章能帮你走出从“复制脚本”到“设计系统”的第一步。
返回列表