Práticas responsáveis de raspagem

Por que a raspagem responsável é importante

Verificando robots.txt

import urllib.robotparser

rp = urllib.robotparser.RobotFileParser()
rp.set_url('https://example.com/robots.txt')
rp.read()

# Check if our bot can fetch a specific path
can_fetch = rp.can_fetch('MyAgent/1.0', 'https://example.com/public-data')
print(f'Can fetch: {can_fetch}')  # True or False

cannot_fetch = rp.can_fetch('*', 'https://example.com/private/admin')
print(f'Admin allowed: {cannot_fetch}')  # Often False

Configurando um agente do usuário descritivo

import httpx

headers = {
    # Honest, descriptive User-Agent
    'User-Agent': 'MyResearchBot/1.0 (contact: me@example.com; purpose: academic research)'
}

response = httpx.get(
    'https://example.com/data',
    headers=headers,
    timeout=10.0
)

print(response.status_code)

Limitando a taxa com time.sleep()

import time
import random
import httpx

urls = [
    'https://example.com/page/1',
    'https://example.com/page/2',
    'https://example.com/page/3'
]

for url in urls:
    response = httpx.get(url, timeout=10.0)
    print(f'Fetched {url}: {response.status_code}')

    # Wait between 1 and 3 seconds before the next request
    delay = random.uniform(1, 3)
    print(f'Sleeping {delay:.1f}s...')
    time.sleep(delay)

Respeitando os cabeçalhos de nova tentativa

import time
import httpx

def fetch_with_retry(url: str) -> httpx.Response:
    while True:
        response = httpx.get(url, timeout=10.0)

        if response.status_code == 429:
            retry_after = int(response.headers.get('Retry-After', 5))
            print(f'Rate limited. Waiting {retry_after}s...')
            time.sleep(retry_after)
            continue  # retry

        response.raise_for_status()
        return response

Armazenando respostas em cache para evitar novas buscas

# pip install requests-cache
import requests_cache

# Install the cache globally — cached responses expire after 1 hour
requests_cache.install_cache(
    'agent_cache',
    backend='sqlite',
    expire_after=3600  # seconds
)

import requests

# First call: fetches from network and caches
response = requests.get('https://api.example.com/data')
print(response.from_cache)  # False

# Second call: returns cached result instantly
response2 = requests.get('https://api.example.com/data')
print(response2.from_cache)  # True

Armazenamento manual em cache baseado em arquivos com httpx

import hashlib
import json
import os
import time
import httpx

CACHE_DIR = '/tmp/agent_cache'
os.makedirs(CACHE_DIR, exist_ok=True)

def cached_get(url: str, ttl_seconds: int = 3600) -> dict:
    key = hashlib.md5(url.encode()).hexdigest()
    cache_file = os.path.join(CACHE_DIR, key + '.json')

    if os.path.exists(cache_file):
        age = time.time() - os.path.getmtime(cache_file)
        if age < ttl_seconds:
            with open(cache_file) as f:
                return json.load(f)

    response = httpx.get(url, timeout=10.0)
    response.raise_for_status()
    data = response.json()
    with open(cache_file, 'w') as f:
        json.dump(data, f)
    return data

Lendo os termos de serviço

Recuo exponencial após erros

import time
import httpx

def fetch_with_backoff(url: str, max_retries: int = 4) -> dict:
    for attempt in range(max_retries):
        try:
            response = httpx.get(url, timeout=10.0)
            response.raise_for_status()
            return response.json()
        except (httpx.HTTPStatusError, httpx.RequestError) as e:
            if attempt == max_retries - 1:
                raise
            wait = 2 ** attempt  # 1, 2, 4, 8 seconds
            print(f'Attempt {attempt+1} failed: {e}. Retrying in {wait}s...')
            time.sleep(wait)
    return {}

Fazendo raspagem apenas do que você precisa

import httpx

def has_page_changed(url: str, last_etag: str = None) -> bool:
    headers = {}
    if last_etag:
        headers['If-None-Match'] = last_etag

    response = httpx.head(url, headers=headers, timeout=5.0)

    if response.status_code == 304:  # Not Modified
        return False

    new_etag = response.headers.get('ETag')
    print(f'New ETag: {new_etag}')
    return True

Resumo das regras para uma raspagem responsável

Verificação de conhecimentos: raspagem responsável

Recapitulação: práticas de raspagem responsável