API响应始终判定无效，如何修正call_api函数验证逻辑？

问题背景 #

我通过call_api函数检查API响应JSON的Response_Description字段是否包含字符串"Hazard was successfully processed and at least one value is returned."来验证API有效性，但所有测试的Pass列都是N（对应返回false），怀疑call_api函数存在问题，需要修正代码实现正确验证。

核心问题排查与修正步骤 #

1. 先排查API实际返回内容 #

当前代码无任何日志输出，无法确定API返回的JSON结构、字段是否存在，或返回的描述文本是否与预期一致。需在call_api中添加调试输出，明确实际响应内容：

• 打印响应状态码，确认请求是否成功
• 打印响应文本，排查是否为合法JSON，或字段层级是否错误
• 打印解析后的JSON结构，确认Response_Description的位置

2. 修正验证逻辑与错误处理 #

• 先检查HTTP响应状态码（如200），避免解析错误响应
• 确保匹配的字符串完全符合实际返回内容（注意大小写、空格、标点）
• 处理JSON解析失败的情况（如响应不是JSON格式）

3. 修复计时逻辑错误 #

原代码的计时范围未包含等待API响应的时间，导致统计的Time Taken不准确，需将计时逻辑调整为覆盖从提交请求到获取所有结果的完整过程。

修改后的完整代码 #

import requests
import pandas as pd
import concurrent.futures
import time

def call_api(url):
    try:
        headers = {
            "User-Agent": "Mozilla/5.0 (X11; CrOS x86_64 12871.102.0) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/81.0.4044.141 Safari/537.36"
        }
        response = requests.get(url, headers=headers)
        # 打印调试信息，排查响应情况
        print(f"URL: {url}")
        print(f"Status Code: {response.status_code}")
        print(f"Response Text: {response.text[:500]}")  # 打印前500字符避免过长
        
        # 先检查HTTP状态码是否成功
        if response.status_code != 200:
            print(f"Request failed with status code {response.status_code}")
            return False
        
        json_response = response.json()
        print(f"Parsed JSON keys: {list(json_response.keys())}")  # 打印JSON顶层键
        
        # 检查字段是否存在并匹配目标字符串
        target_str = "Hazard was successfully processed and at least one value is returned."
        if (
            "Response_Description" in json_response
            and target_str in json_response["Response_Description"]
        ):
            print("API validation passed")
            return True
        else:
            # 打印实际的描述文本
            if "Response_Description" in json_response:
                print(f"Actual Response_Description: {json_response['Response_Description']}")
            else:
                print("Response_Description field not found in JSON")
            return False
    except requests.exceptions.RequestException as e:
        print(f"Request error: {str(e)}")
        return False
    except ValueError as e:
        print(f"JSON parsing error: {str(e)}")
        return False

df = pd.read_csv(r"C:\Users\Jose.Moquaimbo\Bulk Calling APIs\dataset.csv")

# 迭代次数
num_iterations = 5

# 创建空DataFrame存储结果
results_df = pd.DataFrame(columns=["Iteration", "Pass", "Time Taken"])

# 跟踪最小、最大、总耗时的变量
min_time = float("inf")
max_time = float("-inf")
total_time = 0

def process_iteration(iteration):
    # 从DataFrame中随机抽取URL
    random_urls = df["url"].sample(n=1).tolist()
    
    # 修正计时逻辑：从提交请求前开始计时
    start_time = time.time()
    # 使用ThreadPoolExecutor并发执行API调用
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        futures = [executor.submit(call_api, url) for url in random_urls]
        # 等待所有任务完成并获取结果
        results = [future.result() for future in futures]
    
    # 获取所有结果后停止计时
    end_time = time.time()
    
    # 计算本次迭代耗时
    iteration_time = end_time - start_time
    
    # 更新最小、最大、总耗时
    global min_time, max_time, total_time
    min_time = min(min_time, iteration_time)
    max_time = max(max_time, iteration_time)
    total_time += iteration_time
    
    # 检查本次迭代中所有API调用是否成功
    passed = "Y" if all(results) else "N"
    
    # 将本次迭代结果添加到DataFrame
    results_df.loc[iteration] = [iteration, passed, iteration_time]

# 执行迭代
for i in range(1, num_iterations + 1):
    process_iteration(i)

# 计算平均耗时
avg_time = total_time / num_iterations

# 打印结果DataFrame
print("\n=== 结果汇总 ===")
print(results_df)

# 统计信息
print("\n=== 统计数据 ===")
print("最短耗时:", min_time)
print("最长耗时:", max_time)
print("平均每次迭代耗时:", avg_time)
print("Y表示响应无错误，N表示响应无效")

关键修改说明 #

• 添加详细调试打印，可直接查看每个请求的状态码、响应内容、JSON结构，快速定位问题（比如字段是否在嵌套层级里，或返回的描述文本与预期不一致）
• 增加HTTP状态码检查，避免解析非成功响应的JSON
• 修复计时逻辑，现在统计的是从提交请求到获取所有结果的完整耗时
• 处理JSON解析失败的异常，避免程序崩溃

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