用Python自动抓取视频热门榜单：从数据采集到趋势分析的完整实践

1. 动手之前先想清楚：热门榜单数据到底能解决什么问题

先聊个挺常见的工作场景：你负责一个内容账号的日常运营，每天早上打开后台第一件事就是看竞品又更新了什么，今天平台上什么话题在涨，哪个方向值得跟。这些信息如果全靠人工刷网页，一天至少耗掉一两个小时，而且刷完还容易漏。我以前就是这种状态，直到某天发现自己把大量时间浪费在“看榜单”而不是“用榜单”上，才下决心做一个能自动抓取视频平台热门榜单的小工具。

这个工具的本质，就是把平台公开的热门数据定时抓下来，存进本地数据库，再按自己的需求做过滤、排序、提醒。它能解决的问题大致有这几类：

• 追热点：每天自动记录全站热门视频，不用半夜盯着榜单变动，第二天起床看前一天的快照即可。
• 竞品监控：把同行账号的视频数据也纳入抓取范围，定期对比播放量、点赞数、弹幕数的增长趋势。
• 选题决策：连续抓一周数据后，统计哪些题材反复出现在高位，哪些是“一日游”型热点，选题时心里更有底。
• 汇报自动化：每周把热门数据的统计结果自动整理成表格，省掉手动截图和抄数字的时间。

适合参考这篇文章的人，我大致分成三类：一是做内容运营和自媒体，想用数据辅助选题的；二是刚开始学爬虫，想找一个真实、可控、又不复杂的练手项目的；三是团队内部想做竞品情报或行业观察，需要低成本方案的技术人员。无论你是哪一类，这篇文章都会从一个完整的工程视角来拆解，从目标设定、技术选型到代码实现、避坑经验都覆盖到。

需要先说明的是，我这里以主流的视频平台公开榜单接口为例子，抓取的都是无需登录就能看到的公开数据，并且抓取频率会控制在一个对平台没有任何压力的范围内。项目本身不涉及任何违规手段，核心目的是把“数据采集—存储—分析”这条链路跑通。

2. 技术选型：为什么用Python轻量组合，而不是重型框架

先说说技术栈。这个项目我用的是 Python 3.10 + httpx + SQLite，解析方面因为返回的是标准 JSON，所以直接用 Python 内置的 json 模块，连 BeautifulSoup 都没用上。很多人一听到“抓取”两个字就想到 Scrapy、Selenium、Playwright 这些重量级工具，但对于抓公开榜单接口这个场景，属于典型的大炮打蚊子。

2.1 请求库：httpx 比 requests 好在哪

requests 是老牌的 HTTP 库，但我在新项目里更倾向于 httpx。核心原因有几个：一是 httpx 支持 HTTP/2，对现代接口的兼容性和性能更好；二是它的接口设计跟 requests 高度相似，迁移成本几乎为零；三是它天然支持异步，万一后面要同时抓多个榜单，写异步并发很顺手。

import httpx headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" } def fetch_json(url: str, params: dict | None = None) -> dict: with httpx.Client( headers=headers, timeout=15.0, follow_redirects=True, ) as client: resp = client.get(url, params=params) resp.raise_for_status() return resp.json()

这段代码是一个最基础的请求封装。注意我设置了 15 秒超时和自动跟随重定向，这些细节在真实环境里很重要，否则一个慢接口就能把整个定时任务卡死。

2.2 存储选型：SQLite 完全够用

有人会习惯性地上 MySQL 或 PostgreSQL，但说实话，个人脚本、小团队内部工具，数据量在百万级以下，SQLite 是体验最好的方案。它不需要单独部署服务，一个文件就是整个数据库，备份直接复制文件搞定。配合 Python 内置的 sqlite3 模块，零依赖。

我在设计表结构时，参考了“排行榜快照”的模式，也就是每次抓取都记录当时榜单的完整状态，而不是只记录增量。这样做的目的是：热门榜单本身是动态变化的，只有记录每一时刻的完整排名，才能分析出“哪些视频一直在榜”“哪些视频是突然后劲爆发”这类趋势信息。

CREATE TABLE if NOT EXISTS hot_rank ( id INTEGER PRIMARY KEY AUTOINCREMENT, video_id TEXT NOT NULL, rank INTEGER NOT NULL, title TEXT, author TEXT, play_count INTEGER, like_count INTEGER, danmaku_count INTEGER, snapshot_time TEXT NOT NULL DEFAULT (datetime('now', 'localtime')), UNIQUE(video_id, snapshot_time) );

这张表的关键在最后一行的联合唯一约束：同一时刻、同一个视频只能有一条快照记录。这样即使脚本意外重复执行，也不会插入脏数据。

2.3 为什么不用 Scrapy

