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)