资讯动态

从抓取皮卡丘到批量数据采集:Python爬虫的工程化实战

发布时间:2026/9/3 17:52:21 来源:尧图企业网站定制
“去吧皮卡丘”喊出这句话的时候大多数人脑子里出现的是动画里那只黄色的电气鼠。但放在开发者眼里它完全可以变成一个真实可运行的小项目你在终端敲下“去吧皮卡丘”脚本就去公开的宝可梦数据接口抓取皮卡丘的数据保存成一份结构化文件甚至批量抓完整个图鉴。这个场景听起来很有趣但真正实操过的人会明白难点从来不在“请求皮卡丘”这个动作上。PokeAPI 是一个公开的宝可梦数据接口访问一次返回完整的 JSON几乎零门槛。真正麻烦的是只抓一个皮卡丘时代码怎么乱都能跑一旦想抓几十只、几百只就不得不面对网络超时、接口限流、字段结构、日志记录、断点重跑这些工程问题。所以这篇文章的主判断很明确这类项目最有价值的产出不是一张皮卡丘数据卡而是一条可以反复使用、不会因为一次异常就全盘崩溃的数据抓取和整理流程。后面所有内容都围绕这个判断展开。1. 这个需求真正卡住的不是“抓皮卡丘”而是“下次怎么抓”只抓单只皮卡丘其实就是一个 GET 请求。在浏览器里访问https://pokeapi.co/api/v2/pokemon/pikachu你甚至不需要写代码就能看到一长串 JSON。很多人第一次看到这种返回结构会觉得“那很简单请求一下然后解析就行”。但我想说这个想法只对了前面一半。为什么因为日常开发里的数据抓取几乎都不会只抓一条。你这次能抓皮卡丘下次就能抓杰尼龟、小火龙如果做一个图鉴项目那至少是一百多只算上地区形态和更多版本数量还要撑大。一旦进入批量抓取阶段问题不是由requests带来的而是由“数量”带来的。1.1 一次性抓取与批量抓取的本质差别一次性抓取的特点是输入固定、错误好修、重来成本低。你发现抓错了改一行代码再跑一次就行。反正只抓一只时间开销可以忽略。批量抓取就完全不一样。最危险的通常不是某一条请求失败而是“跑到一半失败了”且你不知道之前成功的那些有没有被保存。你可能会遇到个别宝可梦的字段结构不一致有的属性缺失有的图片字段为空请求太频繁被服务端限流返回 429抓到一半网络中断已经抓到的数据没有保存全部白跑输出文件格式不统一后续合并数据时光是清洗就要花半天换了一个数据源字段结构完全不一样之前的解析代码几乎要重写。这些问题才是真正决定一个脚本能不能从“能用”变成“好用”的分水岭。如果只是自己玩玩一次性抓一只皮卡丘那确实不需要工程化。但只要你打算把皮卡丘这个例子扩展成真正的数据工具就必须先把“只抓一次”的临时代码改造成“可重复执行、可容错、可追溯”的小流程。1.2 从单点数据到可复用资产需要的三块拼图我把这里的关键能力拆成三块后面每一步基本都在围绕它们展开。第一块是容错。网络请求不能假设永远成功要有超时、重试、跳过异常记录的能力。否则批量任务会因为一条坏数据全部中断。第二块是可追溯。程序跑完以后你要能回答三个问题哪些成功了哪些失败了失败的原因是什么这需要日志和落盘而不是等程序结束后只能看终端上滚过的文字。第三块是可重跑。如果批量任务跑一半挂了下一次启动时最好能接着跑而不是从 1 号重新抓一遍。这个能力在数据量小的时候没什么感觉数据量大以后非常重要。可以说“抓取皮卡丘”只是一个引子真正的目标是锻炼这三块能力。2. 用最小代码跑通单只皮卡丘的数据请求不管后面要做什么样的架构第一步都应该是先跑通单只。先不急着做批量先把最小闭环做完。只抓一只皮卡丘确认请求能成功、解析能通过、字段能拿对然后再谈扩展。2.1 环境准备先让依赖简单到不用怀疑在动手之前先把环境准备好。这里假设你用的是 Python 3.9 以上版本需要安装requests这个第三方库。建议用虚拟环境避免污染系统级 Python。python -m venv venv source venv/bin/activate # Windows 下换成 venv\Scripts\activate pip install requests为什么不建议用标准库里的urllib它也能做 HTTP 请求但代码偏底层异常处理也不如requests直观。对于一次练手项目requests足够清晰也足够普及。如果将来想追求性能或异步能力可以换成httpx但这是后话。2.2 核心请求一次 GET 拿到皮卡丘的骨架下面这段代码是最小闭环。它做的事情很简单向 PokeAPI 发送请求拿到 JSON然后打印几个关键字段。import requests pokemon_name pikachu url fhttps://pokeapi.co/api/v2/pokemon/{pokemon_name} resp requests.get(url, timeout10) print(resp.status_code) data resp.json() print(data[name]) print(data[id]) print(data[height]) print(data[weight])运行后你应该会看到类似这样的结果200 pikachu 25 4 60注意timeout10是必需的。很多人碰到脚本“卡住”不是服务端不返回而是没有设置超时时间。requests默认会一直等下去一旦网络抖动程序就可能挂在resp requests.get(...)这一行。另外PokeAPI 返回的height和weight单位不是我们最直观的公制和克数。height是分米weight是百克。如果要在页面上展示需要换算。这就是“拿到数据不等于拿到可用数据”的典型例子。2.3 字段检查返回的 JSON 不是直接能用的文档第一次请求时皮卡丘的返回数据很长有几十个字段。别急着直接写data[types][0][type][name]。先看看整体结构打印一下所有顶层字段print(list(data.keys()))返回结果里会有abilities、base_experience、forms、game_indices、height、held_items、id、is_default、location_area_encounters、moves、name、order、past_abilities、past_types、species、sprites、stats、types、weight等。对于做图鉴或数据分析通常只需要其中一部分字段说明id全国图鉴编号name宝可梦名称英文types属性列表stats种族值列表abilities特性列表height身高单位分米weight体重单位百克sprites图片 URL 集合types的典型结构是嵌套的{ types: [ { slot: 1, type: { name: electric, url: https://pokeapi.co/api/v2/type/13/ } } ] }如果要提取所有属性名不能只写data[types][0][type][name]因为有些宝可梦有两个属性索引为 0 只能拿到第一个。更稳妥的做法是names [t[type][name] for t in data[types]] print(names)到这里最小闭环就已经完成了请求、解析、打印。只有这一步稳定之后批量处理才有基础。3. 从请求到文件把临时数据改造成可复用资产为什么要把数据落地成文件因为你不应该每次都重新请求接口。落盘之后后续的分析、展示、调试都可以基于本地文件进行不依赖网络也不容易因为接口变化而丢数据。3.1 定义自己的数据模型不做全量存储不建议把整个原始 JSON 原封不动保存下来。原始 JSON 很大包含大量你用不到的嵌套字段保存后处理不方便。一般做法是提取自己关心的字段生成一个统一结构。这里可以用dataclass定义一个简单模型from dataclasses import dataclass from typing import List, Dict dataclass class PokemonSummary: id: int name: str height: int weight: int types: List[str] stats: Dict[str, int]然后编写一个转换函数把 PokeAPI 返回的数据转成我们的模型def convert_to_summary(data: dict) - PokemonSummary: types [t[type][name] for t in data[types]] stats {} for item in data[stats]: stats[item[stat][name]] item[base_stat] return PokemonSummary( iddata[id], namedata[name], heightdata[height], weightdata[weight], typestypes, statsstats, )这个模型的好处是结构稳定。以后即使接口字段有变化只需要修改转换函数下游代码不用跟着改。3.2 输出到 JSON 和 CSV让中间结果可以被复用JSON 适合保留嵌套结构CSV 适合做表格化分析。两种格式都很有用可以按场景选择。输出 JSONimport json from dataclasses import asdict summary convert_to_summary(data) with open(pikachu.json, w, encodingutf-8) as f: json.dump(asdict(summary), f, ensure_asciiFalse, indent2)输出 CSV 时要注意CSV 不支持嵌套结构所以列表字段需要做字符串连接import csv with open(pikachu.csv, w, encodingutf-8, newline) as f: writer csv.writer(f) writer.writerow([id, name, height, weight, types, hp, attack]) writer.writerow([ summary.id, summary.name, summary.height, summary.weight, /.join(summary.types), summary.stats.get(hp, ), summary.stats.get(attack, ) ])这里有一个很容易踩的坑打开 CSV 文件时如果漏了encodingutf-8遇到中文字符就会出现乱码。虽然 PokeAPI 返回的名称大多是英文但代码里可能还会有其他文本统一指定编码总没错。到这里“抓一只皮卡丘并保存下来”已经是一个完整闭环。接下来才是真正的挑战批量抓取。4. 从单只到批量循环只是第一步重试和断点才是关键批量抓取字面上看是给单只代码包一层循环。但事实上一个合格的批量抓取脚本至少要包含三件事分批试跑、失败重试、断点续抓。4.1 先小批试跑再全量执行不要一上来就跑range(1, 899)。先跑 1 到 10确认几件事前 10 只都能正常返回解析函数没有遇到未知字段输出文件能正确生成。通过以后再慢慢扩大范围。我一般会先跑 10 只再跑 151 只最后才跑全量。这样做可以减少无谓的请求也方便定位问题。如果一上来就全量一旦代码里有个字段提取逻辑写错可能跑到第 30 只才报错前面 29 次请求就白费了。4.2 失败重试不要只写 requests.get网络请求比本地程序更容易失败。可能是超时、连接中断也可能是服务端临时限流。如果不做重试整个批次可能因为一次偶发失败而中断。下面是一个简单的带重试的请求函数。关键点是“指数退避”import time import requests def fetch_json(url, max_retries3, base_wait1.0, timeout10): for attempt in range(max_retries): try: resp requests.get(url, timeouttimeout) if resp.status_code 429: wait_time base_wait * (2 ** attempt) print(f收到 429等待 {wait_time} 秒后重试) time.sleep(wait_time) continue resp.raise_for_status() return resp.json() except (requests.Timeout, requests.ConnectionError) as exc: wait_time base_wait * (2 ** attempt) print(f请求失败{exc}等待 {wait_time} 秒后重试) time.sleep(wait_time) raise RuntimeError(f请求最终失败: {url})为什么用指数退避因为服务端限流通常有时间窗口如果用固定间隔重试可能会在窗口期内反复撞墙。每次等待时间翻倍可以有效减少对服务端的压力也能提高重试成功率。如果重试 3 次仍然失败就直接抛异常让上层逻辑决定是跳过这条记录还是结束整个任务。不要做无限重试否则程序可能卡在一个坏 URL 上很久。4.3 断点续抓一批跑挂了之后能接着跑批量跑挂最常见的结果是前面已经成功的数据没保存后面也没跑完。为了避免重头再来通常做法是维护一个“已完成列表”。抓之前先检查这个 ID 是否已经完成完成则跳过新抓到的数据立即写入文件并记录当前 ID。一个朴素但有效的示例completed set() log_file completed_ids.txt try: with open(log_file, encodingutf-8) as f: completed {int(line.strip()) for line in f} except FileNotFoundError: pass for pokemon_id in range(1, 152): if pokemon_id in completed: continue url fhttps://pokeapi.co/api/v2/pokemon/{pokemon_id} try: data fetch_json(url) summary convert_to_summary(data) # 这里写入文件最好每条结果独立保存 with open(fdata/processed/{pokemon_id}.json, w, encodingutf-8) as f: json.dump(asdict(summary), f, ensure_asciiFalse, indent2) with open(log_file, a, encodingutf-8) as f: f.write(f{pokemon_id}\n) except Exception as exc: print(f处理 {pokemon_id} 失败: {exc}) time.sleep(0.5)这个例子的核心不是文件命名而是“进度记录”。即便进程在第 80 只时崩溃下一次运行时也能从completed_ids.txt里读到已完成到哪了直接跳过这些 ID只补抓剩余的。如果项目变得更复杂可以把“已完成列表”换成 SQLite 表用状态字段区分pending、success、failed但核心思路是一样的。这个能力在数据量大的时候能帮你省下大量重复请求。5. 日志、配置和目录长期维护的三件套如果只是本地临时跑一次用print就够了。但如果你打算把这个脚本放到服务器上定时跑或者交给别人使用就必须解决一个问题程序跑起来之后你不在现场时怎么知道发生了什么5.1 把 logging 加上别用 print 输出所有信息print的问题在于无法区分重要级别也无法同时输出到文件和控制台。用 Python 的logging模块可以做到分级记录既能看到终端实时输出也能落一份日志文件备查。import logging logging.basicConfig( levellogging.INFO, format%(asctime)s [%(levelname)s] %(message)s, handlers[ logging.FileHandler(logs/run.log, encodingutf-8), logging.StreamHandler() ] ) logger logging.getLogger(__name__) logger.info(开始抓取 1-151 号宝可梦) logger.error(抓取 25 号失败)这样每天跑完你可以直接看日志文件快速定位哪些记录失败。如果程序部署在服务器上日志文件也是排查问题的第一手资料。5.2 把参数抽出来别在代码里写死这里的“参数”指的是起始 ID、结束 ID、延迟时间、超时时间、输出目录、日志文件路径。这些值如果散落在代码各处改起来容易漏。最简单的方式是单独建一个config.py# config.py START_ID 1 END_ID 151 DELAY 0.5 TIMEOUT 10 OUTPUT_DIR ./data/processed RAW_DIR ./data/raw LOG_FILE ./logs/run.log主脚本里统一引用。以后要改成抓 251 只只需要改两个 ID。如果这个脚本要交给别人别人也不需要翻代码找魔法数字。5.3 用清晰目录区分原始数据和处理结果顺着上面的配置我建议把目录结构整理成下面这样project/ ├── config.py ├── main.py ├── fetch_pokemon.py ├── data/ │ ├── raw/ │ └── processed/ └── logs/ └── run.lograw目录存放原始返回响应万一之后想重新提取另一个字段不用重新请求接口processed目录放转换后的 JSON 或 CSV方便下游直接使用。把这两种文件分开开发和排查都会更轻松。工程化不是玄学它只是用一系列简单规则让脚本从“能跑”变成“能长期跑”。6. 常见报错的排查链路一步步缩小问题范围在这个场景里最常见的报错大概有下面几类。现象可能原因优先排查动作requests.exceptions.ConnectionError网络不通或目标接口暂时不可达先请求一个普通网页确认本地网络HTTP 404宝可梦名称或 ID 写错把输入换成已知正确的 ID比如 25HTTP 429请求过于频繁被限流增大间隔开启重试逻辑KeyError: xxx字段名拼错或返回结构变化先打印resp.json().keys()文件乱码写入时没有指定 utf-8写文件时统一使用encodingutf-8光有表还不够还要有排查顺序。遇到问题先不要急着改代码按下面这条链路来先看现象是完全没有输出还是报错还是输出空值再看输入URL 里的名称或 ID 是否正确参数类型对不对再看环境requests装了吗Python 版本对吗网络环境有没有限制再看参数timeout设置了吗间隔够不够是不是触发了限流最后看工具边界PokeAPI 是不是真的返回了这个字段是不是某一只宝可梦的特殊情况很多人一看到报错就去改解析逻辑结果发现其实是网络超时也有人反复查代码却忘了先把 URL 粘到浏览器里看一眼。先确认“服务端返回了什么”往往能省下大量时间。代码里还可以加一点防御式写法if types not in data: logger.warning(数据里没有 types 字段跳过处理) return None这不算过度设计。真实接口中总会有个别记录字段不全或者返回的数据结构和你预期不完全一致。用防御式写法至少不会让一个异常把整批任务干掉。7. 皮卡丘只是开始这套思路能搬到其他数据源上最后聊一聊这类项目的长期价值。为什么用 PokeAPI 练手非常合适因为它足够清晰有文档、有稳定域名、返回结构统一、没有复杂鉴权流程。你可以把精力放在“请求-解析-存储-重试-日志”这整套流程上而不是被各种认证和签名消耗掉。但底层的工作流不只是 PokeAPI 适用。换一个公开 API比如天气、城市、开源项目仓库信息流程骨架几乎是一样的先确定数据源的 URL 规则再定义目标数据模型然后实现拉取、解析、校验接着落地到文件或数据库最后加上重试、日志、进度恢复。如果你能把这个流程做扎实下一次面对“帮我把某个外部数据同步到本地”的需求时你写的不再是一次性脚本而是一个小系统的第一版。7.1 为什么用公共 API 练手最合适公共 API 的数据结构往往经过设计字段命名也相对规范。用这类接口练手你可以在“数据本身没有问题”的前提下去写请求和管理流程更容易判断自己的逻辑到底哪里出了问题。如果一上来就面对一个文档缺失、字段混乱、还经常限流的私有接口你会被各种意外状况淹没。不是说不可以学但对新手来说先用 PokeAPI 这类接口把流程跑顺是性价比最高的路径。7.2 向正式工作流过渡缓存、数据库、调度、监控往更正式的方向过渡你还可以补上这些能力缓存层已经成功抓取的数据不再重复请求数据库用 SQLite 或 PostgreSQL 存储避免大量 JSON 文件难以查询调度用 cron 或 APScheduler 定时执行让数据保持更新监控记录成功率、耗时、数据量变化异常时通知自己。对大多数入门项目来说一次性全做会非常重。我更建议的路径是先跑通最小闭环然后做批量、重试和日志等真的有需要再加数据库和调度。不要一开始就追求“完美架构”。“去吧皮卡丘”这句话真正的价值不是把皮卡丘叫出来而是提醒我们技术里的很多复杂问题都是从最简单的一小步开始放大出来的。先请求一只皮卡丘再请求一百只先打印到终端再写入文件先靠print再靠logging先随缘中断再做断点续抓。如果你现在正准备练手我建议你真的打开终端建一个虚拟环境从请求皮卡丘开始先把最小闭环跑通。跑通之后不要急着全量抓取先定义好你的数据模型加上日志、重试、进度记录再做批量。这个过程本身比“抓到皮卡丘”那个结果更值得。当你有一天把整个流程扩展成自己的数据管道再回头看那句“去吧皮卡丘”它已经不再是一句台词而是一段代码里的函数名或者你自动化流程的第一个入口。这才是我觉得这类项目真正迷人的地方。

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价