Raspagem de Web Assíncrona em Python: Escale para 10.000+ URLs com aiohttp e Scrapeless

pip install aiohttp scrapeless playwright
playwright install chromium

export SCRAPELESS_API_KEY="seu_token_api_aqui"
export SCRAPELESS_CHANNEL_ID="seu_id_de_canal"          # impresso no início do nome de usuário
export SCRAPELESS_PROXY_PASS="sua_senha_do_canal"
export SCRAPELESS_PROXY_GATEWAY="gw-us.scrapeless.io"   # veja abaixo para gateways regionais

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"Buscou {len(html):,} caracteres via saída residencial dos EUA")

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 solicitações concorrentes por host. Ajuste pelo alvo — catálogos públicos
# toleram mais, origens protegidas contra bots querem 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 do catálogo
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        pages = await asyncio.gather(*tasks)
    print(f"Obtivemos {len(pages)} páginas, total de {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"Bem-sucedidos: {len(ok)}   Falhados: {len(failed)}")

    # Arquivo de cartas mortas para revisão separada — o pipeline nunca bloqueia por falhas
    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()  # lê SCRAPELESS_API_KEY do env
    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/ é o sandbox canônico "precisa de JS".
    # HTTP simples retorna 0 elementos de citação; renderizado na nuvem retorna 10.
    html = await render_via_cloud_browser("https://quotes.toscrape.com/js/")
    print(f"Renderizado {len(html):,} caracteres incluindo o DOM pós-renderização")

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)      # camada aiohttp
BROWSER_LIMIT = asyncio.Semaphore(3)    # camada de navegador na nuvem

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) Hosts conhecidos com muito JS sempre elevam — sinal mais confiável.
JS_HEAVY_HOSTS = {"quotes.toscrape.com"}

def should_escalate(url: str, html: str | None) -> bool:
    # (a) Acerto da lista permitida — host explícito com muito JS.
    if urlparse(url).hostname in JS_HEAVY_HOSTS:
        return True
    # (b) Sinal pós-análise — corpo vazio ou shell de aplicativo reconhecível.
    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 — camada aiohttp
        "https://quotes.toscrape.com/js/",   # JS — eleva
    ]
    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
  }
]

- **O custo da conexão fria é real.** A primeira solicitação em uma nova `ClientSession` paga por TLS + DNS; solicitações subsequentes na mesma sessão reutilizam a conexão. Não recrie a sessão a cada solicitação.
- **Os limites de concorrência dependem do destino, não do aiohttp.** Cinco por host é um ponto de partida seguro para catálogos públicos; três é mais seguro para origens protegidas contra bots; dez é realista para APIs amigáveis.
- **As sessões de navegador em nuvem duram mais do que carregamentos de página únicos.** Se um pipeline precisar de login mais navegação mais extração, crie uma sessão por unidade de trabalho lógica e reutilize-a nas páginas dentro dessa unidade — `context.new_page()` é barato dentro da mesma sessão.
- **A resolução de DNS permanece dentro do aiohttp.** O conector armazena em cache os IPs resolvidos durante a vida útil da `ClientSession`. Para rastreadores de longa duração, recicle a sessão a cada poucas horas para que o DNS não fique desatualizado.
- **`ClientTimeout(total=30)` é por solicitação, não por `gather`.** Um fan-out de 1.000 URLs não expira em 30 s — cada solicitação recebe seu próprio orçamento de 30 segundos.

---

## Conclusão: escale seus scrapers Python assíncronos

O padrão assíncrono se resume a quatro movimentos: inicie uma `ClientSession`, limite a concorrência com um `Semaphore`, roteie o fan-out através de proxies residenciais Scrapeless e eleve a minoria renderizada em JS para o Scrapeless Scraping Browser via o SDK Python mais a API assíncrona do Playwright.

