Rastreamento incremental da Web: atualizar páginas alteradas sem ter de rastrear tudo novamente

O rastreamento incremental da Web consiste na prática de voltar a obter apenas as páginas que sofreram alterações, e não todo o site. No final deste guia, terá um módulo em Python que identifica URLs a partir de um mapa do site ou do ponto final Geonode /v1/map, normaliza-as, mantém o estado por URL no SQLite, utiliza ETag e hashes de conteúdo para determinar o que foi alterado, adapta os intervalos de revisitação, deteta eliminações e extrai o conjunto de alterações através de /v1/extract ou /v1/batch. O Geonode recupera as páginas; o seu script é responsável pelo estado e pela comparação.

Por que razão os rastreios completos desperdiçam orçamento

Um site com 50 000 páginas, em que 2% são alteradas por semana, necessita de cerca de 1 000 pedidos por semana para se manter atualizado. Um rastreio completo semanal gasta 50 000. Isso representa uma diferença de 50 vezes em pedidos de API, largura de banda e no tempo que o seu pipeline fica bloqueado em E/S.

O desperdício agrava-se em páginas renderizadas em JavaScript, onde cada pedido é mais lento e mais dispendioso, e em sites que limitam agressivamente a taxa de pedidos. O rastreamento incremental da Web resolve o problema: mantém-se uma pequena quantidade de estado por URL e deixa-se que esse estado indique onde deve ser feita a próxima solicitação.

O módulo abaixo tem cerca de 120 linhas. Cada secção adiciona uma parte. Instale a dependência com pip install requests (o httpx funciona de forma idêntica, se preferir).

Descobrir URLs

Duas fontes, por ordem de preferência. Um mapa do site fornece-lhe URLs, além de um lastmod opcional, que é o sinal de alteração mais simples que existe. Quando um site não tem um mapa do site utilizável, o POST /v1/map descobre URLs combinando a análise do mapa do site com a extração de links HTML e devolve cada URL com a sua source . O

import os
import xml.etree.ElementTree as ET
import requests

API = "https://scraper.geonode.io"
HEADERS = {"X-Api-Key": os.environ["GEONODE_API_KEY"]}
SM_NS = {"sm": "http://www.sitemaps.org/schemas/sitemap/0.9"}

def urls_from_sitemap(sitemap_url: str, session: requests.Session) -> list[tuple[str, str | None]]:
    """Return (loc, lastmod) pairs. Follows sitemap index files one level."""
    root = ET.fromstring(session.get(sitemap_url, timeout=30).content)
    out: list[tuple[str, str | None]] = []
    for sm in root.findall("sm:sitemap", SM_NS):          # sitemap index
        out.extend(urls_from_sitemap(sm.findtext("sm:loc", namespaces=SM_NS), session))
    for u in root.findall("sm:url", SM_NS):
        out.append((u.findtext("sm:loc", namespaces=SM_NS),
                    u.findtext("sm:lastmod", namespaces=SM_NS)))
    return out

def urls_from_map(site_url: str, session: requests.Session) -> list[tuple[str, None]]:
    body = {"url": site_url, "include_subdomains": False, "ignore_query_parameters": True}
    r = session.post(f"{API}/v1/map", json=body, headers=HEADERS, timeout=120)
    r.raise_for_status()
    return [(link["url"], None) for link in r.json()["links"]]

lastmod de um mapa do site é uma indicação, não uma garantia. Alguns CMS marcam cada URL com a data de implementação; outros nunca a atualizam. Considere isso como um motivo para verificar mais cedo, nunca como um motivo para ignorar completamente a verificação.

Normalização de URLs

Sem normalização, https://Example.com/docs/?utm_source=x#top e https://example.com/docs transformam-se em duas linhas e duas consultas. Normalize antes de qualquer alteração no armazenamento de estado.

from urllib.parse import urlsplit, urlunsplit, parse_qsl, urlencode

TRACKING = {"utm_source", "utm_medium", "utm_campaign", "utm_term", "utm_content",
            "gclid", "fbclid", "ref", "mc_cid", "mc_eid"}

