Python — API de datos de WhatsApp

Instalar

pip install requests
# or for async
pip install httpx

Búsqueda de perfil único

import os
import requests

BASE = "https://whatsapp-proxy.checkleaked.cc"
HEADERS = {
    # Direct proxy uses your direct-purchase key (issued in our dashboard).
    # For RapidAPI marketplace: BASE = "https://whatsapp-data1.p.rapidapi.com"
    # plus add: "x-rapidapi-host": "whatsapp-data1.p.rapidapi.com"
    "x-rapidapi-key": os.environ["WHATSAPP_API_KEY"],
}

def lookup(number: str) -> dict:
    r = requests.get(f"{BASE}/number/{number}", headers=HEADERS, timeout=10)
    r.raise_for_status()
    return r.json()

if __name__ == "__main__":
    data = lookup("13105551234")
    print(f"On WhatsApp: {data.get('isWAContact')}")
    print(f"Business: {data.get('isBusiness')}")
    print(f"About: {data.get('about')}")

Búsqueda asíncrona/concurrente

import asyncio
import os
import httpx

BASE = "https://whatsapp-proxy.checkleaked.cc"
HEADERS = {
    # Direct proxy uses your direct-purchase key (issued in our dashboard).
    # For RapidAPI marketplace: BASE = "https://whatsapp-data1.p.rapidapi.com"
    # plus add: "x-rapidapi-host": "whatsapp-data1.p.rapidapi.com"
    "x-rapidapi-key": os.environ["WHATSAPP_API_KEY"],
}

async def lookup_many(numbers: list[str]) -> list[dict]:
    async with httpx.AsyncClient(timeout=15, headers=HEADERS) as client:
        tasks = [client.get(f"{BASE}/number/{n}") for n in numbers]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
    return [r.json() if not isinstance(r, Exception) else {"error": str(r)} for r in responses]

if __name__ == "__main__":
    out = asyncio.run(lookup_many(["13105551234", "447911123456", "34612345678"]))
    print(out)

Verificación masiva (postal único)

import os
import requests

BASE = "https://whatsapp-proxy.checkleaked.cc"
HEADERS = {
    "x-rapidapi-key": os.environ["WHATSAPP_API_KEY"],
    "Content-Type": "application/json",
}

def bulk_check(numbers: list[str]) -> list[dict]:
    r = requests.post(
        f"{BASE}/bulk_check",
        json={"numbers": numbers},
        headers=HEADERS,
        timeout=60,
    )
    r.raise_for_status()
    # Returns ARRAY: [{success, number, hasWhatsapp, isBanned, formatted, country, ...}, ...]
    return r.json()

Reintentos y gestión de límites de velocidad

import os
import time
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

BASE = "https://whatsapp-proxy.checkleaked.cc"
HEADERS = {
    # Direct proxy uses your direct-purchase key (issued in our dashboard).
    # For RapidAPI marketplace: BASE = "https://whatsapp-data1.p.rapidapi.com"
    # plus add: "x-rapidapi-host": "whatsapp-data1.p.rapidapi.com"
    "x-rapidapi-key": os.environ["WHATSAPP_API_KEY"],
}

session = requests.Session()
retry = Retry(
    total=5,
    backoff_factor=1.5,
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=["GET"],
    respect_retry_after_header=True,
)
session.mount("https://", HTTPAdapter(max_retries=retry))

def lookup(number: str) -> dict:
    r = session.get(f"{BASE}/number/{number}", headers=HEADERS, timeout=10)
    if r.status_code == 429:
        wait = int(r.headers.get("Retry-After", "60"))
        time.sleep(wait)
        return lookup(number)
    r.raise_for_status()
    return r.json()

Procesamiento masivo de archivos CSV (por número, con control de ritmo)

import csv
import os
import time
import requests

BASE = "https://whatsapp-proxy.checkleaked.cc"
HEADERS = {
    # Direct proxy uses your direct-purchase key (issued in our dashboard).
    # For RapidAPI marketplace: BASE = "https://whatsapp-data1.p.rapidapi.com"
    # plus add: "x-rapidapi-host": "whatsapp-data1.p.rapidapi.com"
    "x-rapidapi-key": os.environ["WHATSAPP_API_KEY"],
}

with open("numbers.csv") as f_in, open("results.csv", "w", newline="") as f_out:
    reader = csv.DictReader(f_in)
    writer = csv.DictWriter(f_out, fieldnames=["number", "isWAContact", "isBusiness", "about"])
    writer.writeheader()
    for row in reader:
        r = requests.get(
            f"{BASE}/number/no_picture/{row['number']}",
            headers=HEADERS,
            timeout=10,
        )
        d = r.json() if r.ok else {}
        writer.writerow({
            "number": row["number"],
            "isWAContact": d.get("isWAContact", False),
            "isBusiness": d.get("isBusiness", False),
            "about": d.get("about", ""),
        })
        time.sleep(0.5)  # 2 req/sec — respect tier burst limit

Relacionado