Как проверить список email из CSV через API

Для проверки списка email через API запустите задачу, которая читает CSV, отправляет адреса с вашего сервера и записывает ответы рядом с исходными строками. Сохраняйте прогресс по ходу работы: запрос может сразу завершиться, остаться в pending или вернуть ошибку без вердикта об адресе.

Ниже приведен Python-скрипт для такой задачи. Файл и запуск остаются у вас; Emvero получает один адрес на запрос проверки. Страница массовой проверки описывает этот сценарий сервиса. Загрузчика CSV и отдельного batch-эндпоинта нет.

Подготовьте файл и учетные данные API

В CSV нужны строка заголовков и столбец email. Это синтетический пример: запятая внутри кавычек относится к заметке, а в последней строке адрес отсутствует.

contact_id,email,note
c-001,ada@example.com,"trial, invited"
c-002,ops@example.org,newsletter
c-003,,missing address

Сохраните исходный файл и выберите другой путь для результата. Скрипт оставляет исходные столбцы и значение адреса. Пробелы по краям он удаляет только у отправляемой копии, не переводя адрес в нижний регистр и не переписывая его. Пустой адрес получает ошибку входных данных в своей строке.

Понадобятся Python 3, учетные данные в EMVERO_API_TOKEN и права на создание выходного файла и checkpoint. Осторожно ограничьте число параллельных обработчиков. Пример загружает весь CSV в память. Если файл туда не помещается, измените задачу так, чтобы она читала и сохраняла данные небольшими группами.

Запросы идут на https://api.emvero.tech/v1/verifications с заголовком Authorization: Bearer. Запускайте код локально или на сервере, чтобы учетные данные не попали в браузер. Контракт запросов описан на странице API проверки email, обработка данных — на странице Безопасность.

Сохраните постоянную связь между задачей и строками

Скрипт вычисляет хеш из байтов входного файла и имени выбранного столбца с адресами. Этот идентификатор задачи и позиция строки образуют emvero_row_id и постоянный Idempotency-Key. Неизмененный файл дает те же ключи. Редактирование, перестановка строк или выбор другого столбца создают другую задачу: старые записи checkpoint не попадут в новые строки.

В пределах поддерживаемого окна идемпотентности повтор ключа возвращает исходную проверку вместо создания новой. Пример не зависит от конкретной длительности окна. Скрипт также сохраняет ID каждой проверки pending. После перезапуска он запрашивает этот ID и лишь затем рассматривает новый POST.

Каждая строка обрабатывается отдельно. Одинаковые адреса в разных строках получают разные ключи; скрипт не удаляет дубли. Если в вашей системе уже есть неизменяемые ID записей и задач, на их основе можно построить собственную схему связи.

Записывайте свидетельства отдельно от ошибок

В выходном файле остаются входные столбцы и появляются поля ниже. Структурированные объекты хранятся как компактный JSON внутри ячейки CSV. Так вложенные поля и массивы кодов причин сохраняются при записи и повторном чтении.

Если следующая система принимает JSON или записи базы данных, сохраняйте объекты в исходном виде. Отдельные столбцы под сегодняшние поля проверки могут незаметно потерять поля, добавленные позже.

Запустите задачу на Python

Сохраните код как verify_csv.py. Парсер Python csv правильно читает поля в кавычках и пустые ячейки. Повторяющиеся заголовки и зарезервированные имена столбцов результата скрипт отклоняет до обработки, чтобы не перезаписать входные значения.

Checkpoint хранится в файле JSON Lines, куда скрипт только дописывает события. При продолжении берется последнее событие каждой строки. Окончательные результаты используются повторно, а незавершенные проверки запрашиваются по ID. Возобновленная задача не проверяет все готовые строки заново с новыми свидетельствами.

#!/usr/bin/env python3
import argparse
import csv
import hashlib
import json
import os
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.parse import quote
from urllib.request import Request, urlopen

