| 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 |
| |
| |
| |
| 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") |
|
|
| |
| if not company_name or not str(company_name).strip(): |
| return None |
| |
| if not city or not str(city).strip(): |
| return None |
|
|
| |
| if not is_valid_email(email): |
| if website: |
| fallback = guess_fallback_email(website) |
| if fallback: |
| lead["contact_email"] = fallback |
| else: |
| return None |
| else: |
| return None |
| |
| return lead |
|
|