基于Excel数据的Python API批量请求（Curl转Python）问题

核心需求 #

• 读取含ReceiptID、FirstName等字段的输入Excel
• 为每行数据构造符合API要求的JSON请求体
• 批量调用目标API并捕获返回结果
• 将原数据与API返回结果合并后，追加写入输出Excel

参考单次请求的正确核心代码 #

假设你的单次请求代码（可正常运行）类似如下：

import requests
import pandas as pd

# API配置
API_URL = "https://your-api-endpoint.com/query"
HEADERS = {
    "Content-Type": "application/json",
    "Authorization": "Bearer YOUR_AUTH_TOKEN"
}

# 构造单个请求体
single_payload = {
    "ReceiptID": "R00123",
    "FirstName": "Alice"
    # 补充API要求的其他字段
}

# 发送请求并解析结果
response = requests.post(API_URL, json=single_payload, headers=HEADERS)
response_data = response.json()

# 导出单次结果
result_df = pd.DataFrame([{**single_payload, **response_data}])
result_df.to_excel("single_result.xlsx", index=False)

批量代码常见失败原因及修正方案 #

批量代码失败通常源于请求体构造错误、未处理API异常、Excel读写逻辑错误这三类问题，以下是修正后的完整可运行代码：

import requests
import pandas as pd
from requests.exceptions import RequestException

# 全局配置
INPUT_EXCEL_PATH = "input_data.xlsx"
OUTPUT_EXCEL_PATH = "batch_results.xlsx"
API_URL = "https://your-api-endpoint.com/query"
API_HEADERS = {
    "Content-Type": "application/json",
    "Authorization": "Bearer YOUR_AUTH_TOKEN"  # 替换为你的实际授权信息
}
REQUIRED_FIELDS = ["ReceiptID", "FirstName"]  # API必填字段

def call_api(payload):
    """封装API调用，处理各类异常"""
    try:
        response = requests.post(API_URL, json=payload, headers=API_HEADERS, timeout=15)
        response.raise_for_status()  # 主动触发HTTP状态码错误（如400、500）
        return response.json()
    except RequestException as e:
        print(f"请求失败 | 数据: {payload} | 错误: {str(e)}")
        return {"api_error": str(e)}

def main():
    # 读取输入Excel并校验字段
    try:
        input_df = pd.read_excel(INPUT_EXCEL_PATH)
    except FileNotFoundError:
        print(f"错误：未找到输入文件 {INPUT_EXCEL_PATH}")
        return

    missing_fields = [f for f in REQUIRED_FIELDS if f not in input_df.columns]
    if missing_fields:
        print(f"错误：输入Excel缺少必填字段: {', '.join(missing_fields)}")
        return

    # 批量处理每行数据
    all_results = []
    for _, row in input_df.iterrows():
        # 严格按照API要求构造请求体（注意字段类型匹配，比如ReceiptID转字符串）
        payload = {
            "ReceiptID": str(row["ReceiptID"]),
            "FirstName": row["FirstName"].strip() if pd.notna(row["FirstName"]) else ""
            # 补充其他API需要的字段，如 row["LastName"] 等
        }
        # 调用API并合并结果
        api_res = call_api(payload)
        combined_data = {**row.to_dict(), **api_res}
        all_results.append(combined_data)

    # 写入输出Excel（支持追加模式）
    result_df = pd.DataFrame(all_results)
    try:
        # 如果输出文件已存在，追加数据
        existing_df = pd.read_excel(OUTPUT_EXCEL_PATH)
        final_df = pd.concat([existing_df, result_df], ignore_index=True)
    except FileNotFoundError:
        # 输出文件不存在则直接写入
        final_df = result_df

    final_df.to_excel(OUTPUT_EXCEL_PATH, index=False)
    print(f"批量处理完成，结果已写入 {OUTPUT_EXCEL_PATH}")

if __name__ == "__main__":
    main()

关键注意事项 #

• 字段类型匹配：确认API要求的字段类型（比如ReceiptID是字符串还是数字），必要时做类型转换
• 异常隔离：call_api函数单独处理请求异常，避免单个请求失败导致整个批量任务终止
• Excel追加逻辑：自动检测输出文件是否存在，避免覆盖原有数据
• 空值处理：对Excel中的空值做默认值填充（如空字符串），避免构造请求体时出错

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