API_BASE = "https://api.emvero.tech"
TERMINAL_STATUSES = {"deliverable", "risky", "undeliverable", "unknown"}
RETRYABLE_HTTP_STATUSES = {429, 500, 503}
MAX_REQUEST_ATTEMPTS = 4

RESULT_COLUMNS = [
    "emvero_row_id",
    "emvero_submitted_email",
    "emvero_state",
    "emvero_verification_id",
    "emvero_status",
    "emvero_action",
    "emvero_checks",
    "emvero_reason_codes",
    "emvero_confidence",
    "emvero_freshness",
    "emvero_normalized",
    "emvero_error_http_status",
    "emvero_error_code",
    "emvero_error_message",
]

checkpoint_lock = threading.Lock()


class ApiError(Exception):
    def __init__(self, http_status, code, message, retry_after_seconds=0):
        super().__init__(message or code)
        self.http_status = http_status
        self.code = code
        self.message = message
        self.retry_after_seconds = retry_after_seconds


def retry_after_seconds(value):
    if not value:
        return 0
    value = value.strip()
    if value.isdigit():
        return int(value)
    try:
        return max(0, (parsedate_to_datetime(value) - datetime.now(timezone.utc)).total_seconds())
    except (TypeError, ValueError):
        return 0


def compact_json(value):
    if value is None:
        return ""
    return json.dumps(value, ensure_ascii=False, separators=(",", ":"))


def request_json(method, path, token, body=None, idempotency_key=None):
    headers = {
        "Accept": "application/json",
        "Authorization": f"Bearer {token}",
    }
    data = None

    if body is not None:
        data = json.dumps(body).encode("utf-8")
        headers["Content-Type"] = "application/json"
    if idempotency_key:
        headers["Idempotency-Key"] = idempotency_key

    request = Request(
        f"{API_BASE}{path}",
        data=data,
        headers=headers,
        method=method,
    )

    try:
        with urlopen(request, timeout=30) as response:
            return json.load(response)
    except HTTPError as error:
        raw = error.read()
        try:
            payload = json.loads(raw)
        except (json.JSONDecodeError, UnicodeDecodeError):
            payload = {}
        raise ApiError(
            error.code,
            payload.get("error", "http_error"),
            payload.get("message", ""),
            retry_after_seconds(error.headers.get("Retry-After")),
        ) from error


def call_with_backoff(method, path, token, body=None, idempotency_key=None):
    for attempt in range(MAX_REQUEST_ATTEMPTS):
        delay = min(2**attempt, 8)
        try:
            return request_json(method, path, token, body, idempotency_key)
        except ApiError as error:
            retryable = error.http_status in RETRYABLE_HTTP_STATUSES
            if not retryable or attempt == MAX_REQUEST_ATTEMPTS - 1:
                raise
            delay = max(delay, error.retry_after_seconds)
        except (URLError, TimeoutError):
            if attempt == MAX_REQUEST_ATTEMPTS - 1:
                raise

        # Keep local retries bounded, and honour a longer server Retry-After.
        time.sleep(delay)


def load_checkpoint(path):
    latest = {}
    if not path.exists():
        return latest

    with path.open("rb") as handle:
        for line in handle:
            try:
                event = json.loads(line)
                latest[event["row_id"]] = event
            except (json.JSONDecodeError, UnicodeDecodeError, KeyError):
                # Ignore an incomplete final line left by an interrupted write.
                continue
    return latest


