Python异步XHR抓取实战：批量导出在线编程平台已完成代码

1. 项目缘起与整体设计思路

在线编程练习平台这几年越来越多，很多开发者习惯在上面刷题、提交代码、积累自己的解题记录。用得久了就会冒出一个很自然的需求：把自己在平台上已经完成的题目代码批量导出到本地，方便离线查阅、整理成个人知识库，或者迁移到自己的笔记系统里。手动一题一题复制粘贴显然不现实，几十上百道题下来手都要废掉，所以用脚本自动化抓取就成了顺理成章的选择。

这个项目的核心目标很明确：针对某在线编程网站（下文统一用"阿尔法Coding"代称）中"已完成"的代码记录，通过分析其异步XHR接口，用Python实现批量抓取并落地保存。它解决的是"个人解题数据无法便捷导出"这个痛点，适合有一定Python基础、了解HTTP协议、想学习异步接口抓取实战的开发者参考。哪怕你之前只写过简单的同步爬虫，跟着思路走也能理解异步XHR抓取的门道。

为什么强调"异步XHR"？因为现代前端几乎都是前后端分离架构，页面骨架先加载出来，真正的数据是通过JavaScript发起的XHR（XMLHttpRequest）或Fetch请求，从后端接口异步拉取JSON再渲染的。你直接请求页面URL拿到的HTML里根本没有代码内容，全是空壳。所以抓取这类站点的关键，不是解析HTML，而是找到那个真正返回数据的接口，然后模拟它发请求。这也是很多人第一次写爬虫时最困惑的地方——明明浏览器里能看到内容，代码一跑就啥也抓不到，原因就在这里。

整体设计上，我把项目拆成四个层次：第一层是接口分析，搞清楚数据从哪个URL来、带什么参数、返回什么结构；第二层是会话与鉴权，因为"已完成的代码"属于个人数据，必须登录后才能访问；第三层是异步并发抓取，用异步HTTP客户端批量请求接口，提升效率；第四层是数据落地，把抓到的代码按题目分类保存成本地文件。这四层环环相扣，任何一层出问题都会导致抓取失败，所以下面我会逐层拆开讲。

在方案选型上，我最终选择了httpx配合asyncio来做异步抓取，而不是传统的requests加多线程。原因很简单：这类接口是典型的IO密集型任务，等待网络响应的时间远大于CPU计算时间，异步模型在单线程内就能通过事件循环调度成百上千个并发请求，资源占用低、代码也更清爽。requests虽然生态成熟，但同步阻塞的特性决定了它必须靠线程池硬扛并发，线程切换和内存开销都更大。当然，如果你对异步不熟，用requests加ThreadPoolExecutor也能实现，只是效率上限低一些，这个取舍后面会细说。

提示：本文所有域名、接口路径、字段名均为示意性代称，实际抓取时请以你自己分析出的真实接口为准。抓取个人自己的数据用于本地备份是合理需求，但务必控制请求频率，不要对目标站点造成压力。

2. 接口分析：找到真正的数据来源

2.1 用浏览器开发者工具定位XHR请求

抓取异步接口的第一步永远是"抓包"。打开目标网站的已完成代码列表页，按下F12打开开发者工具，切到Network（网络）面板，然后勾选筛选器里的"Fetch/XHR"，这样就能过滤掉图片、CSS、JS这些静态资源，只留下异步数据请求。接着刷新页面，你会看到一串请求列表冒出来。

这时候要做的是一一排查。通常列表页会有一个请求返回的是题目列表数据，点进去看Response（响应）标签，如果是一段JSON，里面包含题目ID、标题、完成状态等字段，那它就是我们要找的列表接口。再点开某一题的详情，找到返回具体代码内容的那个请求，它往往带着题目ID作为参数。把这两个接口记下来，整个抓取链路就清晰了：先调列表接口拿到所有已完成题目的ID，再逐个调详情接口拿代码。

