Listmonk 订阅者批量导入 API 完整指南：CSV/ZIP 上传、状态轮询与源码级实现解析

Listmonk 订阅者批量导入 API 完整指南：CSV/ZIP 上传、状态轮询与源码级实现解析

【免费下载链接】listmonkHigh performance, self-hosted, newsletter and mailing list manager with a modern dashboard. Single binary app.项目地址: https://gitcode.com/GitHub_Trending/li/listmonk

导读

本文是 listmonk 中订阅者批量导入（Bulk Subscriber Import）API 的实战指南，覆盖/api/import/subscribers系列四个端点（上传、状态查询、日志查询、停止导入）的请求格式、参数语义与典型调用方式。在 listmonk 中，当需要将历史订阅者数据、第三方导出的 CSV 或跨系统迁移的用户列表一次性写入系统时，这套 API 是唯一面向自动化场景的官方入口。读完本文，你将掌握：如何构造 multipart 上传请求、理解params中每个 JSON 字段（含文档未展开的subscription_status、overwrite_userinfo、overwrite_subscription_status）的实际效果，以及导入器在后台如何处理 CSV、批量落库与状态反馈——并辅以仓库源码佐证每个结论。

一、端点总览与权限说明

listmonk 将导入功能封装为同一资源路径下的四个 REST 端点，全部位于 API 分组中，定义在 cmd/handlers.go：

从路由注册代码可见，四个端点统一挂在subscribers:import权限（permission）之下，均通过pm(...)中间件做认证与授权。这意味着调用方必须是具备该权限的用户，通常以-u "api_user:token"形式的 HTTP Basic Auth 通过认证（api_user为 listmonk 管理员用户名，token为其 API token）。

在系统内部，这四个端点分别对应 cmd/import.go 中的四个 Handler：

• ImportSubscribers（POST）：校验参数 → 保存上传文件 → 启动导入会话；
• GetImportSubscribers（GET）：返回a.importer.GetStats()；
• GetImportSubscriberStats（GET 日志）：返回a.importer.GetLogs()；
• StopImportSubscribers（DELETE）：调用a.importer.Stop()后返回当前统计。

二、导入文件格式准备：CSV 列、分隔符与 ZIP

在发起请求前，先理解 listmonk 期望的文件结构，这决定了导入能否成功。

CSV 列约定

导入器只识别三个表头列（定义于 internal/subimporter/importer.go 的csvHeaders映射）：

列顺序不固定。源码中的mapCSVHeaders会按表头名称动态建立“列名 → 列索引”的映射（importer.go），因此email,name,attributes与attributes,name,email都能正确解析。表头中的未知列会被忽略并写入日志（ignoring unknown header），不会导致导入失败。

一个可用的示例文件（该示例同时出现在管理后台导入页 frontend/src/views/Import.vue 中）：

email,name,attributes user1@mail.com,"User One","{""age"": 42, ""planet"": ""Mars""}" user2@mail.com,"User Two","{""age"": 24, ""job"": ""Time Traveller""}"

注意attributes列中的 JSON 使用双引号包裹转义后的 JSON 文本（CSV 中""表示一个字面双引号）。

分隔符（delimiter）

delim参数指定 CSV 文件使用的单字符分隔符，常见为,。源码在 cmd/import.go 中强制校验len(opt.Delim) != 1即返回import.invalidDelim错误，所以制表符（\t）、分号（;）等单字符分隔符均可使用，但必须是单个字符。

ZIP 压缩包支持

POST 端点接受普通.csv文件，也接受将 CSV 打包后的 ZIP 文件。上传处理逻辑（cmd/import.go）：

• 文件名以.csv结尾（不区分大小写）→ 直接按 CSV 解析；
• 其他文件名 → 视为 ZIP，调用sess.ExtractZIP(out.Name(), 1)解压。

ZIP 处理有几个值得注意的规则（实现于 importer.go）：

• 只提取包内第一个.csv文件参与导入，其余 CSV 文件会被忽略——因此若有多个 CSV，官方建议先自行拼接为一个 CSV 再上传；
• 压缩包内的目录条目与非.csv文件会被跳过并记录日志；
• 文件名经filepath.Base清洗，防止 ZIP slip 路径穿越（安全考虑）；
• 若包内没有任何 CSV 文件，报错no CSV files found in the ZIP并使导入进入failed状态。

三、POST /api/import/subscribers：启动批量导入

这是整个导入流程的入口。请求必须使用multipart/form-data表单，包含两个字段：

3.1params完整参数表

原文档给出的参数为mode、delim、lists、overwrite。结合 internal/subimporter/importer.go 中SessionOpt结构的json标签，完整的参数集如下：

overwrite与两个细粒度字段的关系在NewSession中有明确说明（importer.go）：为了 API 向后兼容，只要设置了旧的overwrite: true，overwrite_userinfo与overwrite_subscription_status会被同时置为true。

3.2 参数校验与默认值（服务端行为）

