Raspado web asíncrono en Python: Escala hasta 10,000+ URLs con aiohttp y Scrapeless

pip install aiohttp scrapeless playwright
playwright install chromium

export SCRAPELESS_API_KEY="tu_token_api_aqui"
export SCRAPELESS_CHANNEL_ID="tu_id_de_canal"          # impreso al inicio del nombre de usuario
export SCRAPELESS_PROXY_PASS="tu_contraseña_del_canal"
export SCRAPELESS_PROXY_GATEWAY="gw-us.scrapeless.io"   # ver más abajo para puertas de enlace regionales

import asyncio
import os
import aiohttp

PROXY = (
    f"http://{os.environ['SCRAPELESS_CHANNEL_ID']}-proxy-country_US"
    f":{os.environ['SCRAPELESS_PROXY_PASS']}"
    f"@{os.environ['SCRAPELESS_PROXY_GATEWAY']}:8789"
)

async def fetch(session: aiohttp.ClientSession, url: str) -> str:
    timeout = aiohttp.ClientTimeout(total=30)
    async with session.get(url, proxy=PROXY, timeout=timeout) as resp:
        resp.raise_for_status()
        return await resp.text()

async def main() -> None:
    async with aiohttp.ClientSession() as session:
        html = await fetch(session, "https://books.toscrape.com/")
        print(f"Obtenido {len(html):,} caracteres a través de salida residencial de EE. UU.")

if __name__ == "__main__":
    asyncio.run(main())

import asyncio
import os
import aiohttp

PROXY = (
    f"http://{os.environ['SCRAPELESS_CHANNEL_ID']}-proxy-country_US"
    f":{os.environ['SCRAPELESS_PROXY_PASS']}"
    f"@{os.environ['SCRAPELESS_PROXY_GATEWAY']}:8789"
)

# Limitar a 5 solicitudes concurrentes por host. Ajustar por objetivo — catálogos públicos
# toleran más, orígenes protegidos contra bots quieren menos.
PER_HOST = asyncio.Semaphore(5)

async def fetch(session: aiohttp.ClientSession, url: str) -> str:
    async with PER_HOST:
        timeout = aiohttp.ClientTimeout(total=30)
        async with session.get(url, proxy=PROXY, timeout=timeout) as resp:

resp.raise_for_status()
            return await resp.text()

