crm-sync

/digital-marketing-pro:crm-sync

脚本位置。 如果你的宿主未设置 ${CLAUDE_PLUGIN_ROOT}，脚本位于此插件的 scripts/ 文件夹中，与 skills/ 文件夹并列。请使用 Python 3.10 或更高版本运行：python；如果找不到 python，则使用 python3（macOS 和许多 Linux 系统）。

用途

将营销数据同步到品牌已连接的 CRM 平台，或从该平台同步数据。支持创建联系人、更新交易和关联营销活动，并具备自动字段映射、去重和合规检查功能。支持 Salesforce、HubSpot、Zoho 和 Pipedrive，可双向同步，确保营销团队和销售团队基于同一可信数据源开展工作，无需手动录入或反复导入/导出数据。适用于一次性批量同步和定期自动传输，并提供完整的审计跟踪和回滚功能，满足企业级数据治理要求。

当你需要去重、合规验证或审计日志时，请使用此命令，而不是手动导入 CSV。需要导入带有评分的潜在客户时，请改用 /digital-marketing-pro:lead-import。

执行门禁（强制 — 不可跳过）

1. 在操作任何线上系统前，先完整展示预览 — 接收者 / 支出 / 变更 / 合规情况 — 作为执行摘要。
2. 用户必须输入 yes（或明确表示同意的等效内容）。任何其他输入 — 含糊、暗示性、部分同意或未回复 — 都会取消本次运行。
3. 不得在用户输入含糊时继续。执行失败后不得自动重试；再次运行前必须由人工审核。
4. 只有在用户输入 yes 后，才记录该同意：python "${CLAUDE_PLUGIN_ROOT}/scripts/approval-manager.py" --brand {slug} --action create-approval --data '{"type":"crm-sync","platform":"<platform>","content_summary":"<one line from the Execution Summary>","risk_level":"<tier>"}'，然后运行 python "${CLAUDE_PLUGIN_ROOT}/scripts/approval-manager.py" --brand {slug} --action approve --id {approval_id}。如果任一命令报错，请停止并报告错误；不得使用其他工具绕过。如果该记录能够证明此操作已执行同意步骤，但无法证明是谁输入了 yes。
5. 执行操作。通过 connector_executor.py 发送的写入操作需要 --approval-id，且仅会针对匹配、未使用且未过期的记录执行（参见 /digital-marketing-pro:execute-action）。通过已连接的 MCP 服务工具执行的写入操作不受此代码检查约束：它依赖用户输入的 yes 和宿主的权限提示。操作完成后，运行 python "${CLAUDE_PLUGIN_ROOT}/scripts/approval-manager.py" --brand {slug} --action mark-executed --id {approval_id} --data '{"execution_result":"success"}'（或 failure）。

所需输入

用户必须提供（或系统将提示用户提供）：

• 同步类型：要同步的内容——联系人、交易、营销活动，或在单次操作中组合多种对象类型
• 数据源：数据的来源——CSV 文件路径、JSON 数组、手动输入，或来自其他已连接 MCP 的数据（例如 Google Sheets、电子邮件平台、广告平台）
• 目标 CRM 平台：Salesforce、HubSpot、Zoho 或 Pipedrive——以及适用时的具体对象类型（例如 Salesforce Leads 与 Contacts、HubSpot Contacts 与 Companies）
• 同步方向：单向推送（营销平台到 CRM）、单向拉取（CRM 到营销平台），或双向合并，并指定冲突解决优先级（以源数据为准、以 CRM 为准，或以最新数据为准）
• 字段映射覆盖项（可选）：默认字段映射与 CRM 架构不匹配时使用的自定义映射——例如，“company_name”映射到 Salesforce 中的“Account Name”，或“phone_mobile”映射到“MobilePhone”
• 去重策略：按电子邮件、电话、CRM ID 或复合键匹配——以及是更新现有记录、跳过重复项，还是合并两边来源的字段
• 同意与合规要求：同步前是否必须验证 GDPR 同意字段、选择加入状态或数据处理依据——以及适用哪些市场的法规
• 记录筛选条件（可选）：限制同步记录的条件——日期范围、细分群体成员身份、来源营销活动、生命周期状态或自定义字段值
• 批次大小偏好（可选）：每个 API 批次包含的记录数——较小的批次便于安全回滚，较大的批次则适合高数据量同步，速度更快
• 通知偏好：新联系人或交易同步到销售代表的销售管道时，是否向其发送提醒，以及通过何种渠道发送（CRM 通知、电子邮件、Slack）
• 关联关系链接：是否将同步的联系人关联到 CRM 中的现有客户账户、交易或营销活动——以及用于匹配的查询键（域名、客户账户名称、交易 ID）
• 标签或列表分配（可选）：要应用于同步记录的标签、列表或营销活动成员身份，以便进行后续细分和报告归因
• 错误处理偏好：单条记录失败时的处理方式——跳过并继续（记录失败项以便稍后处理）、停止整个同步（回滚已完成的记录），或隔离失败记录以供人工审核
• 同步计划（可选）：这是一次性同步，还是应保存为定期同步模板——如果是定期同步，请指定频率（每小时、每天、每周），以及是仅同步自上次运行后修改的记录（增量同步），还是所有符合条件的记录（全量同步）