def append_checkpoint(path, event):
    line = json.dumps(event, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
    with checkpoint_lock:
        with path.open("a+b") as handle:
            handle.seek(0, os.SEEK_END)
            if handle.tell():
                handle.seek(-1, os.SEEK_END)
                if handle.read(1) != b"\n":
                    handle.write(b"\n")
            handle.write(line + b"\n")
            handle.flush()
            os.fsync(handle.fileno())


def row_id(index, job_id):
    return f"{job_id}-row-{index:08d}"


def idempotency_key(job_id, item_id):
    return f"csv-{item_id}"


def read_input(path, email_column, job_id):
    with path.open("r", encoding="utf-8-sig", newline="") as handle:
        reader = csv.DictReader(handle)
        if not reader.fieldnames:
            raise ValueError("the CSV file has no header row")
        if len(reader.fieldnames) != len(set(reader.fieldnames)):
            raise ValueError("the CSV file has duplicate column names")
        if email_column not in reader.fieldnames:
            raise ValueError(f"the CSV file has no {email_column!r} column")
        reserved = set(reader.fieldnames) & set(RESULT_COLUMNS)
        if reserved:
            raise ValueError(f"input already has result columns: {sorted(reserved)}")

        items = []
        for index, source_row in enumerate(reader, start=1):
            if None in source_row:
                raise ValueError(
                    f"data row {index} has more fields than the header"
                )
            raw_email = source_row.get(email_column) or ""
            items.append(
                {
                    "row_id": row_id(index, job_id),
                    "source_row": source_row,
                    "submitted_email": raw_email.strip(),
                }
            )
        return list(reader.fieldnames), items


def pending_event(item_id, key, result):
    return {
        "row_id": item_id,
        "state": "pending",
        "idempotency_key": key,
        "verification_id": result.get("id", ""),
        "retry_after_ms": result.get("retry_after_ms", 1000),
    }


def terminal_event(item_id, key, result):
    return {
        "row_id": item_id,
        "state": "terminal",
        "idempotency_key": key,
        "result": result,
    }


def poll_until_terminal(result, token, poll_seconds, checkpoint_path, item_id, key):
    deadline = time.monotonic() + poll_seconds

    while result.get("status") == "pending":
        verification_id = result.get("id")
        if not verification_id:
            raise ValueError("a pending result has no verification id")

        append_checkpoint(checkpoint_path, pending_event(item_id, key, result))
        delay_ms = result.get("retry_after_ms") or 1000
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            return result, False

        time.sleep(min(max(delay_ms / 1000, 0.25), remaining))
        result = call_with_backoff(
            "GET",
            f"/v1/verifications/{quote(verification_id, safe='')}",
            token,
        )

    return result, True


def verify_item(item, job_id, token, poll_seconds, checkpoint_path, saved):
    item_id = item["row_id"]
    email = item["submitted_email"]
    key = idempotency_key(job_id, item_id)

    if not email:
        return {"kind": "input_error", "code": "empty_email", "message": "email is empty"}

    previous = saved.get(item_id)
    if previous and previous.get("state") == "terminal":
        return {"kind": "terminal", "result": previous["result"]}

    result = None
    if previous and previous.get("state") == "pending" and previous.get("verification_id"):
        try:
            result = call_with_backoff(
                "GET",
                f"/v1/verifications/{quote(previous['verification_id'], safe='')}",
                token,
            )
        except ApiError as error:
            if error.http_status != 404:
                raise
            # The stored result is no longer retrievable. Reuse the same stable
            # idempotency key and let the API decide whether it still owns it.

    if result is None:
        result = call_with_backoff(
            "POST",
            "/v1/verifications",
            token,
            body={"email": email},
            idempotency_key=key,
        )

    result, finished = poll_until_terminal(
        result,
        token,
        poll_seconds,
        checkpoint_path,
        item_id,
        key,
    )

    if not finished:
        return {
            "kind": "pending",
            "result": result,
            "code": "client_poll_deadline",
            "message": "verification is still pending; resume the job later",
        }

    status = result.get("status")
    if status not in TERMINAL_STATUSES:
        return {
            "kind": "protocol_error",
            "result": result,
            "code": "unexpected_status",
            "message": f"unexpected verification status: {status!r}",
        }

    append_checkpoint(checkpoint_path, terminal_event(item_id, key, result))
    return {"kind": "terminal", "result": result}


def run_item(item, job_id, token, poll_seconds, checkpoint_path, saved):
    try:
        return verify_item(
            item,
            job_id,
            token,
            poll_seconds,
            checkpoint_path,
            saved,
        )
    except ApiError as error:
        return {
            "kind": "api_error",
            "http_status": error.http_status,
            "code": error.code,
            "message": error.message,
        }
    except (URLError, TimeoutError) as error:
        return {
            "kind": "network_error",
            "code": "network_error",
            "message": str(getattr(error, "reason", error)),
        }
    except (KeyError, TypeError, ValueError) as error:
        return {
            "kind": "protocol_error",
            "code": "invalid_response",
            "message": str(error),
        }


def output_fields(item, outcome):
    result = outcome.get("result") or {}
    return {
        "emvero_row_id": item["row_id"],
        "emvero_submitted_email": item["submitted_email"],
        "emvero_state": outcome["kind"],
        "emvero_verification_id": result.get("id", ""),
        "emvero_status": result.get("status", ""),
        "emvero_action": result.get("action", ""),
        "emvero_checks": compact_json(result.get("checks")),
        "emvero_reason_codes": compact_json(result.get("reason_codes")),
        "emvero_confidence": result.get("confidence", ""),
        "emvero_freshness": compact_json(result.get("freshness")),
        "emvero_normalized": compact_json(result.get("normalized")),
        "emvero_error_http_status": outcome.get("http_status", ""),
        "emvero_error_code": outcome.get("code", ""),
        "emvero_error_message": outcome.get("message", ""),
    }


def write_output(path, input_columns, items, outcomes):
    columns = input_columns + [name for name in RESULT_COLUMNS if name not in input_columns]
    temporary_path = path.with_name(path.name + ".tmp")

    with temporary_path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=columns, extrasaction="ignore")
        writer.writeheader()
        for item in items:
            row = dict(item["source_row"])
            row.update(output_fields(item, outcomes[item["row_id"]]))
            writer.writerow(row)

    os.replace(temporary_path, path)


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("input_csv", type=Path)
    parser.add_argument("output_csv", type=Path)
    parser.add_argument("--email-column", default="email")
    parser.add_argument("--workers", type=int, default=4)
    parser.add_argument("--poll-seconds", type=int, default=300)
    parser.add_argument("--checkpoint", type=Path)
    args = parser.parse_args()

    if args.workers < 1:
        parser.error("--workers must be at least 1")
    if args.poll_seconds < 1:
        parser.error("--poll-seconds must be at least 1")
    if args.input_csv.resolve() == args.output_csv.resolve():
        parser.error("input_csv and output_csv must be different files")

    token = os.environ.get("EMVERO_API_TOKEN")
    if not token:
        parser.error("set EMVERO_API_TOKEN before running the job")

    checkpoint_path = args.checkpoint or Path(str(args.output_csv) + ".checkpoint.jsonl")
    if checkpoint_path.resolve() in {args.input_csv.resolve(), args.output_csv.resolve()}:
        parser.error("checkpoint must be separate from input and output")
    job_id = hashlib.sha256(
        args.input_csv.read_bytes() + b"\0" + args.email_column.encode("utf-8")
    ).hexdigest()[:20]
    input_columns, items = read_input(args.input_csv, args.email_column, job_id)
    saved = load_checkpoint(checkpoint_path)

    outcomes = {}
    with ThreadPoolExecutor(max_workers=args.workers) as executor:
        futures = {
            executor.submit(
                run_item,
                item,
                job_id,
                token,
                args.poll_seconds,
                checkpoint_path,
                saved,
            ): item["row_id"]
            for item in items
        }
        for future in as_completed(futures):
            outcomes[futures[future]] = future.result()

    write_output(args.output_csv, input_columns, items, outcomes)

    counts = {}
    for outcome in outcomes.values():
        counts[outcome["kind"]] = counts.get(outcome["kind"], 0) + 1
    print(json.dumps(counts, sort_keys=True))


