如何使用脚本(Python、Node.js、cURL)将 CSV 或 Excel 联系人导入 Brevo
使用 import-contacts API 将 CSV 或 Excel 文件中的联系人批量导入 Brevo。包含可直接运行的 Python 和 Node.js 脚本、cURL 单行命令、错误处理,以及大于 10 MB 文件的提示。
如果你曾经需要把几千个联系人从电子表格推送到 Brevo,手动 UI 路径很快就会变得痛苦:选文件、映射列、选列表、等待、重复。一个脚本几秒就能完成同样的工作,而且更重要的是,你可以按计划运行它,可以在数据库导出后运行,也可以在你的 CRM 输出新 CSV 时运行。
本指南介绍 Brevo 为此提供的 API 端点,配有用 Python、Node.js 编写的完整可运行脚本,以及 cURL 单行命令。本指南中的所有内容都调用 POST /v3/contacts/import。
端点概览
POST https://api.brevo.com/v3/contacts/importContent-Type: application/jsonapi-key: YOUR_API_KEY
{ "listIds": [42], "updateExistingContacts": true, "emailBlacklist": false, "smsBlacklist": false}响应很快返回:
{ "processId": 78 }这是一个 202 Accepted,Brevo 已接受该导入并在后台处理。如果你想轮询完成情况或设置 notifyUrl webhook,processId 就是你的追踪句柄。
写代码前需要注意的几点:
- 请求体可以是 CSV(
fileBody)、JSON(jsonBody)或远程 URL(fileUrl)。 选一个。CSV 形式使用分号(;)来分隔列,而不是逗号,这是一个常见的坑。 fileBody和jsonBody上限为 10 MB。 Brevo 建议保持在 8 MB 左右以确保安全。对于任何更大的内容,将文件上传到 S3、GCS 或任意 HTTPS 主机,然后传递fileUrl。- 账户中不存在的自定义属性会被静默忽略。 在导入使用它们的行之前,先在 Brevo 的 UI 中(或通过 Attributes API)创建这些属性,否则数据就消失了。
updateExistingContacts默认为true。 如果你设为false,Brevo 会跳过邮箱已存在的联系人。对”仅添加新联系人”任务很有用。emptyContactsAttributes控制 CSV 中的空单元格是否清除现有值。默认false(空单元格被忽略)。如果你的 CSV 是真相之源,并希望空白单元格清除过期数据,请设为true。
获取 API 密钥
登录 Brevo → Settings → SMTP & API → API Keys → 创建新密钥。它看起来像 xkeysib-...。你将在每个请求中通过 api-key HTTP 头传递它。把它当作密码对待,它对你的账户拥有完全读写权限。
一个干净的模式是把它放进环境变量,这样它就永远不会进入你的源代码树。
export BREVO_API_KEY="xkeysib-..."Python:读取 CSV,推送到 Brevo
这是尽可能简单的脚本:用标准库读取本地 CSV,把请求体直接发送给 Brevo。除了 requests 之外不需要第三方 SDK。
import csvimport ioimport osimport sysimport requests
API_KEY = os.environ["BREVO_API_KEY"]LIST_ID = 42 # the Brevo list to add contacts toINPUT_FILE = sys.argv[1] if len(sys.argv) > 1 else "contacts.csv"
# Brevo wants semicolon-separated CSV. If your file uses commas (most do),# convert it on the fly so you don't have to edit the source spreadsheet.def to_brevo_csv(path: str) -> str: with open(path, newline="", encoding="utf-8") as f: reader = csv.reader(f) # default: comma-separated out = io.StringIO() writer = csv.writer(out, delimiter=";") for row in reader: writer.writerow(row) return out.getvalue()
body = { "fileBody": to_brevo_csv(INPUT_FILE), "listIds": [LIST_ID], "updateExistingContacts": True, "emptyContactsAttributes": False, "emailBlacklist": False, "smsBlacklist": False,}
resp = requests.post( "https://api.brevo.com/v3/contacts/import", json=body, headers={"api-key": API_KEY, "Content-Type": "application/json"}, timeout=60,)
if resp.status_code != 202: print(f"Import failed: {resp.status_code} {resp.text}") sys.exit(1)
process_id = resp.json()["processId"]print(f"Import accepted. Process ID: {process_id}")运行它:
python import_csv_to_brevo.py contacts.csvCSV 的第一行是表头。列名按大写、精确匹配映射到 Brevo 属性:EMAIL、FIRSTNAME、LASTNAME,加上你定义的任何自定义属性。EMAIL 是必需的。
EMAIL,FIRSTNAME,LASTNAME,COMPANY,CITY这就是整个流程。脚本一秒内返回;实际的导入在服务器端运行,根据数据量从几秒到几分钟不等。
Excel 文件:三条可行路径
Brevo 的导入端点不直接读取 .xlsx,所以根据工作需要在哪里完成,你有三种真正可用的选项:
- 在工作簿内运行 VBA 宏:从 Excel 本身一键同步,无需外部脚本。当文件位于桌面,并且你的团队希望在表格上有一个按钮时,这是正确答案。完整代码见 Excel 到 Brevo 的 VBA 宏指南。
- Office Scripts + Power Automate:如果工作簿在 OneDrive/SharePoint 中,并且你想要无人值守的定时同步。也在 Excel 指南 中有所介绍。
- 从脚本将
.xlsx转换为 CSV:本节其余部分涵盖的内容。如果你已经在服务器上运行 Python 或 Node,且只需每天拉一次工作簿,这是最佳选择。
对于选项 3,pandas 一行代码就能搞定:
# pip install pandas openpyxlimport pandas as pd
def xlsx_to_brevo_csv(path: str, sheet_name: str | int = 0) -> str: df = pd.read_excel(path, sheet_name=sheet_name) return df.to_csv(sep=";", index=False)
body["fileBody"] = xlsx_to_brevo_csv("contacts.xlsx")如果你不想把 pandas 作为依赖,单独使用 openpyxl 也可以:
from openpyxl import load_workbookimport csv, io
def xlsx_to_brevo_csv(path: str) -> str: wb = load_workbook(path, read_only=True) ws = wb.active out = io.StringIO() writer = csv.writer(out, delimiter=";") for row in ws.iter_rows(values_only=True): writer.writerow(["" if v is None else v for v in row]) return out.getvalue()Node.js:同样的工作,官方 SDK
Brevo 为 Node 发布了 @getbrevo/brevo。它处理认证、重试和有类型的请求结构:
// npm install @getbrevo/brevoimport fs from 'node:fs/promises';import { BrevoClient } from '@getbrevo/brevo';
const API_KEY = process.env.BREVO_API_KEY;const LIST_ID = 42;const INPUT_FILE = process.argv[2] ?? 'contacts.csv';
const client = new BrevoClient({ apiKey: API_KEY });
// Read the CSV and convert commas to semicolons if neededconst raw = await fs.readFile(INPUT_FILE, 'utf-8');const csv = raw.includes(';') ? raw : raw.replace(/,/g, ';');
const result = await client.contacts.importContacts({ fileBody: csv, listIds: [LIST_ID], updateExistingContacts: true, emptyContactsAttributes: false,});
console.log(`Import accepted. Process ID: ${result.processId}`);或者,如果你不想用 SDK,普通的 fetch 也以同样方式工作:
const resp = await fetch('https://api.brevo.com/v3/contacts/import', { method: 'POST', headers: { 'api-key': process.env.BREVO_API_KEY, 'Content-Type': 'application/json', }, body: JSON.stringify({ fileBody: csv, listIds: [42], updateExistingContacts: true, }),});
if (resp.status !== 202) { throw new Error(`Import failed: ${resp.status} ${await resp.text()}`);}
const { processId } = await resp.json();console.log(`Process ID: ${processId}`);cURL:用于即兴导入的单行命令
当你只想测试端点或手动塞入一个小文件时:
# Convert commas to semicolons, then send the file inlinecsv=$(sed 's/,/;/g' contacts.csv | jq -Rs .)
curl -X POST https://api.brevo.com/v3/contacts/import \ -H "api-key: $BREVO_API_KEY" \ -H "Content-Type: application/json" \ -d "{\"fileBody\": $csv, \"listIds\": [42], \"updateExistingContacts\": true}"jq -Rs . 会吞下整个文件并对其进行 JSON 转义,让你免于手动转义换行和引号。
大于 10 MB 的文件:使用 fileUrl
fileBody 上限为 10 MB。如果你的联系人转储更大,请将文件托管在 Brevo 可获取的 URL 上并传递它:
body = { "fileUrl": "https://files.example.com/contacts/2026-04-30.csv", "listIds": [42], "updateExistingContacts": True,}任何可通过公共 HTTPS URL 访问的内容都可以:预签名 S3 URL、GCS、你自己的静态主机,甚至用于一次性导入的 GitHub raw URL。Brevo 从你的 URL 获取文件,然后运行导入。接受的格式:.csv、.txt、.json。
发送 JSON 而不是 CSV
如果你已经在从数据库拉取联系人,就不需要绕路通过 CSV,直接以 JSON 发送:
body = { "jsonBody": [ { "attributes": { "FIRSTNAME": "Jane", "LASTNAME": "Doe", "COMPANY": "Acme", }, }, { "attributes": { "FIRSTNAME": "John", "LASTNAME": "Smith", "COMPANY": "Globex", }, }, ], "listIds": [42],}同样的 10 MB 上限。同样的异步行为。
轮询完成状态
导入是异步的,processId 是你追踪它的句柄。有一个独立的端点用于检查进程状态:
def wait_for_import(process_id: int, timeout_s: int = 600) -> dict: import time deadline = time.time() + timeout_s while time.time() < deadline: r = requests.get( f"https://api.brevo.com/v3/processes/{process_id}", headers={"api-key": API_KEY}, timeout=30, ) r.raise_for_status() status = r.json() if status["status"] in ("completed", "failed"): return status time.sleep(5) raise TimeoutError(f"Import {process_id} did not finish in {timeout_s}s")
result = wait_for_import(process_id)print(f"Import {result['status']}: {result}")对于长时间运行的任务,更干净的模式是 notifyUrl:传递一个 HTTPS 端点,Brevo 在导入完成时会向其 POST,从而省去轮询。
body["notifyUrl"] = "https://your-app.example.com/webhooks/brevo-import"常见错误及修复方法
没有明显原因的 400 Bad Request:几乎总是 CSV 中分号与逗号的问题。Brevo 的导入期望 ;,而不是 ,。在转换步骤之后再次检查 fileBody。
自定义属性值消失:该属性在你的账户中不存在。在导入之前,到 Contacts → Settings → Contact attributes 创建它,或者使用 Attributes API 把创建步骤作为脚本的一部分。
401 Unauthorized:错误的 header 名称。是 api-key(小写、连字符),不是 Authorization 或 X-API-Key。
导入”成功”但联系人没有出现在列表中:检查 listIds 是否包含正确的列表。另外:如果 updateExistingContacts 是 false 且联系人已存在,Brevo 会静默跳过它们,而不是把它们重新加入列表。
有些行导入了,有些没有:导入完成后,Brevo 会给你发邮件,附上每行的错误报告(除非你设置了 disableNotification: true)。报告会告诉你哪些行有错误的邮箱、缺失的必填字段或格式问题。
何时使用脚本,何时使用 UI
UI 适合一千行以下的一次性导入。一旦满足以下任一条件,脚本就胜出:
- 你需要多次导入(例如,从 CRM 每周导出)
- 源数据在落入 Brevo 之前需要清洗(去重、格式化电话号码、把全名拆成名和姓)
- 你希望它按计划运行,不需要任何人点按钮
- 文件比 UI 的舒适区更大
把上面的脚本包装在 cron 任务或 GitHub Action 中,你就有了自动联系人同步。本系列的下一篇文章展示如何直接从 Google Sheet 使用 Apps Script 做同样的事,无需服务器。