Как проверить базу email-адресов на валидность самому, без сторонних сервисов

[Itnull]

Как проверить базу email-адресов на валидность самому, без сторонних сервисов​

Если у вас есть база адресов — пользователи сайта, подписчики, клиенты — в ней со временем копится мусор: опечатки при регистрации, мёртвые домены, удалённые ящики. Рассылка по такой базе бьёт по репутации почтового сервера: много возвратов (bounce) — и ваши письма начинают уходить в спам даже у живых получателей.

Платные сервисы проверки (ZeroBounce, NeverBounce, Hunter и т. п.) решают задачу, но стоят денег, а главное — вы отдаёте им свою базу. В этой статье разберём, как сделать то же самое своими силами: на Python, за один вечер, без передачи данных третьим лицам.

Как вообще проверяют email​

Все сервисы проверки делают одно и то же, просто в несколько этапов:

1. Синтаксис. Адрес вида user@@mail или a@b отсеивается сразу.
2. DNS. Существует ли домен и есть ли у него MX-запись — то есть почтовый сервер, который принимает письма. Это ловит опечатки вроде gmial.com, mail.ry и домены, которые давно не продлевали.
3. SMTP. Подключаемся к почтовому серверу домена и «начинаем» отправку: EHLO → MAIL FROM → RCPT TO:<адрес>. Сервер отвечает, есть ли такой ящик (код 250 — есть, 550 — нет). На этом соединение закрывается, письмо не отправляется.
4. Catch-all. Некоторые серверы говорят «да» на любой адрес. Чтобы это поймать, перед проверкой спрашиваем заведомо несуществующий случайный адрес: если сервер его «принял» — домен catch-all, и подтвердить конкретный ящик невозможно.

Первые два этапа безопасны и работают откуда угодно. Третий требует подготовки — о подводных камнях ниже.

Скрипт​

Скрипт ниже делает все четыре этапа. Что в нём есть:

• читает адреса из .txt или .csv в любом виде — сам вытаскивает их из текста и убирает дубли;
• группирует адреса по доменам и проверяет по одной SMTP-сессии на домен, а не по соединению на каждый адрес;
• кэширует DNS-ответы, проверяет несколько доменов параллельно;
• делает паузы между запросами и переподключается при обрывах;
• отличает «ящика нет» от «нас заблокировали» — второе не считается невалидным адресом;
• перед стартом проверяет, работает ли DNS и открыт ли исходящий порт 25, чтобы не получить базу, целиком помеченную как мёртвую;
• пишет результат в CSV (открывается в Excel, кириллица не ломается).

Единственная зависимость — библиотека dnspython.

Спойлер: email_check.py

#!/usr/bin/env python3
"""
email_check.py — проверка базы email-адресов без сторонних сервисов.

Этапы проверки:
  1. Синтаксис адреса.
  2. DNS: существует ли домен и есть ли у него MX-записи (почтовые серверы).
  3. SMTP: подключение к почтовому серверу домена и команда RCPT TO
     (сервер отвечает, есть ли такой ящик). Письмо НЕ отправляется.
  4. Проверка catch-all: принимает ли сервер вообще любые адреса.

Установка:
  pip install dnspython

Запуск:
  python email_check.py emails.txt -o result.csv \
      --helo mail.вашдомен.ru --from check@вашдомен.ru

  Только синтаксис + DNS (быстро, без SMTP):
  python email_check.py emails.txt -o result.csv --dns-only

Входной файл: .txt или .csv — скрипт сам вытащит из него все адреса.

Статусы в результате:
  valid            — сервер подтвердил, что ящик существует
  invalid          — сервер ответил, что такого ящика нет
  catch_all        — сервер принимает любые адреса, проверить нельзя
  unknown          — временная ошибка, greylisting, таймаут, блокировка
  blocked          — сервер отказал из-за политики/репутации вашего IP
  invalid_syntax   — адрес написан с ошибкой
  domain_not_found — домена не существует
  no_mx            — у домена нет почтового сервера
  dns_error        — не удалось получить DNS-ответ
  has_mx           — (режим --dns-only) домен принимает почту
"""
import argparse
import csv
import random
import re
import smtplib
import socket
import string
import sys
import threading
import time
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor, as_completed

try:
    import dns.resolver
except ImportError:
    sys.exit("Нужен пакет dnspython: pip install dnspython")

