调用Faceit API获取数据时偶发JSONDecodeError问题求助

问题根源 #

大概率是大数量级匹配数据请求引发的问题：

• 重复运行时，请求频率过高触发Faceit API限流，返回非JSON格式的限流提示（如HTML页面），导致JSONDecodeError
• 全量拉取1605场数据时，API响应可能超时、传输不完整，JSON解析失败
• 首次运行可能命中API缓存，重复运行需重新拉取全量数据，服务器压力更大，更容易出现异常

优化方案 #

1. 增加请求重试与频率控制 #

给API请求添加指数退避重试机制，同时控制请求间隔，避免触发限流：

import time
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# 配置会话与重试策略
session = requests.Session()
retry_strategy = Retry(
    total=3,
    backoff_factor=1,  # 重试间隔依次为1s、2s、4s
    status_forcelist=[429, 500, 502, 503, 504]  # 针对这些状态码重试
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)

def fetch_api_data(url, headers):
    time.sleep(0.5)  # 固定间隔，避免请求过于密集
    response = session.get(url, headers=headers)
    response.raise_for_status()  # 主动抛出HTTP错误
    return response.json()

2. 分页拉取匹配数据 #

Faceit API支持分页参数（offset和limit），分批次拉取数据，不要一次性请求全量：

def get_all_csgo_matches(player_uuid, api_key):
    all_matches = []
    offset = 0
    batch_size = 100  # 每次拉取100场，可根据API限制调整
    
    while True:
        api_url = f"https://open.faceit.com/data/v4/players/{player_uuid}/games/csgo/stats?offset={offset}&limit={batch_size}"
        batch_data = fetch_api_data(api_url, {"Authorization": f"Bearer {api_key}"})
        current_batch = batch_data.get("items", [])
        
        if not current_batch:
            break
        
        all_matches.extend(current_batch)
        offset += batch_size
    
    return all_matches

3. 本地缓存已获取的数据 #

重复运行时优先读取本地缓存，避免重复请求API：

import json
import os

# 创建缓存目录
CACHE_FOLDER = "faceit_data_cache"
os.makedirs(CACHE_FOLDER, exist_ok=True)

def get_cached_uuid(username):
    cache_path = os.path.join(CACHE_FOLDER, f"{username}_uuid.json")
    if os.path.exists(cache_path):
        with open(cache_path, "r", encoding="utf-8") as f:
            return json.load(f)["uuid"]
    return None

def save_uuid_to_cache(username, uuid):
    cache_path = os.path.join(CACHE_FOLDER, f"{username}_uuid.json")
    with open(cache_path, "w", encoding="utf-8") as f:
        json.dump({"uuid": uuid}, f)

def get_cached_matches(player_uuid):
    cache_path = os.path.join(CACHE_FOLDER, f"{player_uuid}_matches.json")
    if os.path.exists(cache_path):
        with open(cache_path, "r", encoding="utf-8") as f:
            return json.load(f)
    return None

def save_matches_to_cache(player_uuid, matches):
    cache_path = os.path.join(CACHE_FOLDER, f"{player_uuid}_matches.json")
    with open(cache_path, "w", encoding="utf-8") as f:
        json.dump(matches, f)

4. 异常捕获与日志记录 #

在JSON解析环节添加异常捕获，记录错误响应内容，方便排查问题：

import logging

# 配置日志
logging.basicConfig(
    filename="faceit_scraper_errors.log",
    level=logging.ERROR,
    format="%(asctime)s - %(levelname)s - %(message)s"
)

def fetch_api_data(url, headers):
    time.sleep(0.5)
    try:
        response = session.get(url, headers=headers)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as e:
        logging.error(f"HTTP请求错误: {str(e)}，响应内容: {response.text}")
        raise
    except json.JSONDecodeError as e:
        logging.error(f"JSON解析失败: {str(e)}，响应内容: {response.text}")
        raise

5. 分批写入Excel #

不要等所有数据拉取完成再写入，每拉取一批就写入一批，降低内存占用：

import pandas as pd
from openpyxl import load_workbook

def write_matches_to_excel(matches, excel_path, append=False):
    df = pd.DataFrame(matches)
    if append:
        book = load_workbook(excel_path)
        with pd.ExcelWriter(excel_path, engine="openpyxl", mode="a", if_sheet_exists="overlay") as writer:
            writer.book = book
            df.to_excel(writer, index=False, header=False, startrow=writer.sheets["Sheet1"].max_row)
    else:
        df.to_excel(excel_path, index=False)

内容的提问来源于stack exchange，提问作者ouindoze