| 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"(?<![\d.])5\.1\.1(?![\d.])", message)
|
| mailbox_rejection_phrases = (
|
| "user unknown",
|
| "no such user",
|
| "mailbox does not exist",
|
| "mailbox not found",
|
| "recipient does not exist",
|
| "recipient not found",
|
| "account does not exist",
|
| )
|
| return bool(enhanced_status) or any(
|
| phrase in message for phrase in mailbox_rejection_phrases
|
| )
|
|
|
|
|
| def calculate_score(
|
| syntax: SyntaxResult,
|
| dns: Optional[DNSResult],
|
| disposable: Optional[DisposableResult],
|
| catch_all: Optional[CatchAllResult],
|
| smtp: Optional[SMTPResult],
|
| *,
|
| mailbox_verification_skipped: bool = False,
|
| ) -> ScoringResult:
|
| """
|
| Layer 7: Comprehensive deliverability scoring.
|
| Score: 0-100 (higher = better deliverability)
|
| """
|
| score = 100
|
| risk_cap = 100
|
| breakdown = {}
|
| warnings = []
|
|
|
|
|
| if not syntax.passed:
|
| breakdown["SYNTAX_INVALID"] = -100
|
| warnings.append("Email syntax is invalid")
|
| return _scoring_result(0, breakdown, warnings)
|
| breakdown["syntax"] = 0
|
|
|
|
|
| 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
|
|
|
|
|
| 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
|
|
|
|
|
| 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_label = getattr(disposable, "ml_label", None)
|
| if ml_label == "DISPOSABLE" and not disposable.is_disposable:
|
| from ..config import settings
|
|
|
| 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})"
|
| )
|
|
|
|
|
| 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")
|
|
|
|
|
| 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")
|
|
|
|
|
| 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:
|
|
|
| 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
|
| )
|
|
|