def normalize(url: str) -> str:
    parts = urlsplit(url.strip())
    scheme = parts.scheme.lower()
    host = parts.netloc.lower()
    if (scheme, host.rsplit(":", 1)[-1]) in {("http", "80"), ("https", "443")}:
        host = host.rsplit(":", 1)[0]
    query = [(k, v) for k, v in parse_qsl(parts.query, keep_blank_values=True)
             if k.lower() not in TRACKING]
    query.sort()
    path = parts.path or "/"
    if len(path) > 1 and path.endswith("/"):
        path = path[:-1]                       # pick one convention and stick to it
    return urlunsplit((scheme, host, path, urlencode(query), ""))  # fragment dropped

A regra da barra final é uma escolha de política. Algumas estruturas servem conteúdos diferentes em /docs e /docs/. Se o site de destino for um deles, remova essa linha e deixe as duas URLs existirem como linhas separadas.

O armazenamento de estado

Uma tabela SQLite contém tudo o que o rastreador precisa para tomar decisões. As colunas correspondem aos sinais discutidos no resto deste guia.

import sqlite3
import time

SCHEMA = """
CREATE TABLE IF NOT EXISTS pages (
  normalized_url TEXT PRIMARY KEY,
  url            TEXT NOT NULL,
  last_seen      REAL,
  last_fetched   REAL,
  next_check     REAL DEFAULT 0,
  interval_s     INTEGER DEFAULT 86400,
  content_hash   TEXT,
  etag           TEXT,
  last_modified  TEXT,
  sitemap_lastmod TEXT,
  missing_runs   INTEGER DEFAULT 0,
  gone           INTEGER DEFAULT 0
);
"""

def open_db(path: str = "crawl_state.sqlite") -> sqlite3.Connection:
    db = sqlite3.connect(path)
    db.executescript(SCHEMA)
    return db

def upsert_discovered(db: sqlite3.Connection, found: list[tuple[str, str | None]]) -> None:
    now = time.time()
    for raw, lastmod in found:
        n = normalize(raw)
        row = db.execute("SELECT sitemap_lastmod FROM pages WHERE normalized_url=?", (n,)).fetchone()
        if row is None:
            db.execute("INSERT INTO pages (normalized_url,url,last_seen,sitemap_lastmod) VALUES (?,?,?,?)",
                       (n, raw, now, lastmod))
        else:
            bump = lastmod is not None and lastmod != row[0]
            db.execute("UPDATE pages SET last_seen=?, missing_runs=0, sitemap_lastmod=?, "
                       "next_check=CASE WHEN ? THEN 0 ELSE next_check END WHERE normalized_url=?",
                       (now, lastmod, bump, n))
    db.commit()

last_seen regista a última execução de descoberta que incluiu o URL, o que é distinto de last_fetched . Uma alteração em sitemap_lastmod redefine next_check para zero, para que a página seja incluída na próxima fila de alterações.

Pedidos condicionais: ETag e If-Modified-Since

Quando um site envia ETag ou Last-Modified, um pedido condicional permite que a origem responda com um código de estado 304 sem corpo. Envia-se esses cabeçalhos através do campo headers de /v1/extract e lê-se o estado da origem em metadata.http_status. A solicitação continua a ser contabilizada no seu plano, mas é mais rápida do que uma renderização completa.

Nem todos os sites respeitam os validadores, e as CDNs, por vezes, removem-nos. O código abaixo envia-os quando estão presentes e recorre ao hash quando a resposta é 200.

Cálculo do hash do conteúdo após a remoção de blocos voláteis

Um hash bruto do Markdown altera-se a cada execução em páginas que exibem a hora atual, um token CSRF ou um bloco de «artigos relacionados». Remova esses elementos antes de calcular o hash. Mantenha as regras de remoção num único local para que possa ajustá-las por site.

import hashlib
import re

VOLATILE = [
    re.compile(r"\b\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}(:\d{2})?\S*"),   # ISO timestamps
    re.compile(r"(?i)(csrf|nonce|session)[_-]?token[^\s]*"),
    re.compile(r"(?i)^#+\s*related (posts|articles).*?(?=^#|\Z)", re.M | re.S),
    re.compile(r"[ \t]+"),
]

def content_fingerprint(markdown: str) -> str:
    text = markdown
    for rx in VOLATILE:
        text = rx.sub(" ", text)
    text = "\n".join(line.strip() for line in text.splitlines() if line.strip())
    return hashlib.sha256(text.encode("utf-8")).hexdigest()