if __name__ == "__main__":
    main()

Укажите разные пути для входного и выходного файлов:

export EMVERO_API_TOKEN="emv_your_credential"
python3 verify_csv.py contacts.csv contacts.verified.csv --workers 4

--workers ограничивает число одновременно обрабатываемых строк. --poll-seconds задает срок опроса одной строки, а --checkpoint позволяет выбрать файл прогресса. Его путь должен отличаться от обоих CSV-файлов.

Для 429, 500, 503 и сетевых ошибок число повторов ограничено. Если серверный Retry-After больше локальной задержки, скрипт выдерживает серверный интервал. Сохранившаяся ошибка записывается рядом со строкой. Перед продолжением исправьте учетные данные, баланс или проблему сервиса. При ограничении частоты запросов снизьте параллельность.

Продолжайте ту же проверку при pending

POST /v1/verifications возвращает HTTP 200 и для окончательного результата, и для pending. Сохраните id незавершенной проверки, подождите retry_after_ms и вызовите GET /v1/verifications/{id}. Отсутствие готового результата не повод создавать новую проверку.

Если срок опроса в скрипте истек, он записывает client_poll_deadline и оставляет emvero_status равным pending. Код описывает предел ожидания клиента, а не свойство адреса. Позже запустите ту же команду с тем же входным файлом и checkpoint.

