如何用Python将带自定义字段的用户数据推送至Campaign Monitor

前置步骤：在Campaign Monitor后台创建自定义字段 #

你需要先在目标订阅列表中手动添加对应自定义字段，API无法自动创建：

• 登录Campaign Monitor后台，进入你的订阅列表
• 找到「自定义字段」选项，添加total_bookings（建议选数字类型）和first_booking（建议选日期类型），字段名要和后续API中使用的完全一致（大小写敏感）

步骤1：处理Redshift返回的数据 #

你的creator_details是元组列表，先转成字典列表方便后续构造请求：

# 定义字段名顺序（需与SQL查询的select顺序完全匹配）
field_names = ["name", "email", "total_bookings", "first_booking"]
# 将元组列表转换为字典列表
subscribers = []
for row in creator_details:
    subscriber_dict = dict(zip(field_names, row))
    subscribers.append(subscriber_dict)

步骤2：用Python调用Campaign Monitor API推送数据 #

以下提供两种实现方式，按需选择：

方式1：使用官方SDK #

1. 先安装官方SDK：

pip install campaignmonitor

2. 编写推送代码：

import campaignmonitor

# 初始化客户端，替换为你的API密钥和目标列表ID
client = campaignmonitor.Client(api_key="YOUR_API_KEY")
target_list_id = "YOUR_TARGET_LIST_ID"

# 批量推送订阅者数据
for sub in subscribers:
    # 构造自定义字段数组
    custom_fields = [
        {"Key": "total_bookings", "Value": sub["total_bookings"]},
        {"Key": "first_booking", "Value": sub["first_booking"].strftime("%Y-%m-%d") if sub["first_booking"] else None}
    ]
    # 发送请求
    response = client.subscribers.add(
        list_id=target_list_id,
        email_address=sub["email"],
        name=sub["name"],
        custom_fields=custom_fields,
        resubscribe=True  # 允许已订阅用户更新数据，设为False则拒绝重复订阅
    )
    # 处理响应结果
    if response["Code"] == 201:
        print(f"订阅者 {sub['email']} 推送成功")
    else:
        print(f"订阅者 {sub['email']} 推送失败：{response['Message']}")

方式2：直接使用requests库 #

若不想安装SDK，可直接发送HTTP请求：

import requests
import base64

API_KEY = "YOUR_API_KEY"
TARGET_LIST_ID = "YOUR_TARGET_LIST_ID"
API_URL = f"https://api.createsend.com/api/v3.2/subscribers/{TARGET_LIST_ID}.json"

# 构造认证头（API密钥需按 "api_key:" 格式Base64编码）
auth_str = base64.b64encode(f"{API_KEY}:".encode()).decode()
headers = {
    "Content-Type": "application/json",
    "Authorization": f"Basic {auth_str}"
}

# 批量推送数据
for sub in subscribers:
    payload = {
        "EmailAddress": sub["email"],
        "Name": sub["name"],
        "CustomFields": [
            {"Key": "total_bookings", "Value": str(sub["total_bookings"])},
            {"Key": "first_booking", "Value": sub["first_booking"].strftime("%Y-%m-%d") if sub["first_booking"] else ""}
        ],
        "Resubscribe": True
    }
    response = requests.post(API_URL, json=payload, headers=headers)
    # 处理响应
    if response.status_code == 201:
        print(f"订阅者 {sub['email']} 推送成功")
    else:
        error_info = response.json()
        print(f"订阅者 {sub['email']} 推送失败：{error_info['Message']}")

关键注意事项 #

• 自定义字段的Key必须与后台创建的字段名完全匹配，大小写敏感
• 日期类型字段需转换为Campaign Monitor支持的格式（如YYYY-MM-DD）
• 若数据量较大，建议分批次推送，避免触发API速率限制
• Resubscribe=True会覆盖已有订阅者的数据，根据业务需求调整

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