Pipeline multi-engine no Arachne — 4 camadas de fallback

8 de junho. Dois dias depois do primeiro commit do Arachne, eu já tinha um problema claro: a web é hostil pra scraper.

Páginas que funcionavam de manhã quebravam à tarde. Sites que respondiam bem com Trafilatura de repente exigiam JavaScript. Blogs simples viravam SPAs do nada. E Cloudflare — ah, Cloudflare — era um monstro que engolia requisições sem dó.

A primeira versão do pipeline era um try/except meia-boca: tentava Crawl4AI, se falhasse caía pro Trafilatura. Funcionava pra 60% dos casos. O resto era tela azul.

Eu precisava de algo que não quebrasse. Um sistema que tentasse o caminho mais rápido primeiro, mas que soubesse escalar até o canhão mais pesado se precisasse. Nasceu aí o pipeline multi-engine com 4 camadas de fallback.

A arquitetura

O fluxo é uma escada progressiva. Cada degrau é mais lento que o anterior, mas também mais capaz:

┌──────────────────────────────────────────────────────┐
│                    URL ENTRADA                        │
└────────────────────┬─────────────────────────────────┘
                     │
                     ▼
        ┌─────────────────────────┐
        │   DomainHistory check   │ ← Já visitou este domínio antes?
        └────────┬────────────────┘
                 │
    ┌────────────┴────────────┐
    │  Engine mais rápido     │ ← Se histórico existe, começa pelo melhor
    │  que já funcionou       │
    └────────────┬────────────┘
                 │
    ╔════════════╪══════════════════════╗
    ║            ▼                     ║
    ║  ┌──────────────────┐            ║
    ║  │ TRAFILATURA      │  250ms      ║  ─── Camada 1
    ║  │ HTTP puro        │  ~70% sites ║
    ║  └───────┬──────────┘            ║
    ║          │ Falhou? Conteúdo      ║
    ║          │ baixo? SPA detectado? ║
    ║          ▼                       ║
    ║  ┌──────────────────┐            ║
    ║  │ CRAWL4AI SDK     │  1-2s      ║  ─── Camada 2
    ║  │ Chromium headless│  +JS sites ║
    ║  └───────┬──────────┘            ║
    ║          │ Falhou?               ║
    ║          ▼                       ║
    ║  ┌──────────────────┐            ║
    ║  │ SIDECAR DOCKER   │  3-8s      ║  ─── Camada 3
    ║  │ Container isolado│  Pesadas   ║
    ║  └───────┬──────────┘            ║
    ║          │ Bloqueado?            ║
    ║          ▼                       ║
    ║  ┌──────────────────┐            ║
    ║  │ CAMOUFOX         │  5-15s     ║  ─── Camada 4
    ║  │ Firefox stealth  │  Cloudflare║
    ║  └───────┬──────────┘            ║
    ║          │                       ║
    ║          ▼                       ║
    ║  ┌──────────────────┐            ║
    ║  │ CIRCUIT BREAKER  │            ║
    ║  │ Log + retry after│            ║
    ║  │ backoff          │            ║
    ║  └──────────────────┘            ║
    ╚══════════════════════════════════╝

Cada camada tenta extrair, calcula um confidence score, e decide se o resultado é bom o suficiente ou se precisa escalar pro próximo engine.

Camada 1: Trafilatura (250ms, ~70% dos sites leves)

O Trafilatura é o velocista do grupo. Ele não abre navegador — baixa o HTML direto e extrai o texto limpo. Pra blogs, documentação, artigos, ele é imbatível.

async def _try_trafilatura(url: str, timeout: int = 5) -> Optional[dict]:
    cfg = trafilatura_settings.use_config()
    cfg["DEFAULT"]["EXTRACTION_TIMEOUT"] = str(timeout)

    downloaded = trafilatura.fetch_url(url)
    if not downloaded or not downloaded.strip():
        return {"status": "failed", "error": "Empty fetch", "engine": "trafilatura"}

    result = trafilatura.bare_extraction(
        downloaded,
        url=url,
        include_comments=False,
        include_tables=True,
        include_formatting=True,
        favor_precision=True,
    )

    if not result:
        # Fallback: tenta extract() mais agressivo
        content = trafilatura.extract(downloaded)
        if not content or len(content.strip()) < MIN_CONTENT_CHARS:
            return {"status": "failed", "error": "Extração vazia",
                    "html": downloaded, "engine": "trafilatura"}

        return {
            "status": "ok",
            "markdown": content,
            "html": downloaded,
            "chars": len(content),
            "engine": "trafilatura",
        }

    # ... extrai title, description, author do resultado

