Como transformar Web Scrapers em APIs de dados

Web scraping é uma técnica poderosa para extrair grandes quantidades de dados de sites. No entanto, extrair dados sob demanda pode ser um desafio. Neste guia abrangente, aprenderemos como transformar um web scraper Python em uma API de dados em tempo real usando FastAPI.

Por que construir uma API de scraping?

Aqui estão alguns dos principais benefícios de fornecer dados extraídos por meio de uma API:

• Dados em tempo real – As APIs podem extrair dados sob demanda, em vez de depender de lotes obsoletos e armazenados em cache.

• Consultas personalizadas – As APIs permitem consultas personalizadas de dados, em vez de apenas despejar sites inteiros.

• AMPLIAR – APIs lidam com picos de tráfego e escalam para milhares de usuários.

• Confiabilidade – APIs repetem solicitações com falha e implementam uma lógica robusta de raspagem.

• Flexibilidade – As APIs podem implementar vários métodos de entrega de dados, como webhooks.

• Abstração – As APIs ocultam lógica e infraestrutura de scraping complexas dos consumidores de API.

Arquitetura de API de raspagem

Em alto nível, uma API de web scraping segue esta arquitetura:

1. API recebe solicitação
2. Raspador busca dados
3. Os dados são armazenados em cache
4. API retorna dados extraídos

Os principais componentes são:

• Servidor de API – Lida com solicitações de clientes e entrega dados extraídos.

• raspador – Busca dados sob demanda de sites de destino.

• Esconderijo – Armazena dados copiados para evitar raspagem redundante.

• banco de dados – Opcionalmente, armazena dados extraídos para análise histórica.

Por que usar FastAPI?

Existem muitas estruturas de API excelentes por aí. Para raspadores, recomendo FastAPI Porque:

• FastAPI é muito rápido – perfeito para APIs de coleta de dados.

• Ele fornece documentos automáticos, validação, serialização, etc.

• Suporta asyncio para raspagem assíncrona.

• Simples de usar e aprender. Flexível para APIs pequenas e grandes.

• Possui um ótimo ecossistema de ferramentas de scraping como httpx, parsel, etc.

Configuração da API de raspagem

Usaremos as seguintes bibliotecas principais:

• FastAPI – Nossa estrutura de API

• httpx – Cliente HTTP assíncrono

• parcela – analisador HTML/XML

• loguru – Utilitário de registro

Instale os pacotes:

pip install fastapi uvicorn httpx parsel loguru

Para este guia, extrairemos alguns dados básicos de ações do Yahoo Finance.

Crie a API

Vamos configurar o aplicativo FastAPI inicial com um único endpoint:

from fastapi import FastAPI

app = FastAPI() 

@app.get("/stock/{symbol}")  
async def get_stock(symbol: str):
    return { "stock": symbol } 

Esta API básica apenas retorna o símbolo de ação que fornecemos.

Vamos iniciar o servidor e testá-lo:

uvicorn main:app --reload

import httpx

print(httpx.get("http://localhost:8000/stock/AAPL").json())

# {‘stock‘: ‘AAPL‘}

Nossa API está pronta para receber solicitações. A seguir, vamos extrair alguns dados.

Extração de dados de estoque

Para obter os dados de uma ação, iremos:

1. Crie o URL do Yahoo Finance a partir do símbolo
2. Busque a página HTML
3. Analisar valores usando XPath

from parsel import Selector # for xpath parsing