流程

1. 加载品牌上下文：读取 ~/.claude-marketing/brands/_active-brand.json 获取当前启用的 slug，然后加载 ~/.claude-marketing/brands/{slug}/profile.json。应用品牌语气、目标市场的合规规则（skills/context-engine/compliance-rules.md）和行业背景。另请检查是否有指南：查看 ~/.claude-marketing/brands/{slug}/guidelines/_manifest.json；如果存在，加载其中的限制和相关类别文件。检查 ~/.claude-marketing/sops/ 中是否有代理机构的 SOP。如果不存在品牌，请询问：“先设置品牌？(/digital-marketing-pro:brand-setup)”——或者使用默认设置继续。
2. 检查已连接 CRM 的状态：运行 python "${CLAUDE_PLUGIN_ROOT}/scripts/crm-sync.py" --brand {slug} --action get-crm-status，验证平台连接、API 凭据、速率限制余量和可用对象。如果未连接 CRM，请引导用户配置其 CRM MCP 集成，并提供对应平台的设置说明。
3. 验证输入数据：解析源数据并验证必填字段——检查电子邮件格式（RFC 5322）、电话号码规范化格式（E.164）、必填字段是否存在、数据类型是否一致，以及字符编码。标记无效记录并说明具体错误原因，将其与有效记录分开。
4. 将字段映射到 CRM 架构：参考 skills/context-engine/crm-integration-guide.md 中各平台的标准字段映射。自动映射匹配的字段名称，应用用户覆盖设置，识别未映射的源字段，并标记没有源字段映射的必填 CRM 字段。继续之前，请提供完整的字段映射表供用户确认。
5. 检查重复记录：运行 python "${CLAUDE_PLUGIN_ROOT}/scripts/crm-sync.py" --brand {slug} --action check-dedup --data '{"email":"[email protected]","phone":"...","name":"...","company":"..."}'——此脚本每次调用接收一个联系人字典，其中 email 为必填项（phone、name、company 为可选项）；批量检查时，逐条遍历记录。识别完全匹配、模糊匹配（根据姓名和域名计算 Levenshtein 距离）以及真正的新记录。提供去重结果，并为每条记录推荐操作（创建、更新、跳过、合并）。
6. 验证合规要求：根据品牌目标市场的合规规则检查同意字段、选择加入状态和数据处理依据。标记缺少必要同意或违反数据保留政策的记录。对于 GDPR 市场，验证每条记录都已记录合法依据。
7. 准备适用于 CRM 的负载数据：将已验证并去重的记录转换为特定平台的 API 负载，确保字段名称、数据类型、选项列表值、关联引用（例如，将联系人关联到账户）和负责人分配均正确；负责人根据区域或轮转规则分配。
8. 设置审批关卡：评估风险级别——少于 100 条记录时为中等，达到或超过 100 条时为高。提供同步预览，展示记录总数、创建／更新／跳过数量明细、去重结果、字段映射摘要、合规状态，以及任何需要处理的验证警告。
9. 获批后，通过 CRM MCP 执行同步：通过平台 MCP 以分批请求的方式将负载推送到目标 CRM。使用指数退避处理速率限制，并且仅在能够确认写入请求从未到达平台时重试（连接被拒绝、DNS 故障或因 429 速率限制被拒绝）；发生超时或回复丢失后，先在平台上查找记录（按电子邮件或外部 ID），然后仅重新发送缺失内容。记录每条记录的结果（已创建并附 ID、已更新并列出变更字段、已跳过并说明原因、失败并附错误代码）。
10. 验证记录数量和数据完整性：同步后查询 CRM，确认已同步记录数与预期总数一致。随机抽查 5 条记录，验证字段值是否正确传输。发现数量不符或数据不一致时，立即标记以供调查。
11. 建立关联并添加标签：对于已同步的记录，执行关联操作——按域名匹配将联系人关联到账户，按电子邮件将联系人关联到交易，并添加营销活动成员关系。添加用户指定的标签和列表分配，以便后续细分。
12. 生成回滚清单：创建可逆变更清单，列出每条已创建或修改的记录 ID，并保存之前的字段值，以便在同步后出现数据质量问题时，可在 CRM 中针对性回滚。
13. 记录结果并通知：将完整的同步执行信息——时间戳、记录数量、字段映射、错误、已建立的关联和执行时长——记录到 ~/.claude-marketing/brands/{slug}/logs/crm-sync-log.json。如果已设置通知偏好，则提醒相关销售代表其销售线索中有新记录。

