Примеры использования

Практические примеры интеграции с API Sync.

Содержание

• Примеры на Python
• Примеры на JavaScript/Node.js
• Примеры с cURL
• Обработка ошибок
• Расширенное использование

---

Примеры на Python

Базовый импорт

import httpx
import uuid
import asyncio

API_KEY = "ваш-api-ключ"
BASE_URL = "https://import-api.liddex.ru"

async def import_contacts(contacts: list[dict]) -> dict:
    """Импорт контактов в AI."""
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Idempotency-Key": str(uuid.uuid4()),
        "Content-Type": "application/json",
    }

    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{BASE_URL}/v1/contacts/import",
            json=contacts,
            headers=headers,
            timeout=30.0,
        )
        response.raise_for_status()
        return response.json()

# Использование
contacts = [
    {
        "phone": "+79001234567",
        "tags": ["лид", "сайт"],
        "additionalFields": {
            "source": "landing",
            "campaign": "summer2024",
        }
    },
    {
        "phone": "89001234568",
        "tags": ["клиент"],
        "additionalFields": {
            "source": "referral",
        }
    }
]

result = asyncio.run(import_contacts(contacts))
print(f"ID задачи: {result['job_id']}")
print(f"Валидных контактов: {result['valid_contacts']}")

Опрос статуса задачи

async def wait_for_job_completion(
    job_id: str,
    max_wait: int = 300,
    poll_interval: int = 5
) -> dict:
    """Опрос статуса задачи до завершения или таймаута."""
    headers = {"Authorization": f"Bearer {API_KEY}"}
    elapsed = 0

    async with httpx.AsyncClient() as client:
        while elapsed < max_wait:
            response = await client.get(
                f"{BASE_URL}/v1/contacts/import/{job_id}",
                headers=headers,
            )
            response.raise_for_status()
            status = response.json()

            if status["status"] in ["completed", "failed"]:
                return status

            await asyncio.sleep(poll_interval)
            elapsed += poll_interval

    raise TimeoutError(f"Задача {job_id} не завершилась за {max_wait}с")

# Использование
result = asyncio.run(import_contacts(contacts))
final_status = asyncio.run(wait_for_job_completion(result["job_id"]))
print(f"Статус: {final_status['status']}")
print(f"Отправлено: {final_status['sent_contacts']}/{final_status['valid_contacts']}")

С HMAC подписью

import hmac
import hashlib
from datetime import datetime, timezone
import json

def generate_hmac_signature(secret: str, timestamp: str, body: bytes) -> str:
    """Генерация HMAC-SHA256 подписи."""
    message = f"{timestamp}\n".encode() + body
    return hmac.new(secret.encode(), message, hashlib.sha256).hexdigest()

async def import_with_hmac(contacts: list[dict], hmac_secret: str) -> dict:
    """Импорт контактов с HMAC подписью."""
    timestamp = datetime.now(timezone.utc).isoformat()
    body = json.dumps(contacts).encode()
    signature = generate_hmac_signature(hmac_secret, timestamp, body)

    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Idempotency-Key": str(uuid.uuid4()),
        "X-Signature": signature,
        "X-Timestamp": timestamp,
        "Content-Type": "application/json",
    }

    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{BASE_URL}/v1/contacts/import",
            content=body,
            headers=headers,
        )
        response.raise_for_status()
        return response.json()

Режим отладки (Dry-Run)

async def validate_contacts(contacts: list[dict]) -> dict:
    """Валидация контактов без отправки в AI."""
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Idempotency-Key": str(uuid.uuid4()),
        "X-Debug": "true",
        "Content-Type": "application/json",
    }

    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{BASE_URL}/v1/contacts/import",
            json=contacts,
            headers=headers,
        )
        response.raise_for_status()
        return response.json()

# Использование
result = asyncio.run(validate_contacts([
    {"phone": "+79001234567", "tags": [], "additionalFields": {}},
    {"phone": "invalid", "tags": [], "additionalFields": {}},
]))