В текущем контракте нет исходящего callback-вызова. Задача должна хранить ID и сама получать результат. В работе со списком ожидание можно организовать иначе, чем в форме регистрации; это разбирает сравнение проверки в реальном времени и массовой проверки.

Считайте HTTP-ошибку проблемой выполнения задачи

Успешный ответ HTTP 200 может содержать undeliverable: проверка получила результат об адресе. Ответы 401, 402, 403, 429, 500 и 503 относятся к работе API. Не превращайте их в undeliverable или выдуманный unknown.

Ошибки запроса хранятся в emvero_error_http_status, emvero_error_code и emvero_error_message. Свидетельства об адресе записываются в emvero_status, emvero_action и поля результата. Устраните причину и повторите ту же задачу. Checkpoint сохранит готовые результаты, а строки с ошибками будут обработаны еще раз.

Примените правила списка к сохраненным результатам

Читайте action как рекомендацию для следующего шага вместе с объясняющими свидетельствами. Рекомендации Emvero помогают принимать решения при регистрации; они не дают разрешения на рассылку. Подписки, отписки и запреты отправки храните в своей системе рассылок.

• При deliverable сохраните результат и freshness. Наблюдавшееся принятие адреса не гарантирует будущую доставку или попадание во входящие.
• При risky прочитайте action и reason_codes перед выбором правила. Одноразовая почта и catch-all могут требовать разных решений. Отметка ролевого адреса сама по себе не делает результат рискованным.
• При undeliverable исключите адрес из отправки, если это подтверждают свидетельства. Сохраните исходную строку и остальные данные клиента.
• При unknown держите неоднозначный вердикт отдельно от положительных и отрицательных групп. В нем могут остаться полезные сведения о домене. Решите, отложить адрес или проверить повторно. Если уместно письмо с подтверждением, его отправляет ваш процесс; проверка email писем не отправляет.

Правила отправки разобраны в руководстве по очистке списка. Для экспорта из CRM руководство по проверке email в CRM объясняет, как сохранить запись клиента при непригодном адресе. Руководство по результату проверки описывает поля и показывает, почему confidence не является вероятностью.

Проверьте выходной файл перед передачей

У каждой входной строки должно быть записанное состояние, в том числе у пустых адресов и запросов с ошибками. Сравните число строк и проверьте input_error, api_error, network_error, protocol_error и pending, прежде чем передавать файл другой системе.

Для окончательных результатов сохраните status, action, checks, reason_codes, confidence и freshness, если API их вернул. Архивируйте вместе исходный файл, идентификатор задачи, результат и checkpoint. Оставьте checkpoint, пока может понадобиться продолжение, и удаляйте только после надежного сохранения результата.

Начните с небольшого синтетического CSV. Сценарий массовой проверки поможет встроить задачу в ваш процесс. Он не обещает нулевых возвратов или доставки во входящие.