这里有个经验：接口的命名往往有规律，比如列表接口可能叫/api/submissions?status=completed，详情接口可能叫/api/submissions/{id}/code。观察URL结构能帮你快速推断出其他相关接口。另外注意请求方法，列表和详情通常是GET，但有些平台会用POST传参，这个必须看准，方法错了直接返回405。

2.2 拆解请求头与关键参数

找到接口只是开始，真正决定能不能成功的是请求头和参数。在开发者工具里点开某个请求，切到Headers标签，重点看三块内容：Request Headers里的Cookie、Authorization、User-Agent、Referer，以及Query String Parameters或Request Payload里的业务参数。

Cookie和Authorization是鉴权的核心。个人数据接口几乎都要求登录态，Cookie里通常带着会话标识，Authorization里可能是Bearer Token。这两个值会过期，所以不能写死在代码里长期用，后面讲会话管理时会说怎么处理。User-Agent建议如实填写一个常见浏览器的值，有些站点会校验它，缺失或异常会被拦截。Referer有时也校验，填上列表页的地址比较稳妥。

业务参数方面，列表接口常见的有分页参数page、page_size，状态过滤status，排序order_by等。详情接口一般就是题目ID或提交ID。把这些参数的含义搞清楚，你才能构造出正确的请求。我踩过的一个坑是：某接口的分页是从0开始而不是从1开始，结果第一页数据永远重复，排查了半天才发现是页码基准问题。所以拿到接口后，先用不同参数值手动请求几次，观察返回数据的变化，确认参数语义再写代码。

2.3 分析返回数据的JSON结构

接口返回的JSON结构决定了你后面怎么解析。列表接口通常长这样：一个对象里包含data数组和total、page等分页信息，data里每个元素是一道题的摘要。详情接口返回的可能是{ "code": "...", "language": "python", "title": "..." }这样的结构。

分析时要注意几个点：一是字段的嵌套层级，代码内容可能藏在data.submission.code这种深层路径里，取值时要一层层剥；二是字段可能为null，比如某题没写代码或代码被清空，解析时要做好判空；三是代码里可能包含特殊字符、换行、引号，JSON解析本身能处理，但你保存到文件时要注意编码。我一般会先把一个完整的响应复制到本地，用json.loads解析后打印出来，肉眼确认结构，再动手写解析逻辑，这样比盲写靠谱得多。