Scrapy 是一个很优秀的爬虫框架，但它解决的是“大规模、多站点、需要管道化处理”的复杂问题。对于单接口、单表的小工具，Scrapy 的配置成本反而很高——你要搭项目结构、配置 Item Pipeline、写 Spider，还得理解框架的异步调度机制。这就好比你想煮一碗泡面，结果把整套厨具都搬出来。能用十行代码解决的问题，不要用一百行去解决，这是我一直坚持的原则。等你真的需要抓几百个站点、做分布式调度的时候，再迁移到 Scrapy 也不迟。

3. 抓取链路核心实现：从定位接口到数据落库

这章是动手环节，我按正常的开发顺序来讲：先找到数据从哪来，再分析请求参数，最后写代码把数据解析、清洗、入库。

3.1 通过开发者工具定位真实数据接口

很多教程会直接教你去解析 HTML 页面，用 BeautifulSoup 去抠节点。但稍微正规一点的平台，首屏数据基本都是通过 XHR 异步加载的。这意味着你只要打开浏览器的开发者工具（F12），切到 Network 面板，刷新页面，就能看到浏览器到底向哪个接口发了请求、返回了什么数据。

以 B 站的热门榜单为例，打开页面后 Network 面板里会有一个 popular 相关的请求，响应是标准的 JSON 结构。这种方式找到的接口，请求参数清晰、响应结构稳定，比解析 HTML 高出不止一个维度。

操作步骤很简单：

1. 打开目标榜单页面
2. F12 打开开发者工具，切到 Network
3. 刷新页面，筛选 XHR 或 Fetch 类型的请求
4. 逐个查看响应内容，找到返回视频列表的接口
5. 右键这个请求，复制为 cURL，方便分析请求头

第 5 步特别实用。复制为 cURL 之后，你可以把它粘贴到 Postman 或 https://curlconverter.com 这类工具里，自动转换成 Python 代码，包括所有请求头都能带过来。

3.2 请求参数与频率控制

找到接口后，别急着写代码，先分析一下它的 Query String。通常这些参数里有两个关键信息：一个是分页参数（pagination），一个是时间窗口参数（time window）。

我在实际项目里建议控制抓取频率在每 10 到 15 分钟一次。热门榜单的更新频率本身就不是秒级的，太频繁的请求不仅浪费资源，还容易被平台的反爬机制盯上。如果你只是做每日回顾，甚至每天抓三次就够：早上、中午、晚上各一次，捕捉不同时段的热点变化。

3.3 数据解析与字段清洗

从接口返回的 JSON 里，视频列表通常嵌套在 data 字段下面。解析的时候建议写一个独立的解析函数，方便测试和维护。

def parse_hot_items(raw: dict) -> list[dict]: items = [] video_list = raw.get("data", {}).get("list", []) for item in video_list: # 跳过播放量为空或明显异常的数据 if not item.get("stat", {}).get("view"): continue items.append({ "video_id": item["bvid"], "title": item["title"].strip(), "author": item["owner"]["name"], "play_count": item["stat"]["view"], "like_count": item["stat"]["like"], "danmaku_count": item["stat"]["danmaku"], "rank": int(item.get("rank", 0)), }) return items

清洗这一步很多人会省略，但我觉得非常值得做。比如标题里的首尾空格，比如播放量字段偶尔会出现负数或超大的异常值，这些都是脏数据。如果不处理，后续做趋势分析的时候会出现很离谱的误差。

3.4 主流程串联

把前面的模块串起来，主程序大概长这样：

import sqlite3 from datetime import datetime def main(): url = "你的目标接口地址" params = {"ps": 50, "pn": 1} # 每页50条，取第一页 raw = fetch_json(url, params=params) items = parse_hot_items(raw) conn = sqlite3.connect("hot_data.db") for item in items: item["snapshot_time"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") insert_snapshot(conn, item) conn.commit() conn.close() print(f"成功写入 {len(items)} 条快照数据")

这个主流程看起来简单，但它把“请求—解析—入库”三个环节彻底解耦了。后面的章节里，我会在这个基础上叠加增量更新、异常重试和定时调度，让它从一个“手动跑一次”的脚本变成“每天自动干活”的工具。

4. 增量更新与去重：让工具从“跑一次”变成“天天用”

如果只是在命令行里手动执行一下脚本，那这工具还停留在一半的状态。真正让它产生价值的是长期持续运行，而这背后依赖两件事：去重机制和增量更新。

4.1 基于联合唯一约束的天然去重

我在建表时已经用UNIQUE(video_id, snapshot_time)做了约束。这意味着同一个视频在同一分钟内只会有一条记录。即使定时任务因为某种原因重复执行，二次插入也会触发冲突异常，而不会产生重复数据。

