1. 项目缘起与整体设计思路在线编程练习平台这几年越来越火很多人习惯在上面刷题、提交代码、积累解题记录。用得久了就会冒出一个很实际的需求把自己在平台上已经完成的题目代码批量导出到本地方便离线复习、整理成个人题解库或者迁移到自己的笔记系统里。我这次要聊的就是这么一个实践——用 Python 异步爬虫把某个在线编程网站下面统一叫它“阿尔法Coding”里自己账号下已完成的代码抓下来。先说清楚这个项目的定位。它不是那种大规模分布式采集也不是要搞什么高并发压测而是一个带登录态的、面向个人账号的、以 XHR 接口为数据源的异步抓取脚本。核心关键词有三个异步、XHR、已完成代码。异步是为了在题目数量多的时候不至于一个请求一个请求地干等XHR 是因为现代编程平台几乎都是前后端分离页面上的代码内容基本不会直接写在 HTML 里而是通过接口返回 JSON已完成代码则是我们的目标数据范围只抓自己提交通过的那些不碰别人的、不碰未完成的。为什么选择异步而不是多线程这是很多人第一个会问的问题。传统的requestsThreadPoolExecutor也能做并发但线程模型在 IO 密集场景下开销偏大尤其是当你要维持登录态、处理 cookie、做请求间隔控制的时候线程之间的状态共享和异常处理会比较啰嗦。而asyncioaiohttp这套组合天生就是为大量网络 IO 设计的单线程事件循环就能把成百上千个请求编排得井井有条配合信号量控制并发数既快又稳。实测下来几百道题的抓取异步方案比同步串行快一个数量级比线程池也通常快 30% 到 50%。再说 XHR 这个点。很多人一上来就想着用 Selenium 或者 Playwright 去模拟浏览器点击把页面渲染出来再解析 DOM。这个思路不是不行但太重了。编程平台的题目列表、提交记录、代码详情几乎全都是 XHR 接口驱动的。你打开浏览器的开发者工具切到 Network 面板筛选 XHR/Fetch刷新一下页面就能看到一串返回 JSON 的请求。找到这些接口直接构造请求去拿数据比驱动浏览器快得多也稳定得多。浏览器方案适合接口加密特别狠、参数难以还原的场景但大多数编程平台还没到那个程度XHR 直连完全够用。这个项目适合谁来参考我觉得有三类人。第一类是刚学完 Python 基础、想找一个真实项目练手的初学者这个项目涉及登录态、异步、JSON 解析、文件写入是一个很好的综合练习。第二类是有同步爬虫经验、想过渡到异步的开发者可以拿它当 asyncio 的实战模板。第三类是单纯想把自己题解导出来的普通用户照着步骤改改就能用。不管你属于哪一类下面的内容我都会尽量讲透“为什么这么做”而不只是“怎么做”。在动手之前有一个原则必须先立好只抓取自己账号下的数据控制请求频率尊重平台的负载能力。这不是一句空话而是决定你这个脚本能不能长期稳定跑下去的前提。频率太高轻则被限流重则账号受影响得不偿失。所以整个方案里并发控制、请求间隔、失败重试这三件事我会重点讲。2. 核心细节解析与实操要点2.1 登录态是怎么维持的抓取自己账号的数据第一步永远是解决“我是谁”的问题。编程平台的接口基本都要求登录后才能访问未登录状态下请求提交记录接口返回的要么是 401要么是一个空的列表。所以登录态的获取是整个项目的地基。常见的登录态维持方式有三种。第一种是手动复制 Cookie你登录网站后从开发者工具的 Application 面板里把 Cookie 字符串复制出来直接塞进请求头。这种方式最简单但 Cookie 有有效期过期了就得重新复制适合一次性导出。第二种是用账号密码走登录接口程序化地拿到 Cookie 或 Token。这种方式自动化程度高但需要分析登录接口的参数有些平台还会加验证码、加密签名成本较高。第三种是复用浏览器已保存的登录态比如从本地浏览器配置里读取 Cookie或者用自动化工具登录一次后导出存储状态。我这次采用的是第一种和第二种结合的策略优先尝试程序化登录如果登录接口有额外的风控参数搞不定就退回到手动复制 Cookie。对于大多数只是想导出自己代码的人来说手动复制 Cookie 其实是最省事、最不容易出问题的方案。你只需要在浏览器里正常登录然后按 F12找到任意一个发往该平台的请求把请求头里的Cookie整条复制出来即可。这里有个细节要注意Cookie 里通常包含多个键值对比如会话标识、用户标识、CSRF Token 等。有些平台的关键接口除了 Cookie还要求请求头里带一个X-CSRF-Token或者类似的字段这个值往往和 Cookie 里的某个值对应。你在复制的时候最好把整个请求头都看一遍把非标准的自定义头也一并记下来。我踩过的坑就是只复制了 Cookie结果提交记录接口一直返回 403后来发现是漏了一个自定义的鉴权头。提示Cookie 属于敏感凭证等同于你的登录密码。不要把包含真实 Cookie 的脚本上传到公开仓库也不要在截图里暴露。建议用环境变量或本地配置文件来存放代码里通过os.environ读取。2.2 找到真正的数据接口登录态搞定后下一步是找到返回目标数据的 XHR 接口。这一步没有捷径就是打开开发者工具一个一个看。具体操作是登录平台进入“我的提交”或“已完成题目”页面打开 Network 面板勾选 XHR/Fetch 过滤然后刷新页面或者翻页观察新出现的请求。你要找的接口通常有这么几个特征返回内容是 JSONURL 里带有submission、record、solution、problem之类的关键词返回的 JSON 里包含题目 ID、提交状态、代码内容等字段。以典型的编程平台为例一般会有两类接口一类是列表接口返回你所有提交记录的摘要包含题目 ID、提交时间、通过状态通常支持分页另一类是详情接口根据某个提交 ID 返回具体的代码内容。这里有个关键判断代码内容到底在列表接口里就有还是需要再调一次详情接口有些平台为了减小列表响应体积列表里只给摘要代码要单独请求有些平台则直接在列表里带上代码。你需要实际看一次响应内容才能确定。如果是前者那你的抓取流程就是“先翻页拿列表再逐个请求详情”请求量会翻倍这时候异步的价值就体现出来了。还有一个容易被忽略的点分页参数。列表接口一般用page和page_size或limit/offset来控制。你要先确认平台单页最多返回多少条比如单页上限是 20 条你有 300 道题那就是 15 页。把总页数算出来才能规划抓取任务。有些平台还会返回总条数字段直接读出来除以每页条数向上取整即可。2.3 异步并发的粒度控制确定了接口就要设计并发策略了。异步爬虫最容易犯的错误就是“一把梭”——把所有请求同时发出去。这样做的后果是目标服务器瞬间收到几百个请求触发限流你自己的脚本也可能因为连接数过多而报错。所以并发必须控制粒度。我的做法是用asyncio.Semaphore来控制同时进行的请求数量。信号量的值设多少合适这取决于平台的承受能力和你的账号风险偏好。一般来说5 到 10 是一个比较稳妥的区间。我实测下来并发设为 5 的时候几百道题的抓取大概几分钟就能完成同时对服务器压力很小设到 10 会更快但偶尔会遇到限流返回。如果你特别在意账号安全设成 3 甚至 2 也完全可以无非是多等一会儿。除了信号量请求之间最好再加一个随机间隔。固定间隔容易被识别出机器行为随机间隔更自然。我通常用asyncio.sleep(random.uniform(0.3, 0.8))这样的方式在每个请求前后加一点抖动。注意这个 sleep 要放在信号量内部也就是拿到“通行证”之后再睡否则睡的时候还占着并发名额效率会打折。还有一个细节是连接复用。aiohttp的ClientSession内部维护连接池整个抓取过程应该只创建一个 session所有请求共用它。如果你每个请求都新建 session那连接建立和关闭的开销会吃掉异步带来的大部分收益。这一点和requests里用Session复用连接是一个道理。2.4 数据落盘与去重抓下来的代码怎么存这看似简单其实有不少讲究。最直接的方式是每道题存一个文件文件名用题目 ID 或题目标题。但题目标题里可能有空格、斜杠、特殊符号直接当文件名会出问题需要做一次清洗把非法字符替换掉。我一般用题目 ID 加一个清洗后的标题比如1001_two-sum.py这样既唯一又可读。文件扩展名要根据代码语言来定。编程平台通常会在提交记录里标明语言比如 Python、Java、C。你可以维护一个语言到扩展名的映射表Python 存成.pyJava 存成.javaC 存成.cpp。这样导出来的文件可以直接在本地 IDE 里打开体验好很多。去重是另一个必须考虑的问题。同一道题你可能提交过多次列表接口会把每次提交都返回。如果你只想保留“已完成”的那次就要根据状态字段过滤只保留通过Accepted的记录。如果同一道题有多次通过记录通常保留最新的一次即可。我一般会在内存里用一个字典按题目 ID 去重后出现的覆盖先出现的最后统一写盘。这样既避免了重复请求详情接口也避免了本地文件被反复覆盖。注意写文件的时候建议用utf-8编码并且显式指定newline或者用二进制模式避免不同操作系统下换行符被转换导致代码内容变形。代码里的缩进、换行都是语义的一部分存错了就失去意义了。3. 实操过程与核心环节实现3.1 环境准备与依赖安装先把环境搭起来。Python 版本建议 3.8 以上因为asyncio的一些新特性在低版本上不可用。依赖主要就两个aiohttp负责异步 HTTP 请求aiofiles负责异步写文件可选但推荐避免同步写文件阻塞事件循环。如果你需要解析 HTML比如某些信息只在页面里再加一个beautifulsoup4或lxml。pip install aiohttp aiofiles安装完之后建议先写一个最小的异步请求测试确认能正常拿到数据再往上叠功能。很多人一上来就写完整脚本结果报错都不知道是哪一层的问题。分步验证是省时间的好习惯。3.2 配置与凭证管理把 Cookie、请求头、接口地址这些配置项单独抽出来放在脚本顶部或者一个独立的配置文件里。我习惯用一个字典来管理请求头把 Cookie 从环境变量读取这样脚本本身可以安全地分享出去。import os BASE_URL https://example-coding-site.com LIST_API BASE_URL /api/submissions DETAIL_API BASE_URL /api/submission/{id} HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) ..., Accept: application/json, text/plain, */*, Referer: BASE_URL /my-submissions, Cookie: os.environ.get(CODING_COOKIE, ), }这里User-Agent和Referer不是可有可无的。很多平台会校验Referer是否来自站内缺失或不对会直接拒绝。User-Agent则用于伪装成正常浏览器避免被简单的反爬规则拦掉。这些头信息你都可以从浏览器开发者工具里那个真实请求上原样抄下来。3.3 分页拉取提交列表先实现列表拉取。逻辑是从第一页开始请求拿到当前页的记录和总条数算出总页数然后并发请求剩余页面。这里我用一个fetch_page协程负责单页请求主流程用asyncio.gather并发调度。import asyncio import aiohttp async def fetch_page(session, sem, page, page_size20): params {page: page, page_size: page_size, status: accepted} async with sem: await asyncio.sleep(random.uniform(0.3, 0.8)) async with session.get(LIST_API, paramsparams, headersHEADERS) as resp: if resp.status ! 200: print(fpage {page} failed: {resp.status}) return [] data await resp.json() return data.get(items, [])注意statusaccepted这个参数它让服务端只返回已通过的记录从源头上减少了数据量。不是所有平台都支持这个过滤参数如果不支持就在客户端过滤。拿到第一页后读取响应里的total字段算出总页数再并发拉取其余页。async def fetch_all_list(session, sem): first await fetch_page(session, sem, 1) # 假设响应里有 total 字段这里需要根据实际结构调整 total 300 # 从第一页响应中读取 page_size 20 total_pages (total page_size - 1) // page_size tasks [fetch_page(session, sem, p, page_size) for p in range(2, total_pages 1)] results await asyncio.gather(*tasks) all_items first [item for page_items in results for item in page_items] return all_items3.4 并发抓取代码详情列表拿到后按题目 ID 去重得到需要抓详情的提交 ID 列表。然后同样用信号量控制并发逐个请求详情接口。async def fetch_detail(session, sem, submission_id): url DETAIL_API.format(idsubmission_id) async with sem: await asyncio.sleep(random.uniform(0.3, 0.8)) try: async with session.get(url, headersHEADERS) as resp: if resp.status ! 200: return submission_id, None data await resp.json() return submission_id, data.get(code) except Exception as e: print(fdetail {submission_id} error: {e}) return submission_id, None这里加了try/except因为异步并发下个别请求失败是常态不能让一个异常把整个gather搞崩。失败的任务记录下来后面可以重试。3.5 文件写入与命名规范拿到代码后写盘。文件名清洗函数很关键把路径分隔符、冒号、问号这些非法字符替换成下划线。import re def safe_filename(name): return re.sub(r[\\/:*?|], _, name).strip() LANG_EXT { python: .py, python3: .py, java: .java, cpp: .cpp, c: .cpp, javascript: .js, go: .go, }写文件用aiofiles异步写避免阻塞事件循环。如果题目数量不多用同步写其实也没大问题但既然都上异步了写文件也一并异步化更一致。import aiofiles async def save_code(problem_id, title, lang, code): ext LANG_EXT.get(lang.lower(), .txt) filename f{problem_id}_{safe_filename(title)}{ext} path os.path.join(output, filename) async with aiofiles.open(path, w, encodingutf-8) as f: await f.write(code)3.6 主流程编排把上面的环节串起来主流程大致是创建 session拉列表去重抓详情写文件。整个过程中 session 只创建一次信号量也复用。async def main(): sem asyncio.Semaphore(5) connector aiohttp.TCPConnector(limit10) async with aiohttp.ClientSession(connectorconnector) as session: items await fetch_all_list(session, sem) # 去重保留每题最新一次通过 unique {} for it in items: unique[it[problem_id]] it tasks [fetch_detail(session, sem, it[submission_id]) for it in unique.values()] results await asyncio.gather(*tasks) for (sid, code), it in zip(results, unique.values()): if code: await save_code(it[problem_id], it[title], it[language], code) if __name__ __main__: asyncio.run(main())TCPConnector的limit参数控制连接池大小一般设成和信号量差不多的值就行。设太大没意义因为信号量已经限制了并发。4. 常见问题与排查技巧实录4.1 请求返回 401 或 403 怎么办这是最常见的问题基本都和登录态有关。排查顺序是这样的先确认 Cookie 是否过期重新登录后复制一次试试再确认请求头是否完整特别是自定义的鉴权头有没有漏然后确认Referer和User-Agent是否和真实请求一致。如果都对了还是 403可能是平台对请求频率做了限制把并发降到 2 或 3加大请求间隔再试。还有一种情况是 Cookie 里包含了多个域名的值你复制的时候可能只复制了主域的漏了接口所在子域的。这时候要看清楚接口请求的域名确保 Cookie 覆盖到了。4.2 异步任务卡死或超时异步爬虫跑着跑着不动了通常是某个请求没有设置超时一直挂着。aiohttp默认的超时是 5 分钟太长了。建议显式设置一个较短的超时比如 15 秒。timeout aiohttp.ClientTimeout(total15) async with aiohttp.ClientSession(timeouttimeout) as session: ...另外asyncio.gather默认只要有一个任务抛异常就会整体失败除非用return_exceptionsTrue。我在fetch_detail里已经用try/except兜住了但如果你在别的地方没兜建议给gather加上return_exceptionsTrue让失败的任务返回异常对象而不是中断整个流程。4.3 代码内容乱码或格式错乱如果抓下来的代码打开是乱码先检查响应编码。aiohttp的resp.json()会自动处理 JSON 编码一般不会乱。如果是resp.text()可能需要手动指定编码。写文件时务必用utf-8并且不要做任何换行符转换。有些平台返回的代码里带\r\n你写的时候如果用了文本模式且没指定newlinePython 可能会把它转成\n虽然大多数情况无害但严格来说改变了原始内容。4.4 常见问题速查表问题现象可能原因排查方向401/403登录态失效或请求头不全重新复制 Cookie核对自定义头返回空列表过滤参数不对或未登录检查 status 参数确认登录态任务卡死请求无超时设置 ClientTimeout部分请求失败并发过高被限流降低信号量加大间隔文件名报错标题含非法字符用正则清洗文件名代码乱码编码不一致统一用 utf-8 读写重复抓取未按题目去重用字典按 problem_id 去重4.5 几个踩过坑才明白的经验第一个经验是先小批量验证再全量跑。我一开始图快直接对几百道题全量并发结果跑到一半被限流前面抓的还得重来。后来改成先抓 10 道题验证流程确认没问题再放开稳得多。第二个经验是把中间结果缓存下来。列表数据抓一次不容易抓到后先存成 JSON 文件详情抓取失败时可以基于缓存重跑不用重新拉列表。这个习惯在调试阶段特别有用。第三个经验是日志要打全。异步并发下出问题的时候你根本不知道是哪个请求挂了。我在每个请求里都打了 submission_id 和状态码出问题时一眼就能定位。别嫌日志多排查的时候你会感谢自己。第四个经验是别把并发开太大。很多人觉得异步就是要快信号量设成 50、100结果不是被限流就是自己连接池爆掉。5 到 10 这个区间对个人账号导出这种场景已经足够快了。快不是目的稳定拿到数据才是。5. 方案的可扩展方向这套脚本跑通之后其实还能做不少延伸。比如把抓下来的代码按语言或难度分类归档生成一个本地的题解索引比如把代码同步到自己的笔记软件或代码托管仓库做成自动化的题解备份再比如给脚本加一个增量更新逻辑每次只抓上次导出之后新增的提交避免重复劳动。增量更新的实现思路是本地维护一个已抓取提交 ID 的集合每次运行先拉列表过滤掉已经在集合里的只抓新的。这个集合可以存成一个 JSON 文件简单可靠。对于经常刷题的人来说这个功能能省下大量重复抓取的时间。另一个方向是错误重试机制。目前失败的任务只是打印出来没有自动重试。可以加一个重试队列对失败的提交 ID 再跑一轮最多重试两三次。实现上可以用一个简单的循环把失败列表重新丢进gather直到全部成功或达到重试上限。我个人在实际操作中的体会是爬虫这类脚本稳定性永远比速度重要。你追求极致的并发可能十分钟抓完但中间被限流、被封、数据缺失反而要花更多时间补救。把并发压一压加上合理的间隔和重试让它安安静静跑完才是真正省心的做法。这套异步 XHR 的思路不只适用于编程平台任何前后端分离、数据走接口的网站都可以套用同样的骨架换个接口地址和解析逻辑就行。