EMAIL_RE = re.compile(
    r"^[a-z0-9.!#$%&'*+/=?^_`{|}~-]+@([a-z0-9]([a-z0-9-]*[a-z0-9])?\.)+[a-z]{2,}$"
)
BLOCK_WORDS = ("block", "blacklist", "spam", "policy", "reputation",
               "spamhaus", "denied", "rbl", "banned", "not allowed")

_mx_cache = {}
_mx_lock = threading.Lock()


def load_emails(path):
    emails, seen = [], set()
    with open(path, encoding="utf-8", errors="ignore") as f:
        for line in f:
            for token in re.split(r"[,;\s]+", line):
                e = token.strip().strip("\"'<>()[]").lower()
                if "@" in e and e not in seen:
                    seen.add(e)
                    emails.append(e)
    return emails


def get_mx(domain):
    """Возвращает (статус, [mx-хосты по приоритету])."""
    with _mx_lock:
        if domain in _mx_cache:
            return _mx_cache[domain]
    try:
        ans = dns.resolver.resolve(domain, "MX", lifetime=10)
        hosts = [str(r.exchange).rstrip(".")
                 for r in sorted(ans, key=lambda r: r.preference)]
        hosts = [h for h in hosts if h]          # "Null MX" = домен не принимает почту
        result = ("ok", hosts) if hosts else ("no_mx", [])
    except dns.resolver.NXDOMAIN:
        result = ("domain_not_found", [])
    except dns.resolver.NoAnswer:
        # MX нет — по стандарту почта идёт на A-запись домена
        try:
            dns.resolver.resolve(domain, "A", lifetime=10)
            result = ("ok", [domain])
        except Exception:
            result = ("no_mx", [])
    except Exception:
        result = ("dns_error", [])
    with _mx_lock:
        _mx_cache[domain] = result
    return result


def is_block(text):
    t = text.lower()
    return "5.7." in t or any(w in t for w in BLOCK_WORDS)


class Blocked(Exception):
    pass


def smtp_session(host, domain, pending, out, args):
    """Проверяет адреса из pending (удаляя проверенные). При обрыве бросает исключение."""
    s = smtplib.SMTP(host, 25, local_hostname=args.helo, timeout=args.timeout)
    try:
        s.ehlo()
        code, msg = s.mail(args.mail_from)
        if code != 250:
            text = msg.decode(errors="ignore")
            if is_block(text):
                raise Blocked(f"MAIL FROM: {code} {text[:150]}")
            raise smtplib.SMTPException(f"MAIL FROM: {code} {text[:150]}")

        # Проверка catch-all случайным несуществующим адресом
        rnd = "".join(random.choices(string.ascii_lowercase + string.digits, k=18))
        code, msg = s.rcpt(f"{rnd}@{domain}")
        text = msg.decode(errors="ignore")
        if 500 <= code < 600 and is_block(text):
            raise Blocked(f"{code} {text[:150]}")
        catch_all = code in (250, 251)

        n = 0
        while pending:
            addr = pending[0]
            code, msg = s.rcpt(addr)
            text = msg.decode(errors="ignore").replace("\n", " ")[:150]
            if code == 421:
                raise smtplib.SMTPServerDisconnected(f"421 {text}")
            if code in (250, 251):
                out[addr] = ("catch_all" if catch_all else "valid", text)
            elif 500 <= code < 600:
                if is_block(text):
                    raise Blocked(f"{code} {text}")
                out[addr] = ("invalid", f"{code} {text}")
            else:
                out[addr] = ("unknown", f"{code} {text}")   # 4xx: greylisting/лимиты
            pending.pop(0)
            n += 1
            if pending and n % args.per_session == 0:
                s.rset()
                s.mail(args.mail_from)
            time.sleep(args.delay)
    finally:
        try:
            s.quit()
        except Exception:
            pass


def check_domain(domain, addrs, args):
    out = {}
    status, hosts = get_mx(domain)
    if status != "ok":
        return {a: (status, "") for a in addrs}
    if args.dns_only:
        return {a: ("has_mx", hosts[0]) for a in addrs}

    pending = list(addrs)
    last_err = ""
    attempts = 0
    for host in (hosts[:2] * 2):          # до 2 MX, по 2 попытки
        if not pending or attempts >= 4:
            break
        attempts += 1
        try:
            smtp_session(host, domain, pending, out, args)
        except Blocked as e:
            for a in pending:
                out[a] = ("blocked", str(e)[:200])
            return out
        except Exception as e:
            last_err = f"{host}: {type(e).__name__}: {e}"[:200]
            time.sleep(2)
    for a in pending:
        out[a] = ("unknown", last_err or "нет ответа")
    return out