def fetch_one(session: requests.Session, row: sqlite3.Row | tuple) -> dict:
    """Sync extract with conditional headers. Returns a dict the caller can store."""
    normalized_url, url, etag, last_modified = row
    cond = {}
    if etag:
        cond["If-None-Match"] = etag
    if last_modified:
        cond["If-Modified-Since"] = last_modified
    body = {"url": url, "formats": ["markdown"], "render_js": False,
            "processing_mode": "sync", "headers": cond or None, "timeout": 60000}
    r = session.post(f"{API}/v1/extract", json=body, headers=HEADERS, timeout=90)
    if r.status_code == 429:
        raise RuntimeError(f"rate limited: {r.json().get('error', r.json())}")
    r.raise_for_status()
    payload = r.json()
    status = payload["metadata"].get("http_status")
    md = (payload.get("data") or {}).get("markdown") or ""
    return {"http_status": status, "markdown": md,
            "hash": content_fingerprint(md) if status == 200 else None}

O facto de os cabeçalhos de resposta ETag e Last-Modified da origem estarem expostos nos metadata depende da versão da API que estiver a utilizar; verifique a resposta da sua conta antes de confiar neles. Se não estiverem expostos, o hash do conteúdo continua a fornecer-lhe uma resposta fiável sobre se houve ou não alterações; apenas perde o atalho 304.

A fila de alterações e a revisitação adaptativa

A fila é o cerne do rastreamento incremental da Web e constitui-se numa consulta, não numa estrutura de dados: cada linha cujo next_check esteja no passado e que não esteja marcada como gone. Após cada recolha, o intervalo altera-se. Se for detetada uma alteração: reduza-o para metade, arredondando para o inteiro inferior a uma hora. Sem alterações: multiplique por 1,5, arredondando para cima até 30 dias. Ao longo de algumas semanas, o calendário converge para a frequência de alterações observada em cada página.

MIN_S, MAX_S = 3600, 30 * 86400

def change_queue(db: sqlite3.Connection, limit: int = 500) -> list[tuple]:
    return db.execute(
        "SELECT normalized_url, url, etag, last_modified FROM pages "
        "WHERE gone=0 AND next_check<=? ORDER BY next_check LIMIT ?",
        (time.time(), limit)).fetchall()