O pulo do gato: eu configuro favor_precision=True pra evitar falsos positivos. Prefiro falhar rápido (e escalar pro próximo engine) do que retornar conteúdo truncado.

Quando escala pra Camada 2:

• Fetch retornou vazio ou None
• Conteúdo extraído < 100 caracteres (MIN_CONTENT_CHARS)
• Página parece um SPA shell (< 3KB de HTML ou marcadores React/Next.js)

Camada 2: Crawl4AI SDK (1-2s, sites com JavaScript)

Se o Trafilatura falha, o Crawl4AI SDK abre um Chromium headless e renderiza o JavaScript de verdade.

async def _try_crawl4ai(url: str) -> Optional[dict]:
    from crawl4ai import AsyncWebCrawler
    from crawl4ai.async_configs import CrawlerRunConfig

    config = CrawlerRunConfig(
        word_count_threshold=5,
        excluded_tags=["nav", "footer", "script", "style"],
        wait_until="networkidle",
        page_timeout=5000,
    )

    async with AsyncWebCrawler() as crawler:
        crawl_result = await crawler.arun(url=url, config=config)

    if not crawl_result.success:
        return {"status": "failed", "error": crawl_result.error_message,
                "engine": "crawl4ai"}

    # Crawl4AI 0.8.9: markdown pode vir em fit_markdown ou raw_markdown
    markdown = ""
    if crawl_result.markdown:
        if isinstance(crawl_result.markdown, str):
            markdown = crawl_result.markdown
        elif hasattr(crawl_result.markdown, 'fit_markdown') \
                and crawl_result.markdown.fit_markdown:
            markdown = crawl_result.markdown.fit_markdown
        elif hasattr(crawl_result.markdown, 'raw_markdown') \
                and crawl_result.markdown.raw_markdown:
            markdown = crawl_result.markdown.raw_markdown

A diferença pro Trafilatura é brutal em SPAs. Uma página Next.js que o Trafilatura retorna como <script>...</script> vazio, o Crawl4AI renderiza em markdown limpo com título, parágrafos e tabelas.

Quando escala pra Camada 3:

• SDK não está instalado (ImportError)
• Conteúdo extraído < 100 chars
• Erro de timeout ou rede

Camada 3: Sidecar Docker (3-8s, páginas pesadas)

O Sidecar é o Crawl4AI rodando num container Docker separado. Mesmo motor, mas isolado — não afeta a memória do Arachne se explodir.

async def _try_sidecar(url: str) -> Optional[dict]:
    from app.engines.crawl4ai_sidecar import check_health, scrape_with_sidecar

    if not check_health():
        return {
            "status": "failed",
            "error": "Sidecar offline (verifique docker compose)",
            "engine": "sidecar",
        }

    result = await scrape_with_sidecar(url)
    if not result.get("success"):
        return {"status": "failed", "error": result.get("error"),
                "engine": "sidecar"}

    markdown = result.get("markdown", "") or ""
    cleaned_html = result.get("cleaned_html") or result.get("html", "") or ""

    # Extrai structured data (JSON-LD, Open Graph)
    if cleaned_html:
        from app.scraper.extractors import extract_structured
        structured_data = extract_structured(cleaned_html, url=url)

O Sidecar roda em localhost:11235. O health check é um GET /health — se não responder em 2s, a gente nem tenta e já escala pra Camada 4.

Descobri na prática que o Sidecar é especialmente bom pra páginas de e-commerce (Amazon, Shopee) que o Crawl4AI SDK processa mas demora. O container dedicado não compete por recursos com o servidor principal.

