1. 项目缘起与整体设计思路在线编程练习平台这几年越来越多很多开发者习惯在上面刷题、提交代码、积累自己的解题记录。用得久了就会冒出一个很自然的需求把自己在平台上已经完成的题目代码批量导出到本地方便离线查阅、整理成个人知识库或者迁移到自己的笔记系统里。手动一题一题复制粘贴显然不现实几十上百道题下来手都要废掉所以用脚本自动化抓取就成了顺理成章的选择。这个项目的核心目标很明确针对某在线编程网站下文统一用阿尔法Coding代称中已完成的代码记录通过分析其异步XHR接口用Python实现批量抓取并落地保存。它解决的是个人解题数据无法便捷导出这个痛点适合有一定Python基础、了解HTTP协议、想学习异步接口抓取实战的开发者参考。哪怕你之前只写过简单的同步爬虫跟着思路走也能理解异步XHR抓取的门道。为什么强调异步XHR因为现代前端几乎都是前后端分离架构页面骨架先加载出来真正的数据是通过JavaScript发起的XHRXMLHttpRequest或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?statuscompleted详情接口可能叫/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(headersheaders, timeout15.0)把这段headers抽成一个配置后续所有请求复用既整洁又不容易漏字段。3.2 用Client保持会话与连接复用httpx的Client同步和AsyncClient异步对象会自动管理Cookie和连接池。用Client的好处是一次构造多次请求底层TCP连接会被复用省去反复握手的开销对批量抓取来说性能提升明显。而且如果登录接口设置了新CookieClient会自动更新不用手动维护。异步场景下用AsyncClient配合async with上下文管理器确保请求结束后连接被正确关闭import httpx import asyncio async def main(): async with httpx.AsyncClient(headersheaders, timeout15.0) as client: resp await client.get(https://alpha-coding.example.com/api/submissions?statuscompleted) 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 problems4.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( fhttps://alpha-coding.example.com/api/submissions/{problem[id]}/code ) return problem, resp.json()并发数设多少合适我的经验是5到10比较稳妥具体看站点承受能力和你的网络。设太小效率低设太大风险高。可以先从5开始试观察有没有被限流的迹象比如返回429或响应变慢再酌情调整。4.3 用gather批量调度与异常隔离收集好题目列表后用asyncio.gather把所有抓取任务一起调度。但gather默认遇到一个异常就会中断整体这显然不行——某一题失败不该影响其他题。所以要加return_exceptionsTrue让异常作为结果返回最后统一处理。async def main(): async with httpx.AsyncClient(headersheaders, timeout15.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_exceptionsTrue) 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对应.pyjavascript对应.jsjava对应.javacpp对应.cppgo对应.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, fproblem_{problem[id]})) folder os.path.join(output, f{problem[id]}_{title}) os.makedirs(folder, exist_okTrue) 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, fsolution.{ext}), w, encodingutf-8) as f: f.write(code) with open(os.path.join(folder, meta.json), w, encodingutf-8) as f: json.dump(problem, f, ensure_asciiFalse, indent2)标题截断到80字符是为了避免路径过长某些系统对路径长度有限制。编码统一用utf-8防止中文标题或代码里的中文注释乱码。5.3 增量抓取与去重如果隔一段时间想更新一次没必要全量重抓。可以在本地记录已抓取的题目ID抓取前先读出来跳过已存在的。这样每次只抓新增的题目速度快、请求少。实现上就是维护一个done_ids.txt或done_ids.json每成功保存一题就追加进去。def load_done_ids(pathdone_ids.json): if os.path.exists(path): with open(path, encodingutf-8) as f: return set(json.load(f)) return set() def mark_done(pid, pathdone_ids.json): ids load_done_ids(path) ids.add(pid) with open(path, w, encodingutf-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)。写文件时也务必指定encodingutf-8。6.4 并发过高被限流降速与退避如果发现大量请求返回429或超时说明并发太高了。把信号量从8降到3或5请求间隔从0.2秒加到0.5秒通常就能缓解。配合指数退避重试遇到429时等待更长时间再重试能显著提升成功率。问题现象可能原因排查方向解决手段返回HTML而非JSON请求了页面URL检查URL是否为接口重新抓包定位接口401/403登录态失效检查Cookie/Token重新登录更新凭证429请求过于频繁查看并发与间隔降并发、加延迟、退避重试中文乱码编码不一致检查响应编码手动指定utf-8部分题目失败个别接口异常查看失败详情记录后单独重试文件名为空标题字段缺失检查JSON结构用ID兜底命名6.5 独家避坑心得说几个文档里不会写、只有踩过才知道的点。第一有些站点的接口在返回代码时会做HTML转义比如把变成lt;保存前要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抓取最难的不是写代码而是前期把接口分析透彻。接口摸清了代码就是水到渠成的事。另外把配置和逻辑分离、把失败处理做扎实比追求极致的并发速度更重要——一个能稳定跑完、失败可重试的脚本远比一个跑一半就崩的快脚本有价值。
