利用Python高效获取USDA食品营养数据：API分页处理详解

import requests
import json
import pandas as pd

def get_all_usda_foods(query_term, api_key):
    """
    通过USDA食品数据中心API获取所有匹配的食品营养数据。
    该函数处理API分页，并返回一个包含所有食品项的列表。

    Args:
        query_term (str): 查询的食品名称或关键词，例如 'raw'。
        api_key (str): 您的USDA API密钥。

    Returns:
        list: 包含所有食品项字典的列表，如果请求失败则返回None。
    """
    all_foods = []
    # 构造基础URL，设置pageSize为最大值200以减少请求次数
    base_url = f"https://api.nal.usda.gov/fdc/v1/foods/search?api_key={api_key}&query={query_term}&pageSize=200"

    with requests.Session() as session: # 使用requests.Session提高效率
        try:
            # 首次请求获取总页数
            print(f"Fetching initial page for query: '{query_term}'...")
            initial_response = session.get(base_url, timeout=10)
            initial_response.raise_for_status() # 检查HTTP请求是否成功
            api_response_data = initial_response.json()

            if not api_response_data.get("foods"):
                print(f"No foods found for query: '{query_term}'.")
                return []

            total_pages = api_response_data.get("totalPages", 1)
            print(f"Total pages to retrieve: {total_pages}")

            all_foods.extend(api_response_data["foods"]) # 添加第一页的数据

            # 遍历剩余页面
            for page_num in range(2, total_pages + 1):
                print(f"Fetching page {page_num}/{total_pages}...")
                page_url = f"{base_url}&pageNumber={page_num}"
                page_response = session.get(page_url, timeout=10)
                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: {http_err.response.status_code}")
            return None
        except requests.exceptions.ConnectionError as conn_err:
            print(f"Connection error occurred: {conn_err}")
            return None
        except requests.exceptions.Timeout as timeout_err:
            print(f"Request timed out: {timeout_err}")
            return None
        except requests.exceptions.RequestException as req_err:
            print(f"An unexpected error occurred: {req_err}")
            return None
        except json.JSONDecodeError as json_err:
            print(f"Error decoding JSON response: {json_err}")
            print(f"Response content: {initial_response.text if 'initial_response' in locals() else 'N/A'}")
            return None
    return all_foods

# 替换为您的实际API密钥
API_KEY = "YOUR_USDA_API_KEY" 
food_items = get_all_usda_foods("raw", API_KEY)

if food_items:
    print(f"\nSuccessfully retrieved {len(food_items)} food items.")
    # 后续数据处理...
else:
    print("Failed to retrieve food items or no items found.")

# 假设food_items已经通过get_all_usda_foods函数获取
if food_items:
    table_data = []
    for food_item in food_items:
        row = {
            "Description": food_item.get("description", "N/A"),
            "FDC_ID": food_item.get("fdcId", "N/A") # 添加FDC ID以便追踪
        }

        # 提取营养成分
        for nutrient in food_item.get("foodNutrients", []):
            nutrient_name = nutrient.get("nutrientName")
            nutrient_value = nutrient.get("value")
            if nutrient_name and nutrient_value is not None:
                row[nutrient_name] = nutrient_value
        table_data.append(row)

    # 创建DataFrame
    df = pd.DataFrame(table_data)

    # 打印一些信息，例如数据框的形状和前几行
    print("\nDataFrame created:")
    print(f"Shape: {df.shape}")
    print("First 5 rows:")
    print(df.head())

    # 导出到Excel
    output_filename = 'usda_nutritional_facts_all.xlsx'
    try:
        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.")