AcFun榜单Python爬虫实战：接口分析、多线程与反爬应对

用 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 realmId5rankTypeviewpcursor0pageSize20realmId分区 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, pcursor0, page_size20, rank_typeview): data { realmId: realm_id, rankType: rank_type, pcursor: pcursor, pageSize: page_size, } resp requests.post(BASE_URL, datadata, headersHEADERS, timeout10) 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, pcursor0, max_retries3): 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}, fattempt{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_typerank_type) return result.get(rankList, []) with ThreadPoolExecutor(max_workers4) 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(ftask 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-sigimport pandas as pd df pd.DataFrame(all_items) df.to_csv(acfun_rank.csv, indexFalse, encodingutf-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写入时指定 encodingutf-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, encodingutf-8-sig) realm_count df.groupby(realm_id)[video_id].count().sort_values() plt.figure(figsize(10, 6)) realm_count.plot(kindbarh) plt.title(AcFun 榜单分区分布) plt.tight_layout() plt.savefig(realm_distribution.png, dpi160)图生成后可以直接放进周报或选题会材料里比贴一堆数字直观得多。5.2 把“收割机”升级成持续监控系统单次抓取只能看到某一时刻的快照真正有价值的是持续跟踪热度变化。我后来给脚本加了两层扩展让它从“手动收割机”变成“自动监控台”。第一层是定时触发。直接用操作系统的 crontab 定时任务或者用 Python 里的APScheduler库做进程内调度每天上午 10 点自动抓取一次榜单追加到历史数据表里。第二层是变化检测。把每次抓到的同一视频的排名、播放量等数据与昨天做差值生成一份“上升最快”“下滑最猛”的榜单。这种动态数据比单次榜单有价值得多也是做热点追踪时最想看到的内容。运行一段时间后还可以把这些历史数据导出用折线图看某个分区头部视频的热度生命周期分析一类内容从爆发到衰退的周期规律。爬虫到这里已经不仅仅是抓数据的工具而成了一个小型内容观察系统。如果你感兴趣还可以继续扩展抓取视频详情页的弹幕总数和标签信息完善内容画像对一些重点作者做长期追踪观察其作品在各榜单的排名变化在榜单发生明显波动时通过企业微信机器人或邮件发送通知第一时间感知热点。这些扩展方向都不需要推倒重来在现有清洗结构上追加接口和字段即可。最后分享一点个人体会。做这个项目最大的收获不是代码本身而是养成了一种“先看接口、再写代码”的思维习惯。很多爬虫初学者上来就复制代码、跑通就满足遇到改版就抓瞎。相信我多花五分钟在浏览器里看 Network 面板比你在网上搜三小时旧教程都管用。另外无论项目多小都要把频率控制和异常重试放在和功能实现同等重要的位置这既是专业度的体现也是让你的数据采集任务能稳定跑下去的底线思维。这套“榜单收割机”的代码量不大但每块都踩在真实需求上改造成其他内容平台的榜单采集也是同样的套路。