从 cmd/import.go 可以看到 POST 处理器的完整校验链，调用方值得了解这些约束以避免无谓的 400 错误：

1. 并发限制：若已有导入正在运行（状态为importing），直接返回400 import.alreadyRunning——同一时刻只允许一个导入会话；
2. 列表权限过滤：opt.ListIDs = user.FilterListsByPerm(auth.PermTypeManage, opt.ListIDs)，当前用户无权管理的列表 ID 会被过滤掉；若过滤后为空且模式不是blocklist，返回403权限拒绝；
3. mode 校验：只允许subscribe或blocklist，否则返回400 import.invalidMode；
4. subscription_status 默认值（cmd/import.go）：
   • subscribe模式默认unconfirmed；
   • blocklist模式默认unsubscribed；
   • 显式传入的值必须是unconfirmed/confirmed/unsubscribed三者之一，否则报import.invalidSubStatus；
5. delim 长度必须为 1；
6. file 字段必须存在且可打开，否则返回import.invalidFile。

上传的文件会被复制到系统临时目录（os.CreateTemp("", "listmonk")），随后新建导入会话并异步启动处理：

go sess.Start() go sess.LoadCSV(out.Name(), rune(opt.Delim[0]))

因此 POST 接口会立即返回，真正的导入在后台 goroutine 中执行，需要配合状态查询接口跟踪进度。

3.3 标准调用示例

curl -u "api_user:token" -X POST 'http://localhost:9000/api/import/subscribers' \ -F 'params={"mode":"subscribe", "subscription_status":"confirmed", "delim":",", "lists":[1, 2], "overwrite": true}' \ -F "file=@/path/to/subs.csv"

响应示例（返回当前统计快照，data字段结构与 GET 状态接口一致）：

{ "data": { "name": "subs.csv", "total": 500, "imported": 0, "status": "importing" } }

说明：name为上传文件的原始文件名（由opt.Filename = file.Filename记录）；total为文件总数据行数（不含表头）；imported为已成功落库的记录数；status见下文状态机。

四、GET /api/import/subscribers：查询导入状态与进度

4.1 请求与响应

curl -u "api_user:token" -X GET 'http://localhost:9000/api/import/subscribers'

{ "data": { "name": "", "total": 0, "imported": 0, "status": "none" } }

当没有进行过任何导入时，返回name为空、status为none的初始状态（Importer初始化时的默认状态，见 importer.go）。

4.2 状态机与统计字段语义

status字段的全部取值定义在 importer.go：

统计字段的精确语义（LoadCSV 与 countLines）：

• total：通过统计换行符得到的文件总行数再减 1（排除表头行），用作前端进度条的基数；
• imported：已成功写入数据库的记录数，由incrementImportCount在每次批量提交时累加；
• 由于total是“行数”而非“有效记录数”，被跳过的非法行会导致最终imported略小于total，这是正常现象（详见下文错误处理章节）。

五、GET /api/import/subscribers/logs：查看导入日志

5.1 请求与响应

curl -u "api_user:token" -X GET 'http://localhost:9000/api/import/subscribers/logs'

{ "data": "2020/04/08 21:55:20 processing 'import.csv'\n2020/04/08 21:55:21 imported finished\n" }

5.2 日志内容说明

日志由Session中的log.Logger写入（importer.go），带日期时间与源码位置前缀。管理后台前端在展示时会用正则去掉importer.go:<行号>:前缀（见 Import.vue 的getLogs方法）。

典型的日志事件包括：

• processing '<文件名>'：会话开始；
• skipping line N. ...：某行记录因校验失败或列数不匹配被跳过（原因各不相同）；
• imported <累计数>：每完成一个批量提交输出一次；
• imported finished：全部处理完成；
• stop request received：收到停止信号；
• extracting .../skipping non .csv file ...：ZIP 解压过程的日志。

日志缓冲区在会话开始时重建（NewSession中logBuf: bytes.NewBuffer(nil)），因此该接口返回的是最近一次导入会话的日志。

六、DELETE /api/import/subscribers：停止或清除导入

6.1 请求与响应

curl -u "api_user:token" -X DELETE 'http://localhost:9000/api/import/subscribers'

{ "data": { "name": "", "total": 0, "imported": 0, "status": "none" } }

6.2 双重语义

从StopImportSubscribers的注释与 Importer.Stop 的实现看，该端点有两个行为分支：

• 有导入在运行（importing）：向stop通道发送信号，状态置为stopping。CSV 读取循环在每次迭代时通过select检查该信号（importer.go），收到后关闭队列并停止读取，剩余未处理行不再导入；
• 没有导入在运行：直接重置为none状态，相当于“清除”上一次导入的残留状态，使系统可以开启新导入。

前端行为与之一致：导入完成后按钮文案变为“Done”，点击即调用该接口清除状态（Import.vue 的stopImport方法）。

七、源码级原理：导入会话如何把 CSV 写入数据库

7.1 单例 Importer 与会话模型

