ActiveCampaign 集成 - 在营销自动化中验证邮箱

ActiveCampaign 表单/API
      ↓
Webhook 触发
      ↓
您的后端服务器
      ↓
BillionVerify API 验证
      ↓
ActiveCampaign API（更新联系人）
      ↓
应用标签 / 更新自定义字段 / 触发自动化

// 示例：联系人添加到 ActiveCampaign 时验证邮箱

const axios = require('axios');

// ActiveCampaign webhook 处理器
app.post('/webhook/activecampaign-contact', async (req, res) => {
  const { contact } = req.body;
  const { email, id } = contact;

  try {
    // 步骤 1：使用 BillionVerify 验证邮箱
    const verificationResult = await axios.post(
      'https://api.billionverify.com/v1/verify',
      { email },
      {
        headers: {
          'Authorization': `Bearer ${process.env.BILLIONVERIFY_API_KEY}`,
          'Content-Type': 'application/json'
        }
      }
    );

    const {
      status,
      risk_level,
      is_disposable,
      is_catch_all,
      is_role_account,
      deliverability_score
    } = verificationResult.data;

    // 步骤 2：准备 ActiveCampaign 更新
    const acApiKey = process.env.ACTIVECAMPAIGN_API_KEY;
    const acAccount = process.env.ACTIVECAMPAIGN_ACCOUNT; // 例如：'yourcompany'
    const acApiUrl = `https://${acAccount}.api-us1.com/api/3`;

    let tagsToAdd = [];
    let customFields = {
      'email_status': status,
      'email_risk_level': risk_level,
      'deliverability_score': deliverability_score,
      'verified_at': new Date().toISOString()
    };

    // 步骤 3：根据验证结果应用逻辑
    if (status === 'valid' && risk_level === 'low') {
      // 有效、安全的邮箱
      tagsToAdd = ['verified', 'high-quality'];

      // 更新联系人
      await axios.put(
        `${acApiUrl}/contacts/${id}`,
        {
          contact: {
            fieldValues: [
              { field: '1', value: 'verified' }, // 自定义字段 ID 1: email_status
              { field: '2', value: 'low' },      // 自定义字段 ID 2: risk_level
              { field: '3', value: deliverability_score.toString() }
            ]
          }
        },
        {
          headers: {
            'Api-Token': acApiKey,
            'Content-Type': 'application/json'
          }
        }
      );

      // 添加标签
      await axios.post(
        `${acApiUrl}/contactTags`,
        {
          contactTag: {
            contact: id,
            tag: 'verified'
          }
        },
        {
          headers: {
            'Api-Token': acApiKey
          }
        }
      );

      // 触发自动化（自动化 ID：5）
      await axios.post(
        `${acApiUrl}/contactAutomations`,
        {
          contactAutomation: {
            contact: id,
            automation: 5 // 您的"已验证联系人"自动化
          }
        },
        {
          headers: {
            'Api-Token': acApiKey
          }
        }
      );

    } else if (status === 'invalid') {
      // 无效邮箱 - 取消订阅
      tagsToAdd = ['invalid-email'];

      await axios.put(
        `${acApiUrl}/contacts/${id}`,
        {
          contact: {
            fieldValues: [
              { field: '1', value: 'invalid' }
            ]
          }
        },
        {
          headers: {
            'Api-Token': acApiKey
          }
        }
      );

      // 从所有列表取消订阅
      await axios.delete(
        `${acApiUrl}/contacts/${id}/contactLists`,
        {
          headers: {
            'Api-Token': acApiKey
          }
        }
      );

    } else {
      // 有风险的邮箱 - 标记以供审核
      tagsToAdd = ['needs-review'];

      if (is_disposable) tagsToAdd.push('disposable');
      if (is_catch_all) tagsToAdd.push('catch-all');
      if (is_role_account) tagsToAdd.push('role-account');

      await axios.put(
        `${acApiUrl}/contacts/${id}`,
        {
          contact: {
            fieldValues: [
              { field: '1', value: 'risky' },
              { field: '2', value: risk_level }
            ]
          }
        },
        {
          headers: {
            'Api-Token': acApiKey
          }
        }
      );

      // 应用标签
      for (const tag of tagsToAdd) {
        await axios.post(
          `${acApiUrl}/contactTags`,
          {
            contactTag: {
              contact: id,
              tag: tag
            }
          },
          {
            headers: {
              'Api-Token': acApiKey
            }
          }
        );
      }
    }

    res.status(200).send({ success: true, status });
  } catch (error) {
    console.error('Verification error:', error);
    res.status(500).send({ error: 'Verification failed' });
  }
});

