import re from typing import Optional from ..models import ( SyntaxResult, DNSResult, DisposableResult, CatchAllResult, SMTPResult, ScoringResult, RiskLevel, ) def _is_explicit_mailbox_rejection(smtp: SMTPResult) -> bool: if smtp.smtp_code != 550 or not smtp.smtp_message: return False message = smtp.smtp_message.casefold() enhanced_status = re.search(r"(? ScoringResult: """ Layer 7: Comprehensive deliverability scoring. Score: 0-100 (higher = better deliverability) """ score = 100 risk_cap = 100 breakdown = {} warnings = [] # ---- Syntax (hard fail → score 0) -------------------------------- if not syntax.passed: breakdown["SYNTAX_INVALID"] = -100 warnings.append("Email syntax is invalid") return _scoring_result(0, breakdown, warnings) breakdown["syntax"] = 0 # ---- Domain/MX ------------------------------------------------------- if dns: if dns.error_code in {"timeout", "resolver_error"}: breakdown["DNS_TRANSIENT_FAILURE"] = -50 warnings.append("DNS lookup was inconclusive") return _scoring_result(50, breakdown, warnings) elif not dns.domain_exists: breakdown["DNS_MX_MISSING"] = -100 warnings.append("Domain does not exist") return _scoring_result(0, breakdown, warnings) elif dns.error_code == "nxdomain" or ( not dns.has_mx_records and not dns.a_fallback ): breakdown["DNS_MX_MISSING"] = -100 warnings.append("Domain has no valid mail route") return _scoring_result(0, breakdown, warnings) elif dns.a_fallback: risk_cap = min(risk_cap, 74) breakdown["mx_records"] = -45 score -= 45 warnings.append( "No MX records — using A/AAAA fallback; mailbox result is limited" ) else: breakdown["domain"] = 0 breakdown["mx_records"] = 0 # Authentication bonuses/penalties if dns.has_spf: breakdown["spf"] = 3 score += 3 else: warnings.append("No SPF record — increases spam risk") if dns.has_dmarc: breakdown["dmarc"] = 3 score += 3 else: warnings.append("No DMARC record configured") if dns.has_dkim: breakdown["dkim"] = 2 score += 2 # ---- Disposable --------------------------------------------------- if disposable: if disposable.is_disposable: breakdown["DISPOSABLE_DOMAIN"] = -100 warnings.append("Disposable/temporary email domain detected") return _scoring_result(0, breakdown, warnings) elif disposable.is_role_based: risk_cap = min(risk_cap, 74) breakdown["role_based"] = -10 score -= 10 warnings.append( "Role-based email (admin@, info@, etc.) — low engagement risk" ) elif disposable.is_free_provider: breakdown["free_provider"] = -5 score -= 5 # ---- ML advisory (DistilBERT, unknown domains only) --------------- # NEVER a hard verdict: a statistical false positive must not destroy a # legitimate lead. Penalty is small and reversible by promotion gate. ml_label = getattr(disposable, "ml_label", None) if ml_label == "DISPOSABLE" and not disposable.is_disposable: from ..config import settings # local import: scoring stays pure penalty = max(1, int(getattr(settings, "ml_score_penalty", 5))) score -= penalty breakdown["ML_DISPOSABLE_SUSPECT"] = -penalty confidence = getattr(disposable, "ml_confidence", 0.0) or 0.0 warnings.append( f"ML model flagged domain as disposable-like ({confidence:.2f})" ) # ---- Catch-All (-20) --------------------------------------------- if catch_all and catch_all.is_catch_all: risk_cap = min(risk_cap, 74) breakdown["catch_all"] = -20 score -= 20 warnings.append("Catch-all domain — cannot verify specific mailbox") # ---- SMTP Verification ------------------------------------------- if smtp: if smtp.verified is True: breakdown["smtp"] = 10 score += 10 elif smtp.verified is False: if _is_explicit_mailbox_rejection(smtp): breakdown["SMTP_MAILBOX_NOT_FOUND"] = -100 warnings.append("SMTP explicitly reported the mailbox as not found") return _scoring_result(0, breakdown, warnings) breakdown["smtp"] = -30 score -= 30 warnings.append( f"SMTP rejected the recipient (code {smtp.smtp_code}); " "provider policy may make the result inconclusive" ) else: risk_cap = min(risk_cap, 70) breakdown["SMTP_CHECK_INCONCLUSIVE"] = -30 score -= 30 if smtp.error: warnings.append(f"SMTP check inconclusive: {smtp.error}") if mailbox_verification_skipped: risk_cap = min(risk_cap, 70) breakdown["FAST_MODE_SKIPPED_MAILBOX_CHECK"] = -30 score -= 30 warnings.append("Mailbox verification was not performed") # Clamp score between 0 and 100 score = max(0, min(100, score)) if score > risk_cap: breakdown["RISK_CONTEXT_CAP"] = risk_cap - score score = risk_cap return _scoring_result(score, breakdown, warnings) def _scoring_result( score: int, breakdown: dict[str, int], warnings: list[str], ) -> ScoringResult: # Determine risk level if score >= 75: risk = RiskLevel.LOW elif score >= 40: risk = RiskLevel.MEDIUM else: risk = RiskLevel.HIGH return ScoringResult( score=score, risk_level=risk, breakdown=breakdown, warnings=warnings )