import re from typing import Dict, Any, Optional from urllib.parse import urlparse EMAIL_REGEX = re.compile(r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$") COMMON_PREFIXES = ["info", "contact", "sales", "hello", "support", "inquiries"] def is_valid_email(email: str) -> bool: """Check if the provided string is a valid email format.""" if not email: return False return bool(EMAIL_REGEX.match(email)) def extract_domain(url: str) -> Optional[str]: """Extract domain from a URL to use for email guessing.""" if not url: return None if not url.startswith(('http://', 'https://')): url = 'http://' + url try: parsed = urlparse(url) domain = parsed.netloc if domain.startswith('www.'): domain = domain[4:] return domain except Exception: return None def guess_fallback_email(url: str) -> Optional[str]: """Generate a potential contact email based on the website domain.""" domain = extract_domain(url) if not domain: return None # Lightweight pattern matching routine: returns the first common corporate format # In a real-world scenario, you might ping these to check validity via SMTP, # but here we just return the most likely fallback. return f"{COMMON_PREFIXES[0]}@{domain}" def validate_and_enrich_lead(lead: Dict[str, Any]) -> Optional[Dict[str, Any]]: """ Applies the Data Completeness Rule: A lead is dropped unless it contains a company name, city, and a valid email address. Also attempts Email Validation Fallback. """ company_name = lead.get("company_name") city = lead.get("city") email = lead.get("contact_email") website = lead.get("website_url") # Check for basic required fields if not company_name or not str(company_name).strip(): return None if not city or not str(city).strip(): return None # Validate email or attempt fallback if not is_valid_email(email): if website: fallback = guess_fallback_email(website) if fallback: lead["contact_email"] = fallback else: return None # No valid email and couldn't generate fallback else: return None # No valid email and no website for fallback return lead