def record_result(db: sqlite3.Connection, normalized_url: str, res: dict) -> str:
    now = time.time()
    old_hash, interval = db.execute(
        "SELECT content_hash, interval_s FROM pages WHERE normalized_url=?",
        (normalized_url,)).fetchone()
    if res["http_status"] in (404, 410):
        db.execute("UPDATE pages SET gone=1, last_fetched=? WHERE normalized_url=?", (now, normalized_url))
        db.commit()
        return "gone"
    changed = res["http_status"] == 200 and res["hash"] != old_hash
    interval = max(MIN_S, interval // 2) if changed else min(MAX_S, int(interval * 1.5))
    db.execute("UPDATE pages SET last_fetched=?, next_check=?, interval_s=?, "
               "content_hash=COALESCE(?, content_hash) WHERE normalized_url=?",
               (now, now + interval, interval, res["hash"], normalized_url))
    db.commit()
    return "changed" if changed else "unchanged"

Um código de estado 304 chega com http_status == 304 e sem hash, pelo que é encaminhado para o ramo «inalterado» e o intervalo aumenta. Qualquer erro 5xx ou de rede deve deixar a linha inalterada, para que seja repetida na próxima execução; para tal, envolva fetch_one num try/except no seu executor.

Detecção de URLs eliminados

Dois sinais, utilizados em conjunto. Um erro 404 ou 410 direto ativa imediatamente gone em record_result. Um URL que desaparece do mapa do site constitui uma evidência menos forte, uma vez que os mapas do site apresentam atrasos e, por vezes, omitem secções. Conte as execuções consecutivas de deteção em que esteve ausente e só force uma recuperação de confirmação após N dessas execuções.

def mark_missing(db: sqlite3.Connection, run_started: float, n_runs: int = 3) -> None:
    """Call after upsert_discovered. Rows not seen this run get missing_runs+1."""
    db.execute("UPDATE pages SET missing_runs=missing_runs+1 WHERE last_seen<? AND gone=0", (run_started,))
    db.execute("UPDATE pages SET next_check=0 WHERE missing_runs>=? AND gone=0", (n_runs,))
    db.commit()

As linhas marcadas como gone permanecem na tabela. Isso impede que um link de entrada desatualizado reintroduza a URL como nova. Se a página voltar a estar disponível, uma execução de deteção posterior irá listá-la novamente; redefina gone para 0 em upsert_discovered se pretender esse comportamento.

Obter o conjunto de alterações em massa

Para algumas centenas de URLs, o ciclo de sincronização acima é suficiente, e numa configuração madura de rastreamento incremental da Web, o conjunto de alterações é normalmente assim tão pequeno. Para milhares, envie a fila como uma única tarefa /v1/batch e faça a sondagem. O corpo do lote aceita os mesmos campos formats, render_js, proxy, headers e wait_config que uma única extração, aplicados a todas as URLs; por isso, os cabeçalhos condicionais não podem variar por URL num lote. Utilize o lote para o caminho «plain content-hash» e mantenha o caminho de sincronização para as URLs em que possua um ETag.

def batch_fetch(session: requests.Session, urls: list[str], poll_s: int = 10) -> dict[str, dict]:
    body = {"urls": urls, "formats": ["markdown"], "render_js": False, "ignore_invalid_urls": True}
    job = session.post(f"{API}/v1/batch", json=body, headers=HEADERS, timeout=60).json()
    while True:
        st = session.get(f"{API}/v1/batch/{job['job_id']}", headers=HEADERS, timeout=60).json()
        if st["status"] in ("completed", "failed", "cancelled"):
            break
        time.sleep(poll_s)
    out = {}
    for res in st.get("results", []):
        if res["status"] != "completed":
            continue
        md = (res.get("data") or {}).get("markdown") or ""
        out[res["url"]] = {"http_status": res["metadata"].get("http_status"),
                           "markdown": md, "hash": content_fingerprint(md)}
    return out

A alternativa é processing_mode: "async" em chamadas individuais /v1/extract, o que devolve 202 com um job_id que se consulta em GET /v1/extract/{job_id}. Isso mantém os cabeçalhos por URL, mas implica uma solicitação por URL. Seja como for, envie os resultados através de record_result exatamente como faz o caminho de sincronização.

Configurar a execução

Uma única execução consiste em: descobrir, inserir ou atualizar, marcar o que falta, esvaziar a fila e registar. Execute-a a partir do cron ou de um agendador com a frequência necessária para atingir o seu objetivo de atualização; os intervalos adaptativos mantêm o número de consultas baixo, independentemente da frequência com que o executor é acionado.

def run(sitemap_url: str | None, site_url: str) -> None:
    db = open_db()
    session = requests.Session()
    started = time.time()
    found = urls_from_sitemap(sitemap_url, session) if sitemap_url else urls_from_map(site_url, session)
    upsert_discovered(db, found)
    mark_missing(db, started)
    queue = change_queue(db)
    with_etag = [r for r in queue if r[2] or r[3]]
    plain = [r for r in queue if not (r[2] or r[3])]
    for row in with_etag:
        try:
            record_result(db, row[0], fetch_one(session, row))
        except Exception as exc:          # leave row for next run
            print("retry later:", row[1], exc)
    if plain:
        results = batch_fetch(session, [r[1] for r in plain])
        for row in plain:
            if row[1] in results:
                record_result(db, row[0], results[row[1]])

Este é o módulo completo de rastreamento incremental da Web. A tabela abaixo resume o que cada sinal faz, para que possa eliminar as partes de que não necessita.

Erros comuns

Calcular o hash antes de remover o conteúdo volátil

Uma página que exibe «Gerado às 14:02:11» altera-se a cada recolha, pelo que o seu rastreador reduz o intervalo para metade até atingir o limite mínimo e voltar a recolher de hora a hora. Analise as diferenças entre duas recolhas consecutivas de qualquer página que registe alterações em cada execução e adicione o padrão a VOLATILE.

Normalização após a primeira inserção

Se a primeira execução armazenar URLs em bruto e a normalização for adicionada posteriormente, cada linha será uma duplicata em potencial. Normalize desde o primeiro dia e defina normalized_url como chave primária, tal como o esquema indica.

Confiar apenas na ausência do mapa do site

Os mapas do site ficam desatualizados em relação às implementações, estão divididos por ficheiros de índice e, por vezes, deixam de incluir secções inteiras após uma alteração no CMS. Marcar uma página como desaparecida apenas porque não apareceu numa única recolha do mapa do site irá eliminar conteúdo ativo do seu índice. Exija uma confirmação através de um código de erro 404 ou 410.

Executar o processamento em lote com cabeçalhos condicionais por URL

/v1/batch aplica um objeto headers a cada URL na tarefa. Enviar um If-None-Match destinado a uma página para mil outras produz resultados sem sentido. Encaminhe as linhas com ETag pelo caminho de sincronização e o resto pelo processamento em lote, tal como faz run().

Próximo passo

O módulo acima mantém o estado, compara conteúdos e agenda novas visitas, e necessita de uma camada de recuperação que devolva Markdown limpo com um http_status correto para cada URL, incluindo os códigos 304 e 404 de que depende. A API Scraper API do Geonode fornece essa camada: /v1/extract para páginas individuais com cabeçalhos por pedido, /v1/batch para o conjunto de alterações em massa e /v1/map quando não existe um mapa do site. O plano gratuito cobre 1,5 mil pedidos por mês, o suficiente para testar o ciclo incremental num site real antes de escolher um plano. Consulte a página da Scraper API para conhecer os planos e os preços para os níveis de solicitações.

Perguntas frequentes

O que é o rastreamento incremental da Web?

O rastreamento incremental da Web consiste em revisitar apenas as páginas que provavelmente sofreram alterações desde a última execução, em vez de voltar a obter todo o site de cada vez. O rastreador mantém um registo de estado com uma «impressão digital» por URL, identifica as páginas candidatas de forma eficiente (data da última modificação do mapa do site, ETag, Last-Modified) e utiliza o orçamento de recuperação apenas no subconjunto que realmente necessita disso.

Como posso saber quais as páginas que foram alteradas sem as descarregar?

Normalmente, não é possível saber com certeza sem efetuar um pedido, mas é possível classificar as páginas candidatas. A data de última modificação do mapa do site é o indicador mais económico. Uma solicitação condicional com «If-None-Match» ou «If-Modified-Since» retorna um código de estado 304 em sites que suportam validadores. Para todos os outros casos, recupere e compare um hash de conteúdo calculado após remover blocos voláteis, como carimbos de data/hora e tokens de sessão.

A API do GeonodeScraper API trata da deteção de estado e alterações por mim?

Não. A Scraper API recupera páginas e devolve Markdown ou HTML, juntamente com metadados como o http_status. O seu script é responsável pelo armazenamento do estado, pela normalização de URLs, pelo cálculo do hash e pela programação. Essa separação é deliberada: a API mantém-se sem estado e económica de chamar, enquanto a sua lógica de rastreamento incremental da Web reside no código que controla e pode testar.

Como devo agendar novas visitas para o rastreamento incremental da Web?

Comece com um intervalo predefinido e, em seguida, adapte-o por URL. Se uma recuperação detetar uma alteração, reduza o intervalo para metade até um limite mínimo; se não detetar nenhuma alteração, multiplique-o até um limite máximo. As páginas com alta rotatividade são verificadas com frequência, enquanto as páginas estáticas tendem para o limite máximo. Armazene o valor de next_check na tabela de estado e consulte as linhas cujo next_check já tenha expirado.

Como deteto páginas que foram eliminadas?

Utilize dois sinais em conjunto. Um código de erro 404 ou 410 durante a recolha é um sinal forte. A ausência do mapa do site em N execuções consecutivas é um sinal mais fraco, porque os mapas do site estão frequentemente incompletos. Marque a linha como eliminada apenas quando uma consulta direta o confirmar e mantenha a linha para não voltar a descobrir e a consultar a URL a partir de links desatualizados.

Quais são os limites e as regras de conformidade que devo respeitar?

Respeite o ficheiro robots.txt e os termos de serviço do site, mantenha as taxas de pedidos moderadas e nunca contorne a autenticação. No que diz respeito à Scraper API, o plano gratuito cobre 1,5 mil pedidos por mês e os planos pagos variam entre 2 e 100 threads; por isso, adapte o seu conjunto de alterações ao seu plano. Os pedidos condicionais continuam a contar como pedidos.