def port25_open():
    try:
        with socket.create_connection(("gmail-smtp-in.l.google.com", 25), timeout=8):
            return True
    except OSError:
        return False


def main():
    p = argparse.ArgumentParser(description="Проверка email-адресов без сторонних сервисов")
    p.add_argument("input", help="файл с адресами (.txt/.csv)")
    p.add_argument("-o", "--output", default="result.csv")
    p.add_argument("--helo", default=socket.getfqdn(),
                   help="имя сервера для EHLO (должно совпадать с PTR-записью вашего IP)")
    p.add_argument("--from", dest="mail_from", default="",
                   help="адрес для MAIL FROM, на вашем реальном домене")
    p.add_argument("--dns-only", action="store_true", help="только синтаксис и DNS")
    p.add_argument("--workers", type=int, default=8, help="параллельных доменов")
    p.add_argument("--delay", type=float, default=1.0, help="пауза между RCPT, сек")
    p.add_argument("--per-session", type=int, default=20, help="адресов на одну сессию")
    p.add_argument("--timeout", type=float, default=20)
    args = p.parse_args()

    emails = load_emails(args.input)
    print(f"Найдено уникальных адресов: {len(emails)}")

    results = {}
    by_domain = defaultdict(list)
    for e in emails:
        if EMAIL_RE.match(e):
            by_domain[e.rsplit("@", 1)[1]].append(e)
        else:
            results[e] = ("invalid_syntax", "")

    if not args.dns_only:
        if not args.mail_from:
            sys.exit("Укажите --from адрес на вашем домене (или используйте --dns-only)")
        print("Проверяю исходящий порт 25...")
        if not port25_open():
            sys.exit("Порт 25 закрыт: провайдер/хостинг блокирует исходящий SMTP. "
                     "Запустите на сервере с открытым портом 25 или используйте --dns-only.")

    # Защита от ложных "domain_not_found", если на машине не работает DNS
    if get_mx("gmail.com")[0] != "ok":
        sys.exit("DNS не работает: не удалось получить MX для gmail.com. "
                 "Проверьте интернет/DNS и запустите снова.")
    _mx_cache.pop("gmail.com", None)

    print(f"Доменов: {len(by_domain)}. Проверка...")
    done = 0
    with ThreadPoolExecutor(max_workers=args.workers) as ex:
        futures = {ex.submit(check_domain, d, a, args): d for d, a in by_domain.items()}
        for f in as_completed(futures):
            d = futures[f]
            try:
                results.update(f.result())
            except Exception as e:
                for a in by_domain[d]:
                    results[a] = ("unknown", f"ошибка: {e}"[:200])
            done += 1
            if done % 10 == 0 or done == len(futures):
                print(f"  доменов проверено: {done}/{len(futures)}")

    with open(args.output, "w", newline="", encoding="utf-8-sig") as f:
        w = csv.writer(f, delimiter=";")
        w.writerow(["email", "status", "detail"])
        for e in emails:
            st, det = results.get(e, ("unknown", ""))
            w.writerow([e, st, det])

    stats = defaultdict(int)
    for st, _ in results.values():
        stats[st] += 1
    print("\nИтого:")
    for st, n in sorted(stats.items(), key=lambda x: -x[1]):
        print(f"  {st:17} {n}")
    print(f"\nРезультат: {args.output}")


if __name__ == "__main__":
    main()

Запуск на Linux / FreeBSD​

Установка зависимостей.

Debian / Ubuntu:

apt install python3 python3-dnspython

FreeBSD:

pkg install python311 py311-dnspython

(на FreeBSD команда называется по версии — python3.11, python3.12 и т. д.)

Или универсально через pip:

pip install dnspython

Сначала — быстрая проверка синтаксиса и доменов:

python3 email_check.py emails.txt -o result.csv --dns-only

Большая база проверяется долго, поэтому на сервере удобнее запускать в фоне, чтобы проверка не оборвалась при отключении SSH:

nohup python3 -u email_check.py emails.txt -o result.csv --dns-only > check.log 2>&1 &
tail -f check.log

Ключ -u нужен, чтобы прогресс сразу писался в лог. Выход из tail — Ctrl+C, сама проверка продолжит работать.

Полная проверка с опросом почтовых серверов:

nohup python3 -u email_check.py emails.txt -o result.csv \
  --helo mail.вашдомен.ru --from check@вашдомен.ru > check.log 2>&1 &

Запуск на Windows​

