Z User
v3.1.0: Part B architecture migration — eliminate Hugging Face inference credit dependency
6c516b6 | import asyncio | |
| import re | |
| import time | |
| import uuid | |
| from datetime import datetime, timezone | |
| from typing import Any, Optional | |
| import structlog | |
| from ..models import ( | |
| ValidationResult, | |
| ValidationStatus, | |
| ScoringResult, | |
| CatchAllResult, | |
| DNSResult, | |
| DisposableResult, | |
| SMTPResult, | |
| SyntaxResult, | |
| ) | |
| from ..config import settings | |
| from .syntax import validate_syntax | |
| from .dns_check import check_dns | |
| from .disposable import check_disposable, domain_in_blocklist, domain_is_known | |
| from .mailbox_provider import MailboxProvider, resolve_provider | |
| from .i18n import EAI_REASON, eai_gate | |
| from ..ml.risk_scorer import AIRiskScorer, ai_scorer, label_for_score, score_many_by_domain | |
| from ..ml.runtime import get_model_async, predict_async | |
| from .mx_intel import mx_is_trusted | |
| from .retry_queue import is_greylist_result | |
| from .scoring import calculate_score | |
| logger = structlog.get_logger() | |
| def _mask_email_for_log(email: str) -> str: | |
| local_part, separator, domain = email.rpartition("@") | |
| if not separator or not domain: | |
| return "[invalid-email]" | |
| return f"{local_part[:1]}***@{domain}" | |
| def _build_checks_run( | |
| syntax: SyntaxResult | None, | |
| dns: DNSResult | None, | |
| disposable: DisposableResult | None, | |
| catch_all: CatchAllResult | None, | |
| smtp: SMTPResult | None, | |
| scoring: ScoringResult | None, | |
| ) -> list[str]: | |
| """Which of the 7-layer taxonomy keys actually executed for this result. | |
| Mirrors the frontend fallback in src/lib/verifier.ts classify() so a layer | |
| the engine did not run is never reported as a pass. DNS without an error | |
| claims both `dns` and `mx`; a DNS error claims only `dns`. | |
| """ | |
| checks: list[str] = [] | |
| if syntax is not None: | |
| checks.append("syntax") | |
| if dns is not None: | |
| if dns.error: | |
| checks.append("dns") | |
| else: | |
| checks.extend(["dns", "mx"]) | |
| if disposable is not None: | |
| checks.append("disposable") | |
| if catch_all is not None: | |
| checks.append("catch_all") | |
| if smtp is not None: | |
| checks.append("smtp") | |
| if scoring is not None: | |
| checks.append("scoring") | |
| return checks | |
| def _build_reason_code( | |
| dns: DNSResult | None, | |
| disposable: DisposableResult | None, | |
| catch_all: CatchAllResult | None, | |
| smtp: SMTPResult | None, | |
| *, | |
| early: str | None = None, | |
| ) -> str | None: | |
| """Machine cause for the verdict. Ordered most-specific/actionable first. | |
| disposable outranks everything (frontend classifies it as Disposable). A | |
| definitive SMTP rejection (verified False + code) outranks the domain-level | |
| catch-all flag: reporting "catch_all_detected" beside status=invalid is | |
| incoherent — the explicit "mailbox does not exist" is the actionable fact. | |
| Catch-all then outranks a merely inconclusive probe. | |
| """ | |
| if early: | |
| return early | |
| if disposable is not None and disposable.is_disposable: | |
| return "disposable" | |
| if dns is not None and dns.a_fallback and not dns.has_mx_records: | |
| return "a_fallback_only" | |
| if smtp is not None and smtp.verified is False and smtp.smtp_code: | |
| if smtp.smtp_code >= 500: | |
| return "rcpt_5xx" | |
| if smtp.smtp_code >= 400: | |
| return "rcpt_4xx" | |
| if smtp.smtp_code >= 200: | |
| return "rcpt_2xx" | |
| if catch_all is not None and catch_all.is_catch_all: | |
| return "catch_all_detected" | |
| if smtp is not None and smtp.verified is None and smtp.error: | |
| return ( | |
| "smtp_timeout" | |
| if re.search(r"timed?\s*out", smtp.error, re.IGNORECASE) | |
| else "smtp_unavailable" | |
| ) | |
| return None | |
| def _deterministic_verdict( | |
| disposable: DisposableResult | None, | |
| catch_all: CatchAllResult | None, | |
| smtp: SMTPResult | None, | |
| ) -> str: | |
| """Pre-AI verdict of the deterministic gauntlet (spec Phase 5). | |
| Order matters: an explicit mailbox rejection outranks the domain-level | |
| catch-all flag (mirrors _build_reason_code), and a catch-all NEVER reads | |
| "valid" even when the probe was accepted — it only proves the domain | |
| accepts everything. | |
| """ | |
| if disposable is not None and disposable.is_disposable: | |
| return "disposable" | |
| if ( | |
| smtp is not None | |
| and smtp.verified is False | |
| and smtp.smtp_code is not None | |
| and smtp.smtp_code >= 500 | |
| ): | |
| return "invalid" | |
| if catch_all is not None and catch_all.is_catch_all: | |
| return "catch_all" | |
| if smtp is not None and smtp.verified is True: | |
| return "valid" | |
| return "unknown" | |
| def _smtp_class(smtp: SMTPResult | None) -> str: | |
| """Coarse SMTP response class for the AI feature string (no PII).""" | |
| if smtp is None: | |
| return "skipped" | |
| if smtp.verified is True: | |
| return "accept" | |
| if smtp.verified is False: | |
| if smtp.smtp_code and smtp.smtp_code >= 500: | |
| return "5xx" | |
| if smtp.smtp_code and smtp.smtp_code >= 400: | |
| return "4xx" | |
| return "reject" | |
| err = (smtp.error or "").lower() | |
| if "worker" in err or "unreachable" in err: | |
| return "transport_error" | |
| return "unavailable" | |
| def _ambiguous(verdict: str) -> bool: | |
| """Only these two reach the AI scorer (spec gate steps 6 and 8).""" | |
| return verdict in {"catch_all", "unknown"} | |
| class EmailValidationEngine: | |
| """ | |
| Master 7-Layer Email Validation Orchestrator. | |
| Layers: | |
| 1. Syntax / RFC 5321 check | |
| 2. DNS domain existence | |
| 3. MX record verification | |
| 4. Disposable / Role-based / Free provider detection | |
| 5. Catch-all domain detection | |
| 6. SMTP mailbox verification | |
| 7. Deterministic heuristic deliverability scoring | |
| """ | |
| def __init__( | |
| self, | |
| mailbox: MailboxProvider | None = None, | |
| cache=None, | |
| retry_queue=None, | |
| scorer: AIRiskScorer | None = None, | |
| ): | |
| self.use_proxies = settings.use_proxies | |
| # Layers 5 (catch-all) and 6 (SMTP) run through the autonomous mailbox | |
| # layer: the built-in SMTP engine on this host, or the remote SMTP | |
| # probe worker when SMTP_WORKER_URL is configured (spec §3.3). | |
| self.mailbox = mailbox if mailbox is not None else resolve_provider() | |
| # DNS/MX and definitive mailbox verdicts are cached across runs — the | |
| # same domain (and often the same address) recurs constantly in bulk | |
| # lists. Transient failures are never cached; see validation_cache.py. | |
| if cache is None: | |
| from .validation_cache import validation_cache as cache | |
| self.cache = cache | |
| # Greylisted (4xx) addresses get a bounded second chance. Without this a | |
| # valid mailbox on a greylisting domain stays permanently "risky". | |
| if retry_queue is None: | |
| from .retry_queue import retry_queue as retry_queue | |
| self.retry_queue = retry_queue | |
| # AI risk scorer (spec §3.4): pattern-based classification for the | |
| # ambiguous rows only; circuit breaker + heuristic fallback inside. | |
| self.scorer = scorer if scorer is not None else ai_scorer | |
| async def validate(self, email: str, deep: bool = True) -> ValidationResult: | |
| start_time = time.monotonic() | |
| run_id = uuid.uuid4().hex[:12] | |
| log = logger.bind(email=_mask_email_for_log(email), run_id=run_id) | |
| log.info("validation_started") | |
| # ---- Layer 1: Syntax ----------------------------------------- | |
| syntax = validate_syntax(email) | |
| if not syntax.passed: | |
| elapsed = (time.monotonic() - start_time) * 1000 | |
| scoring = calculate_score(syntax, None, None, None, None) | |
| log.warning("syntax_failed", error=syntax.error) | |
| return ValidationResult( | |
| email=email, | |
| status=ValidationStatus.INVALID, | |
| syntax=syntax, | |
| scoring=scoring, | |
| processing_time_ms=round(elapsed, 2), | |
| validated_at=datetime.now(timezone.utc).isoformat(), | |
| checks_run=_build_checks_run(syntax, None, None, None, None, scoring), | |
| reason_code=_build_reason_code(None, None, None, None), | |
| run_id=run_id, | |
| validator_version=settings.app_version, | |
| deterministic_verdict="invalid", | |
| ) | |
| domain = syntax.domain | |
| normalized_email = syntax.normalized_email or email | |
| # ---- Layers 2 & 3: DNS + MX --------------------------------- | |
| dns_result = await check_dns(domain, settings.dns_timeout) | |
| if dns_result.error_code in {"timeout", "resolver_error"}: | |
| elapsed = (time.monotonic() - start_time) * 1000 | |
| scoring = calculate_score(syntax, dns_result, None, None, None) | |
| log.warning( | |
| "dns_inconclusive", domain=domain, error_code=dns_result.error_code | |
| ) | |
| return ValidationResult( | |
| email=normalized_email, | |
| status=self._determine_status(scoring), | |
| syntax=syntax, | |
| dns=dns_result, | |
| scoring=scoring, | |
| processing_time_ms=round(elapsed, 2), | |
| validated_at=datetime.now(timezone.utc).isoformat(), | |
| checks_run=_build_checks_run(syntax, dns_result, None, None, None, scoring), | |
| reason_code=_build_reason_code(dns_result, None, None, None, early="dns_timeout"), | |
| run_id=run_id, | |
| validator_version=settings.app_version, | |
| deterministic_verdict="unknown", | |
| ) | |
| if ( | |
| not dns_result.domain_exists | |
| or dns_result.error_code == "nxdomain" | |
| or (not dns_result.has_mx_records and not dns_result.a_fallback) | |
| ): | |
| elapsed = (time.monotonic() - start_time) * 1000 | |
| scoring = calculate_score(syntax, dns_result, None, None, None) | |
| log.warning("dns_failed", domain=domain) | |
| return ValidationResult( | |
| email=normalized_email, | |
| status=ValidationStatus.INVALID, | |
| syntax=syntax, | |
| dns=dns_result, | |
| scoring=scoring, | |
| processing_time_ms=round(elapsed, 2), | |
| validated_at=datetime.now(timezone.utc).isoformat(), | |
| checks_run=_build_checks_run(syntax, dns_result, None, None, None, scoring), | |
| reason_code=_build_reason_code( | |
| dns_result, None, None, None, early="nxdomain" | |
| ), | |
| run_id=run_id, | |
| validator_version=settings.app_version, | |
| deterministic_verdict="invalid", | |
| ) | |
| # ---- Layer 4: Disposable Detection --------------------------- | |
| disposable = check_disposable(normalized_email, domain) | |
| # ML advisory (single path): unknown domains only, fail-open. Mirrors | |
| # the batch pipeline so both surfaces agree on the same address. | |
| # Skipped when the MX runs on a trusted enterprise platform: the model | |
| # learned 'unfamiliar domain string = disposable' from an imbalanced | |
| # dataset and flags legitimate small businesses on Google Workspace / | |
| # M365 — the deterministic MX intelligence is the better signal there. | |
| ml_model = await get_model_async() | |
| mx_host = ( | |
| dns_result.mx_records[0]["host"] | |
| if dns_result.mx_records | |
| else None | |
| ) | |
| if ( | |
| ml_model is not None | |
| and not domain_is_known(domain) | |
| and deep | |
| and not mx_is_trusted(mx_host) | |
| ): | |
| # Off the event loop: CPU inference inline would stall every other | |
| # concurrent request (measured 42.6s cold start before this). | |
| preds = await predict_async(ml_model, [f"john.smith [SEP] {domain}"]) | |
| if preds: | |
| disposable = disposable.model_copy( | |
| update={"ml_label": preds[0][0], "ml_confidence": preds[0][1]} | |
| ) | |
| if not deep: | |
| # Fast path: skip SMTP for quick validation | |
| elapsed = (time.monotonic() - start_time) * 1000 | |
| scoring = calculate_score( | |
| syntax, | |
| dns_result, | |
| disposable, | |
| None, | |
| None, | |
| mailbox_verification_skipped=True, | |
| ) | |
| status = self._determine_status(scoring) | |
| return ValidationResult( | |
| email=normalized_email, | |
| status=status, | |
| syntax=syntax, | |
| dns=dns_result, | |
| disposable=disposable, | |
| scoring=scoring, | |
| processing_time_ms=round(elapsed, 2), | |
| validated_at=datetime.now(timezone.utc).isoformat(), | |
| checks_run=_build_checks_run( | |
| syntax, dns_result, disposable, None, None, scoring | |
| ), | |
| reason_code=_build_reason_code(dns_result, disposable, None, None), | |
| run_id=run_id, | |
| validator_version=settings.app_version, | |
| # Fast path: the mailbox question was never asked, so there is | |
| # no deterministic mailbox verdict to report. | |
| deterministic_verdict=None, | |
| ) | |
| # ---- Layers 5 & 6: Catch-All + SMTP -------------------------- | |
| mx_host = dns_result.mx_records[0]["host"] | |
| # EAI/SMTPUTF8 gate (spec §7.2): a non-ASCII local part can only be | |
| # delivered when the receiving MX advertises SMTPUTF8. Without an SMTP | |
| # dialogue the engine refuses the per-address probe deterministically | |
| # instead of faking a verdict. The domain-level catch-all probe (a | |
| # random ASCII local part) still runs — it is independent of the | |
| # address's own local part. | |
| eai_reason = eai_gate(syntax.local_part or "") | |
| catch_all_task = asyncio.create_task( | |
| self.mailbox.catch_all(domain, mx_host) | |
| ) | |
| if eai_reason: | |
| smtp_result = SMTPResult( | |
| verified=None, | |
| error=( | |
| f"{EAI_REASON}: mailbox probe requires SMTPUTF8 support " | |
| "(deterministic refusal)" | |
| ), | |
| ) | |
| try: | |
| catch_all_raw = await catch_all_task | |
| except Exception as exc: | |
| # Same contract as the gather path: a failed probe is UNKNOWN. | |
| catch_all_raw = exc | |
| else: | |
| smtp_task = asyncio.create_task( | |
| self.mailbox.verify(normalized_email, mx_host) | |
| ) | |
| catch_all_raw, smtp_result = await asyncio.gather( | |
| catch_all_task, smtp_task, return_exceptions=True | |
| ) | |
| # Handle catch-all result. An exception means the probe could not run | |
| # (e.g. port-25 egress blocked) — that is UNKNOWN, not "not catch-all". | |
| if isinstance(catch_all_raw, Exception): | |
| catch_all = CatchAllResult( | |
| is_catch_all=None, confidence=0.0, error=str(catch_all_raw) | |
| ) | |
| else: | |
| is_catch_all, confidence = catch_all_raw | |
| catch_all = CatchAllResult(is_catch_all=is_catch_all, confidence=confidence) | |
| # Handle SMTP result | |
| if isinstance(smtp_result, Exception): | |
| smtp_result = SMTPResult(verified=None, error=str(smtp_result)) | |
| # ---- Layer 7: Scoring ---------------------------------------- | |
| scoring = calculate_score( | |
| syntax, dns_result, disposable, catch_all, smtp_result | |
| ) | |
| status = self._determine_status(scoring) | |
| elapsed = (time.monotonic() - start_time) * 1000 | |
| # ---- Normalized verdict (spec Phase 5): deterministic first, then AI -- | |
| deterministic = _deterministic_verdict(disposable, catch_all, smtp_result) | |
| ai_score: float | None = None | |
| ai_label: str | None = None | |
| fallback_used = False | |
| if deep and _ambiguous(deterministic): | |
| features = AIRiskScorer.build_features( | |
| domain=domain, | |
| mx_host=mx_host, | |
| smtp_class=_smtp_class(smtp_result), | |
| catch_all=catch_all.is_catch_all if catch_all else None, | |
| local_part=syntax.local_part or "", | |
| dns_inconsistent=dns_result.dns_inconsistent, | |
| ) | |
| scored = await self.scorer.score(features) | |
| ai_score, ai_label, fallback_used = ( | |
| scored.score, | |
| scored.label, | |
| scored.fallback_used, | |
| ) | |
| if scored.label == "high_risk": | |
| scoring.warnings.append("ai_high_risk") | |
| if dns_result.dns_inconsistent: | |
| scoring.warnings.append("dns_inconsistent: resolvers disagree on MX") | |
| reason = _build_reason_code(dns_result, disposable, catch_all, smtp_result) | |
| if eai_reason: | |
| reason = EAI_REASON | |
| log.info( | |
| "validation_complete", | |
| status=status, | |
| score=scoring.score, | |
| time_ms=round(elapsed, 2), | |
| ) | |
| return ValidationResult( | |
| email=normalized_email, | |
| status=status, | |
| syntax=syntax, | |
| dns=dns_result, | |
| disposable=disposable, | |
| catch_all=catch_all, | |
| smtp=smtp_result, | |
| scoring=scoring, | |
| processing_time_ms=round(elapsed, 2), | |
| validated_at=datetime.now(timezone.utc).isoformat(), | |
| checks_run=_build_checks_run( | |
| syntax, dns_result, disposable, catch_all, smtp_result, scoring | |
| ), | |
| reason_code=reason, | |
| run_id=run_id, | |
| validator_version=settings.app_version, | |
| deterministic_verdict=deterministic, | |
| ai_risk_score=ai_score, | |
| ai_label=ai_label, | |
| fallback_used=fallback_used, | |
| ) | |
| async def validate_batch( | |
| self, emails: list[str], deep: bool = True | |
| ) -> list[ValidationResult]: | |
| """Validate many addresses in one pass, reusing per-domain work. | |
| Bulk throughput comes from doing DNS/MX **once per domain** and handing | |
| every (email, mx_host) pair to the mailbox provider in a **single** call, | |
| so the SMTP engine can reuse one session per receiving host instead of a | |
| fresh TCP handshake per address. Providers that cannot batch fall back to | |
| the per-address path, so behaviour is never worse than before. | |
| Results are returned in the same order as `emails`. | |
| """ | |
| if not emails: | |
| return [] | |
| run_id = uuid.uuid4().hex[:12] | |
| out: list[ValidationResult | None] = [None] * len(emails) | |
| def _stamp(**kw) -> ValidationResult: | |
| return ValidationResult( | |
| processing_time_ms=0.0, | |
| validated_at=datetime.now(timezone.utc).isoformat(), | |
| run_id=run_id, | |
| validator_version=settings.app_version, | |
| **kw, | |
| ) | |
| # ---- Layer 1: syntax (per address) --------------------------- | |
| parsed: list[tuple[str, SyntaxResult]] = [] | |
| domain_index: dict[str, list[int]] = {} | |
| for idx, raw in enumerate(emails): | |
| syntax = validate_syntax(raw) | |
| parsed.append((raw, syntax)) | |
| if not syntax.passed: | |
| scoring = calculate_score(syntax, None, None, None, None) | |
| out[idx] = _stamp( | |
| email=raw, | |
| status=ValidationStatus.INVALID, | |
| syntax=syntax, | |
| scoring=scoring, | |
| checks_run=_build_checks_run(syntax, None, None, None, None, scoring), | |
| reason_code=_build_reason_code(None, None, None, None), | |
| ) | |
| continue | |
| domain_index.setdefault(syntax.domain, []).append(idx) | |
| # ---- Layers 2 & 3: DNS + MX once per domain (cached) --------- | |
| domains = list(domain_index.keys()) | |
| domain_dns: dict[str, DNSResult] = {} | |
| misses: list[str] = [] | |
| for domain in domains: | |
| cached_dns = await self.cache.get_dns(domain) | |
| if cached_dns is not None: | |
| domain_dns[domain] = cached_dns | |
| else: | |
| misses.append(domain) | |
| if misses: | |
| resolved = await asyncio.gather( | |
| *[check_dns(d, settings.dns_timeout) for d in misses], | |
| return_exceptions=True, | |
| ) | |
| for domain, dns in zip(misses, resolved): | |
| result = ( | |
| dns | |
| if isinstance(dns, DNSResult) | |
| else DNSResult( | |
| domain_exists=False, | |
| has_mx_records=False, | |
| error=str(dns), | |
| error_code="resolver_error", | |
| ) | |
| ) | |
| domain_dns[domain] = result | |
| # put_dns ignores transient failures internally. | |
| await self.cache.put_dns(domain, result) | |
| pending: list[tuple[int, str, SyntaxResult, str, DNSResult]] = [] | |
| for domain, idxs in domain_index.items(): | |
| dns = domain_dns[domain] | |
| inconclusive = dns.error_code in {"timeout", "resolver_error"} | |
| dead = ( | |
| not dns.domain_exists | |
| or dns.error_code == "nxdomain" | |
| or (not dns.has_mx_records and not dns.a_fallback) | |
| ) | |
| for idx in idxs: | |
| raw, syntax = parsed[idx] | |
| normalized = syntax.normalized_email or raw | |
| if inconclusive: | |
| scoring = calculate_score(syntax, dns, None, None, None) | |
| out[idx] = _stamp( | |
| email=normalized, | |
| status=self._determine_status(scoring), | |
| syntax=syntax, | |
| dns=dns, | |
| scoring=scoring, | |
| checks_run=_build_checks_run( | |
| syntax, dns, None, None, None, scoring | |
| ), | |
| reason_code=_build_reason_code( | |
| dns, None, None, None, early="dns_timeout" | |
| ), | |
| deterministic_verdict="unknown", | |
| ) | |
| elif dead: | |
| scoring = calculate_score(syntax, dns, None, None, None) | |
| out[idx] = _stamp( | |
| email=normalized, | |
| status=ValidationStatus.INVALID, | |
| syntax=syntax, | |
| dns=dns, | |
| scoring=scoring, | |
| checks_run=_build_checks_run( | |
| syntax, dns, None, None, None, scoring | |
| ), | |
| reason_code=_build_reason_code( | |
| dns, None, None, None, early="nxdomain" | |
| ), | |
| deterministic_verdict="invalid", | |
| ) | |
| else: | |
| pending.append((idx, normalized, syntax, domain, dns)) | |
| # ---- Layer 4 fast path (no mailbox probe) -------------------- | |
| if pending and not deep: | |
| for idx, normalized, syntax, domain, dns in pending: | |
| disposable = check_disposable(normalized, domain) | |
| scoring = calculate_score( | |
| syntax, dns, disposable, None, None, | |
| mailbox_verification_skipped=True, | |
| ) | |
| out[idx] = _stamp( | |
| email=normalized, | |
| status=self._determine_status(scoring), | |
| syntax=syntax, | |
| dns=dns, | |
| disposable=disposable, | |
| scoring=scoring, | |
| checks_run=_build_checks_run( | |
| syntax, dns, disposable, None, None, scoring | |
| ), | |
| reason_code=_build_reason_code(dns, disposable, None, None), | |
| deterministic_verdict=None, | |
| ) | |
| pending = [] | |
| # ---- Layers 5 & 6: ONE batched mailbox call ------------------ | |
| cached_verdicts: dict[str, tuple[SMTPResult, Optional[bool]]] = {} | |
| catch_all_map: dict[str, Optional[bool]] = {} | |
| if pending: | |
| # Serve anything already answered (definitive verdicts only) so we | |
| # never re-probe an address we already know the answer to. | |
| to_probe: list[tuple[int, str, SyntaxResult, str, DNSResult]] = [] | |
| for item in pending: | |
| _idx, normalized, _syntax, _domain, _dns = item | |
| hit = await self.cache.get_verdict(normalized, deep) | |
| if hit is None: | |
| to_probe.append(item) | |
| else: | |
| cached_verdicts[normalized] = hit | |
| if hit[1] is not None: | |
| catch_all_map[item[3]] = hit[1] | |
| smtp_map: dict[str, SMTPResult] = {} | |
| pairs: list[tuple[str, str]] = [] | |
| for _idx, normalized, _syntax, _domain, _dns in to_probe: | |
| if eai_gate(_syntax.local_part or ""): | |
| # Deterministic EAI refusal (spec §7.2): the probe requires | |
| # an SMTPUTF8 advertisement we cannot observe, so refuse | |
| # instead of guessing. Cache/check semantics unchanged. | |
| smtp_map[normalized] = SMTPResult( | |
| verified=None, | |
| error=( | |
| f"{EAI_REASON}: mailbox probe requires SMTPUTF8 " | |
| "(deterministic refusal)" | |
| ), | |
| ) | |
| continue | |
| if _dns.mx_records: | |
| pairs.append((normalized, _dns.mx_records[0]["host"])) | |
| else: | |
| smtp_map[normalized] = SMTPResult( | |
| verified=None, error="no MX host available for probe" | |
| ) | |
| if pairs: | |
| try: | |
| if hasattr(self.mailbox, "verify_many"): | |
| smtp_map, caught = await self.mailbox.verify_many(pairs) | |
| for domain, value in (caught or {}).items(): | |
| catch_all_map.setdefault(domain, value) | |
| else: | |
| # Provider cannot batch: per-address fallback. | |
| for email, mx in pairs: | |
| smtp_map[email] = await self.mailbox.verify(email, mx) | |
| for domain in {e.rsplit("@", 1)[-1].lower() for e, _ in pairs}: | |
| mx = next( | |
| m for e, m in pairs | |
| if e.rsplit("@", 1)[-1].lower() == domain | |
| ) | |
| value, _conf = await self.mailbox.catch_all(domain, mx) | |
| catch_all_map.setdefault(domain, value) | |
| except Exception as exc: # one bad stage must not abort the batch | |
| smtp_map = { | |
| e: SMTPResult(verified=None, error=f"mailbox_stage_error: {exc}") | |
| for e, _ in pairs | |
| } | |
| # ---- ML advisory signal (unknown domains, ONE batched call) ---- | |
| # Runs only where the curated lists have no answer - the exact gap | |
| # a static blocklist cannot close. Any failure here is silently | |
| # ignored: the deterministic verdict stands. | |
| # Domains on trusted enterprise MX (Google Workspace / M365 / | |
| # gateways) are excluded: disposable domains essentially never run | |
| # there, and the model's 'unfamiliar string' bias would otherwise | |
| # flag legitimate businesses as disposable. | |
| ml_map: dict[str, tuple[str, float]] = {} | |
| ml_model = await get_model_async() | |
| if ml_model is not None: | |
| unknown_domains = sorted( | |
| { | |
| domain | |
| for _, _, _, domain, dns in pending | |
| if not domain_is_known(domain) | |
| and dns.mx_records | |
| and not mx_is_trusted(dns.mx_records[0]["host"]) | |
| } | |
| ) | |
| if unknown_domains: | |
| # One batched call, off the event loop (CPU inference inline | |
| # stalls concurrent requests), and fail-open by construction: | |
| # predict_async never raises. | |
| preds = await predict_async( | |
| ml_model, [f"john.smith [SEP] {d}" for d in unknown_domains] | |
| ) | |
| if preds: | |
| ml_map = dict(zip(unknown_domains, preds)) | |
| # ---- Layer 7: score each row ----------------------------- | |
| ai_cache: dict[str, Any] = {} # feature string -> AIScore (dedupe) | |
| for idx, normalized, syntax, domain, dns in pending: | |
| cached = cached_verdicts.get(normalized) | |
| if cached is not None: | |
| smtp, ca_cached = cached | |
| else: | |
| smtp = smtp_map.get(normalized) or SMTPResult( | |
| verified=None, error="engine returned no result" | |
| ) | |
| ca_cached = None | |
| # Persist only definitive answers (verified verdict or a | |
| # confirmed catch-all); transient failures are never cached. | |
| await self.cache.put_verdict( | |
| normalized, deep, smtp, catch_all_map.get(domain) | |
| ) | |
| ca_value = ( | |
| ca_cached if ca_cached is not None else catch_all_map.get(domain) | |
| ) | |
| catch_all = CatchAllResult( | |
| is_catch_all=ca_value, | |
| confidence=0.9 if ca_value is not None else 0.0, | |
| ) | |
| disposable = check_disposable(normalized, domain) | |
| ml = ml_map.get(domain) | |
| if ml is not None: | |
| disposable = disposable.model_copy( | |
| update={"ml_label": ml[0], "ml_confidence": ml[1]} | |
| ) | |
| # A 4xx means "come back later", not a verdict. Schedule one | |
| # bounded retry so a real mailbox on a greylisting domain is not | |
| # stranded as permanently risky. The queue never raises. | |
| if cached is None: | |
| try: | |
| if is_greylist_result(smtp): | |
| await self.retry_queue.enqueue( | |
| normalized, dns.mx_records[0]["host"], smtp.smtp_code | |
| ) | |
| elif smtp.verified is not None: | |
| # Definitive answer — cancel any pending retry. | |
| await self.retry_queue.resolve_email(normalized) | |
| except Exception: | |
| pass | |
| scoring = calculate_score( | |
| syntax, dns, disposable, catch_all, smtp | |
| ) | |
| # ---- Normalized verdict (spec Phase 5) ------------------- | |
| deterministic = _deterministic_verdict(disposable, catch_all, smtp) | |
| ai_score: float | None = None | |
| ai_label: str | None = None | |
| fallback_used = False | |
| if deep and _ambiguous(deterministic): | |
| feats = AIRiskScorer.build_features( | |
| domain=domain, | |
| mx_host=dns.mx_records[0]["host"] if dns.mx_records else None, | |
| smtp_class=_smtp_class(smtp), | |
| catch_all=ca_value, | |
| local_part=syntax.local_part or "", | |
| dns_inconsistent=dns.dns_inconsistent, | |
| ) | |
| if feats not in ai_cache: | |
| ai_cache[feats] = await self.scorer.score(feats) | |
| scored = ai_cache[feats] | |
| ai_score, ai_label, fallback_used = ( | |
| scored.score, | |
| scored.label, | |
| scored.fallback_used, | |
| ) | |
| if scored.label == "high_risk": | |
| scoring.warnings.append("ai_high_risk") | |
| if dns.dns_inconsistent: | |
| scoring.warnings.append("dns_inconsistent: resolvers disagree on MX") | |
| reason = _build_reason_code(dns, disposable, catch_all, smtp) | |
| if (syntax.local_part or "") and eai_gate(syntax.local_part or ""): | |
| reason = EAI_REASON | |
| out[idx] = _stamp( | |
| email=normalized, | |
| status=self._determine_status(scoring), | |
| syntax=syntax, | |
| dns=dns, | |
| disposable=disposable, | |
| catch_all=catch_all, | |
| smtp=smtp, | |
| scoring=scoring, | |
| checks_run=_build_checks_run( | |
| syntax, dns, disposable, catch_all, smtp, scoring | |
| ), | |
| reason_code=reason, | |
| deterministic_verdict=deterministic, | |
| ai_risk_score=ai_score, | |
| ai_label=ai_label, | |
| fallback_used=fallback_used, | |
| cache_hit=cached is not None, | |
| ) | |
| # Guarantee one result per input, in order. | |
| for idx, result in enumerate(out): | |
| if result is None: | |
| raw = emails[idx] | |
| syntax = parsed[idx][1] if idx < len(parsed) else validate_syntax(raw) | |
| scoring = calculate_score(syntax, None, None, None, None) | |
| out[idx] = _stamp( | |
| email=raw, | |
| status=ValidationStatus.UNKNOWN, | |
| syntax=syntax, | |
| scoring=scoring, | |
| checks_run=_build_checks_run( | |
| syntax, None, None, None, None, scoring | |
| ), | |
| reason_code="engine_missing_result", | |
| deterministic_verdict="unknown", | |
| ) | |
| return [r for r in out if r is not None] | |
| def _determine_status(self, scoring: ScoringResult) -> ValidationStatus: | |
| if scoring.score >= 75: | |
| return ValidationStatus.VALID | |
| elif scoring.score >= 40: | |
| return ValidationStatus.RISKY | |
| else: | |
| return ValidationStatus.INVALID | |
| # Singleton engine | |
| engine = EmailValidationEngine() | |