free_api_fallback.py

#!/usr/bin/env python3
"""
╔══════════════════════════════════════════════════════════════════════════════╗
║           БЕСПЛАТНЫЕ API ДЛЯ ПРОВЕРКИ EMAIL (Fallback)                       ║
║              Без регистрации | Без API ключей | Безлимитно                  ║
╚══════════════════════════════════════════════════════════════════════════════╝

Быстрое решение без инфраструктуры. Используйте когда:
- Нужно быстро проверить небольшой список
- Нет возможности развернуть свою инфраструктуру
- Нужна резервная проверка
"""

import requests
import json
import time
from typing import Dict, List, Optional
from dataclasses import dataclass
from concurrent.futures import ThreadPoolExecutor, as_completed


@dataclass
class FreeAPIResult:
    email: str
    valid: bool
    disposable: bool
    mx_valid: bool
    provider: str
    confidence: float
    raw_response: Dict


class FreeEmailValidators:
    """
    Коллекция бесплатных API для проверки email.
    Все API работают без регистрации и API ключей.
    """
    
    # ============ API 1: Disify (безлимитно, без регистрации) ============
    
    @staticmethod
    def verify_disify(email: str, timeout: int = 10) -> Optional[FreeAPIResult]:
        """
        Disify API - полностью бесплатно, без регистрации.
        https://disify.com
        
        Features:
        - Syntax validation
        - Domain validation
        - Disposable detection
        - MX record check
        """
        try:
            url = f"https://disify.com/api/email/{email}"
            response = requests.get(url, timeout=timeout)
            data = response.json()
            
            # Определяем валидность
            is_valid = data.get('format', False) and data.get('dns', False)
            is_disposable = data.get('disposable', False)
            
            confidence = 0.9 if is_valid else 0.5
            
            return FreeAPIResult(
                email=email,
                valid=is_valid and not is_disposable,
                disposable=is_disposable,
                mx_valid=data.get('dns', False),
                provider='disify',
                confidence=confidence,
                raw_response=data
            )
        except Exception as e:
            return FreeAPIResult(
                email=email,
                valid=False,
                disposable=False,
                mx_valid=False,
                provider='disify',
                confidence=0.0,
                raw_response={'error': str(e)}
            )
    
    @staticmethod
    def verify_disify_bulk(emails: List[str], timeout: int = 30) -> List[FreeAPIResult]:
        """
        Bulk verification через Disify (до 10,000 email за раз).
        """
        try:
            email_string = ','.join(emails)
            url = f"https://disify.com/api/email/{email_string}/mass"
            response = requests.get(url, timeout=timeout)
            data = response.json()
            
            results = []
            for item in data:
                is_valid = item.get('format', False) and item.get('dns', False)
                is_disposable = item.get('disposable', False)
                
                results.append(FreeAPIResult(
                    email=item.get('email', ''),
                    valid=is_valid and not is_disposable,
                    disposable=is_disposable,
                    mx_valid=item.get('dns', False),
                    provider='disify',
                    confidence=0.9 if is_valid else 0.5,
                    raw_response=item
                ))
            return results
        except Exception as e:
            return [FreeAPIResult(
                email=email,
                valid=False,
                disposable=False,
                mx_valid=False,
                provider='disify',
                confidence=0.0,
                raw_response={'error': str(e)}
            ) for email in emails]
    
    # ============ API 2: Open Source Verifier ============
    
    @staticmethod
    def verify_opensource(email: str, timeout: int = 10) -> Optional[FreeAPIResult]:
        """
        Open Source Email Verifier - бесплатно, GDPR compliant.
        https://rapid-email-verifier.fly.dev
        
        Features:
        - Syntax validation
        - Domain/MX check
        - Disposable detection
        - Typo suggestions
        - Email alias detection
        """
        try:
            url = "https://rapid-email-verifier.fly.dev/api/validate"
            response = requests.get(url, params={'email': email}, timeout=timeout)
            data = response.json()
            
            validations = data.get('validations', {})
            is_valid = validations.get('syntax', False) and validations.get('mx_records', False)
            
            return FreeAPIResult(
                email=email,
                valid=is_valid,
                disposable=False,  # Нет в ответе
                mx_valid=validations.get('mx_records', False),
                provider='opensource',
                confidence=0.85 if is_valid else 0.5,
                raw_response=data
            )
        except Exception as e:
            return FreeAPIResult(
                email=email,
                valid=False,
                disposable=False,
                mx_valid=False,
                provider='opensource',
                confidence=0.0,
                raw_response={'error': str(e)}
            )
    
    @staticmethod
    def verify_opensource_batch(emails: List[str], timeout: int = 30) -> List[FreeAPIResult]:
        """
        Batch verification (до 100 email).
        """
        try:
            url = "https://rapid-email-verifier.fly.dev/api/validate/batch"
            response = requests.post(url, json={'emails': emails}, timeout=timeout)
            data = response.json()
            
            results = []
            for item in data.get('results', []):
                validations = item.get('validations', {})
                is_valid = validations.get('syntax', False) and validations.get('mx_records', False)
                
                results.append(FreeAPIResult(
                    email=item.get('email', ''),
                    valid=is_valid,
                    disposable=False,
                    mx_valid=validations.get('mx_records', False),
                    provider='opensource',
                    confidence=0.85 if is_valid else 0.5,
                    raw_response=item
                ))
            return results
        except Exception as e:
            return [FreeAPIResult(
                email=email,
                valid=False,
                disposable=False,
                mx_valid=False,
                provider='opensource',
                confidence=0.0,
                raw_response={'error': str(e)}
            ) for email in emails]
    
    # ============ API 3: DeBounce Disposable ============
    
    @staticmethod
    def check_disposable_debounce(email: str, timeout: int = 5) -> bool:
        """
        DeBounce Disposable Check - только disposable detection.
        https://disposable.debounce.io
        """
        try:
            url = "https://disposable.debounce.io/"
            response = requests.get(url, params={'email': email}, timeout=timeout)
            data = response.json()
            return data.get('disposable') == 'true'
        except Exception:
            return False
    
    # ============ API 4: Email Provider Lookup ============
    
    @staticmethod
    def lookup_provider(email: str, timeout: int = 5) -> Optional[Dict]:
        """
 Email Provider Lookup - определение провайдера.
        https://api.emailproviderlookup.com
        """
        try:
            url = "https://api.emailproviderlookup.com/v1/lookup"
            response = requests.get(url, params={'email': email}, timeout=timeout)
            return response.json()
        except Exception as e:
            return {'error': str(e)}
    
    # ============ Консенсусная проверка ============
    
    @classmethod
    def verify_consensus(cls, email: str) -> Dict:
        """
        Проверка через несколько API с консенсусом.
        Использует Disify + OpenSource для максимальной точности.
        """
        results = []
        
        # Пробуем Disify
        disify = cls.verify_disify(email)
        if disify:
            results.append(disify)
        
        # Пробуем OpenSource
        oss = cls.verify_opensource(email)
        if oss:
            results.append(oss)
        
        # Проверка на disposable
        is_disposable = cls.check_disposable_debounce(email)
        
        # Консенсус
        valid_votes = sum(1 for r in results if r.valid)
        total_votes = len(results)
        
        if total_votes == 0:
            consensus_valid = False
            confidence = 0.0
        else:
            consensus_valid = valid_votes / total_votes >= 0.5
            confidence = valid_votes / total_votes
        
        # Рекомендация
        if is_disposable:
            recommendation = "REJECT_DISPOSABLE"
        elif consensus_valid and confidence >= 0.8:
            recommendation = "ACCEPT"
        elif consensus_valid and confidence >= 0.5:
            recommendation = "REVIEW"
        else:
            recommendation = "REJECT"
        
        return {
            'email': email,
            'valid': consensus_valid,
            'confidence': confidence,
            'disposable': is_disposable,
            'recommendation': recommendation,
            'details': {
                'disify': disify.raw_response if disify else None,
                'opensource': oss.raw_response if oss else None,
            }
        }
    
    @classmethod
    def verify_bulk_parallel(cls, emails: List[str], max_workers: int = 10) -> List[Dict]:
        """
        Параллельная проверка списка email.
        """
        results = []
        
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = {executor.submit(cls.verify_consensus, email): email for email in emails}
            
            for future in as_completed(futures):
                try:
                    result = future.result()
                    results.append(result)
                except Exception as e:
                    email = futures[future]
                    results.append({
                        'email': email,
                        'valid': False,
                        'confidence': 0.0,
                        'error': str(e),
                        'recommendation': 'REVIEW'
                    })
        
        return results