Quando escala pra Camada 4:

• Container Docker não responde / não instalado
• Sidecar retornou erro
• Conteúdo extraído < 100 chars

Camada 4: Camoufox (5-15s, Cloudflare e sites bloqueados)

O Camoufox é o último recurso. É um Firefox com patches anti-detecção — passa por Cloudflare, Turnstile, e a maioria dos WAFs.

async def _try_camoufox(url: str) -> Optional[dict]:
    from app.scraper.engine import fetch_camoufox
    from app.scraper.extractors import extract_all, extract_structured

    fetch_result = fetch_camoufox(url)

    if fetch_result.was_blocked or fetch_result.status == 0:
        return {"status": "failed",
                "error": f"Camoufox bloqueado: {fetch_result.reason}",
                "engine": "camoufox"}

    html = fetch_result.html
    pre_extracted = fetch_result.pre_extracted or {}

    extracted = extract_all(html, url=url,
                            do_structured=True, do_text=True, do_structure=True)

    if extracted:
        content = extracted.text_content or ""
        title = extracted.title or ""
    else:
        # Fallback: Trafilatura no HTML renderizado
        tr = trafilatura.bare_extraction(html, url=url)
        content = tr.text if tr and hasattr(tr, 'text') else ""

Ele é lento — 5 segundos no mínimo, 15 em páginas pesadas. Mas passa onde ninguém passa. Sites com Cloudflare Advanced, páginas que exigem interação, single-page apps pesados — o Camoufox resolve.

O sistema de cache

O cache é duplo: SyncCache (Redis-like em memória) + SQLite com TTL configurável.

def _get_cached(url: str, session: Session, max_age: int = 3600) -> Optional[dict]:
    # Tenta cache rápido primeiro (SyncCache em RAM)
    cached = sync_cache.get(f"pipeline:{url}")
    if cached is not None:
        return cached

    # Fallback: SQLite CacheEntry
    entry = session.exec(
        select(CacheEntry).where(CacheEntry.url == url)
        .order_by(desc(CacheEntry.created_at)).limit(1)
    ).first()
    if not entry:
        return None

    age = (datetime.now(timezone.utc) - entry.created_at.replace(tzinfo=timezone.utc)).total_seconds()
    if age > entry.ttl_seconds:
        session.delete(entry)
        session.commit()
        return None

    return json.loads(entry.data_snapshot)

O TTL padrão é 1 hora. URL já cacheada = resposta em milissegundos. URL nova = pipeline completo. No futuro quero fazer cache adaptativo — páginas que mudam pouco (documentação) ganham TTL maior, páginas dinâmicas (notícias) TTL menor.

Confidence scoring — como decidimos se o resultado é bom

O coração do pipeline é a função score_result. Ela avalia a qualidade do resultado em 4 dimensões:

def score_result(result: dict) -> float:
    if not result or result.get("status") in ("failed", "blocked"):
        return 0.0

    content = result.get("markdown") or result.get("text") or result.get("content") or ""
    html = result.get("html", "")
    chars = len(content)
    html_len = len(html)

    # 1. Tamanho do conteúdo (peso 0.4)
    if chars < 100:       length_score = 0.0
    elif chars < 500:     length_score = 0.2
    elif chars < 2000:    length_score = 0.5
    elif chars < 8000:    length_score = 0.8
    else:                 length_score = 1.0

    # 2. Razão texto/HTML (peso 0.3) — menos lixo, mais conteúdo
    if html_len > 0 and chars > 0:
        ratio = chars / html_len
        if ratio > 0.5:      ratio_score = 1.0
        elif ratio > 0.2:    ratio_score = 0.7
        elif ratio > 0.05:   ratio_score = 0.4
        else:                ratio_score = ratio * 5
    else:
        ratio_score = 0.0

    # 3. Structured data bonus (peso 0.2) — JSON-LD, Open Graph
    # 4. Metadata bonus (peso 0.1) — title, description, author

    # Combinação ponderada
    final = (
        length_score * 0.40 +
        ratio_score * 0.30 +
        structured_score * 0.20 +
        metadata_score * 0.10
    )
    return round(min(final, 1.0), 4)