Importer被设计为有状态单例（包注释明确说明：每个 Importer 实例同时只允许一个导入，见 importer.go）。初始化发生在 cmd/init.go 的initImporter，它将三张预编译 SQL 语句注入导入器：

• UpsertStmt：来自 queries/subscribers.sql 的upsert-subscriber；
• BlocklistStmt：来自 queries/subscribers.sql 的upsert-blocklist-subscriber；
• UpdateListDateStmt：update-lists-date，导入完成后刷新相关列表的更新时间。

同时注入privacy.domain_blocklist与privacy.domain_allowlist配置（在config.toml.sample中可查看对应配置项），用于导入时的邮箱域名过滤，并通过PostCB回调在导入结束后刷新物化视图统计并向管理员发送通知。

每次 POST 上传会创建一次Session，核心结构是一个容量为commitBatchSize的 channel 队列（subQueue）：

• LoadCSV协程负责解析、清洗、校验每一行，将SubReq压入队列（importer.go）；
• Start协程从队列取记录，攒够commitBatchSize（10000 条，定义于 importer.go）后在一个数据库事务中批量提交（importer.go）。

7.2 两种模式的 SQL 差异

• subscribe 模式执行upsert-subscriber：以email为冲突键做 UPSERT。$7（overwrite_userinfo）控制是否覆盖已存在记录的 name/attributes，$8（overwrite_subscription_status）控制是否覆盖subscriber_lists中订阅状态；插入记录统一为enabled状态（订阅状态由列表关联表表达）。ON CONFLICT (email)意味着同一文件内或与库中重复的邮箱只会产生一条记录；
• blocklist 模式执行upsert-blocklist-subscriber：将订阅者（含已存在者）状态置为blocklisted，并把其所有现有订阅标记为unsubscribed——这正对应“从所有列表中移除并拉黑”的语义。

7.3 行级校验与清洗

每一行在入队前经过ValidateFields/SanitizeEmail（importer.go）：

• 邮箱先经utils.SanitizeEmail校验并归一化，长度超过 1000 视为非法；
• 校验通过后统一转为小写；
• 若命中配置的域名黑名单（或不在白名单内），该行被拒绝；黑/白名单支持*.example.com形式的通配子域名（makeDomainMap，importer.go）；
• 姓名为空时，从邮箱用户名部分派生（按.与空格分词并做 Title Case），如john.doe@example.com→John Doe；
• attributes列内容按 JSON 解析，解析失败仅记录日志并跳过该属性，不中断导入（importer.go）。

7.4 进度与完成回调

每提交一个 10000 条批次，imported计数增加一次；队列关闭后若仍有残留记录则提交最后一个事务。随后：

1. 状态置为finished或failed；
2. 执行UpdateListDateStmt更新目标列表时间戳；
3. 触发PostCB（init.go）：刷新物化视图（订阅者计数、统计）并发送管理员通知，通知模板可在 static/email-templates/import-status.html 中查看。

八、错误处理与失败行跳过策略

导入器对“脏数据”采取跳过并继续的策略，而非整体失败。常见场景及对应日志：

因此调用方应以“日志接口中的跳过提示”作为数据质量排查依据，而imported / total的差值即为被跳过的行数。

九、与前端 / 官方客户端的联动参考

管理后台的导入页面（frontend/src/views/Import.vue）是这套 API 最直接的参考实现，展示了推荐的调用时序：

1. 页面加载即调用 GET 状态接口，none时展示上传表单，否则展示进度条；
2. 上传成功后以250ms 间隔轮询状态接口，直到状态离开importing/stopping；
3. 每次轮询同时拉取日志接口并滚动到底部；
4. 进度条 =Math.ceil((imported / total) * 100)；
5. 完成或失败后，点击按钮调用 DELETE 清除状态。

前端提交的params与本文 3.1 节参数表一一对应（见 Import.vue 的onSubmit方法），其中lists传的是所选列表的 ID 数组。若需在自有脚本中复刻这一流程，直接照此顺序调用四个端点即可。

十、实战要点速查

• 认证：所有端点都需要subscribers:import权限，使用 Basic Auth；
• 并发：同一时刻仅允许一个导入会话，重复 POST 会收到 400；
• 文件格式：CSV 表头只需email（必填）、name、attributes，列序任意；支持单 CSV 或含单个 CSV 的 ZIP；
• subscribe模式必须指定lists，否则 403；blocklist模式无需列表；
• 覆盖语义：默认不覆盖已存在订阅者的资料与状态；overwrite: true等价于两个细粒度覆盖开关同时开启；
• 进度判断：total是“总行数减表头”，非法行会被跳过，最终imported可能小于total，请结合日志接口确认原因；
• 状态清理：导入结束后调用 DELETE 将状态重置为none，才能开启下一次导入。

【免费下载链接】listmonkHigh performance, self-hosted, newsletter and mailing list manager with a modern dashboard. Single binary app.项目地址: https://gitcode.com/GitHub_Trending/li/listmonk