import requests
from flask import Flask, request

app = Flask(__name__)

BILLIONVERIFY_API_KEY = 'your_billionverify_api_key'
ACTIVECAMPAIGN_API_KEY = 'your_activecampaign_api_key'
ACTIVECAMPAIGN_ACCOUNT = 'yourcompany'
AC_API_URL = f'https://{ACTIVECAMPAIGN_ACCOUNT}.api-us1.com/api/3'

@app.route('/webhook/activecampaign-contact', methods=['POST'])
def verify_activecampaign_contact():
    data = request.json
    contact = data.get('contact', {})
    email = contact.get('email')
    contact_id = contact.get('id')

    # 步骤 1：使用 BillionVerify 验证邮箱
    verification_response = requests.post(
        'https://api.billionverify.com/v1/verify',
        json={'email': email},
        headers={
            'Authorization': f'Bearer {BILLIONVERIFY_API_KEY}',
            'Content-Type': 'application/json'
        }
    )
    result = verification_response.json()

    # 步骤 2：处理结果并更新 ActiveCampaign
    ac_headers = {
        'Api-Token': ACTIVECAMPAIGN_API_KEY,
        'Content-Type': 'application/json'
    }

    if result['status'] == 'valid' and result['risk_level'] == 'low':
        # 更新联系人为已验证状态
        update_data = {
            'contact': {
                'fieldValues': [
                    {'field': '1', 'value': 'verified'},
                    {'field': '2', 'value': 'low'},
                    {'field': '3', 'value': str(result.get('deliverability_score', 100))}
                ]
            }
        }
        requests.put(
            f'{AC_API_URL}/contacts/{contact_id}',
            json=update_data,
            headers=ac_headers
        )

        # 添加"已验证"标签
        requests.post(
            f'{AC_API_URL}/contactTags',
            json={
                'contactTag': {
                    'contact': contact_id,
                    'tag': 'verified'
                }
            },
            headers=ac_headers
        )

        # 触发自动化（自动化 ID：5）
        requests.post(
            f'{AC_API_URL}/contactAutomations',
            json={
                'contactAutomation': {
                    'contact': contact_id,
                    'automation': 5
                }
            },
            headers=ac_headers
        )

    elif result['status'] == 'invalid':
        # 标记为无效并取消订阅
        update_data = {
            'contact': {
                'fieldValues': [
                    {'field': '1', 'value': 'invalid'}
                ]
            }
        }
        requests.put(
            f'{AC_API_URL}/contacts/{contact_id}',
            json=update_data,
            headers=ac_headers
        )

        # 从所有列表取消订阅
        requests.delete(
            f'{AC_API_URL}/contacts/{contact_id}/contactLists',
            headers=ac_headers
        )

    else:
        # 有风险的邮箱 - 标记以供审核
        tags = ['needs-review']
        if result.get('is_disposable'):
            tags.append('disposable')
        if result.get('is_catch_all'):
            tags.append('catch-all')
        if result.get('is_role_account'):
            tags.append('role-account')

        update_data = {
            'contact': {
                'fieldValues': [
                    {'field': '1', 'value': 'risky'},
                    {'field': '2', 'value': result['risk_level']}
                ]
            }
        }
        requests.put(
            f'{AC_API_URL}/contacts/{contact_id}',
            json=update_data,
            headers=ac_headers
        )

        # 应用标签
        for tag in tags:
            requests.post(
                f'{AC_API_URL}/contactTags',
                json={
                    'contactTag': {
                        'contact': contact_id,
                        'tag': tag
                    }
                },
                headers=ac_headers
            )

    return {'success': True, 'status': result['status']}, 200

if __name__ == '__main__':
    app.run(port=5000)