不过这里有个细节要注意：SQLite 的唯一约束冲突会抛出IntegrityError，如果你让它裸奔，程序会直接崩溃。正确的做法是在插入时捕获这个异常，或者使用INSERT OR IGNORE语句：

def insert_snapshot(conn: sqlite3.Connection, item: dict) -> None: sql = """ INSERT OR IGNORE INTO hot_rank (video_id, rank, title, author, play_count, like_count, danmaku_count, snapshot_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """ conn.execute(sql, ( item["video_id"], item["rank"], item["title"], item["author"], item["play_count"], item["like_count"], item["danmaku_count"], item["snapshot_time"], ))

用INSERT OR IGNORE之后，重复执行就变成静默跳过，脚本天然具备幂等性。这个经验在几乎所有数据采集任务里都适用。

4.2 快照机制与趋势分析

为什么我要强调“快照机制”？因为热门榜单是一个动态排名，并且视频本身也是涨数据的。如果你只在第一次看到某个视频时记录一个静态条目，那你看不到它的增长曲线。

快照机制记录了每一个时间点的排名和互动数据，这样你可以很方便地跑出类似这样的分析：

-- 查询最近7天出现在榜单超过3次的视频 SELECT video_id, title, COUNT(*) AS appear_cnt FROM hot_rank WHERE snapshot_time >= datetime('now', '-7 days') GROUP BY video_id HAVING COUNT(*) >= 3 ORDER BY appear_cnt DESC;

这个查询能帮你找出真正的“常青树”视频——连续多天都在榜单上，这种内容比那些昙花一现的视频更有参考价值。如果你想看一个具体视频的排名变化，只要按 video_id 查快照记录，就能画出它的排名波动曲线。

4.3 定时调度的三种方式

脚本写好后，定时调度是一个绕不开的问题。我试过不少方案，给你列个对比：

我个人推荐：如果你有云服务器，直接用 cron 是最省心的方案；如果不想管服务器，云函数定时触发也不错。本机跑 cron 的问题是电脑一关就罢工，不适合长期任务。

下面是一个 cron 例子，每天早上 8 点到晚上 11 点，每隔 2 小时抓一次：

0 */2 8-23 * * * cd /path/to/project && /usr/bin/python3 main.py >> logs/hot.log 2>&1

记得把日志输出到文件里，不然脚本出错时你根本不知道什么情况。

5. 反爬机制与边界意识：我踩过的几个真实坑

公开接口虽然门槛低，但不代表没有任何防护。这个项目我前前后后跑了一两个月，遇到过几个典型问题，整理出来给你做个参考。

5.1 频率过高：从正常请求到访问异常

我第一次跑的时候为了拿到更细的时间粒度，设置了每 3 分钟请求一次，结果跑了不到半天，接口开始随机返回空数据。当时第一反应是代码出了 bug，排查了半天才发现是请求过于频繁被服务端限流了。

这个问题的背后是服务端通常有 QPS 限制和单位时间请求数限制，你短暂超过阈值不会立刻被封，但会进入一种“软限制”状态——请求能发出去，但返回的数据要么是空的，要么是固定的缓存数据。

解决方式很简单：降低频率。我把抓取间隔从 3 分钟改成 15 分钟之后，连续跑了一周再也没有出现过空数据。这个经验给我的教训是：不要贪多，抓取频率够用就好，追求秒级实时更新对热门榜单没有任何意义。

5.2 User-Agent 和请求头伪装

有些接口会对没有浏览器特征的请求做拦截。我在初版代码里随便写了个 User-Agent，结果部分接口返回 403。后来把请求头补全成浏览器标配（UA、Accept、Accept-Language），问题就解决了。

这里分享一个日常的组合头：

