---
title: "Проверка списка email из CSV через API: Python | Emvero"
description: "Проверьте список email из CSV через API с помощью Python. Сохраните исходные строки, продолжите pending-проверки и запишите результаты отдельно от ошибок."
url: https://emvero.io/ru/blog/validate-email-list-csv-api/
language: ru
updated: 2026-10-10
---

Блог

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

Скрипт безопасно читает CSV, отправляет по одному серверному запросу на строку и записывает статус, действие, коды причин и ошибки, не теряя исходную запись.

Опубликовано 2026-09-11

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

Ниже приведен Python-скрипт для такой задачи. Файл и запуск остаются у вас; Emvero получает один адрес на запрос проверки. Страница [массовой проверки](https://emvero.io/ru/bulk-email-verification/) описывает этот сценарий сервиса. Загрузчика 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](https://emvero.io/ru/email-verification-api/), обработка данных — на странице [Безопасность](https://emvero.io/ru/security/).

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

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

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

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

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

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

| Столбец | Что в нем хранится |
| --- | --- |
| `emvero_row_id` | Отпечаток входных данных и позиция строки |
| `emvero_submitted_email` | Отправленная в API копия адреса без пробелов по краям |
| `emvero_state` | Окончательный результат, `pending` или ошибка входных данных, API, сети либо протокола |
| `emvero_verification_id` | ID для получения еще не завершенной проверки |
| `emvero_status` | `deliverable`, `risky`, `undeliverable`, `unknown` или транспортное состояние `pending` |
| `emvero_action` | `allow`, `allow_with_email_confirmation` или `reject` |
| `emvero_checks` | Полный объект `checks` в JSON |
| `emvero_reason_codes` | Полный массив `reason_codes` в JSON |
| `emvero_confidence` | Оценка силы свидетельств из API, а не вероятность доставки |
| `emvero_freshness` | Моменты наблюдения DNS и SMTP из ответа API |
| `emvero_normalized` | Формы домена из объекта `normalized` |
| `emvero_error_*` | Ошибки запроса или сервиса отдельно от вердикта об адресе |

Если следующая система принимает 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 и сама получать результат. В работе со списком ожидание можно организовать иначе, чем в форме регистрации; это разбирает [сравнение проверки в реальном времени и массовой проверки](https://emvero.io/ru/compare/real-time-vs-bulk/).

## Считайте 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 писем не отправляет.

Правила отправки разобраны в [руководстве по очистке списка](https://emvero.io/ru/guides/cleaning-an-email-list/). Для экспорта из CRM [руководство по проверке email в CRM](https://emvero.io/ru/guides/crm-data-cleaning/) объясняет, как сохранить запись клиента при непригодном адресе. [Руководство по результату проверки](https://emvero.io/ru/blog/reading-a-verification-result/) описывает поля и показывает, почему `confidence` не является вероятностью.

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

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

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

Начните с небольшого синтетического CSV. [Сценарий массовой проверки](https://emvero.io/ru/bulk-email-verification/) поможет встроить задачу в ваш процесс. Он не обещает нулевых возвратов или доставки во входящие.

## Термины, которые здесь встречаются

### Жёсткий и мягкий возврат

Жесткий возврат обычно связан с постоянным отказом доставки, мягкий — с временным. Причину нужно читать в ответе сервера.

### Catch-all-домен

Почтовый сервер принимает произвольных получателей в домене. Положительный ответ не подтверждает отдельный ящик.

### Троттлинг и лимиты

Ограничение частоты обращений к почтовому серверу. Оно может помешать проверке, но не определяет существование ящика.

## Ситуации, к которым это относится

### Чистка списка перед рассылкой

Вы защищаете репутацию отправляющего домена, поэтому вопрос не в том, какие адреса плохие, а в том, какие эта рассылка может себе позволить.

### Чистка CRM без удаления хороших записей

Запись в CRM это история, а не адрес, поэтому здесь ничего не удаляется: проверка решает, кому вы можете писать, но никогда - кого вы можете хранить.

[Все статьи](https://emvero.io/ru/blog/)

## Выберите, что проверить дальше.

Запустите проверку в личном кабинете и посмотрите данные в результате.

Начать бесплатно