深入理解与实现USDA FDC API数据分页获取

import requests import json import pandas as pd  def get_all_food_data(query_term: str, api_key: str, max_page_size: int = 200) -> list:     """     通过迭代请求USDA FDC API，获取指定查询词的所有食物营养数据。      Args:         query_term (str): 搜索的食物名称或关键词。         api_key (str): 您的USDA FDC API密钥。         max_page_size (int): 每页请求的最大结果数量，USDA FDC API最大支持200。      Returns:         list: 包含所有食物营养数据的列表，每个元素是一个字典。               如果请求失败或无数据，则返回空列表。     """     all_foods = []     # 构建基础URL，设置API密钥和查询词，并指定最大页大小     base_url = f"https://api.nal.usda.gov/fdc/v1/foods/search?api_key={api_key}&query={query_term}&pageSize={max_page_size}"      with requests.Session() as session:  # 使用requests.Session保持连接，提高效率         try:             # 1. 获取第一页数据，并确定总页数             print(f"Fetching page 1 for query: '{query_term}'...")             response = session.get(base_url, timeout=15)             response.raise_for_status()  # 检查HTTP请求是否成功             api_response = response.json()              if not api_response.get("foods"):                 print(f"No foods found for query: '{query_term}'.")                 return []              all_foods.extend(api_response["foods"])             total_pages = api_response.get("totalPages", 1)             print(f"Total pages to retrieve: {total_pages}")              # 2. 迭代获取后续页面数据             for page_num in range(2, total_pages + 1):                 print(f"Fetching page {page_num}...")                 page_url = f"{base_url}&pageNumber={page_num}"                 page_response = session.get(page_url, timeout=15)                 page_response.raise_for_status()                 page_data = page_response.json()                 all_foods.extend(page_data["foods"])          except requests.exceptions.HTTPError as http_err:             print(f"HTTP error occurred: {http_err} - Status Code: {response.status_code}")             return []         except requests.exceptions.ConnectionError as conn_err:             print(f"Connection error occurred: {conn_err}")             return []         except requests.exceptions.Timeout as timeout_err:             print(f"Timeout error occurred: {timeout_err}")             return []         except requests.exceptions.RequestException as req_err:             print(f"An unexpected error occurred: {req_err}")             return []         except json.JSONDecodeError:             print(f"Failed to decode JSON from response: {response.text[:200]}...")             return []      return all_foods  # --- 示例使用 --- if __name__ == "__main__":     # 请替换为您的实际API密钥     YOUR_API_KEY = "YOUR_USDA_API_KEY"     search_query = "raw" # 例如，搜索所有“生”的食物      # 1. 获取所有食物数据     food_items = get_all_food_data(search_query, YOUR_API_KEY)      if food_items:         print(f"nSuccessfully retrieved {len(food_items)} food items.")          # 2. 将获取到的数据转换为DataFrame并进行处理         table_data = []         for food_item in food_items:             row = {                 "Description": food_item.get("description", "N/A"),                 "FDC_ID": food_item.get("fdcId", "N/A")             }             # 提取主要营养素             for nutrient in food_item.get("foodNutrients", []):                 nutrient_name = nutrient.get("nutrientName")                 if nutrient_name:                     row[nutrient_name] = nutrient.get("value")             table_data.append(row)          df = pd.DataFrame(table_data)          # 3. 数据清洗和处理（示例）         # 移除完全为空的列（如果有很多稀疏营养素）         df.dropna(axis=1, how='all', inplace=True)         # 填充NaN值，例如用0或平均值         df.fillna(0, inplace=True)          print("nDataFrame Head:")         print(df.head())         print(f"nDataFrame Shape: {df.shape}")          # 4. 导出到excel         try:             output_filename = f'{search_query}_food_data.xlsx'             df.to_excel(output_filename, index=False)             print(f"nData successfully exported to '{output_filename}'")         except Exception as e:             print(f"Error exporting to Excel: {e}")     else:         print("No data to process or export.")