O Arachne nasceu de uma frustração — scrapers quebrados e dados perdidos

Se o Dogwalk nasceu de uma necessidade, o Arachne nasceu de uma frustração profunda com ferramentas quebradas.

Eu estava cansado de manter scrapers que paravam de funcionar do nada. Um site mudava uma classe CSS, e pronto — dia de dados perdido. Serviços como ScrapingBee e Crawlbase resolviam parte do problema, mas custavam caro e não me davam controle fino sobre o processo.

O estalo

3 de junho de 2026. Eu estava debugando um scraper que quebrou pela terceira vez na mesma semana quando pensei: “Vou fazer meu próprio sistema de extração.”

Na época parecia loucura. Construir uma plataforma de scraping do zero? Mas eu já tinha experiência com Crawl4AI, conhecia as limitações de cada abordagem (Trafilatura, Playwright, requests puro), e sabia exatamente o que um bom sistema precisava ter.

O primeiro arquivo

Curiosamente, o primeiro arquivo do projeto Arachne não foi um scraper. Foi um __init__.py vazio em tests/.

# tests/__init__.py — o primeiro arquivo do Arachne
# 03/06/2026, 22:47
# Ainda não tinha nada para testar, mas já sabia que precisaria

Parece bobo, mas abrir o projeto com um diretório de testes definiu o tom: esse projeto seria feito com disciplina. Não era um experimento de fim de semana — era uma plataforma que eu pretendia usar de verdade.

A arquitetura inicial

O plano era simples em teoria, complexo na prática:

1. Camada de extração: Crawl4AI como engine principal, Trafilatura como fallback leve
2. Cache inteligente: Evitar re-buscar páginas já processadas
3. Saída estruturada: Markdown limpo, não HTML cheio de lixo

# A primeira versão do extrator (simplificada)
async def extrair(url: str) -> str | None:
    try:
        resultado = await crawl4ai.extract(url)
        return resultado.markdown
    except Exception:
        # Fallback pra Trafilatura se Crawl4AI falhar
        return await trafilatura.extract(url)

Essa lógica de fallback, que hoje é automática e tem 4 camadas, começou exatamente assim: um try/except simples.

A matrix de engines — quando usar cada uma

Uma das primeiras coisas que fiz foi mapear cada engine de extração disponível e entender onde cada uma brilha. O resultado virou uma tabela que até hoje guia as decisões de roteamento:

A grande sacada foi não competir entre as engines — cada uma tem um nicho. O problema real era rotear inteligentemente.

# O roteador de engines — versão que foi pra produção
class EngineRouter:
    def __init__(self):
        self.engines = {
            "trafilatura": TrafilaturaEngine(),
            "crawl4ai": Crawl4AIEngine(),
            "playwright": PlaywrightEngine(),
            "browser_evasion": BrowserEvasionEngine(),
        }
        self.cache = CacheEngine()

    async def extract(self, url: str, force: bool = False) -> ExtractionResult:
        # 1. Cache check primeiro — sempre
        if not force:
            cached = await self.cache.get(url)
            if cached:
                return cached

        # 2. Try Crawl4AI (best effort, melhor markdown)
        try:
            return await self.engines["crawl4ai"].extract(url)
        except (ConnectionError, TimeoutError):
            pass

        # 3. Fallback pra Trafilatura (leve, sem JS)
        try:
            return await self.engines["trafilatura"].extract(url)
        except Exception:
            pass

        # 4. Playwright se precisar de JS
        try:
            return await self.engines["playwright"].extract(url)
        except (TimeoutError, CloudflareBlock):
            pass

        # 5. Browser com evasão como último recurso
        return await self.engines["browser_evasion"].extract(url)

Cada chamada de fallback custa tempo, mas é melhor que retornar vazio. A progressão natural (leve → pesado) garante que 80% das páginas são resolvidas na primeira ou segunda tentativa.

O sistema de cache que economiza horas

O cache foi uma das decisões mais acertadas do projeto inicial. Cada requisição repetida consome recursos e tempo — especialmente as que passam pelo Playwright (4-8s cada).

# CacheEngine — SQLite com FTS5 pra buscas no conteúdo
import sqlite3
import json
import hashlib
from datetime import datetime, timedelta

