Il web scraping è una tecnica potente per estrarre grandi quantità di dati dai siti web. Tuttavia, raccogliere dati su richiesta può essere difficile. In questa guida completa impareremo come trasformare un web scraper Python in un'API di dati in tempo reale utilizzando FastAPI.
Perché creare un'API di scraping?
Ecco alcuni vantaggi chiave derivanti dalla fornitura di dati ricavati tramite un'API:
Dati in tempo reale – Le API possono acquisire dati su richiesta anziché fare affidamento su batch obsoleti e memorizzati nella cache.
Query personalizzate – Le API consentono interrogazioni personalizzate dei dati anziché limitarsi a scaricare interi siti Web.
Scalabilità – Le API gestiscono i picchi di traffico e si adattano a migliaia di utenti.
Affidabilità – Le API riprovano le richieste non riuscite e implementano una solida logica di scraping.
Flessibilità – Le API possono implementare vari metodi di consegna dei dati come i webhook.
Astrazione – Le API nascondono logiche e infrastrutture di scraping complesse ai consumatori API.
Scraping dell'architettura API
Ad alto livello, un'API di web scraping segue questa architettura:
- L'API riceve la richiesta
- Il raschiatore recupera i dati
- I dati vengono memorizzati nella cache
- L'API restituisce dati raschiati
I componenti chiave sono:
Server API – Gestisce le richieste dei clienti e fornisce dati raschiati.
Raschietto – Recupera i dati su richiesta dai siti di destinazione.
Cache – Memorizza i dati raschiati per evitare raschiamenti ridondanti.
Banca Dati – Memorizza facoltativamente i dati raschiati per l'analisi storica.
Perché utilizzare FastAPI?
Esistono molti framework API eccellenti. Per i raschiatori, lo consiglio API veloce perché:
FastAPI è molto veloce, perfetta per le API di scraping dei dati.
Fornisce documenti automatici, convalida, serializzazione, ecc.
Supporta asyncio per lo scraping asincrono.
Semplice da usare e imparare. Flessibile per API di piccole e grandi dimensioni.
Ha un ottimo ecosistema di strumenti di scraping come httpx, parsel, ecc.
Configurazione dell'API di scraping
Utilizzeremo le seguenti librerie principali:
API veloce – Il nostro framework API
httpx – Client HTTP asincrono
pacco – Analizzatore HTML/XML
loguru – Utilità di registrazione
Installa i pacchetti:
pip install fastapi uvicorn httpx parsel loguruPer questa guida, estrarremo alcuni dati azionari di base da Yahoo Finance.
Crea l'API
Impostiamo l'app FastAPI iniziale con un singolo endpoint:
from fastapi import FastAPI
app = FastAPI()
@app.get("/stock/{symbol}")
async def get_stock(symbol: str):
return { "stock": symbol } Questa API di base restituisce semplicemente il simbolo azionario che forniamo.
Avviamo il server e testiamolo:
uvicorn main:app --reloadimport httpx
print(httpx.get("http://localhost:8000/stock/AAPL").json())
# {‘stock‘: ‘AAPL‘}La nostra API è pronta a ricevere richieste. Quindi facciamo in modo che racimola alcuni dati.
Raschiatura dei dati di magazzino
Per ottenere i dati di un titolo:
- Crea l'URL di Yahoo Finance dal simbolo
- Recupera la pagina HTML
- Analizzare i valori utilizzando 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 dataQuesto raschietto restituisce un dizionario contenente i dati analizzati. Colleghiamolo alla nostra 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 dataOra, quando chiamiamo la nostra API, recupererà i dati più recenti:
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"
}Aggiunta della memorizzazione nella cache
Raschiare su ogni richiesta è uno spreco. Aggiungiamo la memorizzazione nella cache in modo da raschiare uno stock solo una volta ogni 5 minuti.
Useremo un semplice dict per memorizzare i dati raschiati digitati dal simbolo di borsa:
STOCK_CACHE = {}
async def scrape_stock(symbol):
if symbol in STOCK_CACHE:
return STOCK_CACHE[symbol]
data = ... # scrape
STOCK_CACHE[symbol] = data
return dataOra le richieste ripetute restituiranno i dati memorizzati nella cache invece di eseguire lo scraping ogni volta.
Possiamo anche svuotare periodicamente la vecchia cache:
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()) Ciò garantisce che la nostra cache non cresca senza limiti.
Aggiunta di webhook
Per lavori di scraping lunghi, possiamo utilizzare i webhook per restituire i risultati in modo asincrono:
@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)Ora, invece di attendere il completamento dello scraping, la nostra API restituirà immediatamente uno stato e consegnerà i dati in modo asincrono al webhook di callback.
Possiamo testarlo utilizzando uno strumento come Webhook.sito.
Ridimensionare l'API di scraping
Man mano che il traffico aumenta, ecco alcune tecniche di ridimensionamento:
Aggiungi la memorizzazione nella cache dell'API – Utilizza una cache come Redis per ridurre il carico di scraping.
Esegui più processi – Scalabilità su core/server con gunicorn.
Scaricare la raschiatura – Sposta i raschiatori su lavoratori come Celery o RabbitMQ.
Utilizza i servizi di raschiamento – Sfrutta le API di scraping come Scrapfly o ScraperAPI.
Ottimizza i raschiatori – Garantire che gli scraper siano efficienti ed evitare i divieti.
Aggiungi database – Archiviare i dati raschiati nei database per ulteriori analisi.
Conclusione
In questa guida abbiamo creato un'API di web scraping in Python utilizzando FastAPI. I punti chiave sono:
FastAPI fornisce un framework eccellente per lo scraping delle API.
Possiamo generare scraper per recuperare dati su richiesta.
La memorizzazione nella cache e i webhook aiutano a superare le limitazioni dello scraping.
Esistono molte strategie di ottimizzazione e ridimensionamento man mano che il traffico cresce.
Le API di scraping sbloccano la ricchezza di dati sui siti Web. Generando API anziché lo scraping statico, possiamo fornire dati personalizzati a bassa latenza su larga scala.
