E-Mail-Liste aus einer CSV per API verifizieren

Um eine E-Mail-Liste per API zu prüfen, lesen Sie die CSV in einem eigenen Job ein, senden die Adressen von Ihrem Server und schreiben die Ergebnisse neben die Quellzeilen. Speichern Sie den Fortschritt während des Laufs: Eine Anfrage kann sofort fertig sein, als pending weiterlaufen oder ohne Urteil über die Adresse scheitern.

Diese Anleitung enthält ein Python-Skript für den Ablauf. Die Datei und der Job bleiben bei Ihnen; Emvero erhält eine Adresse pro Verifizierungsanfrage. Die Seite zur Massenprüfung beschreibt diesen Produktumfang. Einen gehosteten CSV-Uploader oder Batch-Endpunkt gibt es nicht.

Datei und API-Zugang vorbereiten

Die CSV benötigt eine Kopfzeile und eine Spalte email. Das folgende Beispiel ist synthetisch. Das Komma im quotierten Feld gehört zur Notiz; der letzten Zeile fehlt die Adresse:

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

Behalten Sie die Quelldatei und wählen Sie einen anderen Pfad für die Ausgabe. Das Skript erhält die Spalten und den ursprünglichen Adresswert. Nur bei der Kopie, die es sendet, entfernt es Leerraum an den Rändern. Es schreibt die Adresse weder klein noch anderweitig um. Eine leere Adresse wird als Eingabefehler an der Zeile festgehalten.

Sie benötigen Python 3, einen API-Zugang in EMVERO_API_TOKEN und Schreibrechte für Ausgabe und Checkpoint. Begrenzen Sie die parallelen Worker vorsichtig. Das Beispiel lädt die gesamte CSV in den Arbeitsspeicher. Passt die Datei dort nicht hinein, passen Sie den Job so an, dass er kleinere Gruppen einliest und speichert.

Die Anfragen gehen an https://api.emvero.tech/v1/verifications mit Authorization: Bearer. Führen Sie den Code lokal oder auf einem Server aus, damit der Zugang nicht im Browsercode landet. Den Anfragevertrag finden Sie bei der E-Mail-Verifizierungs-API, die Datenverarbeitung auf der Security-Seite.

Job und Zeilen eindeutig zuordnen

Das Skript bildet einen Hash aus den Bytes der Eingabedatei und dem Namen der gewählten E-Mail-Spalte. Diese Job-Kennung und die Zeilenposition ergeben emvero_row_id und einen stabilen Idempotency-Key. Unveränderte Eingaben erzeugen dieselben Schlüssel. Bei Änderungen, einer neuen Sortierung oder einer anderen Spalte entsteht ein neuer Job; alte Checkpoint-Einträge werden keiner neuen Zeile zugeordnet.

Innerhalb des unterstützten Idempotenzfensters ruft derselbe Schlüssel den ursprünglichen Vorgang ab, statt eine zweite Verifizierung anzulegen. Diese Anleitung setzt keine bestimmte Fensterdauer voraus. Das Skript speichert zusätzlich die ID jeder pending-Prüfung. Beim Fortsetzen fragt es diese ID ab, bevor es einen weiteren POST erwägt.

Jede Quellzeile ist ein eigener Eintrag im Job. Gleiche Adressen in verschiedenen Zeilen erhalten unterschiedliche Schlüssel; das Skript entfernt keine Duplikate. Wenn Ihr System bereits unveränderliche Datensatz- und Job-IDs hat, können Sie daraus eine eigene Zuordnung bilden.

Belege und Fehler getrennt speichern

Die Ausgabe erhält die Eingangsspalten und ergänzt die folgenden Felder. Strukturierte Objekte bleiben als kompaktes JSON in einer CSV-Zelle. So bleiben verschachtelte Angaben und Arrays mit Reason Codes beim Schreiben und Einlesen erhalten.

Akzeptiert das Zielsystem JSON oder Datenbankeinträge, speichern Sie die Objekte dort in ihrer ursprünglichen Form. Eine eigene Spalte für jedes heute bekannte Check-Feld kann spätere Ergänzungen unbemerkt verwerfen.

Python-Job starten

Speichern Sie das Skript als verify_csv.py. Der Python-Parser csv liest quotierte Felder und leere Zellen korrekt. Doppelte Spaltennamen und reservierte Ergebnisnamen weist das Skript vor der Verarbeitung zurück, damit keine Eingabewerte überschrieben werden.

Der Checkpoint ist eine JSON-Lines-Datei, die nur ergänzt wird. Beim Fortsetzen zählt der letzte Eintrag pro Zeile. Endgültige Ergebnisse werden wiederverwendet, laufende Prüfungen über ihre ID abgerufen. Ein fortgesetzter Job prüft also nicht alle fertigen Zeilen mit neuen Belegen.

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

Verwenden Sie verschiedene Pfade für Ein- und Ausgabe:

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

--workers begrenzt die gleichzeitig verarbeiteten Zeilen. --poll-seconds legt das Polling-Budget einer Zeile fest, --checkpoint kann den Checkpoint-Pfad vorgeben. Dieser Pfad muss von beiden CSV-Pfaden getrennt sein.