print(f"Валидных: {result['valid_contacts']}")
print(f"Невалидных: {result['invalid_contacts']}")

for contact in result['validation_results']:
    if not contact['is_valid']:
        print(f"Ошибка: {contact['original_phone']} - {contact['error']}")

---

Примеры на JavaScript/Node.js

Базовый импорт

const axios = require('axios');
const { v4: uuidv4 } = require('uuid');

const API_KEY = 'ваш-api-ключ';
const BASE_URL = 'https://import-api.liddex.ru';

async function importContacts(contacts) {
    const response = await axios.post(
        `${BASE_URL}/v1/contacts/import`,
        contacts,
        {
            headers: {
                'Authorization': `Bearer ${API_KEY}`,
                'Idempotency-Key': uuidv4(),
                'Content-Type': 'application/json',
            },
            timeout: 30000,
        }
    );
    return response.data;
}

// Использование
const contacts = [
    {
        phone: '+79001234567',
        tags: ['лид', 'сайт'],
        additionalFields: {
            source: 'landing',
            campaign: 'summer2024',
        }
    }
];

importContacts(contacts)
    .then(result => {
        console.log('ID задачи:', result.job_id);
        console.log('Валидных контактов:', result.valid_contacts);
    })
    .catch(error => {
        console.error('Ошибка:', error.response?.data || error.message);
    });

С логикой повторов

const axios = require('axios');
const axiosRetry = require('axios-retry');

const client = axios.create({
    baseURL: BASE_URL,
    timeout: 30000,
});

// Настройка повторов
axiosRetry(client, {
    retries: 3,
    retryDelay: axiosRetry.exponentialDelay,
    retryCondition: (error) => {
        // Повтор при сетевых ошибках или 5xx
        return axiosRetry.isNetworkOrIdempotentRequestError(error) ||
               (error.response && error.response.status >= 500);
    },
});

async function importContactsWithRetry(contacts) {
    const response = await client.post('/v1/contacts/import', contacts, {
        headers: {
            'Authorization': `Bearer ${API_KEY}`,
            'Idempotency-Key': uuidv4(),
            'Content-Type': 'application/json',
        },
    });
    return response.data;
}

Пример на TypeScript

import axios, { AxiosInstance } from 'axios';
import { v4 as uuidv4 } from 'uuid';

interface Contact {
    phone: string;
    tags: string[];
    additionalFields: Record<string, any>;
}

interface ImportResponse {
    job_id: string;
    correlation_id: string;
    idempotency_status: string;
    status: string;
    total_contacts: number;
    valid_contacts: number;
    invalid_contacts: number;
    chunks_count: number;
    created_at: string;
}

class SyncClient {
    private client: AxiosInstance;

    constructor(apiKey: string, baseURL: string = 'https://import-api.liddex.ru') {
        this.client = axios.create({
            baseURL,
            headers: {
                'Authorization': `Bearer ${apiKey}`,
                'Content-Type': 'application/json',
            },
            timeout: 30000,
        });
    }

    async importContacts(contacts: Contact[]): Promise<ImportResponse> {
        const response = await this.client.post<ImportResponse>(
            '/v1/contacts/import',
            contacts,
            {
                headers: {
                    'Idempotency-Key': uuidv4(),
                },
            }
        );
        return response.data;
    }

    async getJobStatus(jobId: string): Promise<any> {
        const response = await this.client.get(`/v1/contacts/import/${jobId}`);
        return response.data;
    }
}

// Использование
const client = new SyncClient('ваш-api-ключ');

const contacts: Contact[] = [
    {
        phone: '+79001234567',
        tags: ['тест'],
        additionalFields: { source: 'typescript' },
    }
];

client.importContacts(contacts)
    .then(result => console.log('ID задачи:', result.job_id))
    .catch(error => console.error('Ошибка:', error.message));

---

Примеры с cURL

Базовый импорт

#!/bin/bash

API_KEY="ваш-api-ключ"
BASE_URL="https://import-api.liddex.ru"
IDEMPOTENCY_KEY=$(uuidgen)

