用 Python 写过爬虫的朋友应该都清楚，真正推动你进步的不是教科书里的 demo，而是带着真实业务诉求的小项目。前阵子我在做内容选题盘点，需要快速掌握 AcFun 上各分区当前哪些视频在霸榜、热度集中在哪些题材、头部作者的产出情况，于是动手写了一个 Python 爬虫脚本，把 AcFun 全站榜单和分区榜单自动抓下来，整理成结构化数据再落库。项目做下来后，我顺手把这个脚本叫成了“全站硬核榜单收割机”，因为它确实一个分区一个分区地“收割”榜单，逻辑清晰、产出直接。
这篇博文就把整个项目从头到尾拆开讲：为什么选 requests 而不是 Selenium，怎么在浏览器里找到榜单的真实数据接口，多线程抓取怎么做，字段清洗有哪些坑，以及被访问频率限制拦住时该如何自查。适合三类人看：刚学完 Python 基础想练手爬虫的新人、打算做弹幕视频站数据分析的爱好者，以及想研究轻量级数据采集工程化的朋友。项目本身不难，但每一步都有容易被忽略的细节，我把踩过的坑和排查思路也一并写了进来。
1. 项目目标与方案设计：先想清楚要“收割”什么
1.1 需求拆解：榜单数据到底能做什么
AcFun 的榜单页并不算复杂，但它同时承载了几类关键数据：全站热榜、分区热榜、新作榜，排序维度有播放、评论、投蕉（类似点赞）、收藏等。做内容运营和创作者分析时，这些数据就是判断平台风向的“温度计”。
我这次的需求主要有这么几个：
- 拿到全站排名前列的视频基本信息，包括标题、作者、链接、分区、排名、播放量、评论数、投蕉数；
- 按分区分别抓取，方便观察不同内容板块的头部生态；
- 能和上一次抓取的数据做对比，判断热度是在涨还是跌；
- 数据要能落到本地，方便后续用 Excel 或 SQL 查询，而不是每次都重新抓。
需求明确之后，“全站”并不等于“所有视频”。榜单页给出的本来就是 Top 序列，全站榜单抓前 100、分区榜单抓前 50，在数据量和信息量之间比较平衡。真要爬到全站所有视频，那是另一个数量级的工程，需要走更完整的视频列表接口和分页策略，对本次需求来说属于过度设计。
1.2 技术选型：requests 够用，为什么不直接上 Selenium
很多新手在写爬虫时会陷入一个误区：看到页面是动态渲染的，第一反应就是上 Selenium 模拟浏览器操作。其实这是最重的方案，能用轻量方案解决就不要轻易动用重型武器。
当时我做了个对比，三个候选方案各自的优缺点如下：
| 方案 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| requests + JSON 接口 | 速度快、内存占用小、代码简单、易于批量抓取 | 接口改版时需要重新分析 | 目标站有可直接调用的数据接口 |
| Selenium 模拟浏览器 | 无需分析接口，可见即可得 | 慢、吃 CPU/内存、容易被识别、维护成本高 | 页面数据全在 JS 渲染且无接口可用 |
| Scrapy 框架 | 并发控制完善、有中间件和管道，适合大规模抓取 | 学习成本略高，小任务显得重 | 分布式抓取、长期运行的采集系统 |
这次的榜单页虽然前端是动态渲染的，但我用开发者工具查看网络请求后发现，页面数据都是通过一个 JSON 接口返回的，结构非常干净。这种情况下用 requests 直接请求接口，效率远高于 Selenium。
另外，requests 方案在环境搭建上几乎零成本，只要 Python 环境里有 requests 库就能跑。团队协作时给别人看代码，也比丢一个在跑的浏览器窗口更清爽。
提示：技术选型的核心原则是“够用就好”。榜单抓取这种轻量任务，先尝试找数据接口，永远比模拟浏览器更值得优先考虑。
1.3 合规边界：爬虫不是“抢数据”
这部分我必须专门写一节。爬虫技术本身是中性的，但用法必须有边界。这次爬取严格限定在榜单页返回的公开汇总数据上，不涉及用户个人主页、评论正文等个人信息，也没有对目标站点造成压力。
实操前我还做了两件基础工作：一是确认目标站点对爬虫的友好程度，尽量控制在合理访问频率区间；二是明确数据用途仅限个人学习和内容趋势观察，不会打包售卖、不会批量抓取非公开数据、不会给目标服务器带来明显负担。
做技术分享也是这样，讲清楚“怎么做”的同时，也要强调“什么不能做”。下面所有代码思路，都建立在合规、克制、公开数据的前提下。如果你在做类似项目，也请自己把握好尺度。
2. 核心实现：从接口定位到并发抓取
2.1 在浏览器里找到榜单的真实数据接口
写爬虫最忌一上来就照着网上旧教程抄代码。网站改版频繁，别人的代码很可能已经失效。正确顺序是先自己抓包，找到当前的真实数据入口。
我当时用的方法很简单，直接在 Chrome 里打开 AcFun 榜单页，按 F12 进入开发者工具，切到 Network（网络）面板，筛选 XHR/Fetch 请求，然后手动切换榜单的“全站”“分区”“日榜”“周榜”等选项。每切换一次，面板里就会多出几个请求，挨个查看响应内容，很快就能定位到返回视频列表数据的那个 JSON 接口。
以那个版本为例，核心请求大概长这样：
POST https://www.acfun.cn/rest/pc-direct/rank/list Content-Type: application/x-www-form-urlencoded realmId=5&rankType=view&pcursor=0&pageSize=20realmId：分区 ID，不同数字对应不同内容板块；rankType：榜单类型，常见有 view、comment、banana 等；pcursor：分页游标，从 0 开始递增；pageSize：每页条数，实测一般支持 10 到 30。
响应体是标准 JSON，核心数据在一个 result 对象里，里面有 rankList 列表，列表里每个元素就是一条视频的完整信息：标题、作者昵称、作者 ID、视频链接、封面图、播放数、评论数、投蕉数、发布时间戳等等。
注意：接口地址、参数名、字段名都可能在改版中变化。我写这篇文章时用的路径是这个，你实际操作时一定要以自己抓到的为准，但分析的思路完全通用。
2.2 编写基础版爬虫：请求、解析、重试一样都不能少
接口定位之后，写基础版爬虫就顺理成章了。我习惯把代码分成两层：底层是“请求函数”，负责带请求头发 POST 请求并返回 JSON；上层是“解析函数”，负责从 JSON 里抽取字段并做类型处理。
第一版请求函数大概长这样：
import requests import time import random BASE_URL = "https://www.acfun.cn/rest/pc-direct/rank/list" 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", "Referer": "https://www.acfun.cn/rank/list", "Origin": "https://www.acfun.cn", } def fetch_rank_page(realm_id, pcursor=0, page_size=20, rank_type="view"): data = { "realmId": realm_id, "rankType": rank_type, "pcursor": pcursor, "pageSize": page_size, } resp = requests.post(BASE_URL, data=data, headers=HEADERS, timeout=10) resp.raise_for_status() payload = resp.json() return payload.get("result", {})这里有几个非常容易踩的细节：
- headers 里必须带 User-Agent，而且最好模拟真实浏览器版本，裸 requests 默认 UA 很容易被限制；
- Referer 和 Origin 也要带上，很多站点会校验来源；
- 必须设置 timeout，否则某个请求卡住时整个程序都会停在那里；
- 返回数据要先判断是不是 JSON，因为被拦截时接口返回的可能是普通告警页面文本。
单页请求只是第一步，榜单有分页，所以还要处理游标。返回的 result 里通常会有一个表示是否还有下一页的字段，比如 hasNext 或 nextPage。当 pcursor 递增后拿不到新数据，说明已经翻到底，就该停下来了。
重试机制也不能省。网络抖动、偶发的 5xx 错误都会让单次请求失败，简单粗暴的做法是重试三次，每次失败后按指数退避策略等待：
def fetch_with_retry(realm_id, pcursor=0, max_retries=3): for attempt in range(max_retries): try: result = fetch_rank_page(realm_id, pcursor) if result: return result except Exception as e: print(f"[retry] realm_id={realm_id}, pcursor={pcursor}, " f"attempt={attempt + 1}, error={e}") time.sleep(2 ** attempt + random.random()) return {}指数退避的意思是，第一次失败等 2 秒，第二次等 4 秒，第三次等 8 秒，给目标服务器一个喘息时间。实测下来这个策略对偶发错误很管用，比失败后立刻重试更不容易触发访问频率限制。
2.3 多线程并发抓取：把单线程变成小水管
基础版跑通后，逐分区抓取是可行的，但真的太慢了。AcFun 分区数量不算多，每个分区也就两三页，可如果每次都单线程串行，首页切一次、日榜切一次、周榜再切一次，耗时立刻翻倍。
并发改造我选的是 Python 标准库里的concurrent.futures.ThreadPoolExecutor。没有引入 Celery 这类重量级任务框架，因为在这里根本用不上。
思路很简单：把要抓的“分区 ID + 榜单类型 + 页数”组合成一个任务清单，线程池负责并发执行，主线程负责汇总结果：
from concurrent.futures import ThreadPoolExecutor, as_completed TASKS = [ (5, "view", 0), (5, "view", 1), # 游戏区 番剧区 前两页 (7, "view", 0), (7, "view", 1), # 科技区 (9, "view", 0), # 娱乐区 ] all_items = [] def work(task): realm_id, rank_type, pcursor = task result = fetch_with_retry(realm_id, pcursor, rank_type=rank_type) return result.get("rankList", []) with ThreadPoolExecutor(max_workers=4) as pool: future_map = {pool.submit(work, t): t for t in TASKS} for future in as_completed(future_map): try: items = future.result() all_items.extend(items) except Exception as e: print(f"task failed: {future_map[future]}, error={e}")这里要注意两点。第一，max_workers 不要开太大。爬虫并发不是越大越好，4 到 6 个线程对这类轻量任务已经足够，开二三十个线程很容易让目标服务器响应变慢，甚至直接把你的 IP 请进黑名单。第二，每个任务都要做好异常兜底，别让某个分区失败拖垮整个任务列表。
多线程改造后，完整跑一轮全站加分区的榜单抓取，从几分钟缩短到了几十秒，体验提升非常明显。
3. 数据清洗与落库：抓到不是终点，能用才是目标
3.1 字段处理与编码陷阱
接口返回的字段名和 Python 变量的命名习惯往往不一致，比如它可能叫videoName，我们习惯叫title；它可能叫userName，我们统一成author_name。第一步就是做字段映射，把原始 JSON 里的键改成自己后面方便处理的字段名。
这一步看起来机械，但很容易出问题。不同接口返回的字段可能同名却不同含义，比如某个版本的响应里同时有viewCount和playCount，实测下来一个代表播放，一个代表播放页曝光，区分不清就会让数据失真。
另外，原始字段里的时间戳通常是一长串 Unix 时间，直接落库不直观，我会统一转成可读的日期时间；播放量如果返回的是已经格式化好的“1.2万”这类文本，也得解析成数字，否则没法做排序和聚合。
编码问题是另一个大坑。用 pandas 写 CSV 时，默认编码在 Windows 下用 Excel 打开会乱码，解决办法是存成utf-8-sig：
import pandas as pd df = pd.DataFrame(all_items) df.to_csv("acfun_rank.csv", index=False, encoding="utf-8-sig")注意utf-8-sig和utf-8的区别：前者会在文件开头写入 BOM 头，Excel 打开才不乱码。这个细节没处理好的话，辛苦抓到的数据在同事手里直接变成乱码，非常影响观感。
3.2 去重与增量更新：别让数据越跑越脏
榜单数据有个特殊之处——同一时间内，同一个视频可能同时出现在全站榜和分区榜里。如果简单地把多线程结果拼在一起，必然会出现重复行。
我的处理方式是在解析阶段就以视频 ID 为唯一键进行去重：
seen_ids = set() unique_items = [] for item in all_items: vid = item.get("videoId") or item.get("contentId") if vid in seen_ids: continue seen_ids.add(vid) unique_items.append(item)如果是长期跑定时任务，还需要考虑增量更新。比如每周抓一次，同一视频的排名和热度发生了变化，库里应该更新这条记录而不是新增一条。一个比较稳妥的落库设计是：SQLite 表里给视频 ID 建唯一索引，插入时用INSERT OR REPLACE或先查后插。
如果你对 SQL 不熟，也可以用最简单的方式：每次抓完都生成带日期后缀的文件，比如acfun_rank_20250108.csv，要对比历史就按日期读取对应文件，文件名本身充当了时间维度。这个方法虽然没有数据库那样高效的查询能力，但对轻量项目已经够用。
3.3 数据入库：SQLite 足够，别迷信重型数据库
我这次选择 SQLite 而不是 MySQL，理由很直接：单机脚本、数据量不大、没有多端并发写入需求。SQLite 的整个数据库就是一个文件，备份、迁移、分享都非常方便。
建表和批量写入的示例：
CREATE TABLE IF NOT EXISTS acfun_rank ( video_id TEXT PRIMARY KEY, title TEXT, author_name TEXT, realm_id INTEGER, rank_type TEXT, rank INTEGER, view_count INTEGER, comment_count INTEGER, banana_count INTEGER, crawled_at TEXT );导入数据时用executemany批量插入，比循环单条execute快很多。再加上事务控制，几百条数据几乎是瞬间完成。
心得：做爬虫项目时，数据落库方案不值得过度设计。SQLite 在大多数个人项目和中小团队内部工具里都够用，等真到了需要多人实时读写、并发写入量大的阶段，再迁移到 PostgreSQL 或 MySQL 也不迟。
4. 反爬应对与高频请求问题排查实录
4.1 高频请求后遭遇拦截的三种典型表现
我在调试过程中反复触发过访问频率限制，总结下来，被拦截时通常有三种表现：
- 接口正常返回 200，但 body 不是 JSON，而是一段带提示文字的 HTML，内容大致是访问过于频繁；
- 返回 JSON 但数据为空，
rankList变成了空数组，这种情况下往往需要等一段时间才能恢复； - 个别请求返回 403 或 503 状态码，配合重试机制通常能缓解，但如果持续高频请求，就会升级成更严格的风控。
遇到这些不要慌，先确认是不是自己的请求频率太高，再按下面的三板斧排队处理。
4.2 降频、伪装、退避：反爬应对三板斧
第一板斧是降频。这是最基础也最有效的方案，多线程跑起来后，给每个任务之间加入随机 sleep，让请求间隔呈现自然波动。比如：
time.sleep(random.uniform(0.3, 1.2))随机间隔比固定间隔好，固定间隔的周期性太强，反而容易被识别。
第二板斧是伪装。维护一个常见的浏览器 User-Agent 列表，每次请求随机取一个；同时保证请求头里的 Referer、Accept、Accept-Language 等字段看起来像一个正常浏览器发出的请求。
USER_AGENTS = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) ...", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) ...", "Mozilla/5.0 (X11; Linux x86_64) ...", ] def get_headers(): return { "User-Agent": random.choice(USER_AGENTS), "Referer": "https://www.acfun.cn/rank/list", "Accept": "application/json, text/plain, */*", }第三板斧是退避。一旦检测到响应异常，立刻停止当前任务，等待时间拉长，而不是头铁继续硬请求。退避机制的代码在 2.2 节已经给出，这里不再重复，但我要强调的是：退避不是“破解策略”，而是“相处之道”，把抓取频率控制在对目标服务器友善的水平，长期运行才可持续。
这里多说一句：爬虫圈里讨论的那些“高强度对抗”方案，在我看来既不合适也不必要。一个成熟的项目，核心永远是数据质量和采集稳定性，而不是在风控边缘反复试探。合规、克制、可维护，这才是值得学习的能力。
4.3 常见问题速查表
| 现象 | 可能原因 | 处理方式 |
|---|---|---|
| 返回内容不是 JSON | 会话被限制，返回了告警页面 | 拉长 sleep，等待几分钟再试 |
| 某个分区一直抓不到 | 分区 ID 可能已改版 | 重新抓包确认最新分区 ID |
| 视频 ID 字段为空 | 字段名与实际返回不一致 | 打印完整 JSON，核对字段名 |
| 写入 CSV 后 Excel 乱码 | 编码不是 utf-8-sig | 写入时指定 encoding="utf-8-sig" |
| 多线程后请求大量失败 | 并发开得太猛 | 降低 max_workers，增加随机 sleep |
| 时间戳看起来不对 | Unix 时间戳单位是秒或毫秒 | 先判断位数，再统一转换 |
这个表看起来不起眼，却是我这次项目里最实用的产出之一。排查问题最耗时间的往往不是问题本身，而是找原因的方向错了。先把现象对应到可能原因，再去验证，效率会高很多。
5. 成果落地与扩展思路：从“能跑”到“好用”
5.1 一轮榜单抓下来的数据怎么用
脚本跑完后，一张acfun_rank.csv文件里躺着几百条视频数据。数据本身不会说话，得靠分析和可视化让它变得可读。
我做了两个最基础的分析：一是按分区统计 Top 榜单里视频分布的数量，直观看出哪个分区的内容供给最旺盛；二是做播放量与评论量的散点对比，找“高播放低评论”和“低播放高评论”的内容样本，这在内容选题时特别有价值。
可视化用 matplotlib 就能搞定，不用上太重的东西：
import matplotlib.pyplot as plt import pandas as pd df = pd.read_csv("acfun_rank.csv", encoding="utf-8-sig") realm_count = df.groupby("realm_id")["video_id"].count().sort_values() plt.figure(figsize=(10, 6)) realm_count.plot(kind="barh") plt.title("AcFun 榜单分区分布") plt.tight_layout() plt.savefig("realm_distribution.png", dpi=160)图生成后可以直接放进周报或选题会材料里，比贴一堆数字直观得多。
5.2 把“收割机”升级成持续监控系统
单次抓取只能看到某一时刻的快照，真正有价值的是持续跟踪热度变化。我后来给脚本加了两层扩展，让它从“手动收割机”变成“自动监控台”。
第一层是定时触发。直接用操作系统的 crontab 定时任务，或者用 Python 里的APScheduler库做进程内调度，每天上午 10 点自动抓取一次榜单，追加到历史数据表里。
第二层是变化检测。把每次抓到的同一视频的排名、播放量等数据与昨天做差值，生成一份“上升最快”“下滑最猛”的榜单。这种动态数据比单次榜单有价值得多，也是做热点追踪时最想看到的内容。
运行一段时间后，还可以把这些历史数据导出，用折线图看某个分区头部视频的热度生命周期，分析一类内容从爆发到衰退的周期规律。爬虫到这里已经不仅仅是抓数据的工具，而成了一个小型内容观察系统。
如果你感兴趣，还可以继续扩展：
- 抓取视频详情页的弹幕总数和标签信息，完善内容画像；
- 对一些重点作者做长期追踪，观察其作品在各榜单的排名变化；
- 在榜单发生明显波动时，通过企业微信机器人或邮件发送通知，第一时间感知热点。
这些扩展方向都不需要推倒重来，在现有清洗结构上追加接口和字段即可。
最后分享一点个人体会。做这个项目最大的收获不是代码本身，而是养成了一种“先看接口、再写代码”的思维习惯。很多爬虫初学者上来就复制代码、跑通就满足，遇到改版就抓瞎。相信我，多花五分钟在浏览器里看 Network 面板，比你在网上搜三小时旧教程都管用。另外，无论项目多小，都要把频率控制和异常重试放在和功能实现同等重要的位置，这既是专业度的体现，也是让你的数据采集任务能稳定跑下去的底线思维。这套“榜单收割机”的代码量不大，但每块都踩在真实需求上，改造成其他内容平台的榜单采集也是同样的套路。