注意：有些接口返回的不是纯JSON，而是JSONP（外面包了一层回调函数），或者返回体前面带了防跨站前缀。遇到这种情况需要先做字符串裁剪，把外层剥掉再解析。判断方法很简单，看响应体第一个字符是不是{或[，不是的话就要处理。

3. 会话管理与鉴权处理

3.1 登录态的获取与复用

个人数据接口绕不开登录。最省事的做法是：在浏览器里正常登录后，从开发者工具里把Cookie或Authorization头复制出来，直接塞进代码里用。这个方法适合一次性导出，简单粗暴。但它的缺点是会话会过期，可能几小时或几天后就失效，届时需要重新复制。

如果想做得更优雅，可以模拟登录流程：分析登录接口，用账号密码换取会话凭证，再带着凭证去请求数据接口。不过模拟登录涉及验证码、加密参数、风控策略等复杂问题，不同平台差异很大，投入产出比不一定高。对于"导出自己数据"这种低频需求，我通常建议直接用复制Cookie的方式，把精力放在抓取逻辑上。

复制Cookie时有个细节：不要只复制某一个Cookie键值，而是把整个Cookie字符串完整带上，因为会话可能依赖多个Cookie协同。用httpx的话，可以这样构造：

import httpx headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Cookie": "sessionid=你的会话值; csrftoken=你的令牌值", "Referer": "https://alpha-coding.example.com/problems", "Accept": "application/json, text/plain, */*", } client = httpx.Client(headers=headers, timeout=15.0)

把这段headers抽成一个配置，后续所有请求复用，既整洁又不容易漏字段。

3.2 用Client保持会话与连接复用

httpx的Client（同步）和AsyncClient（异步）对象会自动管理Cookie和连接池。用Client的好处是：一次构造，多次请求，底层TCP连接会被复用，省去反复握手的开销，对批量抓取来说性能提升明显。而且如果登录接口设置了新Cookie，Client会自动更新，不用手动维护。

异步场景下用AsyncClient配合async with上下文管理器，确保请求结束后连接被正确关闭：

import httpx import asyncio async def main(): async with httpx.AsyncClient(headers=headers, timeout=15.0) as client: resp = await client.get("https://alpha-coding.example.com/api/submissions?status=completed") print(resp.status_code) print(resp.json()) asyncio.run(main())

这里timeout一定要设，默认不设的话某些请求可能一直挂着，批量抓取时一个卡住会拖垮整个流程。我一般设10到20秒，具体看站点响应速度。

3.3 处理鉴权失效与重试

会话过期是必然会遇到的。表现是接口返回401或403，或者返回体里带一个"未登录"的错误码。代码里要能识别这种情况并给出明确提示，而不是默默失败。我的做法是封装一个请求函数，检查状态码和返回体，遇到鉴权失败就抛出带说明的异常，提醒用户更新Cookie。

对于偶发的网络抖动或5xx错误，可以加简单的重试逻辑。用tenacity库或者手写一个带退避的重试装饰器都行。退避策略建议用指数退避，比如第一次等1秒，第二次2秒，第三次4秒，避免短时间内反复冲击接口。但要注意，鉴权失败（401/403）不该重试，重试也没用，只会白白增加请求量，所以重试逻辑里要把这类状态码排除掉。

提示：把Cookie这类敏感信息放在单独的配置文件或环境变量里，不要硬编码进主逻辑，更不要提交到公开仓库。这是基本的安全习惯。

4. 异步并发抓取的核心实现

4.1 先拿列表再拿详情：两阶段抓取

整个抓取流程分两个阶段。第一阶段调列表接口，把所有"已完成"题目的ID和元信息收集起来。第二阶段遍历这些ID，逐个调详情接口拿代码。为什么要分两阶段？因为列表接口一次能返回多条摘要，请求次数少；详情接口必须一题一次，是请求量的主要来源。分开处理能让逻辑更清晰，也方便做进度统计。

列表阶段要注意分页。假设每页返回20条，总共200道题，那就要翻10页。写个循环，从第一页开始请求，直到返回的data为空或累计数量达到total为止。这里要防止死循环，加一个最大页数上限兜底。收集到的题目信息存成一个列表，每个元素至少包含题目ID和标题，标题后面用来做文件名。

async def fetch_problem_list(client): problems = [] page = 1 while True: resp = await client.get( "https://alpha-coding.example.com/api/submissions", params={"status": "completed", "page": page, "page_size": 20}, ) data = resp.json() items = data.get("data", []) if not items: break problems.extend(items) if len(problems) >= data.get("total", 0): break page += 1 if page > 100: # 兜底，防止异常情况下无限翻页 break return problems

4.2 用信号量控制并发数

异步抓取最容易犯的错就是"一把梭"——把所有请求同时发出去。几百个并发请求瞬间打过去，轻则被限流，重则被封IP，对站点也不友好。正确做法是用asyncio.Semaphore控制同时进行的请求数量，比如限制在5到10个之间。

信号量的原理像一个停车场的车位：车位满了，后来的车就得在门口等，等有车开走才能进。代码里就是每发起一个请求前先await semaphore.acquire()，请求结束后semaphore.release()。用async with semaphore更简洁：

sem = asyncio.Semaphore(8) async def fetch_code(client, problem): async with sem: resp = await client.get( f"https://alpha-coding.example.com/api/submissions/{problem['id']}/code" ) return problem, resp.json()

并发数设多少合适？我的经验是5到10比较稳妥，具体看站点承受能力和你的网络。设太小效率低，设太大风险高。可以先从5开始试，观察有没有被限流的迹象（比如返回429或响应变慢），再酌情调整。

4.3 用gather批量调度与异常隔离

收集好题目列表后，用asyncio.gather把所有抓取任务一起调度。但gather默认遇到一个异常就会中断整体，这显然不行——某一题失败不该影响其他题。所以要加return_exceptions=True，让异常作为结果返回，最后统一处理。

async def main(): async with httpx.AsyncClient(headers=headers, timeout=15.0) as client: problems = await fetch_problem_list(client) print(f"共找到 {len(problems)} 道已完成题目") tasks = [fetch_code(client, p) for p in problems] results = await asyncio.gather(*tasks, return_exceptions=True) for r in results: if isinstance(r, Exception): print(f"抓取失败: {r}") else: problem, data = r save_code(problem, data)

这样即使个别题目抓取失败，其他题目照样能正常保存。失败的那些可以记录下来，稍后单独重试。

4.4 请求间隔与礼貌抓取

即便是异步并发，也建议在请求之间加一点微小延迟，模拟人类浏览节奏，降低被风控的概率。可以在信号量内部加await asyncio.sleep(0.2)，或者用更精细的令牌桶限速。别小看这零点几秒，它能让你的抓取行为看起来温和很多。

另外，抓取时间尽量避开站点的高峰时段，选择深夜或凌晨，既减少对站点的压力，也能获得更稳定的响应速度。这些都是"礼貌抓取"的基本素养，也是长期稳定抓取的前提。

5. 数据解析与本地落地

5.1 从JSON中提取代码与元信息

详情接口返回的JSON结构需要仔细解析。假设返回体是{"data": {"code": "...", "language": "python", "title": "两数之和"}}，那提取逻辑就是逐层取值。要注意代码字段可能包含各种语言的代码，语言类型决定了保存时的文件扩展名。常见映射关系：python对应.py，javascript对应.js，java对应.java，cpp对应.cpp，go对应.go。

如果接口没返回语言类型，可以根据代码特征粗略判断，比如有def和:的多半是Python，有function或=>的可能是JavaScript。但更靠谱的还是从接口字段里拿，拿不到就统一存成.txt，别硬猜。

5.2 文件命名与目录组织

保存文件时，命名和目录结构直接影响后续查阅体验。我的做法是按题目ID或标题建目录，代码文件放在里面，同时把题目描述、提交时间等元信息存成一个meta.json。文件名要处理掉非法字符，Windows下\ / : * ? " < > |都不能出现在文件名里，用正则替换成下划线。

import re import os import json def safe_filename(name): return re.sub(r'[\\/:*?"<>|]', "_", name).strip()[:80] def save_code(problem, data): title = safe_filename(problem.get("title", f"problem_{problem['id']}")) folder = os.path.join("output", f"{problem['id']}_{title}") os.makedirs(folder, exist_ok=True) code = data.get("data", {}).get("code", "") lang = data.get("data", {}).get("language", "txt") ext_map = {"python": "py", "javascript": "js", "java": "java", "cpp": "cpp", "go": "go"} ext = ext_map.get(lang, "txt") with open(os.path.join(folder, f"solution.{ext}"), "w", encoding="utf-8") as f: f.write(code) with open(os.path.join(folder, "meta.json"), "w", encoding="utf-8") as f: json.dump(problem, f, ensure_ascii=False, indent=2)

标题截断到80字符是为了避免路径过长，某些系统对路径长度有限制。编码统一用utf-8，防止中文标题或代码里的中文注释乱码。

5.3 增量抓取与去重

如果隔一段时间想更新一次，没必要全量重抓。可以在本地记录已抓取的题目ID，抓取前先读出来，跳过已存在的。这样每次只抓新增的题目，速度快、请求少。实现上就是维护一个done_ids.txt或done_ids.json，每成功保存一题就追加进去。

def load_done_ids(path="done_ids.json"): if os.path.exists(path): with open(path, encoding="utf-8") as f: return set(json.load(f)) return set() def mark_done(pid, path="done_ids.json"): ids = load_done_ids(path) ids.add(pid) with open(path, "w", encoding="utf-8") as f: json.dump(list(ids), f)

增量抓取不仅省事，也是对站点友好的表现。毕竟你只是备份自己的数据，没必要反复拉取同样的内容。

6. 常见问题与排查技巧实录

6.1 抓不到数据：先确认接口对不对

新手最常见的问题就是"代码跑通了但没数据"。排查顺序是：先确认请求的URL是不是真正的数据接口，而不是页面URL；再确认请求方法（GET还是POST）对不对；然后检查请求头是否完整，尤其是Cookie和User-Agent；最后看返回体的状态码和内容。如果状态码是200但返回的是HTML，说明你请求的还是页面而不是接口，回去重新抓包。

6.2 返回401或403：鉴权问题

这两个状态码基本就是登录态失效或权限不足。解决办法是重新登录，更新Cookie。如果更新后还是403，可能是Referer或Origin头缺失，补上试试。还有一种情况是接口对请求频率做了限制，短时间内请求太多被临时拒绝，这时候降低并发、加长间隔，等一会儿再试。

6.3 中文乱码：编码问题

保存的文件打开是乱码，通常是编码没统一。请求响应可能用的是utf-8，也可能有站点用gbk。httpx会自动根据响应头推断编码，但偶尔会猜错。可以手动指定：resp.encoding = "utf-8"，或者直接用resp.content.decode("utf-8")。写文件时也务必指定encoding="utf-8"。

6.4 并发过高被限流：降速与退避

如果发现大量请求返回429或超时，说明并发太高了。把信号量从8降到3或5，请求间隔从0.2秒加到0.5秒，通常就能缓解。配合指数退避重试，遇到429时等待更长时间再重试，能显著提升成功率。

6.5 独家避坑心得

说几个文档里不会写、只有踩过才知道的点。第一，有些站点的接口在返回代码时会做HTML转义，比如把<变成<，保存前要html.unescape还原，否则代码没法直接运行。第二，代码里如果包含\r\n和\n混用，保存后在某些编辑器里显示会错乱，统一替换成\n更稳妥。第三，别在循环里反复创建Client，一个Client复用到底，否则连接池形同虚设，性能反而更差。第四，抓取前先手动请求一两个接口验证Cookie有效，别等跑完几百个请求才发现全是401，白白浪费时间。

7. 效率优化与后续扩展

7.1 性能调优的几个抓手

想让抓取更快，可以从三个地方下手。一是提高并发数，但要配合限流，找到效率和风险的平衡点；二是复用连接，用Client保持长连接，减少握手开销；三是减少不必要的请求，比如列表接口一次多拿几条（page_size调大），减少翻页次数。实测下来，把page_size从20调到50，翻页请求能少一半多，整体耗时明显下降。

7.2 断点续抓与失败重试

批量任务最怕中途中断。加个断点续抓机制：每成功一题就记录ID，下次运行自动跳过。失败的任务单独存一个列表，程序结束时提示用户，可以再跑一次只重试失败的。这样即使网络抖动或会话过期，也不用从头再来。

7.3 扩展到其他数据维度

这套框架不只适用于代码导出。同样的思路可以扩展到抓取提交记录、运行结果、题目描述、标签分类等。只要找到对应的接口，把解析和保存逻辑换一下就行。甚至可以做一个本地题库，把题目描述、代码、标签整合起来，配合全文搜索，变成一个私人刷题知识库。我自己就把导出的代码按算法类型重新整理了一遍，复习的时候翻起来特别方便。

7.4 合规与礼貌抓取的再强调

最后再啰嗦一句：抓取自己的数据用于个人备份是合理需求，但一定要控制频率、尊重站点的服务条款。别把并发开到几十上百，别在高峰时段猛拉，别把抓到的数据用于商业用途或公开传播。技术是中性的，怎么用取决于人。做一个有分寸的抓取者，才能让这类工具长期可用。

我个人在实际操作中的体会是，异步XHR抓取最难的不是写代码，而是前期把接口分析透彻。接口摸清了，代码就是水到渠成的事。另外，把配置和逻辑分离、把失败处理做扎实，比追求极致的并发速度更重要——一个能稳定跑完、失败可重试的脚本，远比一个跑一半就崩的快脚本有价值。