async def main() -> None:
    urls = [
        f"https://books.toscrape.com/catalogue/page-{n}.html"
        for n in range(1, 51)  # 50 páginas del catálogo
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        pages = await asyncio.gather(*tasks)
    print(f"Obtenido {len(pages)} páginas, total {sum(len(p) for p in pages):,} caracteres")

if __name__ == "__main__":
    asyncio.run(main())

import asyncio
import json
import aiohttp

async def fetch_safe(session, url):
    try:
        async with session.get(
            url, timeout=aiohttp.ClientTimeout(total=30)
        ) as resp:
            resp.raise_for_status()
            return {"url": url, "html": await resp.text()}
    except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
        return {"url": url, "error": repr(exc)}

async def main(urls):
    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(
            *(fetch_safe(session, u) for u in urls)
        )

    ok = [r for r in results if "html" in r]
    failed = [r for r in results if "error" in r]
    print(f"Exitoso: {len(ok)}   Fallido: {len(failed)}")

    # Archivo de cartas muertas para revisión separada — el flujo de trabajo nunca se bloquea por fallos
    with open("dead_letter.jsonl", "w", encoding="utf-8") as f:
        for r in failed:
            f.write(json.dumps(r) + "\n")

import asyncio
from scrapeless import Scrapeless
from scrapeless.types import ICreateBrowser
from playwright.async_api import async_playwright

async def render_via_cloud_browser(url: str, country: str = "US") -> str:
    client = Scrapeless()  # lee SCRAPELESS_API_KEY del entorno
    session = client.browser.create(
        ICreateBrowser(proxy_country=country, session_ttl=240)
    )

    async with async_playwright() as p:
        browser = await p.chromium.connect_over_cdp(session.browser_ws_endpoint)
        context = browser.contexts[0] if browser.contexts else await browser.new_context()
        page = context.pages[0] if context.pages else await context.new_page()
        await page.goto(url, wait_until="networkidle", timeout=60_000)
        html = await page.content()
        await browser.close()
        return html

async def main():
    # quotes.toscrape.com/js/ es el sandbox canónico "necesita JS".
    # HTTP simple devuelve 0 elementos de citas; renderizado en la nube devuelve 10.
    html = await render_via_cloud_browser("https://quotes.toscrape.com/js/")
    print(f"Renderizado {len(html):,} caracteres incluyendo DOM post-pintura")

if __name__ == "__main__":
    asyncio.run(main())

import asyncio
import os
import aiohttp
from scrapeless import Scrapeless
from scrapeless.types import ICreateBrowser
from playwright.async_api import async_playwright

PROXY = (
    f"http://{os.environ['SCRAPELESS_CHANNEL_ID']}-proxy-country_US"
    f":{os.environ['SCRAPELESS_PROXY_PASS']}"
    f"@{os.environ['SCRAPELESS_PROXY_GATEWAY']}:8789"
)
HTTP_LIMIT = asyncio.Semaphore(10)      # nivel aiohttp
BROWSER_LIMIT = asyncio.Semaphore(3)    # nivel de navegador en la nube

async def http_fetch(session: aiohttp.ClientSession, url: str) -> str | None:
    async with HTTP_LIMIT:
        try:
            async with session.get(
                url, proxy=PROXY,
                timeout=aiohttp.ClientTimeout(total=30),
            ) as resp:
                resp.raise_for_status()
                return await resp.text()
        except (aiohttp.ClientError, asyncio.TimeoutError):
            return None

async def browser_fetch(client: Scrapeless, url: str) -> str:
    async with BROWSER_LIMIT:
        session = client.browser.create(
            ICreateBrowser(proxy_country="US", session_ttl=240)
        )
        async with async_playwright() as p:
            browser = await p.chromium.connect_over_cdp(session.browser_ws_endpoint)
            context = (
                browser.contexts[0] if browser.contexts
                else await browser.new_context()
            )
            page = await context.new_page()
            await page.goto(url, wait_until="networkidle", timeout=60_000)
            html = await page.content()
            await browser.close()
            return html

from urllib.parse import urlparse

# (a) Los hosts pesados en JS siempre escalan: señal más confiable.
JS_HEAVY_HOSTS = {"quotes.toscrape.com"}

def should_escalate(url: str, html: str | None) -> bool:
    # (a) Coincidencia en la lista permitida — host pesado en JS explícito.
    if urlparse(url).hostname in JS_HEAVY_HOSTS:
        return True
    # (b) Señal post-análisis — cuerpo vacío o shell de aplicación reconocible.
    if html is None or len(html) < 2000 or '<div id="root"></div>' in html:
        return True
    return False

async def scrape_one(http_session, client, url):
    html = await http_fetch(http_session, url)
    tier = "http"
    if should_escalate(url, html):
        tier = "browser"
        html = await browser_fetch(client, url)
    return {"url": url, "tier": tier, "html_len": len(html) if html else 0}

async def main(urls):
    client = Scrapeless()
    async with aiohttp.ClientSession() as http_session:
        results = await asyncio.gather(
            *(scrape_one(http_session, client, u) for u in urls)
        )
    return results

if __name__ == "__main__":
    urls = [
        "https://books.toscrape.com/",       # estático — nivel aiohttp
        "https://quotes.toscrape.com/js/",   # JS — escala
    ]
    print(asyncio.run(main(urls)))

[
  {
    "url": "https://books.toscrape.com/",
    "tier": "http",
    "html_len": 51274
  },
  {
    "url": "https://quotes.toscrape.com/js/",
    "tier": "browser",
    "html_len": 9246
  }
]

- **El costo de las conexiones en frío es real.** La primera solicitud en un `ClientSession` nuevo paga TLS + DNS; las solicitudes subsiguientes en la misma sesión reutilizan la conexión. No recrees la sesión por cada solicitud.
- **Los límites de concurrencia dependen del objetivo, no de aiohttp.** Cinco por host es un punto de partida seguro para catálogos públicos; tres es más seguro para orígenes protegidos contra bots; diez es realista para APIs amigables.
- **Las sesiones de navegador en la nube sobreviven a las cargas de páginas individuales.** Si un pipeline necesita iniciar sesión más recorrer más extraer, crea una sesión por unidad de trabajo lógica y reutilízala a través de las páginas dentro de esa unidad — `context.new_page()` es barato dentro de la misma sesión.
- **La resolución DNS se mantiene dentro de aiohttp.** El conector almacena en caché las IPs resueltas durante la vida útil del `ClientSession`. Para rastreadores de larga duración, recicla la sesión cada pocas horas para que DNS no se vuelva obsoleto.
- **`ClientTimeout(total=30)` es por solicitud, no por `gather`.** Un fan-out de 1,000 URL no se agota en 30 s — cada solicitud tiene su propio presupuesto de 30 segundos.

---

## Conclusión: escale sus scrapers asíncronos de Python

El patrón asíncrono se reduce a cuatro movimientos: inicie un `ClientSession`, limite la concurrencia con un `Semaphore`, rote el fan-out a través de proxies residenciales Scrapeless y escale a la minoría renderizada por JS a Scrapeless Scraping Browser a través del SDK de Python más la API asíncrona de Playwright.

Para profundizar en la capa de proxy residencial que gestiona cada recuperación asíncrona, consulte [¿Qué es un proxy SSL?](https://www.scrapeless.com/es/blog/what-is-an-ssl-proxy-2026).

Fije la salida con el sufijo del país en el nombre de usuario del proxy, mantenga los Semáforos por host ajustados, ramifique en la forma del sobre de éxito versus fracaso en lugar de capturar excepciones en línea, y considere una respuesta HTTP vacía como la señal para escalar — no como la respuesta.

---

## ¿Listo para construir su canal de datos potenciado por IA?

Únase a nuestra comunidad para reclamar un plan gratuito y conectarse con desarrolladores que construyen pipelines de scraping asíncronos: [Discord](https://discord.gg/scrapeless) · [Telegram](https://t.me/scrapeless).

Regístrese en [Scrapeless](https://app.scrapeless.com/passport/login/?utm_source=website&utm_medium=blog&utm_campaign=scrapingbrowser&utm_term=async) para obtener el tiempo de ejecución gratuito de Scraping Browser y adaptar los patrones anteriores a los catálogos, feeds y regiones que necesita el pipeline. Detalles de precios en [scrapeless.com/en/pricing](https://www.scrapeless.com/es/pricing); los proxies residenciales están documentados en [scrapeless.com/en/product/proxy-solutions](https://www.scrapeless.com/es/product/proxy-solutions); referencia completa del SDK en [docs.scrapeless.com](https://docs.scrapeless.com).

---

## Preguntas frecuentes

**P1: ¿Cuántas solicitudes concurrentes debo ejecutar por host?**

Para catálogos públicos sin protección contra bots, 10 solicitudes en vuelo por host es un límite seguro. Para orígenes protegidos contra bots, 3 es más realista. El Semáforo es la palanca; comienza bajo, observa las respuestas 429 y ajusta a partir de ahí.

**P2: ¿Necesito proxies residenciales Scrapeless si mi objetivo no está bloqueado desde mi centro de datos?**

Para objetivos HTTP no bloqueados, no — aiohttp funciona sin un proxy. La capa de proxy Scrapeless justifica su uso cuando el objetivo geo-restringe (necesitas salida de EE. UU./Reino Unido/Japón), cuando tu IP de centro de datos está limitada por tasas o bloqueada, o cuando necesitas una IP residencial nueva por solicitud para distribuir el grupo en vuelo a través de muchos orígenes.

**P3: ¿Cuándo debo escalar de aiohttp a Scrapeless Scraping Browser?**

Cuando el HTML que devuelve aiohttp es un shell de aplicación JS sin contenido. Heurística: cuenta los elementos que te interesan después de la recuperación de primer nivel; si la cuenta es cero o muy inferior a la esperada, la página se renderiza del lado del cliente. La capa de navegador en la nube maneja esos casos.

**P4: ¿Es legal el scraping asíncrono?**

Asíncrono es un patrón de transporte; la legalidad depende de qué scrapeas, desde dónde, y bajo qué términos. Los datos visibles públicamente son generalmente accesibles; las jurisdicciones varían; se aplican los términos de servicio del sitio; consulta con un abogado para casos de alto riesgo. Scrapeless accede solo a datos disponibles públicamente.

**P5: ¿Puedo usar aiohttp sin Scrapeless Scraping Browser?**

Sí. La capa de aiohttp (Pasos 3-5) funciona como un scraper asíncrono completo para cualquier objetivo que envíe HTML renderizado. Scrapeless Scraping Browser es la capa de escalado — invocada solo cuando la capa HTTP regresa vacía.

**P6: ¿Cómo puedo fijar la salida a un país específico?**

El país se incluye en el nombre de usuario del proxy residencial Scrapeless como `country_<CC>` (código de dos letras en mayúscula, separado por guiones bajos): `country_US`, `country_UK`, `country_DE`, `country_JP`. Reemplaza el segmento en la cadena de nombre de usuario y el gateway rota cada solicitud a través de IPs residenciales en ese país. Para solicitudes de la capa de navegador, pasa `proxy_country="US"` a `ICreateBrowser(...)` al crear la sesión del navegador en la nube.

**P7: ¿Por qué usar la API asíncrona de Playwright en lugar de un cliente de navegador sincrónico?**
Los clientes de navegador sincrónicos bloquean el bucle de eventos. Todo el propósito de asyncio es mantener el bucle libre; llamar a `page.goto(...)` de forma sincrónica desde dentro de una coroutine detiene todas las demás tareas en vuelo. El `async_playwright` de Playwright es la única opción canónica en Python que mantiene la compatibilidad con coroutine en la capa del navegador en la nube. El SDK de Scrapeless aún genera la sesión; Playwright solo se comunica con CDP a través del `browser_ws_endpoint`.