在 Python 自动化脚本的练习清单里robotbilibili是一个非常典型的项目它把网络请求、数据清洗、文件输出和定时任务串在一起结果又能直接看到。对于刚学完requests和pandas的开发者来说用这样一个工具来定期拉取 B 站公开视频信息、生成 CSV 或 Markdown 报告比单独写练习题更有实际感。robotbilibili这个名字可以拆成两部分robot表示自动化程序bilibili表示数据来源。它不是一个视频客户端也不是一个刷量工具而是一个面向公开数据的内容整理工具。你可以用它来归档自己收藏过的视频、登记某个 UP 主的更新列表、按关键词做内容复盘最终输出成本地文件方便后续用 Excel、Notion 或任意文本编辑器查看。下面会从项目定位开始逐步讲到依赖安装、目录设计、核心代码实现、运行验证、常见问题和生产化建议最后给出一个可以直接落地的项目骨架。1. 先理解 robotbilibili 要解决什么问题1.1 一个把“数据获取”和“报告输出”串起来的自动化项目很多开发者在做个人内容库时会遇到一个重复性场景每周想整理一批视频信息手动复制标题、UP 主、播放量、点赞数再粘贴到表格里非常耗时。如果数据量到了几十条手动整理的出错率也会明显上升。robotbilibili的思路是用一个 Python 脚本完成下面这几件事读取配置文件确定要查询的关键词或视频 ID 列表。调用平台公开接口获取视频元数据。把原始数据标准化成统一的数据模型。按视频 ID 去重写入本地data/raw目录。生成 CSV 和 Markdown 两种格式的报告保留原始数据的同时方便阅读。支持手动执行一次也支持按固定时间自动执行。这个定位决定了项目不需要复杂的前端也不需要数据库。对于个人学习工具来说本地文件已经足够后续如果数据量增长再迁移到 SQLite 也不难。1.2 核心流程采集、清洗、归档、报告整个自动化流程可以拆成六个阶段定时触发脚本到了指定时间自动运行或者用户手动通过命令行触发。参数读取从 YAML 配置和环境变量中读取关键词、接口地址、频率限制等参数。接口请求用requests调用公开接口注意控制请求频率。数据清洗把接口返回的 JSON 字段映射到VideoItem数据类处理缺失值和类型转换。本地归档把原始 JSON 和清洗后的 DataFrame 保存到data目录。报告生成使用 Jinja2 模板生成 Markdown 报告同时输出 CSV 方便表格工具打开。这样设计的核心原因是把“数据获取”和“数据使用”解耦。请求层只负责拿数据存储层只负责落盘报告层只负责格式转换。后续如果接口字段变了只需要修改数据清洗逻辑如果报告样式变了只需要修改模板。1.3 使用前必须明确的合规边界robotbilibili适合以下使用方式只采集平台明确允许公开的数据。只用于个人学习、内容归档、数据分析。请求频率控制在合理范围内不冲击目标服务。不伪造身份信息不绕过登录验证、验证码或风控机制。不把采集到的数据用于商业变现不恶意囤积数据。代码中的接口地址会写成占位符。实际开发时必须查阅平台最新的开放接口文档确认账号权限和调用限制。不要从项目名推测它支持“全自动操作”因为“能自动化”和“被平台允许自动化”是两件完全不同的事。注意robotbilibili的定位是公开数据整理工具。凡是涉及绕过限制、刷量、批量注册、抓取非公开用户信息的功能都不应该在项目里出现。2. 环境准备与项目初始化2.1 技术选型与依赖版本robotbilibili使用 Python 3.10 以上版本依赖尽量保持精简。核心库如下库用途建议版本范围requests发送 HTTP 请求requests2.31,3pandas数据清洗和 CSV 处理pandas2.0,3jinja2渲染 Markdown 报告模板jinja23.1,4python-dotenv读取.env环境变量python-dotenv1.0,2schedule定时任务调度schedule1.2,2pyyaml解析 YAML 配置pyyaml6.0,7pytest测试可选pytest7.4,9pandas在这个项目里主要用于去重、排序和 DataFrame 输出。如果只想保留最小依赖可以去掉pandas用 Python 标准库csv代替但代码会多出不少边界处理。2.2 目录结构设计推荐目录结构如下robotbilibili/ ├── config/ │ ├── settings.yaml │ └── .env.example ├── src/ │ ├── __init__.py │ ├── models.py │ ├── api_client.py │ ├── storage.py │ ├── report.py │ ├── scheduler.py │ └── main.py ├── data/ │ ├── raw/ │ └── reports/ ├── tests/ │ └── test_models.py ├── requirements.txt └── README.md把代码放在src目录下而不是全部堆在根目录是为了避免 “根目录脚本导入根目录脚本” 的混乱。config保存配置模板data保存运行时生成的文件tests放置单元测试。2.3 初始化虚拟环境在项目根目录执行python -m venv .venv source .venv/bin/activate pip install -r requirements.txtWindows 下的激活命令不同python -m venv .venv .venv\Scripts\activate pip install -r requirements.txt检查点运行pip list能看到requests、pandas、jinja2等依赖。如果安装速度慢可以临时使用国内镜像源但生产环境建议锁定固定依赖版本。生成requirements.txt时使用pip freeze requirements.txt不要手写版本号。3. 核心模块实现从请求到报告的完整链路3.1 配置分离YAML 与环境变量分工配置文件解决“哪些参数会频繁变化”的问题环境变量解决“哪些信息不能提交到仓库”的问题。config/settings.yaml示例app: timezone: Asia/Shanghai api: base_url: https://api.example.com # 实际项目替换为平台开放接口地址 timeout: 10 retries: 3 qps: 1 targets: keywords: - 机器人 - Python reports: out_dir: data/reports raw_dir: data/raw密钥和 Token 放在config/.env.example里ROBOTBILI_ACCESS_TOKENyour_token_here实际复制为.env后填写真实值并把.env加入.gitignore。这样做的原因是 YAML 文件容易在文档和分享中泄露而.env通常不会被版本控制工具上传。使用python-dotenv在程序启动时加载from dotenv import load_dotenv load_dotenv()3.2 数据模型用 dataclass 定义统一结构接口返回的字段可能会变但业务层最好只面对一个稳定的模型。这里定义一个VideoItem用于描述一条视频的基本信息。from dataclasses import dataclass dataclass class VideoItem: video_id: str title: str author: str published_at: str duration: int view_count: int like_count: int favorite_count: int classmethod def from_dict(cls, data: dict) - VideoItem: return cls( video_idstr(data.get(video_id) or ), titlestr(data.get(title) or ), authorstr(data.get(author) or unknown), published_atstr(data.get(published_at) or ), durationint(data.get(duration) or 0), view_countint(data.get(view_count) or 0), like_countint(data.get(like_count) or 0), favorite_countint(data.get(favorite_count) or 0), )各字段含义如下字段类型含义video_idstr视频唯一标识用于去重titlestr视频标题authorstrUP 主名称published_atstr发布时间ISO 格式字符串durationint视频时长单位秒view_countint播放量like_countint点赞数favorite_countint收藏数使用dataclass的好处是代码可读性高同时可以避免字典 key 拼写问题。字段默认值使用空字符串和0这样即使接口某个字段缺失也不会直接抛出KeyError。3.3 请求层封装公开接口调用与限流请求层最需要关注三件事超时、重试、限流。超时避免程序卡住重试处理临时网络错误限流避免请求过快。import time import requests from requests.adapters import HTTPAdapter class ApiClient: def __init__(self, base_url: str, token: str, timeout: int 10, retries: int 3): self.base_url base_url.rstrip(/) self.session requests.Session() self.session.headers.update( { User-Agent: robotbilibili/0.1 learning-project, Authorization: fBearer {token}, } ) self.timeout timeout adapter HTTPAdapter(max_retriesretries) self.session.mount(http://, adapter) self.session.mount(https://, adapter) def get_video_items(self, keyword: str, page_size: int 20) - list[dict]: params { keyword: keyword, page_size: page_size, } resp self.session.get( f{self.base_url}/video/search, paramsparams, timeoutself.timeout, ) resp.raise_for_status() data resp.json() time.sleep(1 / self.api_qps if hasattr(self, api_qps) else 0.5) return data[data][items]代码中需要注意几个点Authorization头是否必须取决于具体接口的鉴权方式。如果使用签名参数则改为在params中拼接签名。HTTPAdapter(max_retriesretries)只能处理连接错误不能处理业务错误。如果接口返回 200 但业务码失败需要手动处理。time.sleep是最朴素的限流手段。更精确的做法是在请求前记录时间根据时间差动态调整等待时间。这里的接口路径/video/search是示例实际项目要以平台开放文档为准。3.4 存储层原始数据落盘与去重拿到数据后先保存原始 JSON再进行去重和结构化输出。这样后续如果发现清洗逻辑有误还能用原始数据重新处理。import json import pathlib import pandas as pd from models import VideoItem class Storage: def __init__(self, raw_dir: str, report_dir: str): self.raw_dir pathlib.Path(raw_dir) self.report_dir pathlib.Path(report_dir) self.raw_dir.mkdir(parentsTrue, exist_okTrue) self.report_dir.mkdir(parentsTrue, exist_okTrue) def save_raw(self, items: list[dict], filename: str) - pathlib.Path: path self.raw_dir.joinpath(filename) with path.open(w, encodingutf-8) as f: json.dump(items, f, ensure_asciiFalse, indent2) return path def deduplicate(self, items: list[VideoItem]) - list[VideoItem]: seen set() unique_items [] for item in items: if item.video_id not in seen: seen.add(item.video_id) unique_items.append(item) return unique_itemsdeduplicate使用的是内存集合去重适合数据量在几千条以内的情况。如果数据量很大可以改成分批读取再合并。3.5 报告生成Markdown 与 CSV 一次输出报告生成放在report.py中核心思路是把标准化的VideoItem列表分别转成 DataFrame 和 Markdown 文本。import pathlib import pandas as pd from models import VideoItem def to_dataframe(items: list[VideoItem]) - pd.DataFrame: df pd.DataFrame([item.__dict__ for item in items]) return df def save_csv(df: pd.DataFrame, path: pathlib.Path) - None: df.to_csv(path, indexFalse, encodingutf-8-sig) def save_markdown(df: pd.DataFrame, path: pathlib.Path, title: str robotbilibili 报告) - None: path.parent.mkdir(parentsTrue, exist_okTrue) lines [f# {title}, ] lines.append(| 标题 | 作者 | 发布时间 | 播放量 | 收藏数 |) lines.append(| --- | --- | --- | --- | --- |) for row in df.itertuples(indexFalse): lines.append( f| {row.title} | {row.author} | {row.published_at} f| {row.view_count} | {row.favorite_count} | ) path.write_text(\n.join(lines), encodingutf-8)CSV 使用utf-8-sig编码这样用 Excel 打开时不会出现中文乱码。Markdown 报告使用write_text直接写入不依赖 Jinja2。如果后续报告结构复杂再引入模板也不迟。3.6 命令行入口支持手动执行一次命令行入口使用argparse提供--keyword、--out-dir、--once三个参数。import argparse import sys from pathlib import Path import yaml from api_client import ApiClient from models import VideoItem from storage import Storage from report import to_dataframe, save_csv, save_markdown def load_config(path: str) - dict: with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def main() - int: parser argparse.ArgumentParser(descriptionrobotbilibili public data report tool) parser.add_argument(--config, defaultconfig/settings.yaml) parser.add_argument(--keyword, default) parser.add_argument(--out-dir, default) parser.add_argument(--once, actionstore_true) args parser.parse_args() config load_config(args.config) keyword args.keyword or config[targets][keywords][0] base_url config[api][base_url] token config[api].get(token, ) client ApiClient(base_urlbase_url, tokentoken) raw_items client.get_video_items(keywordkeyword, page_size20) items [VideoItem.from_dict(item) for item in raw_items] storage Storage( raw_dirconfig[reports][raw_dir], report_dirconfig[reports][out_dir], ) unique_items storage.deduplicate(items) storage.save_raw(raw_items, fraw_{keyword}.json) df to_dataframe(unique_items) save_csv(df, Path(config[reports][out_dir]) / f{keyword}.csv) save_markdown(df, Path(config[reports][out_dir]) / f{keyword}.md) print(fprocessed {len(raw_items)} items, unique{len(unique_items)}) return 0 if __name__ __main__: sys.exit(main())注意main()返回状态码方便外部脚本判断是否成功。用一个--once参数区分手动执行和定时执行。4. 运行验证与结果分析4.1 手动执行一次在项目根目录执行python -m src.main --keyword 机器人 --out-dir data/reports --once执行后在终端能看到输出processed 20 items, unique18数字 20 表示接口返回 20 条18 表示去重后剩余 18 条说明有 2 条重复记录。如果项目采用src目录结构且src不是包可能需要先安装项目为本地包。这里简化处理将src下的模块放在同一目录中并在入口模块最上方确认import路径正确。实际项目可以使用pip install -e .完成安装再用robotbilibili命令启动。4.2 预期输出示例CSV 文件data/reports/机器人.csv的内容如下video_id,title,author,published_at,duration,view_count,like_count,favorite_count BV000001,机器人入门教程,示例UP主,2026-01-01T10:00:00,600,12000,800,300 BV000002,Python自动化实践,示例UP主,2026-01-02T11:00:00,900,8000,500,200Markdown 文件data/reports/机器人.md的内容如下# robotbilibili 报告 | 标题 | 作者 | 发布时间 | 播放量 | 收藏数 | | --- | --- | --- | --- | --- | | 机器人入门教程 | 示例UP主 | 2026-01-01T10:00:00 | 12000 | 300 | | Python自动化实践 | 示例UP主 | 2026-01-02T11:00:00 | 8000 | 200 |这里展示的是示例数据实际字段以接口返回为准。4.3 验证结果是否可靠验证不能只看程序是否退出。至少要做四步检查检查退出码是否为 0。检查日志中是否出现processed字样。检查data/raw下的 JSON 文件是否完整。检查 CSV 是否能用 Excel 正常打开且没有中文乱码。命令示例echo $? ls -lh data/raw data/reports head -5 data/reports/机器人.csvhead -5只能看表头和前几行要确认总行数是否接近预期可以用wc -l。注意不要只验证程序能启动还要验证输入、输出、异常分支和日志是否符合预期。定时任务场景中程序正常退出但文件没生成是最容易忽略的问题。5. 常见问题与排查路径5.1 接口返回 412 或 403现象脚本运行时报HTTP 412 Precondition Failed或HTTP 403 Forbidden。可能原因请求频率过高触发了风控。Token 失效或权限不足。缺少必要的请求头或签名参数。接口路径写错请求到了错误的服务。检查方式打印请求 URL 和参数确认没有多余字符。查看平台文档确认请求头是否要求User-Agent、Referer或签名。检查 Token 是否过期。确认qps配置是否过低。处理建议增加time.sleep把 QPS 降到 1 以下。刷新 Token。使用日志记录每次请求的状态码不要只记录最终异常。如果接口需要签名把签名逻辑单独封装成函数方便测试。5.2 CSV 中文乱码现象CSV 文件用 Excel 打开后中文内容变成乱码。原因DataFrame.to_csv默认编码是utf-8Excel 在某些区域语言环境下默认不识别。解决方式写入时使用encodingutf-8-sig。df.to_csv(path, indexFalse, encodingutf-8-sig)这个编码会在文件开头加入 BOMExcel 识别后会正常显示中文。JSON 文件写入时使用ensure_asciiFalse可以让中文直接显示在文件里方便排查。5.3 定时任务不触发现象使用schedule设置每天执行但脚本运行后没有打印执行日志。常见原因schedule的while True循环没有真正进入。时区配置错误导致执行时间不符合预期。脚本抛异常后循环退出后续任务不再执行。在main()中提前return任务循环没有启动。正确的最小调度代码import schedule import time from src.main import main schedule.every().day.at(09:00).do(main) while True: schedule.run_pending() time.sleep(1)注意schedule不是持久化任务队列。进程重启后任务需要重新注册。生产环境建议使用系统cron或容器平台的定时任务。5.4 数据去重失效现象报告中的视频 ID 出现重复。可能原因接口返回字段名不是video_id导致from_dict读到空字符串。去重逻辑只处理了当次请求的数据没有和历史上已经保存的数据做对比。数据从原始 JSON 映射到VideoItem时隐式转换出错。解决方式打印raw_items[0]确认字段名。去重时把历史 CSV 中的video_id也加入seen集合。为VideoItem.from_dict写单元测试覆盖字段缺失的情况。6. 从学习脚本到长期运行的加固方案6.1 学习环境和生产环境的差异robotbilibili可以作为学习项目也可以升级成定时服务。两者关注点完全不同维度学习项目生产运行数据存储本地 CSV / Markdown数据库或对象存储配置YAML .env配置中心或环境变量加密日志print输出结构化日志、日志轮转任务调度schedulecron、CI 定时任务、容器调度异常处理try-except简单打印告警通知 重试 死信处理测试跳过往后补每次代码提交都要跑单测数据备份本机文件定期备份和保留策略不要在本地文件方案上直接加一个while True就部署到服务器。至少要考虑磁盘空间增长、日志文件过大、接口字段变更时如何回滚。6.2 日志与异常可观测print只能用于调试。生产运行需要记录时间、级别、模块和关键上下文。import logging logging.basicConfig( levellogging.INFO, format%(asctime)s %(levelname)s %(name)s - %(message)s, handlers[ logging.FileHandler(robotbilibili.log, encodingutf-8), logging.StreamHandler(), ], ) logger logging.getLogger(__name__)在请求层和报告生成层各增加一条结构化日志logger.info(request video items keyword%s count%s, keyword, len(raw_items))如果一次任务失败至少能看到失败发生在请求阶段、数据处理阶段还是文件写入阶段。6.3 上线前检查清单在正式部署之前按下面清单逐项核对[ ] 是否已经把 Token、密钥移出代码仓库[ ] 是否确认接口调用频率没有超过平台限制[ ] 是否对接口字段缺失做了默认值处理[ ] 是否在请求层加入了超时和重试[ ] 是否对 CSV 使用了utf-8-sig编码[ ] 是否有历史数据去重逻辑而不是只对单批次数据去重[ ] 是否对日志文件做了大小轮转[ ] 是否知道任务启动后如何手动停止[ ] 是否能在模拟接口不可用时不产生脏数据[ ] 是否限制了单次采集条数和总数据量这个清单同时适用于个人项目和团队项目可以防止最常见的“本地能跑部署就挂”问题。6.4 扩展方向robotbilibili做完后可以从以下几个方向继续延伸把 CSV 落盘改成 SQLite 存储支持按时间范围查询。增加多个关键词并行采集但每个并发线程必须独立限流。把报告生成从纯文本升级成 HTML方便浏览器查看。增加趋势统计同一关键词连续采集 7 天后输出播放量变化曲线。把告警接进企业微信、钉钉或邮件任务失败时通知维护者。增加配置校验启动时如果字段缺失就快速失败而不是运行到一半才报错。为VideoItem.from_dict、deduplicate和save_csv补充单元测试保证后续改动不破坏现有功能。最重要的不是把功能堆得多全而是先保持“一次只做一件小事、每个模块都能单独验证”的结构。robotbilibili的代码量不大但足够让你完整经历一次真实的 Python 自动化项目开发流程设计配置、封装请求、清洗数据、生成报告、定时运行、排查问题。把这个流程走通之后再迁移到其他数据源或业务场景都会顺利很多。 Python自动化数据采集与报告生成：基于robotbilibili的实战指南
在 Python 自动化脚本的练习清单里robotbilibili是一个非常典型的项目它把网络请求、数据清洗、文件输出和定时任务串在一起结果又能直接看到。对于刚学完requests和pandas的开发者来说用这样一个工具来定期拉取 B 站公开视频信息、生成 CSV 或 Markdown 报告比单独写练习题更有实际感。robotbilibili这个名字可以拆成两部分robot表示自动化程序bilibili表示数据来源。它不是一个视频客户端也不是一个刷量工具而是一个面向公开数据的内容整理工具。你可以用它来归档自己收藏过的视频、登记某个 UP 主的更新列表、按关键词做内容复盘最终输出成本地文件方便后续用 Excel、Notion 或任意文本编辑器查看。下面会从项目定位开始逐步讲到依赖安装、目录设计、核心代码实现、运行验证、常见问题和生产化建议最后给出一个可以直接落地的项目骨架。1. 先理解 robotbilibili 要解决什么问题1.1 一个把“数据获取”和“报告输出”串起来的自动化项目很多开发者在做个人内容库时会遇到一个重复性场景每周想整理一批视频信息手动复制标题、UP 主、播放量、点赞数再粘贴到表格里非常耗时。如果数据量到了几十条手动整理的出错率也会明显上升。robotbilibili的思路是用一个 Python 脚本完成下面这几件事读取配置文件确定要查询的关键词或视频 ID 列表。调用平台公开接口获取视频元数据。把原始数据标准化成统一的数据模型。按视频 ID 去重写入本地data/raw目录。生成 CSV 和 Markdown 两种格式的报告保留原始数据的同时方便阅读。支持手动执行一次也支持按固定时间自动执行。这个定位决定了项目不需要复杂的前端也不需要数据库。对于个人学习工具来说本地文件已经足够后续如果数据量增长再迁移到 SQLite 也不难。1.2 核心流程采集、清洗、归档、报告整个自动化流程可以拆成六个阶段定时触发脚本到了指定时间自动运行或者用户手动通过命令行触发。参数读取从 YAML 配置和环境变量中读取关键词、接口地址、频率限制等参数。接口请求用requests调用公开接口注意控制请求频率。数据清洗把接口返回的 JSON 字段映射到VideoItem数据类处理缺失值和类型转换。本地归档把原始 JSON 和清洗后的 DataFrame 保存到data目录。报告生成使用 Jinja2 模板生成 Markdown 报告同时输出 CSV 方便表格工具打开。这样设计的核心原因是把“数据获取”和“数据使用”解耦。请求层只负责拿数据存储层只负责落盘报告层只负责格式转换。后续如果接口字段变了只需要修改数据清洗逻辑如果报告样式变了只需要修改模板。1.3 使用前必须明确的合规边界robotbilibili适合以下使用方式只采集平台明确允许公开的数据。只用于个人学习、内容归档、数据分析。请求频率控制在合理范围内不冲击目标服务。不伪造身份信息不绕过登录验证、验证码或风控机制。不把采集到的数据用于商业变现不恶意囤积数据。代码中的接口地址会写成占位符。实际开发时必须查阅平台最新的开放接口文档确认账号权限和调用限制。不要从项目名推测它支持“全自动操作”因为“能自动化”和“被平台允许自动化”是两件完全不同的事。注意robotbilibili的定位是公开数据整理工具。凡是涉及绕过限制、刷量、批量注册、抓取非公开用户信息的功能都不应该在项目里出现。2. 环境准备与项目初始化2.1 技术选型与依赖版本robotbilibili使用 Python 3.10 以上版本依赖尽量保持精简。核心库如下库用途建议版本范围requests发送 HTTP 请求requests2.31,3pandas数据清洗和 CSV 处理pandas2.0,3jinja2渲染 Markdown 报告模板jinja23.1,4python-dotenv读取.env环境变量python-dotenv1.0,2schedule定时任务调度schedule1.2,2pyyaml解析 YAML 配置pyyaml6.0,7pytest测试可选pytest7.4,9pandas在这个项目里主要用于去重、排序和 DataFrame 输出。如果只想保留最小依赖可以去掉pandas用 Python 标准库csv代替但代码会多出不少边界处理。2.2 目录结构设计推荐目录结构如下robotbilibili/ ├── config/ │ ├── settings.yaml │ └── .env.example ├── src/ │ ├── __init__.py │ ├── models.py │ ├── api_client.py │ ├── storage.py │ ├── report.py │ ├── scheduler.py │ └── main.py ├── data/ │ ├── raw/ │ └── reports/ ├── tests/ │ └── test_models.py ├── requirements.txt └── README.md把代码放在src目录下而不是全部堆在根目录是为了避免 “根目录脚本导入根目录脚本” 的混乱。config保存配置模板data保存运行时生成的文件tests放置单元测试。2.3 初始化虚拟环境在项目根目录执行python -m venv .venv source .venv/bin/activate pip install -r requirements.txtWindows 下的激活命令不同python -m venv .venv .venv\Scripts\activate pip install -r requirements.txt检查点运行pip list能看到requests、pandas、jinja2等依赖。如果安装速度慢可以临时使用国内镜像源但生产环境建议锁定固定依赖版本。生成requirements.txt时使用pip freeze requirements.txt不要手写版本号。3. 核心模块实现从请求到报告的完整链路3.1 配置分离YAML 与环境变量分工配置文件解决“哪些参数会频繁变化”的问题环境变量解决“哪些信息不能提交到仓库”的问题。config/settings.yaml示例app: timezone: Asia/Shanghai api: base_url: https://api.example.com # 实际项目替换为平台开放接口地址 timeout: 10 retries: 3 qps: 1 targets: keywords: - 机器人 - Python reports: out_dir: data/reports raw_dir: data/raw密钥和 Token 放在config/.env.example里ROBOTBILI_ACCESS_TOKENyour_token_here实际复制为.env后填写真实值并把.env加入.gitignore。这样做的原因是 YAML 文件容易在文档和分享中泄露而.env通常不会被版本控制工具上传。使用python-dotenv在程序启动时加载from dotenv import load_dotenv load_dotenv()3.2 数据模型用 dataclass 定义统一结构接口返回的字段可能会变但业务层最好只面对一个稳定的模型。这里定义一个VideoItem用于描述一条视频的基本信息。from dataclasses import dataclass dataclass class VideoItem: video_id: str title: str author: str published_at: str duration: int view_count: int like_count: int favorite_count: int classmethod def from_dict(cls, data: dict) - VideoItem: return cls( video_idstr(data.get(video_id) or ), titlestr(data.get(title) or ), authorstr(data.get(author) or unknown), published_atstr(data.get(published_at) or ), durationint(data.get(duration) or 0), view_countint(data.get(view_count) or 0), like_countint(data.get(like_count) or 0), favorite_countint(data.get(favorite_count) or 0), )各字段含义如下字段类型含义video_idstr视频唯一标识用于去重titlestr视频标题authorstrUP 主名称published_atstr发布时间ISO 格式字符串durationint视频时长单位秒view_countint播放量like_countint点赞数favorite_countint收藏数使用dataclass的好处是代码可读性高同时可以避免字典 key 拼写问题。字段默认值使用空字符串和0这样即使接口某个字段缺失也不会直接抛出KeyError。3.3 请求层封装公开接口调用与限流请求层最需要关注三件事超时、重试、限流。超时避免程序卡住重试处理临时网络错误限流避免请求过快。import time import requests from requests.adapters import HTTPAdapter class ApiClient: def __init__(self, base_url: str, token: str, timeout: int 10, retries: int 3): self.base_url base_url.rstrip(/) self.session requests.Session() self.session.headers.update( { User-Agent: robotbilibili/0.1 learning-project, Authorization: fBearer {token}, } ) self.timeout timeout adapter HTTPAdapter(max_retriesretries) self.session.mount(http://, adapter) self.session.mount(https://, adapter) def get_video_items(self, keyword: str, page_size: int 20) - list[dict]: params { keyword: keyword, page_size: page_size, } resp self.session.get( f{self.base_url}/video/search, paramsparams, timeoutself.timeout, ) resp.raise_for_status() data resp.json() time.sleep(1 / self.api_qps if hasattr(self, api_qps) else 0.5) return data[data][items]代码中需要注意几个点Authorization头是否必须取决于具体接口的鉴权方式。如果使用签名参数则改为在params中拼接签名。HTTPAdapter(max_retriesretries)只能处理连接错误不能处理业务错误。如果接口返回 200 但业务码失败需要手动处理。time.sleep是最朴素的限流手段。更精确的做法是在请求前记录时间根据时间差动态调整等待时间。这里的接口路径/video/search是示例实际项目要以平台开放文档为准。3.4 存储层原始数据落盘与去重拿到数据后先保存原始 JSON再进行去重和结构化输出。这样后续如果发现清洗逻辑有误还能用原始数据重新处理。import json import pathlib import pandas as pd from models import VideoItem class Storage: def __init__(self, raw_dir: str, report_dir: str): self.raw_dir pathlib.Path(raw_dir) self.report_dir pathlib.Path(report_dir) self.raw_dir.mkdir(parentsTrue, exist_okTrue) self.report_dir.mkdir(parentsTrue, exist_okTrue) def save_raw(self, items: list[dict], filename: str) - pathlib.Path: path self.raw_dir.joinpath(filename) with path.open(w, encodingutf-8) as f: json.dump(items, f, ensure_asciiFalse, indent2) return path def deduplicate(self, items: list[VideoItem]) - list[VideoItem]: seen set() unique_items [] for item in items: if item.video_id not in seen: seen.add(item.video_id) unique_items.append(item) return unique_itemsdeduplicate使用的是内存集合去重适合数据量在几千条以内的情况。如果数据量很大可以改成分批读取再合并。3.5 报告生成Markdown 与 CSV 一次输出报告生成放在report.py中核心思路是把标准化的VideoItem列表分别转成 DataFrame 和 Markdown 文本。import pathlib import pandas as pd from models import VideoItem def to_dataframe(items: list[VideoItem]) - pd.DataFrame: df pd.DataFrame([item.__dict__ for item in items]) return df def save_csv(df: pd.DataFrame, path: pathlib.Path) - None: df.to_csv(path, indexFalse, encodingutf-8-sig) def save_markdown(df: pd.DataFrame, path: pathlib.Path, title: str robotbilibili 报告) - None: path.parent.mkdir(parentsTrue, exist_okTrue) lines [f# {title}, ] lines.append(| 标题 | 作者 | 发布时间 | 播放量 | 收藏数 |) lines.append(| --- | --- | --- | --- | --- |) for row in df.itertuples(indexFalse): lines.append( f| {row.title} | {row.author} | {row.published_at} f| {row.view_count} | {row.favorite_count} | ) path.write_text(\n.join(lines), encodingutf-8)CSV 使用utf-8-sig编码这样用 Excel 打开时不会出现中文乱码。Markdown 报告使用write_text直接写入不依赖 Jinja2。如果后续报告结构复杂再引入模板也不迟。3.6 命令行入口支持手动执行一次命令行入口使用argparse提供--keyword、--out-dir、--once三个参数。import argparse import sys from pathlib import Path import yaml from api_client import ApiClient from models import VideoItem from storage import Storage from report import to_dataframe, save_csv, save_markdown def load_config(path: str) - dict: with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def main() - int: parser argparse.ArgumentParser(descriptionrobotbilibili public data report tool) parser.add_argument(--config, defaultconfig/settings.yaml) parser.add_argument(--keyword, default) parser.add_argument(--out-dir, default) parser.add_argument(--once, actionstore_true) args parser.parse_args() config load_config(args.config) keyword args.keyword or config[targets][keywords][0] base_url config[api][base_url] token config[api].get(token, ) client ApiClient(base_urlbase_url, tokentoken) raw_items client.get_video_items(keywordkeyword, page_size20) items [VideoItem.from_dict(item) for item in raw_items] storage Storage( raw_dirconfig[reports][raw_dir], report_dirconfig[reports][out_dir], ) unique_items storage.deduplicate(items) storage.save_raw(raw_items, fraw_{keyword}.json) df to_dataframe(unique_items) save_csv(df, Path(config[reports][out_dir]) / f{keyword}.csv) save_markdown(df, Path(config[reports][out_dir]) / f{keyword}.md) print(fprocessed {len(raw_items)} items, unique{len(unique_items)}) return 0 if __name__ __main__: sys.exit(main())注意main()返回状态码方便外部脚本判断是否成功。用一个--once参数区分手动执行和定时执行。4. 运行验证与结果分析4.1 手动执行一次在项目根目录执行python -m src.main --keyword 机器人 --out-dir data/reports --once执行后在终端能看到输出processed 20 items, unique18数字 20 表示接口返回 20 条18 表示去重后剩余 18 条说明有 2 条重复记录。如果项目采用src目录结构且src不是包可能需要先安装项目为本地包。这里简化处理将src下的模块放在同一目录中并在入口模块最上方确认import路径正确。实际项目可以使用pip install -e .完成安装再用robotbilibili命令启动。4.2 预期输出示例CSV 文件data/reports/机器人.csv的内容如下video_id,title,author,published_at,duration,view_count,like_count,favorite_count BV000001,机器人入门教程,示例UP主,2026-01-01T10:00:00,600,12000,800,300 BV000002,Python自动化实践,示例UP主,2026-01-02T11:00:00,900,8000,500,200Markdown 文件data/reports/机器人.md的内容如下# robotbilibili 报告 | 标题 | 作者 | 发布时间 | 播放量 | 收藏数 | | --- | --- | --- | --- | --- | | 机器人入门教程 | 示例UP主 | 2026-01-01T10:00:00 | 12000 | 300 | | Python自动化实践 | 示例UP主 | 2026-01-02T11:00:00 | 8000 | 200 |这里展示的是示例数据实际字段以接口返回为准。4.3 验证结果是否可靠验证不能只看程序是否退出。至少要做四步检查检查退出码是否为 0。检查日志中是否出现processed字样。检查data/raw下的 JSON 文件是否完整。检查 CSV 是否能用 Excel 正常打开且没有中文乱码。命令示例echo $? ls -lh data/raw data/reports head -5 data/reports/机器人.csvhead -5只能看表头和前几行要确认总行数是否接近预期可以用wc -l。注意不要只验证程序能启动还要验证输入、输出、异常分支和日志是否符合预期。定时任务场景中程序正常退出但文件没生成是最容易忽略的问题。5. 常见问题与排查路径5.1 接口返回 412 或 403现象脚本运行时报HTTP 412 Precondition Failed或HTTP 403 Forbidden。可能原因请求频率过高触发了风控。Token 失效或权限不足。缺少必要的请求头或签名参数。接口路径写错请求到了错误的服务。检查方式打印请求 URL 和参数确认没有多余字符。查看平台文档确认请求头是否要求User-Agent、Referer或签名。检查 Token 是否过期。确认qps配置是否过低。处理建议增加time.sleep把 QPS 降到 1 以下。刷新 Token。使用日志记录每次请求的状态码不要只记录最终异常。如果接口需要签名把签名逻辑单独封装成函数方便测试。5.2 CSV 中文乱码现象CSV 文件用 Excel 打开后中文内容变成乱码。原因DataFrame.to_csv默认编码是utf-8Excel 在某些区域语言环境下默认不识别。解决方式写入时使用encodingutf-8-sig。df.to_csv(path, indexFalse, encodingutf-8-sig)这个编码会在文件开头加入 BOMExcel 识别后会正常显示中文。JSON 文件写入时使用ensure_asciiFalse可以让中文直接显示在文件里方便排查。5.3 定时任务不触发现象使用schedule设置每天执行但脚本运行后没有打印执行日志。常见原因schedule的while True循环没有真正进入。时区配置错误导致执行时间不符合预期。脚本抛异常后循环退出后续任务不再执行。在main()中提前return任务循环没有启动。正确的最小调度代码import schedule import time from src.main import main schedule.every().day.at(09:00).do(main) while True: schedule.run_pending() time.sleep(1)注意schedule不是持久化任务队列。进程重启后任务需要重新注册。生产环境建议使用系统cron或容器平台的定时任务。5.4 数据去重失效现象报告中的视频 ID 出现重复。可能原因接口返回字段名不是video_id导致from_dict读到空字符串。去重逻辑只处理了当次请求的数据没有和历史上已经保存的数据做对比。数据从原始 JSON 映射到VideoItem时隐式转换出错。解决方式打印raw_items[0]确认字段名。去重时把历史 CSV 中的video_id也加入seen集合。为VideoItem.from_dict写单元测试覆盖字段缺失的情况。6. 从学习脚本到长期运行的加固方案6.1 学习环境和生产环境的差异robotbilibili可以作为学习项目也可以升级成定时服务。两者关注点完全不同维度学习项目生产运行数据存储本地 CSV / Markdown数据库或对象存储配置YAML .env配置中心或环境变量加密日志print输出结构化日志、日志轮转任务调度schedulecron、CI 定时任务、容器调度异常处理try-except简单打印告警通知 重试 死信处理测试跳过往后补每次代码提交都要跑单测数据备份本机文件定期备份和保留策略不要在本地文件方案上直接加一个while True就部署到服务器。至少要考虑磁盘空间增长、日志文件过大、接口字段变更时如何回滚。6.2 日志与异常可观测print只能用于调试。生产运行需要记录时间、级别、模块和关键上下文。import logging logging.basicConfig( levellogging.INFO, format%(asctime)s %(levelname)s %(name)s - %(message)s, handlers[ logging.FileHandler(robotbilibili.log, encodingutf-8), logging.StreamHandler(), ], ) logger logging.getLogger(__name__)在请求层和报告生成层各增加一条结构化日志logger.info(request video items keyword%s count%s, keyword, len(raw_items))如果一次任务失败至少能看到失败发生在请求阶段、数据处理阶段还是文件写入阶段。6.3 上线前检查清单在正式部署之前按下面清单逐项核对[ ] 是否已经把 Token、密钥移出代码仓库[ ] 是否确认接口调用频率没有超过平台限制[ ] 是否对接口字段缺失做了默认值处理[ ] 是否在请求层加入了超时和重试[ ] 是否对 CSV 使用了utf-8-sig编码[ ] 是否有历史数据去重逻辑而不是只对单批次数据去重[ ] 是否对日志文件做了大小轮转[ ] 是否知道任务启动后如何手动停止[ ] 是否能在模拟接口不可用时不产生脏数据[ ] 是否限制了单次采集条数和总数据量这个清单同时适用于个人项目和团队项目可以防止最常见的“本地能跑部署就挂”问题。6.4 扩展方向robotbilibili做完后可以从以下几个方向继续延伸把 CSV 落盘改成 SQLite 存储支持按时间范围查询。增加多个关键词并行采集但每个并发线程必须独立限流。把报告生成从纯文本升级成 HTML方便浏览器查看。增加趋势统计同一关键词连续采集 7 天后输出播放量变化曲线。把告警接进企业微信、钉钉或邮件任务失败时通知维护者。增加配置校验启动时如果字段缺失就快速失败而不是运行到一半才报错。为VideoItem.from_dict、deduplicate和save_csv补充单元测试保证后续改动不破坏现有功能。最重要的不是把功能堆得多全而是先保持“一次只做一件小事、每个模块都能单独验证”的结构。robotbilibili的代码量不大但足够让你完整经历一次真实的 Python 自动化项目开发流程设计配置、封装请求、清洗数据、生成报告、定时运行、排查问题。把这个流程走通之后再迁移到其他数据源或业务场景都会顺利很多。