Python Requests + JSON 解析实战：3步爬取2822所高校近3年录取分数线
·
Python Requests + JSON 解析实战：3步爬取2822所高校近3年录取分数线
高考志愿填报是每个考生家庭的头等大事，而准确获取高校录取分数线数据则是科学决策的基础。本文将带你用Python的Requests库和JSON解析技术，从公开API接口高效获取全国2822所高校的录取数据。不同于网上零散的爬虫教程，我们将系统讲解API分析、请求构建、JSON响应处理的全流程，并提供优化后的完整代码结构。
1. 环境准备与API分析
在开始编写爬虫之前，我们需要先准备好开发环境并分析目标API的结构特点。这个阶段的工作往往决定了后续爬虫的稳定性和可维护性。
1.1 安装必要依赖
确保你的Python环境已安装以下库：
pip install requests pandas
Requests库将用于发送HTTP请求，而Pandas则能帮助我们高效处理爬取到的结构化数据。
1.2 分析目标API
通过浏览器开发者工具分析，我们发现目标网站采用了前后端分离的架构，数据通过JSON API提供。关键API端点包括：
-
学校列表接口：
https://static-data.gaokao.cn/www/2.0/school/name.json -
学校详情接口：
https://static-data.gaokao.cn/www/2.0/school/{school_id}/info.json
提示：现代网站常采用这种RESTful API设计，通过观察网络请求可以快速找到数据源头，避免解析HTML的复杂性。
1.3 请求头配置
为防止被反爬机制拦截，我们需要配置合理的请求头：
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'Referer': 'https://www.gaokao.cn/',
'Accept': 'application/json, text/plain, */*'
}
2. 核心爬虫实现
有了前期准备，我们现在可以着手实现爬虫的核心功能。这部分将分为三个关键步骤：获取学校列表、获取学校详情、处理录取分数线数据。
2.1 获取学校基础信息
首先获取全国高校的基本信息列表：
import requests
import json
def get_school_list():
url = 'https://static-data.gaokao.cn/www/2.0/school/name.json'
try:
response = requests.get(url, headers=headers, timeout=10)
response.raise_for_status()
data = response.json()
return data['data']
except requests.exceptions.RequestException as e:
print(f"获取学校列表失败: {e}")
return None
这个函数会返回包含所有学校ID和名称的列表，这是后续获取详细数据的基础。
2.2 获取学校详情数据
有了学校ID后，我们可以获取每所学校的详细信息：
def get_school_detail(school_id):
url = f'https://static-data.gaokao.cn/www/2.0/school/{school_id}/info.json'
try:
response = requests.get(url, headers=headers, timeout=15)
response.raise_for_status()
return response.json()['data']
except requests.exceptions.RequestException as e:
print(f"获取学校{school_id}详情失败: {e}")
return None
2.3 处理分数线数据
分数线数据嵌套在学校详情中，我们需要专门处理：
def process_admission_scores(detail_data, province_id):
scores = {}
if 'pro_type_min' in detail_data:
province_data = detail_data['pro_type_min'].get(str(province_id), [])
for entry in province_data:
year = entry['year']
score_info = {}
for score_type, score in entry['type'].items():
# 将分数类型代码转换为可读名称
type_name = {
'1': '理科',
'2': '文科',
'3': '综合类',
'2073': '物理类',
'2074': '历史类'
}.get(score_type, f'未知类型{score_type}')
score_info[type_name] = score
scores[year] = score_info
return scores
3. 数据存储与优化
爬取到的数据需要合理存储以便后续分析。同时，我们还需要考虑爬虫的稳定性和效率优化。
3.1 数据存储方案
我们提供两种存储方式供选择：
CSV存储方案：
import csv
def save_to_csv(data, filename):
with open(filename, 'w', newline='', encoding='utf-8-sig') as f:
writer = csv.DictWriter(f, fieldnames=data[0].keys())
writer.writeheader()
writer.writerows(data)
SQLite数据库方案：
import sqlite3
def init_db():
conn = sqlite3.connect('gaokao.db')
c = conn.cursor()
c.execute('''CREATE TABLE IF NOT EXISTS schools
(id TEXT PRIMARY KEY, name TEXT, province TEXT,
f985 INTEGER, f211 INTEGER, ruanke_rank INTEGER)''')
c.execute('''CREATE TABLE IF NOT EXISTS scores
(school_id TEXT, province_id INTEGER, year INTEGER,
score_type TEXT, score TEXT,
FOREIGN KEY(school_id) REFERENCES schools(id))''')
conn.commit()
return conn
3.2 反反爬策略
为确保爬虫稳定运行，我们需要实现以下防护措施：
- 请求间隔控制：
import time
import random
def random_delay():
time.sleep(random.uniform(1, 3))
- 失败重试机制：
def safe_request(url, max_retries=3):
for attempt in range(max_retries):
try:
response = requests.get(url, headers=headers, timeout=15)
response.raise_for_status()
return response
except requests.exceptions.RequestException as e:
if attempt == max_retries - 1:
raise
time.sleep(2 ** attempt)
- 代理IP池（可选）：
proxies = {
'http': 'http://your_proxy:port',
'https': 'http://your_proxy:port'
}
3.3 完整爬虫代码
将上述组件整合，我们得到完整的爬虫实现：
import requests
import json
import csv
import time
import random
import sqlite3
from tqdm import tqdm
class GaokaoSpider:
def __init__(self):
self.headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'Referer': 'https://www.gaokao.cn/'
}
self.provinces = {
11: "北京", 12: "天津", 13: "河北", 14: "山西", 15: "内蒙古",
21: "辽宁", 22: "吉林", 23: "黑龙江", 31: "上海", 32: "江苏",
33: "浙江", 34: "安徽", 35: "福建", 36: "江西", 37: "山东",
41: "河南", 42: "湖北", 43: "湖南", 44: "广东", 45: "广西",
46: "海南", 50: "重庆", 51: "四川", 52: "贵州", 53: "云南",
54: "西藏", 61: "陕西", 62: "甘肃", 63: "青海", 64: "宁夏",
65: "新疆"
}
def get_school_list(self):
url = 'https://static-data.gaokao.cn/www/2.0/school/name.json'
try:
response = requests.get(url, headers=self.headers, timeout=10)
response.raise_for_status()
return response.json()['data']
except Exception as e:
print(f"获取学校列表失败: {e}")
return None
def get_school_detail(self, school_id):
url = f'https://static-data.gaokao.cn/www/2.0/school/{school_id}/info.json'
try:
response = requests.get(url, headers=self.headers, timeout=15)
response.raise_for_status()
return response.json()['data']
except Exception as e:
print(f"获取学校{school_id}详情失败: {e}")
return None
def run(self):
schools = self.get_school_list()
if not schools:
print("无法获取学校列表，程序终止")
return
all_data = []
for school in tqdm(schools[:100], desc="爬取进度"): # 测试时限制100所
detail = self.get_school_detail(school['school_id'])
if not detail:
continue
for province_id, province_name in self.provinces.items():
scores = self.process_scores(detail, province_id)
if scores:
school_data = {
'school_id': school['school_id'],
'school_name': detail['name'],
'province': province_name,
'985': detail.get('f985', '0') == '1',
'211': detail.get('f211', '0') == '1',
'ruanke_rank': detail.get('ruanke_rank', ''),
**scores
}
all_data.append(school_data)
time.sleep(random.uniform(1, 2))
self.save_to_csv(all_data, 'gaokao_scores.csv')
def process_scores(self, detail, province_id):
scores = {}
if 'pro_type_min' in detail:
province_data = detail['pro_type_min'].get(str(province_id), [])
for entry in province_data:
year = entry['year']
for score_type, score in entry['type'].items():
type_name = {
'1': '理科', '2': '文科', '3': '综合类',
'2073': '物理类', '2074': '历史类'
}.get(score_type, f'类型{score_type}')
scores[f'{year}_{type_name}'] = score
return scores
def save_to_csv(self, data, filename):
if not data:
return
# 动态生成字段名
fieldnames = set()
for item in data:
fieldnames.update(item.keys())
with open(filename, 'w', newline='', encoding='utf-8-sig') as f:
writer = csv.DictWriter(f, fieldnames=sorted(fieldnames))
writer.writeheader()
writer.writerows(data)
if __name__ == '__main__':
spider = GaokaoSpider()
spider.run()
4. 数据分析与应用
获取到原始数据后，我们可以进行进一步的分析和处理，为志愿填报提供更有价值的参考。
4.1 数据清洗与转换
使用Pandas进行数据清洗：
import pandas as pd
def clean_data(df):
# 转换985/211标志
df['985'] = df['985'].astype(bool)
df['211'] = df['211'].astype(bool)
# 处理软科排名
df['ruanke_rank'] = pd.to_numeric(df['ruanke_rank'], errors='coerce')
# 提取分数列
score_cols = [col for col in df.columns if any(year in col for year in ['2020', '2021', '2022'])]
for col in score_cols:
df[col] = pd.to_numeric(df[col], errors='coerce')
return df
4.2 常见分析场景
分析某省录取分数线分布：
def analyze_province_scores(df, province):
province_df = df[df['province'] == province].copy()
score_cols = [col for col in province_df.columns if any(year in col for year in ['2020', '2021', '2022'])]
# 计算各校各年份平均分
for year in ['2020', '2021', '2022']:
year_cols = [col for col in score_cols if year in col]
province_df[f'avg_{year}'] = province_df[year_cols].mean(axis=1)
return province_df.sort_values(by='ruanke_rank')
对比985/211与非985/211院校分数线：
def compare_school_levels(df, province):
province_df = df[df['province'] == province].copy()
# 分组统计
result = province_df.groupby(['985', '211']).agg({
'2022_理科': 'mean',
'2022_文科': 'mean',
'ruanke_rank': 'mean'
}).reset_index()
return result
4.3 数据可视化示例
使用Matplotlib进行简单的可视化：
import matplotlib.pyplot as plt
def plot_score_trends(df, school_name, province):
school_df = df[(df['school_name'] == school_name) &
(df['province'] == province)].iloc[0]
years = ['2020', '2021', '2022']
score_types = ['理科', '文科', '物理类', '历史类']
plt.figure(figsize=(10, 6))
for st in score_types:
scores = [school_df.get(f'{y}_{st}', None) for y in years]
if any(s is not None for s in scores):
plt.plot(years, scores, marker='o', label=st)
plt.title(f'{school_name}在{province}录取分数线趋势')
plt.xlabel('年份')
plt.ylabel('分数线')
plt.legend()
plt.grid()
plt.show()
5. 高级技巧与注意事项
在实际应用中，我们还需要考虑更多细节来确保爬虫的稳定性、数据的准确性以及使用的合规性。
5.1 增量爬取策略
为避免重复爬取全部数据，可以实现增量爬取：
def get_last_crawl_time():
try:
with open('last_crawl.txt', 'r') as f:
return float(f.read())
except:
return 0
def save_crawl_time():
with open('last_crawl.txt', 'w') as f:
f.write(str(time.time()))
def is_school_updated(school_id, last_crawl):
# 实现检查学校数据是否更新的逻辑
# 可以通过API的Last-Modified头或其他机制
return True
5.2 数据校验机制
确保爬取的数据质量：
def validate_school_data(data):
required_fields = ['school_id', 'name', 'province_name']
if not all(field in data for field in required_fields):
return False
if not isinstance(data.get('pro_type_min', {}), dict):
return False
return True
5.3 法律与伦理考量
在开发和使用爬虫时，务必注意：
- 遵守robots.txt ：检查目标网站的爬虫政策
- 控制请求频率 ：避免对目标服务器造成过大压力
- 数据使用限制 ：确保爬取的数据仅用于个人学习和研究
- 用户代理标识 ：明确标识你的爬虫，如添加联系信息
headers = {
'User-Agent': 'GaokaoDataCrawler/1.0 (+https://example.com/crawler)',
'From': 'your_email@example.com'
}
5.4 性能优化建议
对于大规模爬取，可以考虑以下优化：
- 异步请求 ：使用aiohttp代替requests
- 分布式爬取 ：使用Scrapy-Redis等框架
- 缓存机制 ：对不变的数据进行本地缓存
- 连接池 ：复用HTTP连接
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
session = requests.Session()
retries = Retry(total=5, backoff_factor=1)
session.mount('https://', HTTPAdapter(max_retries=retries))
6. 扩展应用场景
获取到高校录取数据后，这些数据可以应用于多种有价值的场景，为考生和家长提供更多决策支持。
6.1 志愿填报辅助系统
基于历史数据构建预测模型：
from sklearn.linear_model import LinearRegression
import numpy as np
def predict_score(school_id, province, score_type):
# 获取历史数据
history = get_history_scores(school_id, province, score_type)
if len(history) < 2:
return None
# 简单线性回归预测
X = np.array([int(year) for year in history.keys()]).reshape(-1, 1)
y = np.array(list(history.values()))
model = LinearRegression()
model.fit(X, y)
return model.predict([[2023]])[0]
6.2 院校对比工具
实现多所院校的对比功能：
def compare_schools(school_ids, province):
comparison = []
for sid in school_ids:
school = get_school_data(sid)
scores = get_scores(sid, province)
comparison.append({
'name': school['name'],
'province': school['province_name'],
'985': school['f985'] == '1',
'211': school['f211'] == '1',
'rank': school.get('ruanke_rank', ''),
**scores
})
return pd.DataFrame(comparison)
6.3 录取概率计算
结合位次数据计算录取概率：
def calculate_admission_probability(school_id, province, score, rank):
school_data = get_school_data(school_id)
province_data = get_province_data(province)
# 获取近3年录取位次
past_ranks = [
school_data.get(f'{year}_rank', None)
for year in ['2020', '2021', '2022']
]
if not any(past_ranks):
return None
# 简单加权计算
avg_rank = sum(r for r in past_ranks if r is not None) / len([r for r in past_ranks if r is not None])
# 计算位次百分比
rank_percent = rank / province_data['total_candidates']
school_rank_percent = avg_rank / province_data['total_candidates']
# 简单概率模型
probability = max(0, min(1, 1 - (rank_percent - school_rank_percent)))
return probability
7. 常见问题解决方案
在实际使用过程中，你可能会遇到以下常见问题，这里提供相应的解决方案。
7.1 API变更应对
当API结构发生变化时，可以这样处理：
def get_school_list():
# 尝试多种可能的API端点
endpoints = [
'https://static-data.gaokao.cn/www/2.0/school/name.json',
'https://api.gaokao.cn/v2/schools',
'https://data.gaokao.cn/school/list'
]
for endpoint in endpoints:
try:
response = requests.get(endpoint, headers=headers, timeout=5)
data = response.json()
if 'data' in data and len(data['data']) > 0:
return data['data']
except:
continue
raise Exception("所有API端点尝试失败")
7.2 数据缺失处理
处理部分学校或省份数据缺失的情况：
def get_scores_with_fallback(school_id, province_id):
# 主接口尝试
scores = get_scores_from_main_api(school_id, province_id)
if scores:
return scores
# 备用接口尝试
scores = get_scores_from_secondary_api(school_id, province_id)
if scores:
return scores
# 最终回退
return {
'2022': {'理科': None, '文科': None},
'2021': {'理科': None, '文科': None},
'2020': {'理科': None, '文科': None}
}
7.3 验证码处理方案
遇到验证码时可以：
- 降低请求频率
- 使用OCR识别简单验证码
- 切换用户代理和IP
- 寻找无需验证码的API端点
def solve_captcha(image_url):
# 简单示例，实际可能需要更复杂的OCR服务
import pytesseract
from PIL import Image
import io
import requests
response = requests.get(image_url)
img = Image.open(io.BytesIO(response.content))
return pytesseract.image_to_string(img)
7.4 性能瓶颈优化
当爬取大量数据时，注意以下优化点：
- 批量请求 ：合并多个学校的请求
- 并行处理 ：使用多线程或异步IO
- 内存管理 ：及时清理不用的数据
- 断点续爬 ：保存爬取进度
from concurrent.futures import ThreadPoolExecutor
def batch_get_school_details(school_ids, workers=5):
with ThreadPoolExecutor(max_workers=workers) as executor:
results = list(executor.map(get_school_detail, school_ids))
return [r for r in results if r is not None]
8. 项目结构建议
对于完整的志愿填报分析系统，推荐如下项目结构：
gaokao-analyzer/
├── crawler/ # 爬虫核心代码
│ ├── __init__.py
│ ├── api.py # API请求封装
│ ├── storage.py # 数据存储处理
│ └── utils.py # 工具函数
├── analysis/ # 数据分析
│ ├── __init__.py
│ ├── trends.py # 分数线趋势分析
│ └── probability.py # 录取概率计算
├── config/ # 配置文件
│ ├── __init__.py
│ └── settings.py
├── data/ # 数据存储
│ ├── raw/ # 原始数据
│ └── processed/ # 处理后的数据
├── docs/ # 文档
├── tests/ # 测试代码
├── requirements.txt # 依赖列表
└── main.py # 主入口
这种结构清晰分离了不同功能模块，便于维护和扩展。每个模块都有明确的职责，例如：
-
crawler处理所有数据获取逻辑 -
analysis包含各种数据分析功能 -
config集中管理配置项 -
data按照原始和处理后的状态组织数据文件
9. 进一步学习资源
要深入掌握Python爬虫和数据分析技术，可以参考以下资源：
9.1 推荐书籍
- 《Python网络数据采集》 - Ryan Mitchell
- 《用Python写网络爬虫》 - Richard Lawson
- 《Python数据分析实战》 - Fabio Nelli
9.2 在线课程
- Coursera: "Python for Everybody"
- Udemy: "Python for Data Science and Machine Learning Bootcamp"
- 慕课网: "Python爬虫工程师"
9.3 相关工具库
- 爬虫框架 ：Scrapy, BeautifulSoup, Selenium
- 数据处理 ：Pandas, NumPy, OpenRefine
- 可视化 ：Matplotlib, Seaborn, Plotly
- 数据库 ：SQLAlchemy, Dataset, MongoDB
9.4 社区资源
- Stack Overflow的Python和Web Scraping标签
- GitHub上的开源爬虫项目
- 知乎、掘金等技术社区的相关话题
10. 实际应用案例
最后，我们来看几个实际应用这些技术的案例，展示如何将爬取的数据转化为有价值的洞察。
10.1 院校录取线波动分析
def analyze_score_fluctuation(df, province):
# 计算各校各年份分数变化
fluctuation = []
for _, school in df.iterrows():
scores = []
for year in ['2020', '2021', '2022']:
year_scores = [school[f'{year}_理科'], school[f'{year}_文科']]
avg_score = sum(s for s in year_scores if pd.notna(s)) / len([s for s in year_scores if pd.notna(s)])
scores.append(avg_score)
if len(scores) == 3:
change_2021 = scores[1] - scores[0]
change_2022 = scores[2] - scores[1]
fluctuation.append({
'school': school['school_name'],
'change_2021': change_2021,
'change_2022': change_2022,
'stability': np.std([change_2021, change_2022])
})
return pd.DataFrame(fluctuation).sort_values('stability')
10.2 性价比院校推荐
结合排名和录取分数找出性价比高的学校：
def find_best_value_schools(df, province, score):
province_df = df[df['province'] == province].copy()
# 计算分数与排名的比值
province_df['value'] = province_df['ruanke_rank'] / province_df['2022_理科']
# 筛选可达学校
reachable = province_df[province_df['2022_理科'] <= score]
return reachable.sort_values('value').head(10)
10.3 跨省报考策略分析
比较同一学校在不同省份的录取差异：
def compare_cross_province(school_name):
school_data = df[df['school_name'] == school_name]
if school_data.empty:
return None
comparison = []
for _, row in school_data.iterrows():
comparison.append({
'province': row['province'],
'2022_理科': row['2022_理科'],
'2022_文科': row['2022_文科'],
'difference': row['2022_理科'] - row['2022_文科']
})
return pd.DataFrame(comparison).sort_values('2022_理科')
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
8
0- 0
已为社区贡献16条内容
所有评论(0)