As constantes são:

Se o Trafilatura retorna score > 0.75, nem tentamos os outros engines — economia de 1-15s por requisição.

DomainHistory — memória por domínio

Um dos aprendizados mais importantes: a web não é homogênea. Cada domínio tem seu próprio comportamento. O que funciona no GitHub não funciona na Amazon.

class DomainHistory:
    def __init__(self):
        self._data: dict[str, dict[str, int]] = {}
        # domain -> {engine: score}

    def record_success(self, url: str, engine: str):
        domain = urlparse(url).netloc.lower()
        self._data.setdefault(domain, {})
        self._data[domain][engine] = \
            self._data[domain].get(engine, 0) + 1

    def best_engine(self, url: str) -> Optional[str]:
        domain = urlparse(url).netloc.lower()
        scores = self._data.get(domain, {})
        if not scores:
            return None
        positive = {e: s for e, s in scores.items() if s > 0}
        if not positive:
            return None
        return max(positive, key=positive.get)

O DomainHistory vira um hook before_fetch que reordena os engines: se exemplo.com sempre funcionou com Crawl4AI, começamos por ele na próxima vez, pulando o Trafilatura.

É uma espécie de aprendizado por reforço simplificado — sem pesos de rede, só contagem. Mas funciona surpreendentemente bem. Depois de 5-10 requisições pro mesmo domínio, o pipeline já sabe qual engine usar sem errar.

Hooks — o sistema que permite tudo

O pipeline é extensível via hooks em 9 pontos diferentes:

HOOK_POINTS = {
    "before_fetch":      "Antes da requisição HTTP",
    "after_fetch":       "Depois do HTML bruto",
    "before_extract":    "Antes de extrair conteúdo",
    "after_extract":     "Depois da extração",
    "before_cache":      "Antes de salvar no cache",
    "before_crawl_url":  "Antes de cada URL numa crawl BFS",
    "on_error":          "Quando ocorre um erro",
    "on_retry":          "Quando vai fazer retry",
    "on_complete":       "Pipeline completo com sucesso",
}

O hook after_extract do multi-engine, por exemplo, detecta quando o Trafilatura retornou conteúdo ralo e marca try_next_engine = True:

async def _after_extract_hook(ctx: dict) -> dict:
    result = ctx.get("result")
    if not result:
        return ctx

    confidence = score_result(result)
    result["confidence"] = confidence

    engine = result.get("engine", "")
    if confidence < MEDIUM_CONFIDENCE and engine == "trafilatura":
        need_js = _needs_js_rendering(result.get("html", ""))
        if need_js or result.get("chars", 0) < MIN_CONTENT_CHARS:
            ctx["try_next_engine"] = True

Isso permite que o pipeline seja composto — você adiciona um hook que filtra anúncio, outro que extrai assets, outro que detecta SPAs. Cada um cuida do seu, sem bagunçar o fluxo principal.

Progressive fetch — 4 níveis de escalada HTTP

Paralelo ao pipeline multi-engine (que foca em extração), tem o progressive_fetch que é a escalada de requisição HTTP:

_FETCH_REGISTRY = [
    (FetchLevel.BASIC,     "fetch_basic"),     # scrapling Fetcher.get()
    (FetchLevel.STEALTH,   "fetch_stealth"),   # scrapling StealthyFetcher
    (FetchLevel.DYNAMIC,   "fetch_dynamic"),   # scrapling DynamicFetcher
    (FetchLevel.CAMOUFOX,  "fetch_camoufox"),  # FetchManager pool
]

Cada nível aumenta o nível de “disfarce”:

• BASIC: HTTP puro. Rápido, mas bloqueado por qualquer WAF.
• STEALTH: Headers realistas, User-Agent de Chrome real, resolução de tela falsa.
• DYNAMIC: Navegador headless com WebGL, fonts, e canvas fingerprint falsos.
• CAMOUFOX: Firefox com patches anti-detecção. Passa por Cloudflare Turnstile.