输出

结构化的 CRM 同步报告，包含：

• 同步摘要：已处理记录总数及明细——已创建、已更新、已跳过（重复项）和失败的数量与百分比，并与预期总数进行比较
• 去重报告：发现的重复记录及每组重复记录的匹配类型（精确／模糊）、使用的匹配字段、置信度分数和采取的操作（已更新、已跳过或已合并）
• 字段映射表：使用的完整源字段到 CRM 字段映射，包括自动映射字段、用户覆盖项、未映射的源字段，以及为必填 CRM 字段应用的任何默认值
• 合规审计：标记了缺少同意、选择加入缺漏或数据处理依据问题的记录——针对每条记录提供补救建议，并注明适用法规
• 验证错误报告：列出无效记录及具体的字段级错误——电子邮件格式错误、缺少必填字段、数据类型不匹配或编码问题——并提供建议的修正方法
• 执行日志：每次 API 调用的带时间戳记录——批次编号、每批记录数、响应状态、重试次数（仅限可确证请求从未到达平台的写入操作）、速率限制暂停情况，以及每批处理时长
• 回滚参考：本次同步创建或修改的记录 ID 及其先前字段值，以便在同步后发现问题时进行定向回滚
• 数据完整性验证：同步后抽查结果——随机抽取 5 条记录，对比源数据与 CRM 中的记录，确认所有映射字段准确无误
• 关系链接摘要：关联到客户的联系人、关联到营销活动的交易，以及因缺少目标记录而失败的关系链接——附手动解决步骤
• 性能指标：同步吞吐量（每秒记录数）、消耗的 API 调用总数、速率限制使用率，以及未来执行类似规模同步的预计耗时
• 标签与列表分配确认：已应用的标签、已分配的列表和已创建的营销活动成员关系——提供每个标签／列表的数量，并注明任何分配失败情况
• 冲突解决日志（仅适用于双向同步）：记录遇到的每个字段级冲突——源数据值、CRM 值、采用的解决方式（以源数据为准、以 CRM 为准、以较新值为准），以及任何需要人工审核的冲突
• 下次同步建议：建议的后续操作——重新同步失败的记录、安排定期同步、清理未映射字段、更新字段映射默认值，或配置自动双向同步

使用的 Agent

• crm-manager — 字段映射、去重逻辑、数据验证、CRM 架构解析、合规检查、负载准备、负责人分配、关系链接和同步执行管理
• execution-coordinator — 审批工作流、风险评估、分批执行编排、速率限制处理、仅针对可确证从未到达平台的写入操作进行重试、生成回滚清单、发送通知和记录执行日志