curl -X POST "${BASE_URL}/v1/contacts/import" \
  -H "Authorization: Bearer ${API_KEY}" \
  -H "Idempotency-Key: ${IDEMPOTENCY_KEY}" \
  -H "Content-Type: application/json" \
  -d '[
    {
      "phone": "+79001234567",
      "tags": ["лид", "сайт"],
      "additionalFields": {
        "source": "landing",
        "campaign": "summer2024"
      }
    }
  ]'

Сохранение ID задачи и проверка статуса

#!/bin/bash

# Импорт и извлечение ID задачи
RESPONSE=$(curl -s -X POST "${BASE_URL}/v1/contacts/import" \
  -H "Authorization: Bearer ${API_KEY}" \
  -H "Idempotency-Key: $(uuidgen)" \
  -H "Content-Type: application/json" \
  -d '[{"phone": "+79001234567", "tags": [], "additionalFields": {}}]')

JOB_ID=$(echo $RESPONSE | jq -r '.job_id')
echo "ID задачи: $JOB_ID"

# Ожидание и проверка статуса
sleep 5

curl -s -X GET "${BASE_URL}/v1/contacts/import/${JOB_ID}" \
  -H "Authorization: Bearer ${API_KEY}" | jq

---

Обработка ошибок

Python

from httpx import HTTPStatusError

async def import_with_error_handling(contacts: list[dict]) -> dict:
    try:
        return await import_contacts(contacts)
    except HTTPStatusError as e:
        if e.response.status_code == 400:
            print("Ошибка валидации:", e.response.json())
        elif e.response.status_code == 401:
            print("Ошибка аутентификации - проверьте API ключ")
        elif e.response.status_code == 422:
            print("Конфликт идемпотентности - ключ использован с другими данными")
        elif e.response.status_code == 429:
            print("Превышен лимит запросов - повторите позже")
        else:
            print(f"Неожиданная ошибка: {e.response.status_code}")
        raise

JavaScript

async function importWithErrorHandling(contacts) {
    try {
        return await importContacts(contacts);
    } catch (error) {
        if (error.response) {
            const status = error.response.status;
            const data = error.response.data;

            switch (status) {
                case 400:
                    console.error('Ошибка валидации:', data);
                    break;
                case 401:
                    console.error('Ошибка аутентификации - проверьте API ключ');
                    break;
                case 422:
                    console.error('Конфликт идемпотентности:', data);
                    break;
                case 429:
                    console.error('Превышен лимит запросов - повторите позже');
                    break;
                default:
                    console.error(`Неожиданная ошибка: ${status}`, data);
            }
        } else {
            console.error('Сетевая ошибка:', error.message);
        }
        throw error;
    }
}

---

Расширенное использование

Пакетная обработка

async def import_in_batches(
    all_contacts: list[dict],
    batch_size: int = 1000,
) -> list[str]:
    """Импорт контактов пакетами."""
    job_ids = []

    for i in range(0, len(all_contacts), batch_size):
        batch = all_contacts[i:i + batch_size]
        result = await import_contacts(batch)
        job_ids.append(result['job_id'])

        print(f"Пакет {i // batch_size + 1}: Задача {result['job_id']}")

        # Ограничение частоты
        await asyncio.sleep(0.5)

    return job_ids

Мониторинг прогресса

async def monitor_jobs(job_ids: list[str]) -> dict:
    """Мониторинг нескольких задач."""
    results = {"completed": 0, "failed": 0, "pending": 0}

    async with httpx.AsyncClient() as client:
        for job_id in job_ids:
            response = await client.get(
                f"{BASE_URL}/v1/contacts/import/{job_id}",
                headers={"Authorization": f"Bearer {API_KEY}"},
            )
            status = response.json()

            if status["status"] == "completed":
                results["completed"] += 1
            elif status["status"] == "failed":
                results["failed"] += 1
            else:
                results["pending"] += 1

    return results

---

Смотрите также

• Руководство по API - Полный справочник по API
• Руководство по интеграции - Технические детали