Email Validation in CRM: Bitrix24, amoCRM, and HubSpot Integration

<?php
// Webhook endpoint: POST /webhooks/bitrix24-contact

$contactId = $_POST['data']['FIELDS']['ID'] ?? null;
if (!$contactId) { http_response_code(400); exit; }

$bitrixWebhook = getenv('BITRIX24_WEBHOOK_URL');
// e.g. https://your-domain.bitrix24.ru/rest/1/abc123xyz/

// 1. Fetch contact email from Bitrix24
$contact = json_decode(file_get_contents(
    $bitrixWebhook . "crm.contact.get?ID=" . $contactId
), true)['result'];

$email = $contact['EMAIL'][0]['VALUE'] ?? null;
if (!$email) exit;

// 2. Validate via uChecker API
$ch = curl_init('https://api.uchecker.net/api/v1/validate/single');
curl_setopt_array($ch, [
    CURLOPT_POST => true,
    CURLOPT_HTTPHEADER => [
        'x-api-key: ' . getenv('UCHECKER_API_KEY'),
        'Content-Type: application/json',
    ],
    CURLOPT_POSTFIELDS => json_encode(['email' => $email]),
    CURLOPT_RETURNTRANSFER => true,
]);
$result = json_decode(curl_exec($ch), true);
curl_close($ch);

$taskId = $result['task_id'];

// 3. Poll for result (in production use webhook_url instead)
sleep(5);
$ch = curl_init("https://api.uchecker.net/api/v1/tasks/$taskId");
curl_setopt_array($ch, [
    CURLOPT_HTTPHEADER => [
        'x-api-key: ' . getenv('UCHECKER_API_KEY'),
    ],
    CURLOPT_RETURNTRANSFER => true,
]);
$task = json_decode(curl_exec($ch), true);
curl_close($ch);

$status = $task['result']['status'] ?? 'unknown';
// Map to your custom field values
$statusMap = [
    'deliverable'   => 'valid',
    'undeliverable' => 'invalid',
    'risky'         => 'risky',
    'unknown'       => 'unknown',
];

// 4. Write result back to Bitrix24
file_get_contents($bitrixWebhook . "crm.contact.update?" . http_build_query([
    'ID'     => $contactId,
    'FIELDS' => [
        'UF_EMAIL_VALID'   => $statusMap[$status] ?? 'unknown',
        'UF_EMAIL_CHECKED' => date('Y-m-d'),
    ],
]));

import os, requests
from flask import Flask, request, jsonify

app = Flask(__name__)

UCHECKER_KEY = os.environ["UCHECKER_API_KEY"]
AMO_TOKEN    = os.environ["AMOCRM_ACCESS_TOKEN"]
AMO_DOMAIN   = os.environ["AMOCRM_DOMAIN"]  # e.g. mycompany

def validate_email(email: str) -> dict:
    """Send email to uChecker, return validation result."""
    resp = requests.post(
        "https://api.uchecker.net/api/v1/validate/single",
        headers={"x-api-key": UCHECKER_KEY},
        json={"email": email},
    )
    task_id = resp.json()["task_id"]

    # Poll until done (use webhook_url in production)
    import time
    for _ in range(10):
        time.sleep(3)
        r = requests.get(
            f"https://api.uchecker.net/api/v1/tasks/{task_id}",
            headers={"x-api-key": UCHECKER_KEY},
        )
        data = r.json()
        if data.get("status") == "completed":
            return data["result"]
    return {"status": "unknown"}

def update_amo_contact(contact_id: int, validation_status: str):
    """Write validation result back to amoCRM custom field."""
    requests.patch(
        f"https://{AMO_DOMAIN}.amocrm.ru/api/v4/contacts/{contact_id}",
        headers={
            "Authorization": f"Bearer {AMO_TOKEN}",
            "Content-Type": "application/json",
        },
        json={
            "custom_fields_values": [
                {
                    "field_id": 123456,  # your custom field ID
                    "values": [{"value": validation_status}],
                }
            ]
        },
    )

@app.route("/webhooks/amocrm", methods=["POST"])
def handle_amocrm_webhook():
    payload = request.json or {}
    contacts = payload.get("contacts", {}).get("add", [])
    contacts += payload.get("contacts", {}).get("update", [])

    for contact in contacts:
        contact_id = contact["id"]
        # Fetch full contact to get email
        r = requests.get(
            f"https://{AMO_DOMAIN}.amocrm.ru/api/v4/contacts/{contact_id}",
            headers={"Authorization": f"Bearer {AMO_TOKEN}"},
        )
        fields = r.json().get("custom_fields_values", [])
        email = None
        for f in fields:
            if f["field_code"] == "EMAIL":
                email = f["values"][0]["value"]
                break

        if not email:
            continue

        result = validate_email(email)
        status = result.get("status", "unknown")
        update_amo_contact(contact_id, status)

    return jsonify({"ok": True})

// HubSpot Custom Coded Action
// Input: contact email (configured in workflow)
// Secrets: UCHECKER_API_KEY (add in workflow settings)

const hubspot = require("@hubspot/api-client");
const axios = require("axios");

exports.main = async (event, callback) => {
  const email = event.inputFields["email"];
  if (!email) return callback({ outputFields: {} });

  const apiKey = process.env.UCHECKER_API_KEY;

  // 1. Submit for validation
  const { data: task } = await axios.post(
    "https://api.uchecker.net/api/v1/validate/single",
    { email },
    { headers: { "x-api-key": apiKey } }
  );

  // 2. Poll for result
  let result = null;
  for (let i = 0; i < 10; i++) {
    await new Promise((r) => setTimeout(r, 3000));
    const { data } = await axios.get(
      `https://api.uchecker.net/api/v1/tasks/${task.task_id}`,
      { headers: { "x-api-key": apiKey } }
    );
    if (data.status === "completed") {
      result = data.result;
      break;
    }
  }

  const status = result?.status || "unknown";

  // 3. Update contact property
  const hsClient = new hubspot.Client({
    accessToken: process.env.HUBSPOT_TOKEN,
  });

  await hsClient.crm.contacts.basicApi.update(
    event.object.objectId,
    {
      properties: {
        email_validation_status: status,
        email_validated_date: new Date().toISOString().split("T")[0],
      },
    }
  );

  callback({ outputFields: { validation_status: status } });
};

import os, time, requests

API_KEY = os.environ["UCHECKER_API_KEY"]
BASE    = "https://api.uchecker.net"
HEADERS = {"x-api-key": API_KEY, "Content-Type": "application/json"}

def bulk_validate(emails: list[str]) -> dict:
    """Submit bulk validation and wait for results."""
    # 1. Create bulk task
    resp = requests.post(
        f"{BASE}/api/v1/validate/bulk",
        headers=HEADERS,
        json={"emails": emails},
    )
    task_id = resp.json()["task_id"]
    print(f"Bulk task created: {task_id}, emails: {len(emails)}")

    # 2. Poll until complete
    while True:
        time.sleep(10)
        r = requests.get(f"{BASE}/api/v1/tasks/{task_id}", headers=HEADERS)
        data = r.json()
        if data["status"] == "completed":
            return data["results"]  # {email: status, ...}
        if data["status"] == "failed":
            raise RuntimeError(f"Task {task_id} failed")
        print(f"  status: {data['status']}, progress: {data.get('progress', '?')}")

# Usage example:
# contacts = fetch_contacts_from_crm()
# emails = [c["email"] for c in contacts if c.get("email")]
# results = bulk_validate(emails)
# for contact in contacts:
#     status = results.get(contact["email"], "unknown")
#     update_crm_contact(contact["id"], status)