如何用Python3 DataFrame生成Qualtrics API更新所需JSON payload
批量更新Qualtrics调查嵌入式数据的Python实现
核心思路
遍历你的DataFrame每条记录,为每个response_id生成符合Qualtrics API要求的JSON payload,通过循环逐个调用更新接口;如果Qualtrics支持批量更新接口,也可以一次性构造批量请求payload,提升效率。
单条循环更新代码示例
import pandas as pd import json import requests import time # 假设你已经加载好数据到df # df = pd.read_csv("your_source_data.csv") # 配置Qualtrics API参数 API_TOKEN = "你的Qualtrics API令牌" DATA_CENTER = "你的数据中心前缀(如ca1、eu等)" BASE_URL = f"https://{DATA_CENTER}.qualtrics.com/API/v3/responses" headers = { "X-API-TOKEN": API_TOKEN, "Content-Type": "application/json" } # 遍历每条记录更新 for _, row in df.iterrows(): response_id = row["response_id"] division_value = row["division"] # 构造单条更新的payload payload = { "responseId": response_id, "embeddedData": { "Division": division_value } } # 发送PUT请求更新 api_url = f"{BASE_URL}/{response_id}" try: response = requests.put(api_url, data=json.dumps(payload), headers=headers) response.raise_for_status() # 抛出HTTP错误 print(f"✅ 成功更新响应 {response_id}") except requests.exceptions.RequestException as e: print(f"❌ 更新响应 {response_id} 失败: {str(e)}") # 加入延迟避免触发速率限制(根据Qualtrics限制调整,比如0.5秒) time.sleep(0.5)
批量更新优化(如果支持)
Qualtrics的API提供批量更新响应的接口,相比单条循环更高效,适合6000条量级的数据:
# 构造批量更新payload bulk_payload = { "responses": [] } for _, row in df.iterrows(): bulk_payload["responses"].append({ "responseId": row["response_id"], "embeddedData": { "Division": row["division"] } }) # 发送批量更新请求 bulk_api_url = f"{BASE_URL}/bulk/update" try: response = requests.put(bulk_api_url, data=json.dumps(bulk_payload), headers=headers) response.raise_for_status() print("✅ 批量更新完成") # 可解析响应查看成功/失败的记录 result = response.json() print(f"成功更新: {len(result['result']['successes'])} 条,失败: {len(result['result']['failures'])} 条") except requests.exceptions.RequestException as e: print(f"❌ 批量更新失败: {str(e)}")
注意事项
- 替换
API_TOKEN和DATA_CENTER为你自己的Qualtrics账户信息 - 注意Qualtrics的API速率限制(一般单接口每分钟最多100次请求),单条循环时务必加入延迟
- 测试阶段先选取少量数据(如前10条)验证格式和接口正确性,再执行全量更新
- 可以添加日志记录到文件,方便后续排查失败的更新记录
内容的提问来源于stack exchange,提问作者dduartem
相关产品推荐
相关产品推荐