1. Установите Python с python.org — в установщике отметьте Add python.exe to PATH.
2. В командной строке: pip install dnspython
3. Положите скрипт и файл с адресами в одну папку, перейдите в неё (cd C:\emails) и запустите:

python email_check.py emails.txt -o result.csv --dns-only

Не закрывайте окно и не давайте компьютеру уснуть, пока идёт проверка.

Важно: с домашнего компьютера полная SMTP-проверка почти наверняка не заработает — об этом следующий раздел. Режим --dns-only работает везде.

Подводные камни SMTP-проверки​

1. Порт 25. Домашние провайдеры и большинство облаков (AWS, Google Cloud, многие VPS) закрывают исходящий 25-й порт — против спама. Проверить:

# Linux / FreeBSD
nc -vz -w 8 gmail-smtp-in.l.google.com 25

# Windows (PowerShell)
Test-NetConnection gmail-smtp-in.l.google.com -Port 25

Если соединения нет — нужен сервер, где хостер открыл порт (у многих — по заявке в поддержку). Скрипт сам проверяет это перед стартом.

2. Репутация IP. Почтовые серверы смотрят, кто к ним пришёл. Нужны:

• обратная DNS-запись (PTR) для вашего IP — настраивается у хостера;
• --helo — то же имя, что в PTR;
• --from — адрес на вашем реальном домене;
• IP не должен быть в чёрных списках.

Без этого серверы будут отвечать отказом — скрипт пометит такие адреса как blocked, а не как невалидные.

3. Крупные почтовики не любят, когда их «простукивают». Gmail, Mail.ru, Yandex, Rambler ограничивают частоту запросов, используют greylisting (временный отказ «попробуйте позже»), часть доменов работает как catch-all. Поэтому по умолчанию в скрипте пауза 1 секунда между адресами — ускорять не советую.

4. Не проверяйте с боевого почтового сервера. Массовые RCPT-запросы с одного IP — верный путь в чёрные списки. Если с сервера уходит почта вашего сайта, запускайте проверку с другого IP.

Что означают статусы​

Статус	Значение	Что делать
valid	Сервер подтвердил, что ящик существует	Оставить
has_mx	(режим --dns-only) Домен принимает почту	Оставить, при желании проверить SMTP
catch_all	Сервер принимает любые адреса	Оставить, проверить нельзя
invalid	Сервер ответил: такого ящика нет	Убрать из рассылок
domain_not_found	Домена не существует	Убрать или исправить опечатку
no_mx	Домен есть, но почту не принимает	Убрать
invalid_syntax	Адрес написан с ошибкой	Убрать
unknown	Временная ошибка, таймаут, greylisting	Перепроверить позже
blocked	Сервер отказал из-за репутации вашего IP	Чинить PTR / IP, перепроверить
dns_error	DNS не ответил вовремя	Перепроверить

Пример из практики​

Прогон --dns-only по реальной базе пользователей примерно в 50 тысяч адресов:

Итого:
  has_mx            49034
  domain_not_found  2206
  no_mx             427
  dns_error         94
  invalid_syntax    7

То есть около 5% адресов мертвы ещё до всякой SMTP-проверки — и это без единого запроса к почтовым серверам.

Разложить результат по файлам:

grep ';has_mx;' result.csv | cut -d';' -f1 > good.txt
grep -E ';(domain_not_found|no_mx|invalid_syntax);' result.csv | cut -d';' -f1 > bad.txt
grep ';dns_error;' result.csv | cut -d';' -f1 > retry.txt

Адреса с dns_error стоит прогнать ещё раз — обычно почти все оказываются рабочими:

python3 email_check.py retry.txt -o retry_result.csv --dns-only

А вот эта команда покажет, какие домены чаще всего встречаются среди мёртвых:

cut -d'@' -f2 bad.txt | sort | uniq -c | sort -rn | head -30

Как правило, наверху окажутся опечатки популярных доменов — gmai.com, yandex.ry, mail.ru.ru. Такие адреса не обязательно выбрасывать: часть можно исправить и вернуть живых пользователей.

Рекомендуемый порядок​

1. Прогнать всю базу в режиме --dns-only — быстро и безопасно, убирает явный мусор.
2. Перепроверить dns_error, посмотреть на опечатки в доменах.
3. Мёртвые адреса не удалять вместе с аккаунтами, а пометить в своей системе как «недоставляемые», чтобы на них просто перестали уходить письма.
4. Полную SMTP-проверку запускать только по оставшимся адресам, порциями, с подготовленного сервера.

Вопросы и результаты своих прогонов — пишите в теме.