class CacheEngine:
    def __init__(self, db_path: str = "cache/arachne_cache.db"):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS cache (
                url_hash TEXT PRIMARY KEY,
                url TEXT NOT NULL,
                engine TEXT NOT NULL,
                result TEXT NOT NULL,
                content_type TEXT,
                status_code INTEGER,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                accessed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                ttl_hours INTEGER DEFAULT 24
            )
        """)
        self.conn.execute("""
            CREATE INDEX IF NOT EXISTS idx_cache_url
            ON cache(url)
        """)
        self.conn.commit()

    def _hash(self, url: str) -> str:
        return hashlib.sha256(url.encode()).hexdigest()[:16]

    async def get(self, url: str) -> ExtractionResult | None:
        url_hash = self._hash(url)
        row = self.conn.execute("""
            SELECT result, engine, created_at, ttl_hours
            FROM cache WHERE url_hash = ?
        """, (url_hash,)).fetchone()

        if not row:
            return None

        result_json, engine, created_at, ttl = row
        created = datetime.fromisoformat(created_at)
        if datetime.now() - created > timedelta(hours=ttl):
            self.conn.execute("DELETE FROM cache WHERE url_hash = ?", (url_hash,))
            self.conn.commit()
            return None

        # Atualiza accessed_at
        self.conn.execute("""
            UPDATE cache SET accessed_at = CURRENT_TIMESTAMP
            WHERE url_hash = ?
        """, (url_hash,))
        self.conn.commit()

        return ExtractionResult(
            url=url,
            markdown=json.loads(result_json),
            engine=engine,
            cached=True
        )

    async def set(
        self,
        url: str,
        result: str,
        engine: str,
        ttl_hours: int = 24
    ) -> None:
        url_hash = self._hash(url)
        self.conn.execute("""
            INSERT OR REPLACE INTO cache
            (url_hash, url, engine, result, ttl_hours)
            VALUES (?, ?, ?, ?, ?)
        """, (url_hash, url, engine, json.dumps(result), ttl_hours))
        self.conn.commit()

    async def stats(self) -> dict:
        stats = self.conn.execute("""
            SELECT
                COUNT(*) as total,
                SUM(CASE WHEN engine = 'crawl4ai' THEN 1 ELSE 0 END) as crawl4ai,
                SUM(CASE WHEN engine = 'trafilatura' THEN 1 ELSE 0 END) as trafilatura,
                SUM(CASE WHEN engine = 'playwright' THEN 1 ELSE 0 END) as playwright,
                SUM(CASE WHEN engine = 'browser_evasion' THEN 1 ELSE 0 END) as evasion,
                AVG(ttl_hours) as avg_ttl
            FROM cache
        """).fetchone()

        return {
            "total_entries": stats[0],
            "by_engine": {
                "crawl4ai": stats[1] or 0,
                "trafilatura": stats[2] or 0,
                "playwright": stats[3] or 0,
                "browser_evasion": stats[4] or 0,
            },
            "avg_ttl_hours": round(stats[5] or 0, 1),
        }

O cache usa hash SHA256 truncado pra 16 caracteres como chave — colisão é virtualmente impossível pro volume que a gente processa. O TTL de 24 horas evita que páginas dinâmicas fiquem obsoletas, mas ainda assim economiza requisições repetidas no mesmo dia.

A fallback chain que salvou n projetos

A arquitetura de fallback em 4 camadas não foi planejada — emergiu dos fracassos. Cada camada cobre um cenário de falha específico:

# Async pipeline com retry exponencial e fallback automático
async def pipeline_extracao(
    url: str,
    max_tentativas: int = 3
) -> ExtractionResult | None:
    """
    Pipeline completo de extração com 4 camadas de fallback.
    Cada camada tem até max_tentativas com backoff exponencial.
    """
    engines_chain = [
        ("crawl4ai", Crawl4AIEngine()),
        ("trafilatura", TrafilaturaEngine()),
        ("playwright", PlaywrightEngine()),
        ("browser_evasion", BrowserEvasionEngine()),
    ]

    for engine_name, engine_instance in engines_chain:
        for tentativa in range(max_tentativas):
            try:
                resultado = await engine_instance.extract(url)
                if resultado and resultado.markdown:
                    logger.info(
                        "%s extraiu %s em %.2fs (tentativa %d)",
                        engine_name, url, resultado.tempo, tentativa + 1
                    )
                    return resultado
            except CloudflareBlock:
                logger.warning("%s bloqueado por Cloudflare: %s", engine_name, url)
                break  # Não adianta tentar de novo — pula pra próxima engine
            except TimeoutError:
                backoff = 2 ** tentativa
                logger.warning(
                    "%s timeout em %s — backoff %ds",
                    engine_name, url, backoff
                )
                await asyncio.sleep(backoff)
            except Exception as exc:
                logger.error(
                    "%s quebrou em %s: %s", engine_name, url, exc
                )
                break  # Erro não recuperável — próxima engine

    logger.error("TODAS as engines falharam para %s", url)
    return None

O que parece código simples resolveu 90% dos problemas de extração. A chave está no break do CloudflareBlock — não adianta insistir numa engine que o anti-bot já detectou. Melhor pular direto pro Playwright com evasão.

Decisões arquiteturais que moldaram o projeto

1. SQLite como cache, não Redis

Eu poderia ter ido de Redis. Seria mais rápido, mais moderno. Mas SQLite resolve o problema com zero dependências, zero processos extras, e performance mais que suficiente pra dezenas de milhares de URLs cacheadas.

# Benchmark informal: 10K hits no cache
# SQLite: 47ms total (4.7µs por hit)
# Redis (via docker): 112ms total (11.2µs por hit)
# Conclusão: SQLite ganha em simplicidade e performance

2. Extração assíncrona desde o primeiro dia

Fiz questão de usar asyncio desde o começo. Scraping é I/O-bound por natureza — esperar resposta HTTP, esperar renderização, esperar escrita no cache. Bloquear a thread principal seria desperdício.

# O Arachne processa N URLs em paralelo com asyncio.gather
async def extrair_lote(urls: list[str]) -> list[ExtractionResult]:
    tarefas = [pipeline_extracao(url) for url in urls]
    resultados = await asyncio.gather(*tarefas, return_exceptions=True)
    return [
        r for r in resultados
        if isinstance(r, ExtractionResult)
    ]

3. Engine como plugin, não acoplamento

Cada engine implementa uma interface comum. Isso permite adicionar (ou remover) engines sem tocar no pipeline central:

class BaseEngine(ABC):
    @abstractmethod
    async def extract(self, url: str) -> ExtractionResult:
        pass

    @property
    @abstractmethod
    def name(self) -> str:
        pass

    @property
    @abstractmethod
    def supported_content_types(self) -> list[str]:
        pass

Hoje o Arachne tem 9 engines registradas, de Trafilatura a PDF Vision. Tudo porque a interface foi desenhada pra ser extensível desde o início.

4. Log estruturado como ferramenta de debug

Não subestime o valor de logs bem feitos. Cada etapa da extração loga: engine usada, tempo decorrido, tamanho do resultado, código HTTP.

logger = structlog.get_logger()
# Exemplo real de log:
# 2026-06-03 22:47:01 [crawl4ai] extraiu https://exemplo.com em 2.34s (2367 chars)
# 2026-06-03 22:47:02 [cache] cache HIT para https://exemplo.com (ttl restante: 18h)

Números reais do primeiro mês

Trinta dias depois do primeiro commit, o Arachne já processava:

44.6% de cache hit no primeiro mês — quase metade das requisições nunca chegou a bater na rede. Cada cache hit economizou ~3 segundos de processamento. Foram 1.892 × 3s = ~94 minutos economizados em um mês, só com cache.

Por que “Arachne”?

Aranha que tece teias. Fazia sentido — scrapers são como aranhas digitais, percorrendo a web atrás de informação. O nome ficou e virou identidade do projeto.

A frustração de junho se transformou no que hoje é uma plataforma com pipeline multi-engine, cache em SQLite, indexação FTS5, e integração com RAG. Tudo porque um scraper quebrou na hora errada.

Comandos úteis

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