深入理解与实现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.")