Cómo verificar una lista de correo desde CSV con una API

Para verificar una lista de correo con una API, ejecute un proceso que lea el CSV, envíe cada dirección desde el servidor y escriba el resultado junto a la fila original. Guarde el progreso durante la ejecución: una petición puede terminar al momento, seguir en pending o fallar sin producir un veredicto sobre la dirección.

Esta guía incluye un script en Python para hacerlo. Usted conserva el archivo y gestiona la ejecución; Emvero recibe una dirección por petición de verificación. La página de verificación masiva explica este alcance del servicio. No hay un cargador de CSV alojado ni un endpoint por lotes.

Prepare el archivo y la credencial

El CSV necesita una cabecera y una columna email. El ejemplo es sintético: la coma entre comillas pertenece a la nota y la última fila no tiene dirección.

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

Conserve el archivo de origen y elija otra ruta para la salida. El script mantiene las columnas y el valor original de la dirección. Solo elimina el espacio de los extremos de la copia que envía, sin convertirla a minúsculas ni reescribirla. Una dirección vacía genera un error de entrada asociado a su fila.

Hace falta Python 3, una credencial en EMVERO_API_TOKEN y permiso para crear la salida y el archivo de checkpoint. Elija un límite prudente de procesos de trabajo paralelos. Este ejemplo carga el CSV completo en memoria; si el archivo no cabe, adapte el proceso para leer y guardar grupos más pequeños.

Las peticiones van a https://api.emvero.tech/v1/verifications con Authorization: Bearer. Ejecute el código de forma local o en un servidor para mantener la credencial fuera del navegador. La API de verificación de correo describe la petición y la página de seguridad explica el tratamiento de los datos.

Mantenga una identidad estable para cada fila

El script calcula un hash de los bytes del archivo de entrada y del nombre de la columna de correo elegida. Ese identificador del proceso y la posición de la fila forman emvero_row_id y una Idempotency-Key estable. Si la entrada no cambia, las claves tampoco. Editar o reordenar el archivo, o elegir otra columna, crea un proceso distinto: las entradas del checkpoint anterior no se asignan a filas nuevas.

Dentro del intervalo de idempotencia admitido, repetir una clave recupera la operación original en vez de crear otra verificación. Esta guía no depende de una duración concreta del intervalo. El script también guarda el ID de cada verificación pending. Al reanudar, recupera ese ID antes de plantear otro POST.

Cada fila es un elemento independiente. Dos filas con la misma dirección reciben claves distintas; el script no elimina duplicados. Si su sistema ya tiene identificadores inmutables de registros y procesos, puede usarlos para definir su propia correspondencia.

Escriba evidencia y errores en columnas separadas

La salida mantiene las columnas de entrada y añade las siguientes. Los objetos estructurados se guardan como JSON compacto dentro de una celda, para conservar campos anidados y arrays de códigos de motivo al escribir y volver a leer el CSV.

Si el siguiente sistema acepta JSON o registros de base de datos, conserve los objetos en su forma nativa. Crear una columna por cada campo conocido hoy puede descartar sin aviso los que se añadan después.

Ejecute el proceso en Python

Guarde el código como verify_csv.py. El parser csv de Python interpreta los campos entre comillas y las celdas vacías. El script rechaza las cabeceras duplicadas y los nombres reservados para resultados antes de procesar, para no sobrescribir los valores de entrada.

El checkpoint es un archivo JSON Lines al que solo se añaden eventos. Al reanudar, prevalece el último de cada fila. Los resultados terminales se reutilizan y los pendientes se recuperan por su ID. Retomar el proceso no vuelve a verificar todas las filas terminadas con evidencia nueva.

#!/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()

Use rutas distintas para entrada y salida:

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

--workers limita cuántas filas se procesan a la vez. --poll-seconds establece el plazo de consulta por fila y --checkpoint permite elegir el archivo de progreso. Su ruta debe ser distinta de las dos rutas CSV.

El script limita los reintentos ante 429, 500, 503 y fallos de red. Respeta un Retry-After del servidor que supere la espera local. Si el error persiste, lo escribe junto a la fila. Antes de reanudar, resuelva la credencial, el saldo o el problema del servicio. Si se limitan las peticiones, reduzca la concurrencia.

Retome la misma verificación pendiente

POST /v1/verifications devuelve HTTP 200 tanto para un resultado terminal como para pending. Guarde el id de la verificación pendiente, espere retry_after_ms y llame a GET /v1/verifications/{id}. Que falte un resultado terminado no justifica crear otra operación.

Si se agota el plazo de consulta del script, registra client_poll_deadline y mantiene emvero_status en pending. Ese código describe el límite de espera del cliente, no la dirección. Más tarde, ejecute el mismo comando con la misma entrada y el mismo checkpoint.

El contrato actual no incluye callbacks salientes. El proceso debe guardar el ID y recuperar el resultado. Una lista puede acomodar esa espera de otra forma que una petición de registro; la comparación entre verificación en tiempo real y masiva explica la elección.

Interprete un error HTTP como un problema operativo

Una respuesta HTTP 200 correcta puede contener undeliverable: la verificación produjo un resultado sobre la dirección. 401, 402, 403, 429, 500 y 503 son respuestas operativas. No las convierta en undeliverable ni invente un resultado unknown.

Los errores de petición aparecen en emvero_error_http_status, emvero_error_code y emvero_error_message. La evidencia de la dirección pertenece a emvero_status, emvero_action y los campos de resultado. Tras resolver la causa, vuelva a ejecutar el mismo proceso. El checkpoint conserva los resultados terminados y se reintentan las filas con errores.

Aplique su política de lista a los resultados guardados

Lea action como recomendación para el siguiente paso, junto a la evidencia que la explica. Las recomendaciones de Emvero sirven para decisiones de registro; no autorizan el envío de campañas. Mantenga suscripciones, bajas y reglas de supresión en el sistema de envío.

• Con deliverable, conserve el resultado y freshness. La aceptación observada no garantiza la entrega futura ni la llegada a la bandeja de entrada.
• Con risky, lea action y reason_codes antes de elegir una regla. La evidencia de correo desechable y de catch-all puede exigir decisiones distintas. La anotación de dirección genérica por sí sola no convierte el resultado en arriesgado.
• Con undeliverable, excluya la dirección del envío cuando la evidencia lo respalde. Conserve la fila y el resto de los datos del cliente.
• Con unknown, separe el veredicto inconcluso de los grupos positivos y negativos. Puede contener evidencia útil sobre el dominio. Decida si retener o volver a comprobar la dirección. Si corresponde enviar una confirmación, debe hacerlo su propio proceso; la verificación no envía mensajes.

La guía de limpieza de listas desarrolla las reglas de envío. Para una exportación del CRM, la guía de validación del correo en CRM explica cómo conservar el registro aunque el campo de correo no sirva. La guía del resultado de verificación explica los campos y por qué confidence no es una probabilidad.

Revise la salida antes de entregarla

Cada fila de entrada necesita un estado registrado, también las direcciones vacías y las peticiones fallidas. Compare el número de filas y revise input_error, api_error, network_error, protocol_error y pending antes de pasar la salida a otro sistema.

En las filas terminales, conserve status, action, checks, reason_codes, confidence y freshness cuando la API los devuelva. Archive juntos la fuente, el identificador del proceso, la salida y el checkpoint. Mantenga el checkpoint mientras pueda necesitar reanudar; elimínelo solo cuando la salida esté guardada de forma segura.

Empiece con el pequeño CSV sintético. El proceso de verificación masiva ayuda a encajar este trabajo en su sistema. No puede prometer cero rebotes ni entrega en la bandeja de entrada.