async def scrape_stock(symbol: str):

    url = f"https://finance.yahoo.com/quote/{symbol}"

    async with httpx.AsyncClient() as client:
       response = await client.get(url)

    sel = Selector(response.text)

    # parse summary values 
    values = sel.xpath(‘//div[contains(@data-test,"summary-table")]//tr‘)
    data = {}
    for value in values:
        label = value.xpath("./td[1]/text()").get()
        val = value.xpath("./td[2]/text()").get()
        data[label] = val

    # parse price 
    data["price"] = sel.xpath(‘//fin-streamer[@data-symbol=$symbol]/@value‘, symbol=symbol).get()

    return data

Este raspador retorna um dicionário contendo os dados analisados. Vamos conectá-lo à nossa API:

from fastapi import FastAPI
from yahoo_finance import scrape_stock

app = FastAPI()

@app.get("/stock/{symbol}")
async def get_stock(symbol: str):
    data = await scrape_stock(symbol) 
    return data

Agora, quando chamarmos nossa API, ela buscará os dados mais recentes:

http http://localhost:8000/stock/AAPL

HTTP/1.1 200 OK
Content-Length: 340

{
  "52 Week Range": "142.00 - 182.94",
  "Beta (5Y Monthly)": "1.25", 
  "Diluted EPS (ttm)": "6.05",
  "Earnings Date": "Oct 27, 2022",
  "Ex-Dividend Date": "Aug 05, 2022",
  "Forward Dividend & Yield": "0.92 (0.59%)",
  "Market Cap": "2.44T",
  "Open": "156.76",
  "PE Ratio (ttm)": "25.60",
  "Previous Close": "153.72",
  "Price": "155.33",  
  "Volume": "53,978,024"
}

Adicionando cache

Raspar cada solicitação é um desperdício. Vamos adicionar cache para que possamos apenas raspar o estoque uma vez a cada 5 minutos.

Usaremos um simples dict para armazenar os dados extraídos digitados pelo símbolo de ação:

STOCK_CACHE = {} 

async def scrape_stock(symbol):

   if symbol in STOCK_CACHE:
       return STOCK_CACHE[symbol] 

   data = ... # scrape 
   STOCK_CACHE[symbol] = data

   return data

Agora, solicitações repetidas retornarão dados em cache em vez de serem coletados todas as vezes.

Também podemos limpar periodicamente o cache antigo:

import time

CACHE_MAX_AGE = 300 # seconds

async def clear_expired_cache():
   curr_time = time.time()
   for symbol, data in STOCK_CACHE.items():
       if curr_time - data["ts"] > CACHE_MAX_AGE:
           del STOCK_CACHE[symbol]

# run every 5 minutes   
clear_cache_task = asyncio.create_task(clear_expired_cache()) 

Isso garante que nosso cache não cresça ilimitadamente.

Adicionando webhooks

Para trabalhos de scraping longos, podemos usar webhooks para retornar resultados de forma assíncrona:

@app.get("/stock/{symbol}")
async def get_stock(symbol: str, webhook: str):

   if webhook:
      task = asyncio.create_task(fetch_stock(symbol, webhook))
      return {"msg": "Fetching data"}

   data = await scrape_stock(symbol)  
   return data

async def fetch_stock(symbol, url):
   data = await scrape_stock(symbol)

   async with httpx.AsyncClient() as client:
      await client.post(url, json=data)

Agora, em vez de esperar a conclusão da extração, nossa API retornará imediatamente um status e entregará os dados de forma assíncrona ao webhook de retorno de chamada.

Podemos testar isso usando uma ferramenta como Webhook.site.

Dimensionando a API Scraping

À medida que o tráfego aumenta, aqui estão algumas técnicas de escalonamento:

• Adicionar cache de API – Use um cache como o Redis para reduzir a carga de raspagem.

• Execute vários processos – Dimensione entre núcleos/servidores com gunicorn.

• Raspagem de descarregamento – Mova raspadores para trabalhadores como Celery ou RabbitMQ.

• Use serviços de raspagem – Aproveite APIs de scraping como Scrapfly ou ScraperAPI.

• Otimize raspadores – Garanta que os raspadores sejam eficientes e evite proibições.

• Adicionar bancos de dados – Armazene dados extraídos em bancos de dados para análise posterior.

Conclusão

Neste guia, construímos uma API de web scraping em Python usando FastAPI. As principais conclusões são:

• FastAPI fornece uma excelente estrutura para raspagem de APIs.

• Podemos gerar scrapers para buscar dados sob demanda.

• O cache e os webhooks ajudam a superar as limitações de raspagem.

• Existem muitas estratégias de otimização e escalonamento à medida que o tráfego cresce.

As APIs de scraping desbloqueiam a riqueza de dados em sites. Ao gerar APIs em vez de raspagem estática, podemos fornecer dados personalizados e de baixa latência em escala.