如何在Python中传入Survey ID列表批量下载Qualtrics问卷？

错误原因 #

• 参数类型不匹配：你的get_qualtrics_survey函数定义时，survey_id参数预期接收单个字符串类型的问卷ID，但调用时传入了列表lst = ['1','2','3']，导致拼接请求体时生成无效格式（比如"surveyId":"['1','2','3']"），Qualtrics API无法识别该问卷ID。
• 进度检查逻辑漏洞：原代码中progress_status变量在循环内从未更新，即使导出进度完成，循环条件progress_status != "complete"始终为真，会导致程序无限循环，无法进入下载步骤。

修复方案 #

方案1：遍历列表逐个调用原函数（最小改动） #

保持原函数核心逻辑，调用时遍历问卷ID列表，逐个传入函数执行：

import requests
import zipfile
import io
import time

def get_qualtrics_survey(dir_save_survey, survey_id):
    # 用户参数配置
    api_token = "token"
    file_format = "csv"
    data_center = 'xx' # 格式："<Organization ID>.<Datacenter ID>"

    # 静态参数设置
    request_check_progress = 0
    progress_status = "in progress"
    base_url = "https://{0}.qualtrics.com/API/v3/responseexports/".format(data_center)
    headers = {
        "content-type": "application/json",
        "x-api-token": api_token,
    }

    # 步骤1：创建数据导出请求
    download_request_url = base_url
    download_request_payload = '{"format":"' + file_format + '","surveyId":"' + str(survey_id) + '"}'
    download_request_response = requests.request("POST", download_request_url, data=download_request_payload, headers=headers)
    # 新增请求失败处理
    if download_request_response.status_code != 200:
        print(f"创建导出请求失败，问卷ID: {survey_id}，错误信息: {download_request_response.text}")
        return
    progress_id = download_request_response.json()["result"]["id"]

    # 步骤2：检查导出进度，直到完成
    while request_check_progress < 100 and progress_status != "complete":
        request_check_url = base_url + progress_id
        request_check_response = requests.request("GET", request_check_url, headers=headers)
        if request_check_response.status_code != 200:
            print(f"检查进度失败，问卷ID: {survey_id}，错误信息: {request_check_response.text}")
            return
        request_check_progress = request_check_response.json()["result"]["percentComplete"]
        # 关键：更新进度状态
        progress_status = request_check_response.json()["result"]["status"]
        # 添加延迟避免API限流
        time.sleep(2)

    # 步骤3：下载文件
    request_download_url = base_url + progress_id + '/file'
    request_download = requests.request("GET", request_download_url, headers=headers, stream=True)
    if request_download.status_code != 200:
        print(f"下载文件失败，问卷ID: {survey_id}，错误信息: {request_download.text}")
        return

    # 步骤4：解压文件
    zipfile.ZipFile(io.BytesIO(request_download.content)).extractall(dir_save_survey)
    print(f'已下载问卷ID: {survey_id}')

# 问卷ID列表
lst = ['1','2','3']
# 保存路径
path = r"C:\Users\Downloads"
# 遍历列表逐个调用函数
for survey_id in lst:
    get_qualtrics_survey(dir_save_survey=path, survey_id=survey_id)

方案2：修改函数支持批量处理（更优雅） #

直接调整函数逻辑，使其支持接收单个ID或ID列表，内部自动遍历处理：

import requests
import zipfile
import io
import time
import json

def get_qualtrics_survey(dir_save_survey, survey_ids):
    # 统一参数格式：单个ID转成列表
    if isinstance(survey_ids, str):
        survey_ids = [survey_ids]
    
    # 用户参数配置
    api_token = "token"
    file_format = "csv"
    data_center = 'xx' # 格式："<Organization ID>.<Datacenter ID>"

    # 静态参数设置
    base_url = "https://{0}.qualtrics.com/API/v3/responseexports/".format(data_center)
    headers = {
        "content-type": "application/json",
        "x-api-token": api_token,
    }

    for survey_id in survey_ids:
        print(f"开始处理问卷ID: {survey_id}")
        # 步骤1：创建数据导出请求
        download_request_url = base_url
        # 使用json.dumps生成请求体，避免字符串拼接错误
        download_request_payload = json.dumps({"format": file_format, "surveyId": survey_id})
        download_request_response = requests.request("POST", download_request_url, data=download_request_payload, headers=headers)
        if download_request_response.status_code != 200:
            print(f"创建导出请求失败，问卷ID: {survey_id}，错误信息: {download_request_response.text}")
            continue
        progress_id = download_request_response.json()["result"]["id"]

        # 步骤2：检查导出进度
        request_check_progress = 0
        progress_status = "in progress"
        while request_check_progress < 100 and progress_status != "complete":
            request_check_url = base_url + progress_id
            request_check_response = requests.request("GET", request_check_url, headers=headers)
            if request_check_response.status_code != 200:
                print(f"检查进度失败，问卷ID: {survey_id}，错误信息: {request_check_response.text}")
                break
            request_check_progress = request_check_response.json()["result"]["percentComplete"]
            progress_status = request_check_response.json()["result"]["status"]
            time.sleep(2)
        
        if progress_status != "complete":
            print(f"问卷ID: {survey_id} 导出未完成，跳过下载")
            continue

        # 步骤3：下载文件
        request_download_url = base_url + progress_id + '/file'
        request_download = requests.request("GET", request_download_url, headers=headers, stream=True)
        if request_download.status_code != 200:
            print(f"下载文件失败，问卷ID: {survey_id}，错误信息: {request_download.text}")
            continue

        # 步骤4：解压文件
        zipfile.ZipFile(io.BytesIO(request_download.content)).extractall(dir_save_survey)
        print(f'已下载问卷ID: {survey_id}')

# 问卷ID列表
lst = ['1','2','3']
# 保存路径
path = r"C:\Users\Downloads"
# 调用批量处理函数
get_qualtrics_survey(dir_save_survey=path, survey_ids=lst)

额外优化建议 #

• 错误处理增强：加入API请求状态码检查，避免程序因网络问题、无效token或错误ID直接崩溃。
• 请求限流避免：在进度检查循环中添加time.sleep()，防止短时间内频繁请求触发Qualtrics API限流。
• JSON序列化替代拼接：使用json.dumps()生成请求体，避免手动拼接字符串导致的格式错误，尤其是当问卷ID包含特殊字符时。

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