CESARSQ/Cqantion-OS
0
1"""2╔══════════════════════════════════════════════════════════════════════╗3║ ANTIGRAVITY EMAIL VALIDATOR v1.0 ║4║ Infrastructure Layer — Lead Hygiene Engine ║5║ ║6║ 3-Layer Validation Pipeline: ║7║ Layer 1 → Syntax (RFC-compliant Regex) ║8║ Layer 2 → DNS/MX Record Verification (dnspython) ║9║ Layer 3 → Disposable/Temporary Domain Blacklist ║10║ ║11║ Security: Zero SMTP connections. DNS queries only. ║12╚══════════════════════════════════════════════════════════════════════╝13"""14 15import os16import re17import sys18import time19import pandas as pd20import dns.resolver21 22# ─────────────────────────────────────────────────────────────────────23# CONFIGURATION24# ─────────────────────────────────────────────────────────────────────25 26DNS_TIMEOUT = 5 # seconds per MX lookup27DNS_LIFETIME = 10 # total resolution lifetime28OUTPUT_FILENAME = "Leads_Validados.csv"29 30# RFC 5322 compliant email regex (covers 99.9% of real-world addresses)31EMAIL_REGEX = re.compile(32 r"^[a-zA-Z0-9.!#$%&'*+/=?^_`{|}~-]+"33 r"@"34 r"[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?"35 r"(?:\.[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)*"36 r"\.[a-zA-Z]{2,}$"37)38 39# ─────────────────────────────────────────────────────────────────────40# LAYER 3 — DISPOSABLE / TEMPORARY DOMAIN BLACKLIST41# ─────────────────────────────────────────────────────────────────────42 43DISPOSABLE_DOMAINS = {44 # Major disposable email services45 "mailinator.com",46 "guerrillamail.com",47 "guerrillamail.net",48 "guerrillamail.org",49 "guerrillamailblock.com",50 "tempmail.com",51 "temp-mail.org",52 "temp-mail.io",53 "throwaway.email",54 "throwaway.com",55 "yopmail.com",56 "yopmail.fr",57 "sharklasers.com",58 "guerrillamail.info",59 "grr.la",60 "dispostable.com",61 "trashmail.com",62 "trashmail.me",63 "trashmail.net",64 "fakeinbox.com",65 "mailnesia.com",66 "maildrop.cc",67 "discard.email",68 "discardmail.com",69 "discardmail.de",70 "10minutemail.com",71 "10minute.email",72 "minutemail.com",73 "tempail.com",74 "tempr.email",75 "tempmailaddress.com",76 "burnermail.io",77 "mailcatch.com",78 "inboxbear.com",79 "spamgourmet.com",80 "mytemp.email",81 "mohmal.com",82 "emailondeck.com",83 "getnada.com",84 "nada.email",85 "tmpmail.net",86 "tmpmail.org",87 "binkmail.com",88 "bobmail.info",89 "chammy.info",90 "devnullmail.com",91 "letthemeatspam.com",92 "mailexpire.com",93 "mailforspam.com",94 "safetymail.info",95 "spam4.me",96 "spamfree24.org",97 "trashymail.com",98 "wegwerfmail.de",99 "wegwerfmail.net",100 "wh4f.org",101 "meltmail.com",102 "harakirimail.com",103 "mailnull.com",104 "jetable.org",105 "trash-mail.com",106 "getairmail.com",107 "filzmail.com",108 "crazymailing.com",109 "tmail.ws",110 "mailsac.com",111 "luxusjacht.de",112}113 114 115# ─────────────────────────────────────────────────────────────────────116# CORE VALIDATION ENGINE117# ─────────────────────────────────────────────────────────────────────118 119class EmailValidator:120 """Three-layer email validation pipeline. Zero SMTP connections."""121 122 def __init__(self):123 self.mx_cache: dict[str, bool] = {}124 self.stats = {125 "total": 0,126 "valid": 0,127 "invalid_syntax": 0,128 "invalid_dns": 0,129 "invalid_disposable": 0,130 "dns_errors": 0,131 }132 133 # ── Layer 1: Syntax ──────────────────────────────────────────────134 135 @staticmethod136 def _validate_syntax(email: str) -> bool:137 """RFC 5322 regex validation."""138 if not isinstance(email, str):139 return False140 return bool(EMAIL_REGEX.match(email.strip().lower()))141 142 # ── Layer 2: DNS / MX ────────────────────────────────────────────143 144 def _validate_mx(self, domain: str) -> bool:145 """146 Query DNS for MX records. Results are cached per domain147 to avoid redundant lookups across duplicate domains.148 """149 if domain in self.mx_cache:150 return self.mx_cache[domain]151 152 try:153 resolver = dns.resolver.Resolver()154 resolver.timeout = DNS_TIMEOUT155 resolver.lifetime = DNS_LIFETIME156 answers = resolver.resolve(domain, "MX")157 has_mx = len(answers) > 0158 self.mx_cache[domain] = has_mx159 return has_mx160 161 except dns.resolver.NXDOMAIN:162 # Domain does not exist at all163 self.mx_cache[domain] = False164 return False165 166 except dns.resolver.NoAnswer:167 # Domain exists but has no MX records168 self.mx_cache[domain] = False169 return False170 171 except dns.resolver.NoNameservers:172 # No nameservers available for domain173 self.mx_cache[domain] = False174 return False175 176 except dns.resolver.LifetimeTimeout:177 # DNS query timed out178 self.stats["dns_errors"] += 1179 self.mx_cache[domain] = False180 return False181 182 except Exception:183 # Catch-all for unexpected DNS errors184 self.stats["dns_errors"] += 1185 self.mx_cache[domain] = False186 return False187 188 # ── Layer 3: Disposable Check ────────────────────────────────────189 190 @staticmethod191 def _is_disposable(domain: str) -> bool:192 """Check domain against internal disposable blacklist."""193 return domain in DISPOSABLE_DOMAINS194 195 # ── Pipeline Orchestrator ────────────────────────────────────────196 197 def validate(self, email: str) -> str:198 """199 Run email through all 3 validation layers.200 Returns 'Valid' or 'Invalid'.201 """202 self.stats["total"] += 1203 204 # Normalize205 if not isinstance(email, str) or not email.strip():206 self.stats["invalid_syntax"] += 1207 return "Invalid"208 209 email_clean = email.strip().lower()210 211 # Layer 1 — Syntax212 if not self._validate_syntax(email_clean):213 self.stats["invalid_syntax"] += 1214 return "Invalid"215 216 # Extract domain217 domain = email_clean.split("@")[1]218 219 # Layer 3 — Disposable (check before DNS to save time)220 if self._is_disposable(domain):221 self.stats["invalid_disposable"] += 1222 return "Invalid"223 224 # Layer 2 — DNS/MX225 if not self._validate_mx(domain):226 self.stats["invalid_dns"] += 1227 return "Invalid"228 229 self.stats["valid"] += 1230 return "Valid"231 232 def get_stats_report(self) -> str:233 """Return formatted statistics report."""234 s = self.stats235 valid_pct = (s["valid"] / s["total"] * 100) if s["total"] > 0 else 0236 return (237 "\n"238 "╔══════════════════════════════════════════════════════════╗\n"239 "║ VALIDATION REPORT ║\n"240 "╠══════════════════════════════════════════════════════════╣\n"241 f"║ Total Processed : {s['total']:>8} ║\n"242 f"║ ✅ Valid : {s['valid']:>8} ({valid_pct:.1f}%) ║\n"243 f"║ ❌ Invalid (Syntax) : {s['invalid_syntax']:>8} ║\n"244 f"║ ❌ Invalid (DNS/MX) : {s['invalid_dns']:>8} ║\n"245 f"║ ❌ Invalid (Dispos.) : {s['invalid_disposable']:>8} ║\n"246 f"║ ⚠️ DNS Errors : {s['dns_errors']:>8} ║\n"247 "╚══════════════════════════════════════════════════════════╝\n"248 )249 250 251# ─────────────────────────────────────────────────────────────────────252# TERMINAL UI — INTERACTIVE CSV SELECTOR253# ─────────────────────────────────────────────────────────────────────254 255def _print_banner():256 """Display startup banner."""257 print(258 "\n"259 "╔══════════════════════════════════════════════════════════╗\n"260 "║ ⚡ ANTIGRAVITY EMAIL VALIDATOR v1.0 ⚡ ║\n"261 "║ Infrastructure Layer — Lead Hygiene Engine ║\n"262 "╠══════════════════════════════════════════════════════════╣\n"263 "║ Layers: Syntax → DNS/MX → Disposable Filter ║\n"264 "║ Policy: Zero SMTP — DNS queries only ║\n"265 "╚══════════════════════════════════════════════════════════╝\n"266 )267 268 269def _discover_csv_files() -> list[str]:270 """Scan current working directory for CSV files."""271 cwd = os.getcwd()272 csv_files = sorted(273 [f for f in os.listdir(cwd) if f.lower().endswith(".csv")],274 key=str.lower,275 )276 return csv_files277 278 279def _select_csv() -> str:280 """Interactive menu for CSV file selection."""281 csv_files = _discover_csv_files()282 283 if not csv_files:284 print(" ❌ No CSV files found in the current directory.")285 print(f" Directory: {os.getcwd()}")286 print(" Please place your CSV files here and try again.\n")287 sys.exit(1)288 289 print(" 📂 CSV files detected in current directory:\n")290 for idx, filename in enumerate(csv_files, start=1):291 size_kb = os.path.getsize(filename) / 1024292 print(f" [{idx}] {filename} ({size_kb:.1f} KB)")293 294 print()295 while True:296 try:297 choice = input(" ➜ Select file number: ").strip()298 index = int(choice) - 1299 if 0 <= index < len(csv_files):300 selected = csv_files[index]301 print(f"\n ✅ Selected: {selected}\n")302 return selected303 else:304 print(f" ⚠️ Enter a number between 1 and {len(csv_files)}.")305 except ValueError:306 print(" ⚠️ Please enter a valid number.")307 except KeyboardInterrupt:308 print("\n\n ❌ Operation cancelled by user.\n")309 sys.exit(0)310 311 312def _detect_email_column(df: pd.DataFrame) -> str:313 """314 Detect the email column. Supports 'EMAIL' and 'CORREO'315 (case-insensitive).316 """317 columns_upper = {col.upper().strip(): col for col in df.columns}318 319 for candidate in ["EMAIL", "CORREO"]:320 if candidate in columns_upper:321 return columns_upper[candidate]322 323 # If not found, display available columns324 print(" ❌ Could not find 'EMAIL' or 'CORREO' column.")325 print(f" Available columns: {list(df.columns)}\n")326 sys.exit(1)327 328 329# ─────────────────────────────────────────────────────────────────────330# MAIN EXECUTION331# ─────────────────────────────────────────────────────────────────────332 333def main():334 _print_banner()335 336 # Step 1 — Select CSV337 csv_file = _select_csv()338 339 # Step 2 — Load DataFrame340 print(" ⏳ Loading CSV...")341 try:342 # Try common separators343 try:344 df = pd.read_csv(csv_file, sep=",", dtype=str, encoding="utf-8", on_bad_lines='skip', engine='python')345 if len(df.columns) <= 1:346 df = pd.read_csv(csv_file, sep=";", dtype=str, encoding="utf-8", on_bad_lines='skip', engine='python')347 except UnicodeDecodeError:348 df = pd.read_csv(csv_file, sep=",", dtype=str, encoding="latin-1", on_bad_lines='skip', engine='python')349 if len(df.columns) <= 1:350 df = pd.read_csv(csv_file, sep=";", dtype=str, encoding="latin-1", on_bad_lines='skip', engine='python')351 except Exception as e:352 print(f" ❌ Failed to read CSV: {e}\n")353 sys.exit(1)354 355 print(f" Rows loaded: {len(df):,}")356 print(f" Columns: {list(df.columns)}\n")357 358 # Step 3 — Detect email column359 email_col = _detect_email_column(df)360 print(f" 📧 Email column detected: '{email_col}'\n")361 362 # Step 4 — Validate363 validator = EmailValidator()364 total = len(df)365 366 print(" 🔄 Running 3-layer validation pipeline...\n")367 start_time = time.time()368 369 results = []370 for idx, email in enumerate(df[email_col], start=1):371 status = validator.validate(email if pd.notna(email) else "")372 results.append(status)373 374 # Progress indicator every 50 rows or at the end375 if idx % 50 == 0 or idx == total:376 pct = idx / total * 100377 bar_filled = int(pct / 2)378 bar = "█" * bar_filled + "░" * (50 - bar_filled)379 print(f"\r [{bar}] {pct:5.1f}% ({idx:,}/{total:,})", end="", flush=True)380 381 elapsed = time.time() - start_time382 print(f"\n\n ✅ Validation complete in {elapsed:.2f}s")383 384 # Step 5 — Append results385 df["Email_Status"] = results386 387 # Step 6 — Export388 output_path = os.path.join(os.getcwd(), OUTPUT_FILENAME)389 df.to_csv(output_path, index=False, encoding="utf-8-sig")390 print(f" 💾 Output saved: {output_path}")391 392 # Step 7 — Stats393 print(validator.get_stats_report())394 395 # Quick summary396 valid_count = results.count("Valid")397 invalid_count = results.count("Invalid")398 print(f" 📊 Quick Summary: {valid_count:,} Valid | {invalid_count:,} Invalid")399 print(f" Unique domains cached: {len(validator.mx_cache):,}")400 print(f"\n ✅ File ready: {OUTPUT_FILENAME}\n")401 402 403if __name__ == "__main__":404 main()405