做数据开发的朋友,平时最常见的练习项目是什么?爬房价、爬天气、爬商品信息,翻来覆去都是那几个方向。今天换个更有趣的题材——用 Python 把宝可梦全图鉴数据拉下来,做成一个可以查询的本地数据库,再提供一个 HTTP 接口给前端用。整个过程会用到 requests、pandas、Flask 这些非常常规的 Python 库,但真正有意思的地方在于数据建模和接口设计。
很多人看到“去吧皮卡丘”这个标题,以为是一篇游戏攻略或者娱乐文章。实际上,这篇文章要做的是一套完整的数据采集与分析流程演示。我们会以一个大家都很熟悉的 IP 为切入点,把“数据获取 → 清洗入库 → 接口封装 → 可视化分析”这条链路完整跑一遍。最后的产出是一个可以运行在本地、支持按名字、属性、种族值查询的宝可梦图鉴 API。
任何一个想巩固 Python 数据处理能力、想了解如何把外部 API 数据落地成本地服务的开发者,都值得花半小时把整个项目过一遍。你的收获不只是学会请求一个公开 API,而是掌握一套可迁移到其他数据项目的工程思路。
1. 这个项目到底要解决什么问题
很多人学 Python 到一定阶段会出现一个尴尬的瓶颈:语法都认识,库也装了,但不知道完整的项目长什么样。单独跑一个爬虫脚本没问题,单独启动一个 Flask 服务也没问题,可要把两者串起来,就不知道从哪里下手了。
这个项目刚好把两个最常见的需求连在了一起:数据获取和数据服务化。
我们不只是写一个脚本抓一次数据就结束,而是会做以下这些事情:
- 从公开的宝可梦 API 获取全量宝可梦数据;
- 对字段进行筛选、清洗、转化,得到一份结构化的数据集;
- 将数据导入 SQLite 数据库,方便后续查询和分析;
- 基于 Flask 构建一个查询接口,支持多条件筛选;
- 用图表做一次简单的数据分布分析,验证数据的可用性。
这套流程在真实项目中每天都在发生。不管是抓商品数据做价格监控,还是抓用户行为数据做画像分析,底层的模式和这里完全一样。区别只是数据源和业务字段变了而已。
所以,这篇文章的真正价值不是“用 Python 玩宝可梦”,而是通过一个有趣、有辨识度的案例,把数据工程的基础流程练扎实。读完以后,你自己就能照着这个模式去处理其他公开数据集。
2. 核心概念:PokéAPI、数据字段与整体架构
2.1 PokéAPI 是什么
PokéAPI 是一个免费、开源的宝可梦数据接口,提供了从第一世代到最新世代的宝可梦资料,包括名称、属性、种族值、特性、进化链、技能等大量信息。它不需要 API Key,也不需要身份认证,基本可以无门槛调用,是学习 HTTP 接口调用非常好的数据源。
官方入口地址是https://pokeapi.co/,接口采用标准的 RESTful 风格,返回 JSON 格式数据。
2.2 主要数据字段
宝可梦的数据非常丰富,单只宝可梦的详情接口会返回几十个字段。实际项目中我们只会保留一部分核心字段用来构建最小可用数据集:
| 字段 | 含义 | 示例 |
|---|---|---|
| id | 全国图鉴编号 | 25 |
| name | 英文名称 | pikachu |
| types | 属性列表 | ["electric"] |
| height | 身高,单位分米 | 4 |
| weight | 体重,单位百克 | 60 |
| base_experience | 基础经验值 | 112 |
| stats | 种族值六维 | HP、攻击、防御等 |
种族值是宝可梦对战中最重要的数据维度,包含 HP、攻击、防御、特攻、特防、速度六项,后续做数据分析时会用到。
2.3 项目架构设计
整个项目采用最简单的分层架构,一共三层:
数据采集层(Python脚本 + requests) ↓ 数据存储层(SQLite + pandas) ↓ 数据服务层(Flask API + 查询逻辑) ↓ 前端或其他客户端这套分层的好处是每一层都可以独立替换。以后想换数据源,只需要改第一层;想换数据库,只需要改第二层;想换 Web 框架,只需要改第三层。项目规模小,但结构规范,对初学者建立工程意识很有帮助。
3. 环境准备与依赖安装
在开始写代码之前,先把开发环境准备好。这个项目不需要复杂的基础设施,一台安装 Python 的电脑就够了。
3.1 Python 版本要求
建议使用 Python 3.9 及以上版本,本文的代码在 Python 3.10 下测试通过。如果你本地有多个 Python 版本,建议使用虚拟环境隔离依赖。
3.2 安装依赖库
项目需要以下 Python 库:
requests:发送 HTTP 请求,获取 API 数据;pandas:数据处理与 DataFrame 操作;flask:构建 Web 查询接口;flask-cors:解决跨域问题,方便前端调试。
创建项目目录并初始化虚拟环境:
mkdir pokedex-project cd pokedex-project python -m venv venv在 Windows 系统下激活虚拟环境:
venv\Scripts\activate在 macOS 或 Linux 系统下激活虚拟环境:
source venv/bin/activate激活后安装依赖包:
pip install requests pandas flask flask-cors如果想锁定版本,可以生成 requirements.txt:
pip freeze > requirements.txt这里特别说明一下为什么使用 SQLite 而不是 MySQL 或 PostgreSQL。项目定位是本地学习项目,SQLite 零配置、单文件、免维护,对新手非常友好,后续想迁移到其他数据库也非常简单。在真实业务中,如果数据量增长到一定程度或者需要并发写入,再考虑升级数据库。
4. 数据采集:调用 PokéAPI 获取全量宝可梦数据
4.1 分析接口结构
PokéAPI 有几个常用的端点:
GET /api/v2/pokemon?limit=10&offset=0:分页获取宝可梦列表;GET /api/v2/pokemon/{id}:获取单个宝可梦的完整详情;GET /api/v2/pokemon-species/{id}:获取物种信息,包括中文名和进化链;GET /api/v2/type/{id}:获取属性信息。
第一步先请求列表接口,看返回的数据长什么样。
import requests url = "https://pokeapi.co/api/v2/pokemon?limit=5&offset=0" response = requests.get(url, timeout=10) data = response.json() print(data.keys()) print(data["count"]) for item in data["results"]: print(item)预期输出效果:第一行打印出dict_keys(['count', 'next', 'previous', 'results']),第二行打印宝可梦总数(当前版本可能是 1302 只左右),然后打印出前五条的name和url。
这里有一个很重要的点:列表接口只返回名字和详情链接,不返回属性、种族值等核心数据。所以真正的字段要再请求一次详情接口才能拿到。
4.2 增量获取详情数据
如果按照最朴素的方式,把 1300 多只宝可梦的详情全部请求下来,需要循环调用 1300 多次接口。速度快的情况下大约 5 到 10 分钟完成,这完全可以接受。
不过要注意频率控制。虽然 PokéAPI 是免费开放的,但太过频繁的请求会给服务器造成压力,而且自己的 IP 也可能被限制。建议在循环中加time.sleep(0.1)的间隔。
下面是最小可用的采集脚本:
import time import requests import pandas as pd BASE_URL = "https://pokeapi.co/api/v2/pokemon" def fetch_pokemon_detail(pokemon_id): """获取单个宝可梦的详情数据""" url = f"{BASE_URL}/{pokemon_id}" response = requests.get(url, timeout=10) if response.status_code == 200: return response.json() return None def parse_pokemon_data(raw_data): """从原始 JSON 中提取需要的字段""" stats = {} for item in raw_data["stats"]: stat_name = item["stat"]["name"] base_value = item["base_stat"] stats[stat_name] = base_value types = [item["type"]["name"] for item in raw_data["types"]] parsed = { "id": raw_data["id"], "name": raw_data["name"], "height": raw_data["height"], "weight": raw_data["weight"], "base_experience": raw_data["base_experience"], "type_1": types[0] if len(types) > 0 else None, "type_2": types[1] if len(types) > 1 else None, "hp": stats.get("hp", 0), "attack": stats.get("attack", 0), "defense": stats.get("defense", 0), "special_attack": stats.get("special-attack", 0), "special_defense": stats.get("special-defense", 0), "speed": stats.get("speed", 0), } return parsed def fetch_all_pokemon(limit=151): """获取指定范围内的宝可梦数据,默认第一世代 151 只""" all_data = [] for pokemon_id in range(1, limit + 1): raw_data = fetch_pokemon_detail(pokemon_id) if raw_data: parsed_data = parse_pokemon_data(raw_data) all_data.append(parsed_data) print(f"已获取: {parsed_data['name']} (ID: {pokemon_id})") time.sleep(0.1) return pd.DataFrame(all_data)执行主函数来测试:
df = fetch_all_pokemon(151) print(df.head()) print(df.shape)这里我们只抓了第一世代的 151 只宝可梦,是为了快速跑通流程并避免给服务器造成不必要的压力。如果想抓全量,只需要把limit参数改成全部数量即可,注意耗时和频率控制。
第一次运行需要一点耐心。每抓一只输出一行进度,看到信息滚动起来,说明一切正常。完成后你会得到一个 151 行的 DataFrame,每一行就是一只宝可梦的完整核心数据。
5. 数据清洗与入库
5.1 为什么需要清洗
API 返回的原始 JSON 是一种嵌套结构,比如属性是一个列表,种族值也是一个列表里面套字典。这种结构适合传输,但不适合直接分析和查询。数据清洗的目的就是把嵌套结构拍平成一张二维表,每一列是一个字段,每一行是一条记录。
在parse_pokemon_data函数中,我们其实已经完成了大部分清洗工作。把types列表拆成了type_1和type_2,把stats列表拆成了六个独立列。这就是最常见的清洗方式:字段展开。
5.2 保存到 CSV 文件
数据量不大时,CSV 是很好的中间存储格式,方便用 Excel 或 pandas 随时查看。
df.to_csv("pokemon_data.csv", index=False, encoding="utf-8")5.3 导入 SQLite 数据库
接下来把 DataFrame 写入 SQLite。之所以要入库,是因为后面要提供查询 API,数据库查询比直接操作内存中的 DataFrame 更规范、更接近真实项目架构。
import sqlite3 def create_database(df, db_path="pokedex.db"): """将 DataFrame 写入 SQLite 数据库""" conn = sqlite3.connect(db_path) df.to_sql("pokemon", conn, if_exists="replace", index=False) conn.close() print(f"数据已写入 {db_path}") create_database(df)运行成功后,项目目录下会出现一个pokedex.db文件。你可以用 SQLite 客户端或者 pandas 验证数据:
conn = sqlite3.connect("pokedex.db") check_df = pd.read_sql_query("SELECT * FROM pokemon LIMIT 10", conn) print(check_df) conn.close()5.4 数据验证
入库后必须要做的验证有三项:
- 总行数是否为 151;
- 是否重复;
- 是否有空值。
这几项检查可以通过 pandas 一行代码完成:
print(f"总行数: {len(df)}") print(f"重复行数: {df.duplicated().sum()}") print(f"空值统计:\n{df.isnull().sum()}")从逻辑上看,第一世代每只宝可梦的编号是唯一的,不应该出现重复行。type_2字段可能为空,因为部分宝可梦只有单一属性,这是正常的,不需要处理。但如果name或id为空,就必须回去检查采集逻辑。
这个小结很关键:没有经过验证的数据是不应该进入下一步的。这是数据工程里非常重要的一条原则,哪怕是一个学习项目,也应该养成验证的习惯。
6. 实战进阶:构建宝可梦图鉴查询 API
6.1 需求定义
数据入库后,下一步是把它变成一个可以调用的服务。我们定义以下查询需求:
- 查询全部宝可梦,支持分页;
- 根据 ID 精确查询;
- 根据名字模糊搜索;
- 根据属性筛选;
- 根据种族值范围筛选。
这些需求覆盖了接口设计里最常见的几种模式:列表分页、ID 查询、模糊搜索、条件筛选。掌握了这些,以后做一个电商商品接口或者用户管理接口,逻辑是一样的。
6.2 Flask 应用整体结构
创建一个app.py文件,把所有接口写在一个文件里。项目不大,暂时不需要复杂的包结构。
from flask import Flask, request, jsonify from flask_cors import CORS import sqlite3 app = Flask(__name__) CORS(app) DB_PATH = "pokedex.db" def query_db(sql, params=()): """执行 SQL 查询并返回字典列表""" conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row cursor = conn.execute(sql, params) rows = cursor.fetchall() conn.close() return [dict(row) for row in rows] @app.route("/api/pokemon", methods=["GET"]) def get_pokemon_list(): """获取宝可梦列表,支持分页和条件筛选""" page = int(request.args.get("page", 1)) limit = int(request.args.get("limit", 20)) offset = (page - 1) * limit type_filter = request.args.get("type") name_search = request.args.get("name") min_attack = request.args.get("min_attack") conditions = [] params = [] if type_filter: conditions.append("(type_1 = ? OR type_2 = ?)") params.extend([type_filter, type_filter]) if name_search: conditions.append("name LIKE ?") params.append(f"%{name_search}%") if min_attack: conditions.append("attack >= ?") params.append(min_attack) where_clause = "" if conditions: where_clause = "WHERE " + " AND ".join(conditions) sql = f"SELECT * FROM pokemon {where_clause} LIMIT ? OFFSET ?" params.extend([limit, offset]) rows = query_db(sql, params) return jsonify({"code": 0, "data": rows, "page": page, "limit": limit}) @app.route("/api/pokemon/<int:pokemon_id>", methods=["GET"]) def get_pokemon_detail(pokemon_id): """根据 ID 获取宝可梦详情""" rows = query_db("SELECT * FROM pokemon WHERE id = ?", (pokemon_id,)) if not rows: return jsonify({"code": 404, "message": "宝可梦不存在"}), 404 return jsonify({"code": 0, "data": rows[0]}) if __name__ == "__main__": app.run(host="0.0.0.0", port=5000, debug=True)6.3 接口功能说明
第一个接口GET /api/pokemon是核心接口,它同时支持分页和多条件组合筛选。查询参数都通过request.args获取,所有筛选条件动态拼接成 SQL 的 WHERE 子句。
这里需要注意 SQL 注入的问题。虽然当前项目是学习性质,但习惯一定要从一开始就养成。所有外部传入的参数都使用问号占位符传参,不要直接拼接字符串到 SQL 中。
第二个接口GET /api/pokemon/<id>是标准详情接口,返回单条记录。如果 ID 不存在,返回 404 状态码和错误信息。
6.4 启动与验证 API
启动服务:
python app.py然后在浏览器或 Postman 中测试接口。
查询第一页数据:
curl "http://127.0.0.1:5000/api/pokemon?page=1&limit=5"按名称模糊搜索:
curl "http://127.0.0.1:5000/api/pokemon?name=pika"按属性筛选:
curl "http://127.0.0.1:5000/api/pokemon?type=fire&limit=10"获取 ID 为 25 的皮卡丘详情:
curl "http://127.0.0.1:5000/api/pokemon/25"只要返回的 JSON 中code为 0,并且data数组里有数据,接口就算跑通了。
7. 数据可视化:用图表发现宝可梦数据的规律
数据接口服务化之后,我们再做一道“验证题”:用数据可视化验证这批数据的可用性。如果一张图能讲清楚宝可梦的种族值分布规律,说明前面的采集、清洗、入库流程是完整且正确的。
这里使用 matplotlib 做一个基础分析,先安装依赖:
pip install matplotlib写一个分析脚本:
import pandas as pd import matplotlib.pyplot as plt from matplotlib import rcParams # 设置中文字体,避免乱码 rcParams["font.sans-serif"] = ["SimHei", "Arial Unicode MS", "sans-serif"] rcParams["axes.unicode_minus"] = False df = pd.read_sql_query("SELECT * FROM pokemon", sqlite3.connect("pokedex.db")) # 1. 六维种族值平均分布 stat_cols = ["hp", "attack", "defense", "special_attack", "special_defense", "speed"] stat_means = df[stat_cols].mean() print(stat_means) # 2. 按第一属性分组,统计平均攻击力 type_attack = df.groupby("type_1")["attack"].mean().sort_values() print(type_attack) # 绘制平均种族值雷达图 import numpy as np angles = np.linspace(0, 2 * np.pi, len(stat_cols), endpoint=False).tolist() values = stat_means.tolist() values += values[:1] angles += angles[:1] fig, ax = plt.subplots(figsize=(8, 8), subplot_kw=dict(polar=True)) ax.fill(angles, values, color="red", alpha=0.25) ax.plot(angles, values, color="red", linewidth=2) ax.set_xticks(angles[:-1]) ax.set_xticklabels(["HP", "攻击", "防御", "特攻", "特防", "速度"]) ax.set_title("第一世代宝可梦平均种族值分布") plt.tight_layout() plt.savefig("stats_radar.png", dpi=200) plt.show()从第一世代的数据来看,宝可梦的平均种族值分布相对均衡,攻击和速度略高,这符合初代游戏偏重速攻的体感。雷达图跑出来以后,你会有一种“自己做了一个小数据产品”的成就感。
注意,不同系统的中文字体配置不同。Windows 一般用SimHei,macOS 可以用Arial Unicode MS或PingFang SC,如果字体没配对,图表里会出现方块乱码,这在可视化项目里是很常见的问题。
8. 常见问题与排查思路
做这个项目的过程中,你大概率会遇到下面这些问题。这里列一份排查清单,按出现频率排序:
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| 请求 PokéAPI 超时 | 网络不稳定或 API 限流 | 查看错误日志,检查网络连通性 | 增加 timeout 参数,降低请求频率,使用重试机制 |
| SQLite 数据库文件不存在 | 没有执行写入脚本 | 检查项目目录下是否有 pokedex.db 文件 | 重新执行 create_database() 函数 |
| Flask 启动失败,端口被占用 | 5000 端口被其他程序占用 | 运行 `netstat -ano | findstr 5000` 查看占用程序 |
| 中文输出乱码 | 终端编码问题 | 查看脚本头部是否指定编码 | Python3 默认 utf-8,Windows 终端可执行chcp 65001 |
| 接口返回 500 错误 | SQL 语句拼写错误或表名错误 | 查看 Flask 控制台堆栈信息 | 检查表名是否为 pokemon,字段名是否匹配 |
| 图表中文显示为方块 | 系统缺少中文字体 | 打印当前字体列表 | 配置系统对应中文字体,如 SimHei、Arial Unicode MS |
| 数据量太少,只有不到 151 条 | 接口请求部分失败被跳过 | 统计失败次数 | 增加日志记录,对失败请求进行重试 |
排错的大原则是先看日志,再猜原因。Flask 控制台会打印完整的堆栈信息,pandas 和 requests 也有详细的报错输出。别急着改代码,先根据报错信息确定是网络问题、数据问题还是代码逻辑问题,再动手修。
9. 最佳实践与生产环境建议
项目能跑通只是第一步。如果希望把这个模式用到真实工作中,下面这些建议值得认真琢磨。
9.1 请求频率与重试机制
真实项目中,外部 API 往往有严格的限流策略。需要引入重试机制,例如使用requests配合指数退避策略,失败后等待 1 秒、2 秒、4 秒再重试,最多重试 3 次。
9.2 增量更新策略
这个项目是一次性拉取全量数据,但在真实场景中,数据每天都在变化。更好的做法是设计增量同步:记录上次同步的最大 ID,下次只拉取新增部分,用更新时间戳标记每条记录,方便追踪。
9.3 配置外置
数据库路径、API 地址、请求频率这些值不应该硬编码在代码里。建议放到环境变量或配置文件中,通过os.getenv读取。这样部署到不同环境时,不需要改代码,只需要改配置。
9.4 接口统一返回格式
接口的返回结构最好全局统一。例如所有接口都返回{"code": 0, "data": ..., "message": "success"},这样前端对接时逻辑简单,不需要为每个接口单独写解析逻辑。
9.5 安全与权限
涉及生产环境时,任何开放接口都必须考虑权限控制。增加 API Key 认证、限制调用频率、对敏感接口做操作日志记录,这三件事缺一不可。同时,数据库连接要坚持最小权限原则,不应该让服务使用 root 或管理员权限。
10. 拓展方向:如何从 151 只扩展到完整图鉴
如果觉得 151 只不过瘾,想拉全量宝可梦数据,需要注意几个问题。
把limit改成全量数量之前,先设置好合理的请求间隔,建议至少保持0.2秒以上,避免对公共 API 造成压力。全量拉取会接近 1300 次请求,按 0.2 秒间隔计算,大约需要 5 到 8 分钟,需要耐心等待。
其次要考虑断点续传。如果拉取到第 500 只时网络中断,重新跑会导致前面 500 只白拉。可以在脚本中记录当前进度,下次启动时从上次失败的位置继续。
最后,全量数据分析会带来更丰富的信息量。你可以按照不同世代比较宝可梦种族值的变化趋势,也可以分析属性组合的分布规律。数据量越大,数据挖掘的乐趣也越多。
代码层面,只需要在fetch_all_pokemon的调用参数上做调整,核心逻辑不需要改动。这也体现了分层设计的价值——数据获取层的改动不会影响存储层和服务层。
除了拉全量数据,还有很多值得尝试的拓展方向:
- 增加宝可梦技能数据,构建技能查询接口;
- 加入特性数据,分析每只宝可梦的推荐特性;
- 利用爬虫抓取宝可梦图片,做一个小型图片网站;
- 把 Flask 接口改成 FastAPI,体验自动生成 Swagger 文档;
- 将 SQLite 迁移到 PostgreSQL,锻炼数据库迁移能力;
- 开发一个简单的宝可梦对战模拟器,用真实数据计算伤害。
每一个方向都能单独写一篇文章。作为学习项目,这个案例的价值在于它把太多数据工程的知识点浓缩在了一个可运行的 Demo 里。建议收藏本文,先把基础流程跑通,再根据自己的兴趣选择拓展方向。数据和接口都准备好了,接下来能玩出什么花样,就看你自己的想象力了。