Para se aprofundar na camada de proxy residencial que roteia cada busca assíncrona, veja [O que é um proxy SSL?](https://www.scrapeless.com/pt/blog/what-is-an-ssl-proxy-2026).

Fixe a saída com o sufixo do país no nome de usuário do proxy, mantenha os Semáforos por host ajustados, ramifique com base na forma do envelope de sucesso versus falha em vez de capturar exceções de forma inline, e trate uma resposta HTTP vazia como o sinal para escalar — não como a resposta.

---

## Pronto para construir seu pipeline de dados alimentado por IA?

Junte-se à nossa comunidade para reivindicar um plano gratuito e conectar-se com desenvolvedores que constroem pipelines de scraping assíncronos: [Discord](https://discord.gg/scrapeless) · [Telegram](https://t.me/scrapeless).

Inscreva-se em [Scrapeless](https://app.scrapeless.com/passport/login/?utm_source=website&utm_medium=blog&utm_campaign=scrapingbrowser&utm_term=async) para um tempo de execução gratuito do Scraping Browser e adapte os padrões acima aos catálogos, feeds e regiões que o pipeline precisa. Detalhes sobre preços em [scrapeless.com/en/pricing](https://www.scrapeless.com/pt/pricing); proxies residenciais são documentados em [scrapeless.com/en/product/proxy-solutions](https://www.scrapeless.com/pt/product/proxy-solutions); referência completa do SDK em [docs.scrapeless.com](https://docs.scrapeless.com).

---

## FAQ

**Q1: Quantas solicitações concorrentes devo executar por host?**

Para catálogos públicos sem pilha anti-bot, 10 solicitações em andamento por host é um teto seguro. Para origens protegidas contra bots, 3 é mais realista. O Semaphore é a alavanca; comece baixo, fique atento a respostas 429, e ajuste a partir daí.

**Q2: Eu preciso de proxies residenciais Scrapeless se meu destino estiver desbloqueado do meu datacenter?**

Para destinos HTTP desbloqueados, não — o aiohttp funciona sem proxy. A camada de proxy Scrapeless se justifica quando o destino geo-restringe (você precisa de saída dos EUA/Reino Unido/Japão), quando o IP do seu datacenter é limitado em taxa ou bloqueado, ou quando você precisa de um IP residencial fresco por solicitação para espalhar o pool em andamento por muitas origens.

**Q3: Quando devo escalar de aiohttp para Scrapeless Scraping Browser?**

Quando o HTML que o aiohttp retorna é um shell de aplicativo JS sem conteúdo. Heurística: conte os elementos que você se importa após a busca de primeiro nível; se a contagem for zero ou muito abaixo do esperado, a página é renderizada no lado do cliente. A camada de navegador em nuvem lida com isso.

**Q4: O scraping assíncrono é legal?**

Assíncrono é um padrão de transporte; a legalidade depende do que você raspa, de onde e sob quais termos. Dados visíveis publicamente são geralmente acessíveis; as jurisdições variam; os termos de serviço do site se aplicam; consulte um advogado para casos de uso de alto risco. A Scrapeless acessa apenas dados publicamente disponíveis.

**Q5: Posso usar aiohttp sem o Scrapeless Scraping Browser?**

Sim. A camada aiohttp (Passos 3–5) funciona como um scraper assíncrono completo para qualquer destino que envie HTML renderizado. O Scrapeless Scraping Browser é a camada de escalonamento — invocada apenas quando a camada HTTP retorna vazia.

**Q6: Como faço para fixar a saída para um país específico?**

O país é inserido no nome de usuário do proxy residencial Scrapeless como `country_<CC>` (código de duas letras em maiúsculas, separado por underscore): `country_US`, `country_UK`, `country_DE`, `country_JP`. Substitua o segmento na string do nome de usuário e o gateway roteia cada solicitação através de IPs residenciais naquele país. Para solicitações de camada de navegador, passe `proxy_country="US"` para `ICreateBrowser(...)` ao criar a sessão do navegador em nuvem.

**Q7: Por que usar a API assíncrona do Playwright em vez de um cliente de navegador síncrono?**
Os clientes de navegador síncronos bloqueiam o loop de eventos. O objetivo principal do asyncio é manter o loop livre; chamar um `page.goto(...)` síncrono de dentro de uma corrotina atrasa todas as outras tarefas em andamento. O `async_playwright` do Playwright é a única opção canônica em Python que mantém a camada de navegador em nuvem amigável para corrotinas. O SDK Scrapeless ainda gera a sessão — o Playwright fala apenas CDP com o `browser_ws_endpoint`.