Cómo convertir Web Scrapers en API de datos

El web scraping es una técnica poderosa para extraer grandes cantidades de datos de sitios web. Sin embargo, extraer datos a pedido puede resultar un desafío. En esta guía completa, aprenderemos cómo convertir un web scraper de Python en una API de datos en tiempo real usando FastAPI.

¿Por qué crear una API de raspado?

Estos son algunos de los beneficios clave de entregar datos extraídos a través de una API:

• Datos en tiempo real – Las API pueden extraer datos bajo demanda en lugar de depender de lotes obsoletos almacenados en caché.

• Consultas personalizadas – Las API permiten consultas personalizadas de datos en lugar de simplemente deshacerse de sitios web completos.

• Escalabilidad – Las API manejan picos de tráfico y escalan a miles de usuarios.

• Fiabilidad – Las API reintentan solicitudes fallidas e implementan una lógica de raspado sólida.

• Flexibilidad – Las API pueden implementar varios métodos de entrega de datos, como webhooks.

• Abstracción – Las API ocultan una infraestructura y una lógica de scraping complejas a los consumidores de API.

Arquitectura de API de raspado

En un nivel alto, una API de web scraping sigue esta arquitectura:

1. API recibe solicitud
2. Scraper obtiene datos
3. Los datos se almacenan en caché
4. API devuelve datos extraídos

Los componentes clave son:

• Servidor API – Maneja las solicitudes de los clientes y entrega datos extraídos.

• Raspador – Obtiene datos a pedido de los sitios de destino.

• cache – Almacena datos extraídos para evitar raspados redundantes.

• Base de datos – Opcionalmente almacena datos extraídos para análisis históricos.

¿Por qué utilizar FastAPI?

Existen muchos marcos API excelentes. Para raspadores, recomiendo FastAPI porque:

• FastAPI es muy rápido, perfecto para API de extracción de datos.

• Proporciona documentos automáticos, validación, serialización, etc.

• Admite asyncio para raspado asíncrono.

• Fácil de usar y aprender. Flexible para API pequeñas y grandes.

• Tiene un gran ecosistema de herramientas de scraping como httpx, parsel, etc.

Configuración de API de raspado

Usaremos las siguientes bibliotecas principales:

• FastAPI – Nuestro marco API

• httpx – Cliente HTTP asíncrono

• parcela – Analizador HTML/XML

• logurú – Utilidad de registro

Instalar los paquetes:

pip install fastapi uvicorn httpx parsel loguru

Para esta guía, extraeremos algunos datos bursátiles básicos de Yahoo Finance.

Crear la API

Configuremos la aplicación FastAPI inicial con un único punto final:

from fastapi import FastAPI

app = FastAPI() 

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

Esta API básica simplemente devuelve el símbolo bursátil que proporcionamos.

Iniciemos el servidor y probémoslo:

uvicorn main:app --reload

import httpx

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

# {‘stock‘: ‘AAPL‘}

Nuestra API está lista para recibir solicitudes. A continuación, hagamos que extraiga algunos datos.

Raspado de datos bursátiles

Para obtener los datos de una acción, haremos lo siguiente:

1. Cree la URL de Yahoo Finance a partir del símbolo
2. Obtener la página HTML
3. Analizar 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 devuelve un diccionario que contiene los datos analizados. Conectémoslo a nuestra 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

Ahora, cuando llamemos a nuestra API, obtendrá los datos más recientes:

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"
}

Agregar almacenamiento en caché

Rascar cada solicitud es un desperdicio. Agreguemos almacenamiento en caché para que solo extraigamos un stock una vez cada 5 minutos.

Usaremos un simple dict para almacenar los datos extraídos codificados por símbolo bursátil:

STOCK_CACHE = {} 

async def scrape_stock(symbol):

   if symbol in STOCK_CACHE:
       return STOCK_CACHE[symbol] 

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

   return data

Ahora las solicitudes repetidas devolverán datos almacenados en caché en lugar de rasparlos cada vez.

También podemos borrar periódicamente el caché antiguo:

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()) 

Esto garantiza que nuestro caché no crecerá ilimitadamente.

Agregar webhooks

Para trabajos de scraping largos, podemos usar webhooks para devolver resultados de forma asincrónica:

@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)

Ahora, en lugar de esperar a que se complete el raspado, nuestra API devolverá inmediatamente un estado y entregará los datos de forma asincrónica al webhook de devolución de llamada.

Podemos probar esto usando una herramienta como Webhook.sitio.

Escalando la API de raspado

A medida que aumenta el tráfico, aquí se presentan algunas técnicas de escalado:

• Agregar almacenamiento en caché de API – Utilice un caché como Redis para reducir la carga de raspado.

• Ejecutar múltiples procesos – Escale a través de núcleos/servidores con gunicorn.

• Descargar raspado – Mover raspadores a trabajadores como Celery o RabbitMQ.

• Utilice servicios de scraping – Aproveche las API de scraping como Scrapfly o ScraperAPI.

• Optimizar raspadores – Garantizar que los scrapers sean eficientes y evitar prohibiciones.

• Agregar bases de datos – Almacenar datos extraídos en bases de datos para su posterior análisis.

Conclusión

En esta guía, creamos una API de raspado web en Python usando FastAPI. Las conclusiones clave son:

• FastAPI proporciona un marco excelente para extraer API.

• Podemos generar raspadores para recuperar datos a pedido.

• El almacenamiento en caché y los webhooks ayudan a superar las limitaciones del scraping.

• Existen muchas estrategias de optimización y escalamiento a medida que crece el tráfico.

Las API de scraping desbloquean la gran cantidad de datos en los sitios web. Al generar API en lugar de scraping estático, podemos ofrecer datos personalizados y de baja latencia a escala.