#!/usr/bin/env python3 """ Mails.so Validator - Python CLI Advanced email validation with pattern analysis and reporting """ import sys import json import argparse import requests from typing import List, Dict, Optional from datetime import datetime from collections import Counter API_KEY = "b872fcef-1290-4a84-94a7-a1a50a77d0c9" API_URL = "http://localhost:3002/v1/validate" class Colors: """ANSI color codes""" GREEN = '\033[92m' RED = '\033[91m' YELLOW = '\033[93m' BLUE = '\033[94m' MAGENTA = '\033[95m' CYAN = '\033[96m' WHITE = '\033[97m' BOLD = '\033[1m' RESET = '\033[0m' class EmailValidator: def __init__(self, api_key: str = API_KEY, api_url: str = API_URL): self.api_key = api_key self.api_url = api_url self.results: List[Dict] = [] def validate(self, email: str) -> Optional[Dict]: """Validate a single email""" try: response = requests.get( self.api_url, params={"email": email}, headers={"x-mails-api-key": self.api_key}, timeout=10 ) response.raise_for_status() data = response.json() if data.get("error"): return None result = data.get("data", {}) result["_timestamp"] = datetime.now().isoformat() self.results.append(result) return result except Exception as e: print(f"{Colors.RED}Error validating {email}: {e}{Colors.RESET}") return None def validate_bulk(self, emails: List[str], verbose: bool = True) -> List[Dict]: """Validate multiple emails""" results = [] total = len(emails) for i, email in enumerate(emails, 1): if verbose: print(f"[{i}/{total}] Validating: {email}...", end=" ") result = self.validate(email) if result: results.append(result) if verbose: status = self._get_status_emoji(result) print(f"{status} {result['result']} ({result['score']}/100)") else: if verbose: print(f"{Colors.RED}✗ Failed{Colors.RESET}") return results def _get_status_emoji(self, result: Dict) -> str: """Get emoji for result status""" result_type = result.get("result", "unknown") if result_type == "deliverable": return f"{Colors.GREEN}✓{Colors.RESET}" elif result_type == "undeliverable": return f"{Colors.RED}✗{Colors.RESET}" elif result_type == "risky": return f"{Colors.YELLOW}⚠{Colors.RESET}" else: return f"{Colors.BLUE}?{Colors.RESET}" def analyze(self) -> Dict: """Analyze validation results""" if not self.results: return {} analysis = { "total": len(self.results), "by_result": Counter(r["result"] for r in self.results), "by_provider": Counter(r.get("provider") for r in self.results if r.get("provider")), "disposable_count": sum(1 for r in self.results if r.get("is_disposable")), "free_count": sum(1 for r in self.results if r.get("is_free")), "avg_score": sum(r["score"] for r in self.results) / len(self.results), "score_distribution": { "0-20": sum(1 for r in self.results if 0 <= r["score"] < 20), "20-40": sum(1 for r in self.results if 20 <= r["score"] < 40), "40-60": sum(1 for r in self.results if 40 <= r["score"] < 60), "60-80": sum(1 for r in self.results if 60 <= r["score"] < 80), "80-100": sum(1 for r in self.results if 80 <= r["score"] <= 100), } } return analysis def print_report(self): """Print analysis report""" analysis = self.analyze() if not analysis: print(f"{Colors.YELLOW}No results to analyze{Colors.RESET}") return print(f"\n{Colors.BOLD}{Colors.CYAN}╔══════════════════════════════════════════════╗{Colors.RESET}") print(f"{Colors.BOLD}{Colors.CYAN}║ EMAIL VALIDATION REPORT ║{Colors.RESET}") print(f"{Colors.BOLD}{Colors.CYAN}╚══════════════════════════════════════════════╝{Colors.RESET}\n") # Summary print(f"{Colors.BOLD}📊 SUMMARY{Colors.RESET}") print(f" Total Emails: {Colors.BOLD}{analysis['total']}{Colors.RESET}") print(f" Average Score: {Colors.BOLD}{analysis['avg_score']:.1f}/100{Colors.RESET}\n") # Results breakdown print(f"{Colors.BOLD}📈 RESULTS BREAKDOWN{Colors.RESET}") for result_type, count in analysis['by_result'].items(): percentage = (count / analysis['total']) * 100 color = Colors.GREEN if result_type == "deliverable" else Colors.RED if result_type == "undeliverable" else Colors.YELLOW print(f" {color}{result_type.capitalize():15}{Colors.RESET} {count:3} ({percentage:5.1f}%)") # Score distribution print(f"\n{Colors.BOLD}📊 SCORE DISTRIBUTION{Colors.RESET}") for range_name, count in analysis['score_distribution'].items(): if count > 0: bar = "█" * int((count / analysis['total']) * 30) print(f" {range_name:8} {bar:30} {count}") # Provider breakdown if analysis['by_provider']: print(f"\n{Colors.BOLD}🏢 TOP PROVIDERS{Colors.RESET}") for provider, count in analysis['by_provider'].most_common(5): print(f" {provider or 'Unknown':20} {count}") # Flags print(f"\n{Colors.BOLD}🚩 FLAGS{Colors.RESET}") print(f" Disposable Emails: {Colors.YELLOW}{analysis['disposable_count']}{Colors.RESET}") print(f" Free Providers: {Colors.BLUE}{analysis['free_count']}{Colors.RESET}") print() def export_json(self, filename: str): """Export results to JSON""" with open(filename, 'w') as f: json.dump({ "timestamp": datetime.now().isoformat(), "total": len(self.results), "results": self.results, "analysis": self.analyze() }, f, indent=2) print(f"{Colors.GREEN}✓ Exported to {filename}{Colors.RESET}") def export_csv(self, filename: str): """Export results to CSV""" import csv if not self.results: return headers = ["email", "result", "score", "provider", "is_disposable", "is_free", "mx_record", "reason"] with open(filename, 'w', newline='') as f: writer = csv.DictWriter(f, fieldnames=headers, extrasaction='ignore') writer.writeheader() writer.writerows(self.results) print(f"{Colors.GREEN}✓ Exported to {filename}{Colors.RESET}") def main(): parser = argparse.ArgumentParser( description="Mails.so Email Validator CLI", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: %(prog)s test@gmail.com %(prog)s -f emails.txt %(prog)s -f emails.txt -o results.json %(prog)s -f emails.txt -r """ ) parser.add_argument('email', nargs='?', help='Email address to validate') parser.add_argument('-f', '--file', help='File with emails (one per line)') parser.add_argument('-o', '--output', help='Output file (JSON or CSV)') parser.add_argument('-r', '--report', action='store_true', help='Show detailed report') parser.add_argument('-q', '--quiet', action='store_true', help='Quiet mode (no progress)') parser.add_argument('--api-key', default=API_KEY, help='API key') parser.add_argument('--api-url', default=API_URL, help='API URL') args = parser.parse_args() # Validate arguments if not args.email and not args.file: parser.print_help() sys.exit(1) # Create validator validator = EmailValidator(args.api_key, args.api_url) # Validate emails if args.email: result = validator.validate(args.email) if result: print(f"\n{Colors.BOLD}Email:{Colors.RESET} {result['email']}") print(f"{Colors.BOLD}Result:{Colors.RESET} {result['result']}") print(f"{Colors.BOLD}Score:{Colors.RESET} {result['score']}/100") print(f"{Colors.BOLD}Provider:{Colors.RESET} {result.get('provider', 'N/A')}") print(f"{Colors.BOLD}Disposable:{Colors.RESET} {result.get('is_disposable', False)}") print(f"{Colors.BOLD}MX Record:{Colors.RESET} {result.get('mx_record', 'N/A')}") if result.get('reason'): print(f"{Colors.BOLD}Reason:{Colors.RESET} {result['reason']}") elif args.file: with open(args.file, 'r') as f: emails = [line.strip() for line in f if line.strip()] print(f"{Colors.BOLD}Validating {len(emails)} emails...{Colors.RESET}\n") validator.validate_bulk(emails, verbose=not args.quiet) # Show report if args.report and validator.results: validator.print_report() # Export results if args.output and validator.results: if args.output.endswith('.json'): validator.export_json(args.output) elif args.output.endswith('.csv'): validator.export_csv(args.output) else: print(f"{Colors.RED}Error: Output file must be .json or .csv{Colors.RESET}") if __name__ == "__main__": main()