headers = { "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", "Accept": "application/json, text/plain, */*", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", "Referer": "https://www.bilibili.com/", }

Referer 这个字段有时候比 UA 还关键，因为服务端会校验请求是不是从本站页面发起的。你直接请求 API 可能 403，带上主页面的 Referer 就放行了。

5.3 登录态与 Cookie 的作用

热门榜单这类公开数据通常不需要登录，但如果你要抓取一些需要登录才能看到的个性化推荐，或者抓取量很大的场景，可能就需要登录后的 Cookie。获取 Cookie 的方式很正统：手动登录一次，从开发者工具的 Application 面板或 Network 请求里复制 Cookie 值，放到代码配置里。

这里有一个坑：Cookie 是会过期的，少则几天，多则几个月。用带 Cookie 的请求时要做好失效预案——通常是设置一个合理的过期时间，到期后重新登录换 Cookie，并在代码里把 Cookie 写到单独的配置文件中，不要硬编码。

5.4 异常重试与指数退避

网络请求不可能 100% 成功，超时、连接重置、返回 5xx 都会发生。一个健壮的工具需要异常重试机制。我的做法是在请求封装里加重试逻辑，每次重试间隔翻倍：

import time def fetch_with_retry(url: str, params: dict | None = None, max_retries: int = 3): for attempt in range(max_retries): try: return fetch_json(url, params=params) except Exception as e: if attempt == max_retries - 1: raise wait_time = 2 ** attempt # 指数退避 print(f"请求失败，{wait_time}秒后重试，错误：{e}") time.sleep(wait_time)

指数退避的核心逻辑是：第一次失败等 2 秒，第二次失败等 4 秒，第三次失败等 8 秒，给服务端足够的恢复时间。这样既不会把服务端打爆，也不至于因为一次临时抖动就丢掉整轮数据。

5.5 合规与边界意识

这个必须单独说。抓取公开数据本身是常见的工程技术，但有两个边界要守住：

• 抓取频率不要对目标平台造成压力，合理频次内使用
• 抓下来的数据不要用于商业转售或大规模分发，尤其是涉及用户隐私的内容

我见过有人把爬虫写成高频并发去抓接口，给平台带来不小的压力，这种做法既不负责任也容易给自己惹麻烦。做工具的人，应该对自己写出去的每一行代码可能产生的影响负责。

6. 从“跑通”到“有用”：数据可视化与下游应用

到这一步，工具已经能稳定地定时采集数据了。但裸数据躺在 SQLite 里，如果不消费，价值等于零。我自己的经验是把数据变成三类实际可用的产物：日报、提醒、周报。

6.1 生成每日热点快报

每天早上抓取完成后，跑一个统计脚本，把昨天的榜单变化汇总成一份 Markdown 或 HTML 报告。比如哪些视频首次进入前 10，哪些视频排名上升最快，这些信息用几条 SQL 就能算出来：

-- 对比昨天和前天同一时间点，找排名上升最快的视频 SELECT cur.title, cur.rank, pre.rank AS prev_rank FROM hot_rank cur JOIN hot_rank pre ON cur.video_id = pre.video_id WHERE cur.snapshot_time >= datetime('now', '-2 hours') AND pre.snapshot_time >= datetime('now', '-1 day') AND pre.snapshot_time < datetime('now', '-22 hours') ORDER BY (pre.rank - cur.rank) DESC LIMIT 10;

这是快照机制真正值钱的地方。没有历史快照，你只能说“今天有这个视频”，有了快照，你能说“这个视频 24 小时内排名从 50 上升到了第 7”，这对内容运营来说是完全不同级别的信息量。

6.2 关键词命中提醒

我在使用中发现，单纯看榜单排名提升还不够，更实用的是按关键词过滤。比如你负责美食类账号，就可以设置一些关键词（“探店”“食谱”“深夜食堂”），每天抓完数据后自动扫描标题，命中关键词的视频自动汇总并推送到你的消息里。

这里的实现也很轻量：

keywords = ["探店", "食谱", "食堂"] def filter_by_keywords(items: list[dict]) -> list[dict]: result = [] for item in items: if any(kw in item["title"] for kw in keywords): result.append(item) return result

推送渠道优先推荐飞书群机器人或钉钉群机器人，Webhook 配置几十行代码就能搞定。如果不想折腾这些，也可以直接发邮件。

6.3 周维度趋势分析

数据积累超过两周后，可以做的事情就更多了。比如分析一周内热度最高的视频类型分布、分析不同视频创作者的霸榜时长、预测下一个可能成为热点的方向。这个阶段已经不算“爬虫”了，更接近数据分析。

我在实际使用中发现一个特别有意思的分析：对比“当天霸榜视频”和“三天后仍在榜的视频”，两类内容有着非常明显的特征差异。前者往往是话题性强的争议内容，后者则是真正有干货或情绪价值的作品。这种洞察靠人工看榜单很难形成体感，但数据会告诉你答案。

6.4 可扩展的方向

如果你觉得 SQLite 不够用了，可以无缝迁移到 PostgreSQL；如果想做更复杂的分析，可以把数据定期导出到 ClickHouse 或 Elasticsearch；如果想把工具产品化，可以加一个简单的 Web 面板，用 FastAPI + ECharts 画排名趋势曲线。

但这些都属于锦上添花。对于大多数场景，现有这套基于 SQLite 的轻量方案，已经能够以极低的成本支撑起一个内容团队对热门趋势的数据需求。

我在实际跑这个工具的过程中，最大的感受是：技术难点并不在爬虫本身，而是在于你有没有想清楚数据要用来干嘛。抓着数据存着不分析，跟没抓没什么区别。建议你动手之前，先给自己的数据规划一个明确的消费场景——哪怕是每周导出一次表格发给同事，也算是有明确用途。