Bei 429, 500, 503 und Netzwerkfehlern versucht das Skript den Aufruf begrenzt oft erneut. Ist Retry-After länger als die lokale Wartezeit, gilt der Serverwert. Bleibt der Fehler bestehen, steht er neben der Zeile. Beheben Sie vor dem Fortsetzen das Zugangs-, Guthaben- oder Dienstproblem. Bei begrenzten Anfragen kann eine niedrigere Parallelität helfen.

Eine laufende Prüfung fortsetzen

POST /v1/verifications liefert HTTP 200 sowohl für ein endgültiges Ergebnis als auch für pending. Speichern Sie bei laufender Arbeit die id, warten Sie retry_after_ms ab und rufen Sie GET /v1/verifications/{id} auf. Ein noch nicht fertiger Aufruf rechtfertigt keinen neuen Vorgang.

Ist das Polling-Budget ausgeschöpft, schreibt das Skript client_poll_deadline und lässt emvero_status auf pending. Der Code beschreibt die Wartegrenze des Clients, nicht die Adresse. Starten Sie später denselben Befehl mit derselben Eingabe und demselben Checkpoint.

Der aktuelle Vertrag enthält keinen ausgehenden Callback. Ihr Job speichert die ID und ruft das Ergebnis ab. Ein Listenlauf kann anders mit dieser Wartezeit umgehen als eine Registrierung; der Vergleich von Echtzeit- und Massenprüfung erklärt die Wahl.

HTTP-Fehler als Betriebsproblem behandeln

Eine erfolgreiche HTTP-200-Antwort kann undeliverable enthalten: Die Prüfung hat ein Adressergebnis ermittelt. 401, 402, 403, 429, 500 und 503 sind dagegen Betriebsantworten. Schreiben Sie daraus weder undeliverable noch ein erfundenes unknown.

Anfragefehler stehen in emvero_error_http_status, emvero_error_code und emvero_error_message. Adressbelege gehören in emvero_status, emvero_action und die Ergebnisfelder. Beheben Sie die betriebliche Ursache und starten Sie denselben Job erneut. Der Checkpoint erhält fertige Ergebnisse; fehlerhafte Zeilen werden erneut versucht.

Listenregeln auf die gespeicherten Ergebnisse anwenden

Lesen Sie action als Empfehlung für den nächsten Schritt zusammen mit den Belegen. Die Empfehlungen von Emvero unterstützen Entscheidungen bei der Registrierung. Sie erlauben keinen Kampagnenversand. Abonnements, Abmeldungen und Versandsperren bleiben Aufgabe Ihres Versandsystems.

• Bei deliverable erhalten Sie Ergebnis und freshness. Beobachtete Annahme garantiert weder künftige Zustellung noch einen Platz im Posteingang.
• Bei risky prüfen Sie action und reason_codes vor der Entscheidung. Wegwerf- und Catch-all-Belege können unterschiedliche Regeln brauchen. Die Kennzeichnung als Rollenpostfach allein macht ein Ergebnis nicht riskant.
• Bei undeliverable nehmen Sie die Adresse aus dem Versand, sofern die Belege das tragen. Die Quellzeile und andere Kundendaten bleiben erhalten.
• Bei unknown bleibt das ungeklärte Urteil von positiven und negativen Gruppen getrennt. Nützliche Domain-Belege können trotzdem vorliegen. Entscheiden Sie über Zurückhalten oder erneutes Prüfen. Eine passende Bestätigungs-E-Mail gehört zu Ihrem eigenen Ablauf; die Verifizierung versendet keine Nachricht.

Der Leitfaden zur Listenbereinigung behandelt Versandregeln. Bei einem CRM-Export erklärt der Leitfaden zur Prüfung von CRM-E-Mail-Adressen, wie Sie den Kundendatensatz trotz unbrauchbarem E-Mail-Feld erhalten. Der Leitfaden zum Prüfergebnis erklärt die Felder und warum confidence keine Wahrscheinlichkeit ist.

Ausgabe vor der Übergabe prüfen

Jede Eingabezeile braucht einen festgehaltenen Zustand, auch bei leerer Adresse oder fehlgeschlagener Anfrage. Vergleichen Sie die Zeilenzahlen und prüfen Sie input_error, api_error, network_error, protocol_error sowie pending, bevor ein anderes System die Ausgabe nutzt.

Erhalten Sie bei endgültigen Zeilen status, action, checks, reason_codes, confidence und freshness, soweit die API sie liefert. Archivieren Sie Quelldatei, Job-Kennung, Ausgabe und Checkpoint zusammen. Behalten Sie den Checkpoint, solange Sie fortsetzen müssen; entfernen Sie ihn erst nach sicherer Speicherung der Ausgabe.

Beginnen Sie mit der kleinen synthetischen CSV. Die Massenprüfung hilft Ihnen, den Job in Ihren Ablauf einzuordnen. Sie verspricht weder null Bounces noch erfolgreiche Zustellung im Posteingang.