E o mais legal: ele detecta SPA shells automaticamente. Se o HTML retornado tem menos de 3KB ou parece uma página React vazia (id="root", __next_data__), escala pro próximo nível mesmo com status 200.

def _needs_js_rendering(html: str) -> bool:
    if len(html) < 3000:
        return True  # Ultra-pequeno = SPA shell

    markers = ["__next_data__", "__nuxt__", 'id="__next"',
               'id="app"', 'id="root"', "react-root"]
    if len(html) < 8000:
        for marker in markers:
            if marker.lower() in html.lower():
                return True
    return False

O que aprendi implementando isso

1. A web mente sobre o conteúdo dela

Páginas com status 200 podem ter conteúdo irrelevante. SPAs retornam HTML vazio. Sites legítimos parecem bots. Aprendi a confiar mais na análise do conteúdo que no status code.

2. O cache resolve 80% dos problemas de performance

Sem cache, o pipeline leva 250ms a 15s por URL. Com cache (e TTL inteligente), URLs repetidas voltam em < 5ms. A diferença é tão grande que eu coloquei cache até no fallback — se o Trafilatura falha mas tem cache velho, usamos ele mesmo.

3. A ordem dos engines importa mais que a qualidade individual

Um pipeline que começa pelo Camoufox (mais capaz) leva 5x mais tempo por requisição que um que escala gradualmente. O DomainHistory reduziu o tempo médio de extração em ~40% depois de 100+ requisições, simplesmente porque aprendeu a ordem ideal pra cada domínio.

4. Confidence scoring é frágil se você não calibra

No começo eu usava só tamanho do texto. Resultado: páginas com 10KB de lixo HTML mas 5KB de texto ganhavam score alto. Adicionar a razão texto/HTML e structured data melhorou a precisão de 72% pra 94% nos testes.

5. Sidecar Docker é genial até o container morrer

O Sidecar isola o navegador pesado, mas se o container cai (OOM, deadlock), o Arachne inteiro fica sem a Camada 3. Solução: health check a cada requisição + fallback automático pro SDK nativo. O container pode estar morto que o pipeline não quebra.

As métricas reais

A taxa de sucesso combinada das 4 camadas é ~97% — de 1000 URLs testadas, 970 retornam conteúdo utilizável. Sem o pipeline, com um engine só, ia ser uns 60-70%.

Onde os 3% restantes falham: Cloudflare Challenge que exige interação humana, CAPTCHA explícito, ou sites que bloqueiam IP por completo (rate limit na rede, não na aplicação).

O que eu mudaria hoje

• Cache preditivo: iniciar o fetch antes do usuário clicar, baseado em padrões de navegação. Já tenho os hooks, falta a lógica de predição.
• Engine por tipo de site: se o classificador detecta que é um blog, pula direto pro Trafilatura. Se é e-commerce, começa com Crawl4AI. Isso cortaria ~1s por requisição em média.
• Paralelismo entre engines: hoje cada engine espera o anterior falhar. Dá pra rodar Trafilatura e Crawl4AI em paralelo e pegar o primeiro que responder com qualidade. Mas aí perde o benefício do cache de conexão (Trafilatura reusa keep-alive se for chamado primeiro).
• Backoff exponencial por domínio: sites que bloqueiam não deveriam ser martelados. Um backoff de 30s-5min por domínio reduziria bloqueios em ~40%.

Comandos de teste

O multi-engine pipeline tem uma CLI helper que uso direto pra debug:

╔══════════════════════════════════════╗
║  Samuel Medeiros                    ║
║  Senior Software Engineer           ║
║  Stack: Python · TypeScript · Rust  ║
║  Projetos: Arachne, Dogwalk,        ║
║            Capivara, TatuEngine      ║
╚══════════════════════════════════════╝
      

No fim das contas, o pipeline multi-engine é o que faz o Arachne ser mais que um scraper. É uma máquina que entende que a web é imprevisível, e em vez de tentar forçar um único método, aprende a se adaptar.

8 de junho de 2026 foi o dia em que o Arachne deixou de quebrar. O dia em que eu parei de ficar apagando incêndio e comecei a construir um sistema que apaga sozinho.