class BounceClassifier:
def __init__(self, db: Database) -> None:
self.db = db
def classify(self, smtp_code: int, error_message: str) -> str:
code_info = SMTP_RESPONSE_MAP.get(smtp_code, {})
if code_info.get("bounce_type"):
return code_info["bounce_type"]
for pattern, bounce_type in _TEXT_PATTERNS:
if re.search(pattern, error_message):
return bounce_type
if 400 <= smtp_code < 500:
return "temporary"
if smtp_code >= 500:
return "permanent"
return "unknown"
def record(self, email: str, smtp_code: int, error_message: str) -> str:
bounce_type = self.classify(smtp_code, error_message)
self.db.record_bounce(email, bounce_type, smtp_code, error_message[:500])
AppLogger.info(f"Bounce recorded: {email} -> {bounce_type} (code {smtp_code})")
return bounce_type
def get_recovery_advice(self, bounce_type: str) -> str:
advice = {
"hard_bounce": "Remove this recipient from the list. Email does not exist.",
"mailbox_full": "Retry later. The recipient's mailbox is full.",
"spam_rejection": "Your email content or domain may be flagged. Review your templates.",
"temporary": "Retry with backoff. The server is temporarily unavailable.",
"greylisting": "Retry later. The server is greylisting unfamiliar senders.",
"authentication": "Check SMTP credentials. The password may have expired.",
"rate_limited": "Slow down. You are sending too fast for this provider.",
"dns": "Check the recipient's domain. It may not exist.",
"timeout": "Check your network connection. Increase timeout if needed.",
"tls": "TLS handshake failed. Check SSL certificates.",
"permanent": "Permanent failure. Review the error message manually.",
"unknown": "Unclassified error. Check the full SMTP log.",
}
return advice.get(bounce_type, "No specific advice available.")