Python处理API返回数据时,遇到json.decoder.JSONDecodeError怎么办?一个真实爬虫案例的完整排错流程
Python处理API返回数据时遇到JSONDecodeError的实战排错指南
上周在抓取某电商平台价格数据时,服务器突然返回了一个HTML登录页面,而我的脚本因为json()方法直接崩溃。这种看似简单的JSON解析错误,背后可能隐藏着网络异常、权限变更、反爬策略等多重陷阱。本文将用一个真实爬虫案例,带你构建完整的防御性编程方案。
1. 从现象到本质:理解JSONDecodeError的根源
当requests的json()方法抛出json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)时,根本原因是响应体不符合JSON格式。常见触发场景包括:
- 服务器返回非JSON内容:HTML错误页(如502 Bad Gateway)、纯文本提示、XML格式等
- 空响应体:Content-Length为0或响应体完全为空
- 编码问题:响应头声明与实际编码不一致导致二进制乱码
- 部分数据截断:网络中断导致响应不完整
通过下面这个诊断流程图可以快速定位问题层级:
开始 │ ├─ 检查response.status_code → 非200则进入HTTP错误处理 │ ├─ 检查response.content长度 → 空内容则进入空响应处理 │ └─ 检查response.headers['Content-Type'] → 非application/json则进入格式转换处理关键验证步骤:
import requests resp = requests.get('https://api.example.com/data') print(resp.status_code) # 状态码检查 print(resp.headers['Content-Type']) # 内容类型验证 print(resp.content[:200]) # 查看原始响应前200字节2. 构建防御性解析体系
2.1 响应预处理框架
完整的防御性处理应该包含以下环节:
def safe_json_parse(response): # 状态码过滤 if response.status_code != 200: raise CustomHTTPError(response.status_code) # 内容非空检查 if not response.content: raise EmptyContentError() # 内容类型验证 content_type = response.headers.get('Content-Type', '') if 'application/json' not in content_type: return handle_non_json_response(response) # 实际JSON解析 try: return response.json() except ValueError as e: raise JSONParseError(str(e))2.2 常见异常处理策略
| 异常类型 | 触发场景 | 处理方案 | 重试建议 |
|---|---|---|---|
| 401/403 | 认证失效 | 刷新token/检查权限 | 立即重试1次 |
| 429 | 请求限速 | 添加延迟/切换IP | 指数退避重试 |
| 500+ | 服务端错误 | 记录错误并跳过 | 固定间隔重试 |
| JSONDecodeError | 格式错误 | 原始数据保存分析 | 视情况重试 |
实战示例:带重试机制的请求封装
from tenacity import retry, stop_after_attempt, wait_exponential @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10) ) def robust_request(url): resp = requests.get(url, timeout=5) resp.raise_for_status() try: return resp.json() except ValueError: with open(f'error_{int(time.time())}.html', 'wb') as f: f.write(resp.content) raise3. 高级调试技巧
3.1 错误数据保存策略
建议建立错误样本库,自动保存异常响应:
ERROR_DIR = 'api_errors' def save_error_response(response, error_type): os.makedirs(ERROR_DIR, exist_ok=True) timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') filename = f"{error_type}_{timestamp}.bin" with open(os.path.join(ERROR_DIR, filename), 'wb') as f: f.write(response.content) meta = { 'url': response.url, 'headers': dict(response.headers), 'status': response.status_code } with open(os.path.join(ERROR_DIR, f"{filename}.meta"), 'w') as f: json.dump(meta, f)3.2 自动化内容类型检测
对于非标准JSON响应,可以尝试智能转换:
def content_type_detector(raw_data): if raw_data.startswith(b'<'): return 'html/xml' elif raw_data.startswith(b'{') or raw_data.startswith(b'['): return 'json' elif b'error' in raw_data.lower(): return 'text' return 'binary' def convert_response(response): content_type = content_type_detector(response.content) if content_type == 'json': return response.json() elif content_type == 'html/xml': return parse_html_error(response.text) else: return {'raw': response.text}4. 生产环境最佳实践
4.1 监控指标设计
建议采集以下关键指标:
API健康度指标:
- 成功率 = (成功请求数 / 总请求数) × 100%
- 平均响应延迟(按状态码分类)
- JSON解析失败率
错误分类统计:
ERROR_TYPES = { 'timeout': 0, 'http_error': 0, 'json_decode': 0, 'validation': 0 } def update_error_stats(error_type): ERROR_TYPES[error_type] += 1 if sum(ERROR_TYPES.values()) % 100 == 0: send_alert_report(ERROR_TYPES)
4.2 熔断机制实现
当错误率超过阈值时自动停止请求:
from circuitbreaker import circuit @circuit(failure_threshold=5, recovery_timeout=60) def call_protected_api(url): return robust_request(url)在长期运行的爬虫系统中,建议采用以下架构设计:
请求队列 → 工作线程池 → 结果处理器 ↓ 错误处理器(重试/记录) ↓ 熔断状态监控最近在处理某金融数据API时,发现其偶尔会在JSON中混入NaN值导致解析失败。最终通过自定义JSON解析器解决了这个问题:
import math from json import JSONDecoder def parse_float(s): if s == 'NaN': return float('nan') return float(s) decoder = JSONDecoder(parse_float=parse_float) data = decoder.decode(response.text)