Phone Number Search

async def search_phone(request):
    """Vista asíncrona que maneja el formulario y ejecuta la búsqueda."""
    form = PhoneSearchForm(request.POST or None)
    
    if request.method == "POST":
        if form.is_valid():
            # Obtener datos limpios
            phone = form.phone
            country_code = form.country_code
            
            try:
                # Ejecutar búsqueda asíncrona
                results = await search_phone_async(phone, country_code)
                
                # Calcular estadísticas
                found_count = sum(
                    1 for r in results 
                    if isinstance(r, dict) and r.get("exists") is True
                )

async def search_phone_async(phone, country_code):
    """Ejecutar búsqueda usando módulos ignorant asíncronamente."""
    websites = await import_ignorant_modules()
    results = []
    
    try:
        async with httpx.AsyncClient(
            timeout=30.0,
            limits=httpx.Limits(
                max_connections=10, 
                max_keepalive_connections=5
            ),
            follow_redirects=True,
        ) as client:
            for website in websites:
                try:
                    await website(phone, country_code, client, results)
                except Exception as e:
                    results.append({
                        "name": website.__name__,
                        "domain": f"{website.__name__}.com",
                        "rateLimit": True,
                        "exists": False,
                        "error": str(e)[:100],
                    })
    
    return results

def clean_query(self):
    query = self.cleaned_data["query"]
    parts = query.strip().split()
    
    # Check format: at least 2 parts (CODE and PHONE)
    if len(parts) < 2:
        raise forms.ValidationError(
            "Formato inválido. Usa: CÓDIGO_PAÍS NÚMERO (ej: 521 5534123456)."
        )
    
    # Extract and validate country code
    country_code = parts[0]
    phone = "".join(parts[1:])  # Join remaining parts
    
    # Validate country code (1-999)
    country_code_clean = re.sub(r"[^0-9]", "", country_code)
    if not (1 <= int(country_code_clean) <= 999):
        raise forms.ValidationError("Código de país inválido (1-999).")
    
    # Validate phone number (minimum 6 digits)
    phone_clean = re.sub(r"[^0-9]", "", phone)
    if len(phone_clean) < 6:
        raise forms.ValidationError(
            "Número telefónico demasiado corto (mínimo 6 dígitos)."
        )
    
    # Store parsed values
    self.country_code = country_code_clean
    self.phone = phone_clean
    
    return query

async def import_ignorant_modules():
    """Importar módulos ignorant dinámicamente."""
    
    def import_submodules(package, recursive=True):
        if isinstance(package, str):
            package = importlib.import_module(package)
        results = {}
        
        for loader, name, is_pkg in pkgutil.walk_packages(package.__path__):
            full_name = package.__name__ + "." + name
            results[full_name] = importlib.import_module(full_name)
            if recursive and is_pkg:
                results.update(import_submodules(full_name))
        return results
    
    modules = import_submodules("ignorant.modules")
    websites = []
    
    for module_name in modules:
        if len(module_name.split(".")) > 3:
            modu = modules[module_name]
            site = module_name.split(".")[-1]
            if site in modu.__dict__:
                websites.append(modu.__dict__[site])
    
    return websites

async with httpx.AsyncClient(
    timeout=30.0,                              # 30-second timeout per request
    limits=httpx.Limits(
        max_connections=10,                     # Max concurrent connections
        max_keepalive_connections=5            # Reuse connections
    ),
    follow_redirects=True,                      # Handle redirects automatically
) as client:

{
    "name": "whatsapp",                    # Platform identifier
    "domain": "whatsapp.com",              # Domain name
    "exists": True,                        # Whether number is registered
    "rateLimit": False,                    # Rate limiting encountered
    "error": None                          # Error message if any
}

# Calculate statistics for template
found_count = sum(
    1 for r in results 
    if isinstance(r, dict) and r.get("exists") is True
)
rate_limited = any(
    r.get("rateLimit") for r in results 
    if isinstance(r, dict)
)

context = {
    "results": results,
    "phone": phone,
    "country": country_code,
    "found_count": found_count,
    "rate_limited": rate_limited,
    "total_count": len(results),
}

@shared_task(bind=True, max_retries=3, default_retry_delay=60)
def search_phone_task(self, phone, country_code):
    """Celery task para búsqueda de teléfonos."""
    try:
        logger.info(f"Iniciando búsqueda para teléfono: {phone}")
        
        # Ejecutar async function sincrónicamente
        results = async_to_sync(search_phone_async)(phone, country_code)
        
        logger.info(f"Búsqueda completada. Resultados: {len(results)}")
        return results
        
    except Exception as exc:
        logger.error(f"Error en task: {exc}")
        raise self.retry(exc=exc, countdown=60)

try:
    await website(phone, country_code, client, results)
except Exception as e:
    results.append({
        "name": website.__name__,
        "domain": f"{website.__name__}.com",
        "rateLimit": True,
        "exists": False,
        "error": str(e)[:100],  # Truncate long errors
    })

pip install ignorant
# or
python -m pip install ignorant