# ==================== CLI ====================

def main():
    import argparse
    
    parser = argparse.ArgumentParser(description='Бесплатная проверка email через API')
    parser.add_argument('--email', help='Проверить один email')
    parser.add_argument('--emails', help='Файл со списком email')
    parser.add_argument('--output', '-o', help='Сохранить результаты в файл')
    parser.add_argument('--json', action='store_true', help='Вывод в JSON')
    parser.add_argument('--workers', type=int, default=10, help='Параллельность')
    
    args = parser.parse_args()
    
    # Сбор email
    emails = []
    if args.email:
        emails = [args.email]
    elif args.emails:
        with open(args.emails, 'r') as f:
            emails = [line.strip() for line in f if line.strip() and '@' in line]
    else:
        # Demo mode
        emails = [
            'test@gmail.com',
            'user@outlook.com',
            'test@mailinator.com',
            'invalid@nonexistent12345.com',
        ]
        print("=== DEMO MODE ===")
        print("Используйте --email или --emails для проверки своих адресов\n")
    
    # Проверка
    print(f"Проверка {len(emails)} email через бесплатные API...\n")
    
    start = time.time()
    results = FreeEmailValidators.verify_bulk_parallel(emails, args.workers)
    elapsed = time.time() - start
    
    # Вывод
    if args.json:
        output = {
            'stats': {
                'total': len(results),
                'valid': sum(1 for r in results if r.get('valid')),
                'invalid': sum(1 for r in results if not r.get('valid')),
                'disposable': sum(1 for r in results if r.get('disposable')),
                'time_seconds': elapsed,
            },
            'results': results
        }
        
        if args.output:
            with open(args.output, 'w') as f:
                json.dump(output, f, indent=2, ensure_ascii=False)
            print(f"Результаты сохранены в {args.output}")
        else:
            print(json.dumps(output, indent=2, ensure_ascii=False))
    else:
        print("=" * 90)
        print(f"{'Email':<35} {'Статус':<10} {'Уверенность':<12} {'Рекомендация':<15}")
        print("=" * 90)
        
        for r in results:
            status = '✓ VALID' if r.get('valid') else '✗ INVALID'
            if r.get('disposable'):
                status = '⚠ DISPOSABLE'
            
            print(f"{r['email']:<35} {status:<10} {r.get('confidence', 0):.0%}          {r.get('recommendation', 'REVIEW')}")
        
        print("=" * 90)
        print(f"\nВсего: {len(results)} | Валидных: {sum(1 for r in results if r.get('valid'))} | ", end="")
        print(f"Невалидных: {sum(1 for r in results if not r.get('valid'))} | ", end="")
        print(f"Disposable: {sum(1 for r in results if r.get('disposable'))}")
        print(f"Время: {elapsed:.2f}s | Скорость: {len(results)/elapsed:.0f} email/сек")
        
        if args.output:
            with open(args.output, 'w') as f:
                json.dump(results, f, indent=2, ensure_ascii=False)
            print(f"\nРезультаты сохранены в {args.output}")


if __name__ == "__main__":
    main()