File size: 5,628 Bytes
de745bf | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 | from pydantic import BaseModel, field_validator
from typing import Optional, List, Dict, Any
from enum import Enum
# ---- Enums -----------------------------------------------------------
class ValidationStatus(str, Enum):
VALID = "valid"
INVALID = "invalid"
RISKY = "risky"
UNKNOWN = "unknown"
class RiskLevel(str, Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
# ---- Layer Results ----------------------------------------------------
class SyntaxResult(BaseModel):
passed: bool
normalized_email: Optional[str] = None
local_part: Optional[str] = None
domain: Optional[str] = None
error: Optional[str] = None
class DNSResult(BaseModel):
domain_exists: bool
has_mx_records: bool
mx_records: List[Dict[str, Any]] = []
a_fallback: bool = False
has_spf: bool = False
spf_record: Optional[str] = None
has_dmarc: bool = False
dmarc_record: Optional[str] = None
has_dkim: bool = False
# DoH dual-resolver signal (spec §7.3): True when Google and Cloudflare
# disagree on this domain's MX set — a deterministic risk signal.
dns_inconsistent: bool = False
# Which resolver layer produced the answer: "doh:dual" | "doh:primary" |
# "legacy" — operational provenance for /engine/status triage.
mx_source: Optional[str] = None
error: Optional[str] = None
error_code: Optional[str] = None
class DisposableResult(BaseModel):
is_disposable: bool
is_free_provider: bool
is_role_based: bool
provider_name: Optional[str] = None
# Advisory ML signal (DistilBERT, unknown domains only). Never a verdict:
# the deterministic layer remains the source of truth until the measured
# precision gate promotes the model (see app/ml/).
ml_label: Optional[str] = None
ml_confidence: Optional[float] = None
class CatchAllResult(BaseModel):
# None = the catch-all probe could not run (e.g. port-25 egress blocked on
# this host). It must NEVER be coerced to False, which would falsely claim
# "domain rejects unknown recipients".
is_catch_all: Optional[bool] = None
confidence: float = 0.0
error: Optional[str] = None
class SMTPResult(BaseModel):
verified: Optional[bool] = None
smtp_code: Optional[int] = None
smtp_message: Optional[str] = None
server_banner: Optional[str] = None
supports_tls: bool = False
error: Optional[str] = None
via_proxy: bool = False
class ScoringResult(BaseModel):
score: int
risk_level: RiskLevel
breakdown: Dict[str, int] = {}
warnings: List[str] = []
# ---- Main Response ----------------------------------------------------
class ValidationResult(BaseModel):
email: str
status: ValidationStatus
syntax: SyntaxResult
dns: Optional[DNSResult] = None
disposable: Optional[DisposableResult] = None
catch_all: Optional[CatchAllResult] = None
smtp: Optional[SMTPResult] = None
scoring: ScoringResult
processing_time_ms: float
validated_at: str
# Provenance the frontend contract expects (src/lib/verifier.ts classify()).
# checks_run lists the 7-layer taxonomy keys that actually executed; an
# unchecked layer must never read as a pass. reason_code is a machine cause
# for Unknown/risky rows. run_id correlates a single + its bulk chunk.
checks_run: List[str] = []
reason_code: Optional[str] = None
run_id: Optional[str] = None
validator_version: Optional[str] = None
# ---- Autonomous-engine normalized verdict (spec Phase 5) ----------------
# deterministic_verdict: the pre-AI verdict of the deterministic gauntlet —
# "valid" | "invalid" | "catch_all" | "unknown" | "disposable".
# ai_risk_score / ai_label: pattern-based risk classification for the
# ambiguous rows (catch-all / unknown) only; None for deterministic rows.
# fallback_used: the AI scorer could not reach inference and the local
# rule-based heuristic decided instead ("scored_fallback" in the spec).
# cache_hit: served from the domain/mailbox verdict cache within TTL.
# INVARIANT: a catch_all is NEVER collapsed into "valid", regardless of
# ai_risk_score — a good score means "likely exists and engages", not
# "safe to send blindly".
deterministic_verdict: Optional[str] = None
ai_risk_score: Optional[float] = None
ai_label: Optional[str] = None
fallback_used: bool = False
cache_hit: bool = False
class SingleEmailRequest(BaseModel):
email: str
@field_validator("email")
@classmethod
def strip_email(cls, v: str) -> str:
return v.strip().lower()
class BulkEmailRequest(BaseModel):
emails: List[str]
webhook_url: Optional[str] = None
# Product promise: bulk runs the SAME 7-layer pipeline as single (1 credit
# per email). deep=False skips catch-all + SMTP, which collapsed every bulk
# row to "Unknown / insufficient depth" — so it defaults to True and is only
# flipped off by an explicit caller that accepts the reduced verdict.
deep: bool = True
@field_validator("emails")
@classmethod
def limit_emails(cls, v: List[str]) -> List[str]:
if len(v) > 10000:
raise ValueError("Max 10,000 emails per bulk request")
return [e.strip().lower() for e in v]
class BulkValidationResponse(BaseModel):
task_id: str
status: str
total: int
message: str
class TaskStatusResponse(BaseModel):
task_id: str
status: str
progress: Optional[int] = None
total: Optional[int] = None
results: Optional[List[ValidationResult]] = None
|