mirror of
https://github.com/idrainformatica/UNRAE-Scraper.git
synced 2026-04-17 19:53:48 +02:00
Add files via upload
This commit is contained in:
GitHub
50
README.md
Normal file
50
README.md
Normal file
@@ -0,0 +1,50 @@
|
||||
# UNRAE PDF Scraper
|
||||
|
||||
Uno script Python **asincrono** e **robusto** progettato per automatizzare il download di report statistici specifici dal sito dell'**Unione Nazionale Rappresentanti Autoveicoli Esteri (UNRAE)**, concentrandosi sui dati di immatricolazione.
|
||||
|
||||
---
|
||||
|
||||
## Obiettivo dello Script
|
||||
|
||||
Il progetto nasce dall'esigenza di monitorare e analizzare specifiche tipologie di report mensili di immatricolazione veicoli, spesso disseminati nelle pagine di listing del sito UNRAE.
|
||||
|
||||
Questo script utilizza la potenza di **Playwright** per la navigazione e **aiohttp** per il download diretto, garantendo velocità ed efficienza.
|
||||
|
||||
### Report Ricercati
|
||||
|
||||
Lo script filtra e scarica solo i documenti PDF che contengono nel titolo i seguenti elementi (ricerca *case-insensitive*):
|
||||
|
||||
* `immatricolazioni di autovetture per gruppi`
|
||||
* `immatricolazioni di autovetture per marca`
|
||||
* `struttura del mercato`
|
||||
* `immatricolazioni di autovetture per provincia di residenza del proprietario`
|
||||
|
||||
---
|
||||
|
||||
## Prerequisiti
|
||||
|
||||
Per eseguire lo script è necessario avere installato **Python 3.8+** e le librerie elencate qui sotto.
|
||||
|
||||
### Installazione delle Dipendenze
|
||||
|
||||
1. **Installa i pacchetti Python:**
|
||||
```bash
|
||||
pip install playwright aiohttp
|
||||
```
|
||||
|
||||
2. **Installa i driver del browser (Playwright):**
|
||||
```bash
|
||||
playwright install
|
||||
```
|
||||
*Nota da sistemista: Questo comando scaricherà i binari necessari (Chromium, Firefox, WebKit) per Playwright. Lo script usa Chromium.*
|
||||
|
||||
---
|
||||
|
||||
## Utilizzo
|
||||
|
||||
### 1. Esecuzione
|
||||
|
||||
Esegui lo script direttamente dalla riga di comando:
|
||||
|
||||
```bash
|
||||
python unrae_scraper.py
|
||||
341
scraper_unrae.py
Normal file
341
scraper_unrae.py
Normal file
@@ -0,0 +1,341 @@
|
||||
import os
|
||||
import asyncio
|
||||
from pathlib import Path
|
||||
from playwright.async_api import async_playwright
|
||||
import re
|
||||
from urllib.parse import urljoin
|
||||
import aiohttp
|
||||
import traceback
|
||||
|
||||
TARGET_TITLES = [
|
||||
"immatricolazioni di autovetture per gruppi",
|
||||
"immatricolazioni di autovetture per marca",
|
||||
"struttura del mercato",
|
||||
"immatricolazioni di autovetture per provincia di residenza del proprietario"
|
||||
]
|
||||
|
||||
DOWNLOAD_DIR = Path("downloads_unrae")
|
||||
DOWNLOAD_DIR.mkdir(exist_ok=True)
|
||||
|
||||
MAX_RETRIES = 3
|
||||
RETRY_DELAY = 5 # secondi
|
||||
|
||||
def sanitize_filename(filename):
|
||||
"""Rimuove caratteri non validi dal nome del file"""
|
||||
return re.sub(r'[<>:"/\\|?*]', '_', filename)
|
||||
|
||||
def matches_target(title):
|
||||
"""Verifica se il titolo corrisponde a uno dei target"""
|
||||
title_lower = title.lower()
|
||||
for target in TARGET_TITLES:
|
||||
if target in title_lower:
|
||||
return True
|
||||
return False
|
||||
|
||||
async def download_pdf_direct(pdf_url, filepath):
|
||||
"""Scarica il PDF direttamente via HTTP con logica di retry"""
|
||||
for attempt in range(MAX_RETRIES):
|
||||
try:
|
||||
async with aiohttp.ClientSession() as session:
|
||||
# Timeout più generoso per i download
|
||||
async with session.get(pdf_url, timeout=60) as response:
|
||||
if response.status == 200:
|
||||
content = await response.read()
|
||||
with open(filepath, 'wb') as f:
|
||||
f.write(content)
|
||||
return True
|
||||
|
||||
# Se l'errore è recuperabile, tenta il retry
|
||||
if response.status >= 500 or response.status == 429: # 429 Too Many Requests
|
||||
print(f" [TENTATIVO {attempt + 1}/{MAX_RETRIES}] Errore HTTP recuperabile {response.status}. Riprovo tra {RETRY_DELAY}s...")
|
||||
await asyncio.sleep(RETRY_DELAY)
|
||||
continue
|
||||
|
||||
# Errore non recuperabile (es. 404, 403)
|
||||
print(f" [ERRORE] Errore HTTP non recuperabile {response.status}.")
|
||||
return False
|
||||
|
||||
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
|
||||
if attempt < MAX_RETRIES - 1:
|
||||
print(f" [TENTATIVO {attempt + 1}/{MAX_RETRIES}] Errore di connessione/timeout: {type(e).__name__}. Riprovo tra {RETRY_DELAY}s...")
|
||||
await asyncio.sleep(RETRY_DELAY)
|
||||
else:
|
||||
print(f" [ERRORE] Tentativi esauriti dopo errore: {type(e).__name__}.")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f" [ERRORE] Errore imprevisto: {e}")
|
||||
return False
|
||||
|
||||
return False # Ritorna False se tutti i tentativi falliscono
|
||||
|
||||
async def download_pdf_from_url(page, article_url):
|
||||
"""Scarica il PDF da una pagina di articolo"""
|
||||
try:
|
||||
print(f" Apertura: {article_url}")
|
||||
|
||||
await page.goto(article_url, wait_until='domcontentloaded', timeout=30000)
|
||||
await asyncio.sleep(2)
|
||||
|
||||
title_selectors = [
|
||||
'h1',
|
||||
'.page-title',
|
||||
'h2'
|
||||
]
|
||||
|
||||
article_title = None
|
||||
for selector in title_selectors:
|
||||
title_element = page.locator(selector).first
|
||||
if await title_element.count() > 0:
|
||||
article_title = await title_element.inner_text()
|
||||
article_title = article_title.strip()
|
||||
break
|
||||
|
||||
if not article_title:
|
||||
article_title = article_url.split('/')[-1].replace('-', ' ')
|
||||
|
||||
print(f" Titolo: {article_title}")
|
||||
|
||||
date_match = re.search(
|
||||
r'(gennaio|febbraio|marzo|aprile|maggio|giugno|luglio|agosto|settembre|ottobre|novembre|dicembre)[\s\-]?(\d{4})',
|
||||
article_url + " " + article_title,
|
||||
re.IGNORECASE
|
||||
)
|
||||
|
||||
period = date_match.group(0) if date_match else "unknown_period"
|
||||
period = period.replace(' ', '_')
|
||||
|
||||
if "per gruppi" in article_title.lower():
|
||||
report_type = "gruppi"
|
||||
elif "per marca" in article_title.lower():
|
||||
report_type = "marca"
|
||||
elif "struttura del mercato" in article_title.lower():
|
||||
report_type = "struttura_mercato"
|
||||
elif "per provincia" in article_title.lower():
|
||||
report_type = "provincia"
|
||||
else:
|
||||
report_type = "altro"
|
||||
|
||||
filename = sanitize_filename(f"{period}_{report_type}.pdf")
|
||||
filepath = DOWNLOAD_DIR / filename
|
||||
|
||||
if filepath.exists():
|
||||
print(f" File già esistente: {filename}")
|
||||
return "skipped"
|
||||
|
||||
pdf_selectors = [
|
||||
'a[href$=".pdf"]',
|
||||
'a[href*=".pdf"]',
|
||||
'.field--type-file a',
|
||||
'.file a'
|
||||
]
|
||||
|
||||
pdf_link = None
|
||||
for selector in pdf_selectors:
|
||||
pdf_link = page.locator(selector).first
|
||||
if await pdf_link.count() > 0:
|
||||
break
|
||||
|
||||
if not pdf_link or await pdf_link.count() == 0:
|
||||
print(f" Nessun link PDF trovato")
|
||||
return "no_pdf"
|
||||
|
||||
pdf_url = await pdf_link.get_attribute('href')
|
||||
|
||||
if not pdf_url.startswith('http'):
|
||||
pdf_url = urljoin(article_url, pdf_url)
|
||||
|
||||
print(f" URL PDF: {pdf_url}")
|
||||
|
||||
print(f" Scaricamento: {filename}")
|
||||
|
||||
success = await download_pdf_direct(pdf_url, filepath)
|
||||
|
||||
if success:
|
||||
print(f" Salvato: {filename}")
|
||||
return "downloaded"
|
||||
else:
|
||||
return "error"
|
||||
|
||||
except Exception as e:
|
||||
print(f" Errore durante il download: {e}")
|
||||
traceback.print_exc()
|
||||
return "error"
|
||||
|
||||
async def get_article_links_from_page(page):
|
||||
"""Estrae tutti i link agli articoli dalla pagina di listing"""
|
||||
print(" Ricerca articoli nella pagina...")
|
||||
|
||||
await page.wait_for_load_state('domcontentloaded')
|
||||
await asyncio.sleep(3)
|
||||
|
||||
article_links = []
|
||||
|
||||
all_links = await page.locator('a[href*="/dati-statistici/immatricolazioni/"]').all()
|
||||
|
||||
print(f" Trovati {len(all_links)} link alle immatricolazioni")
|
||||
|
||||
for link in all_links:
|
||||
try:
|
||||
href = await link.get_attribute('href')
|
||||
text = await link.inner_text()
|
||||
text = text.strip()
|
||||
|
||||
if not text or not href:
|
||||
continue
|
||||
|
||||
if any(skip in text.lower() for skip in ['precedente', 'successiva', 'prima', 'ultima', 'pagina']):
|
||||
continue
|
||||
|
||||
if not re.search(r'/\d+/', href):
|
||||
continue
|
||||
|
||||
if matches_target(text):
|
||||
if not href.startswith('http'):
|
||||
href = f"https://www.unrae.it{href}"
|
||||
|
||||
if not any(a['url'] == href for a in article_links):
|
||||
article_links.append({
|
||||
'url': href,
|
||||
'title': text
|
||||
})
|
||||
print(f" Trovato: {text}")
|
||||
|
||||
except Exception as e:
|
||||
continue
|
||||
|
||||
return article_links
|
||||
|
||||
async def process_page(page, page_num):
|
||||
"""Processa una singola pagina di listing"""
|
||||
print(f"\n{'='*60}")
|
||||
print(f"Elaborazione pagina {page_num}")
|
||||
print(f"{'='*60}")
|
||||
|
||||
article_links = await get_article_links_from_page(page)
|
||||
|
||||
print(f"\nArticoli target trovati: {len(article_links)}")
|
||||
|
||||
if len(article_links) == 0:
|
||||
print("NESSUN articolo target trovato in questa pagina")
|
||||
return {"downloaded": 0, "skipped": 0, "no_pdf": 0, "error": 0}
|
||||
|
||||
stats = {"downloaded": 0, "skipped": 0, "no_pdf": 0, "error": 0}
|
||||
|
||||
for idx, article_info in enumerate(article_links, 1):
|
||||
print(f"\n[{idx}/{len(article_links)}] {article_info['title']}")
|
||||
|
||||
result = await download_pdf_from_url(page, article_info['url'])
|
||||
stats[result] += 1
|
||||
|
||||
await asyncio.sleep(2)
|
||||
|
||||
print(f"\nPagina {page_num} - Scaricati: {stats['downloaded']}, Saltati: {stats['skipped']}, No PDF: {stats['no_pdf']}, Errori: {stats['error']}")
|
||||
return stats
|
||||
|
||||
async def main():
|
||||
"""Funzione principale"""
|
||||
print("Avvio scraper UNRAE PDF")
|
||||
print(f"Directory download: {DOWNLOAD_DIR.absolute()}")
|
||||
print(f"Articoli target:")
|
||||
for title in TARGET_TITLES:
|
||||
print(f" - {title}")
|
||||
|
||||
async with async_playwright() as p:
|
||||
print("\nAvvio browser...")
|
||||
browser = await p.chromium.launch(
|
||||
headless=False,
|
||||
args=['--disable-blink-features=AutomationControlled']
|
||||
)
|
||||
context = await browser.new_context(
|
||||
user_agent='Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
|
||||
viewport={'width': 1920, 'height': 1080}
|
||||
)
|
||||
|
||||
await context.add_init_script("""
|
||||
Object.defineProperty(navigator, 'webdriver', {
|
||||
get: () => false
|
||||
});
|
||||
""")
|
||||
|
||||
page = await context.new_page()
|
||||
|
||||
base_url = "https://www.unrae.it/dati-statistici/immatricolazioni"
|
||||
page_num = 0
|
||||
total_stats = {"downloaded": 0, "skipped": 0, "no_pdf": 0, "error": 0}
|
||||
max_pages = 50
|
||||
consecutive_empty = 0
|
||||
|
||||
while page_num <= max_pages:
|
||||
url = f"{base_url}?page={page_num}" if page_num > 0 else base_url
|
||||
|
||||
try:
|
||||
print(f"\nNavigazione a: {url}")
|
||||
response = await page.goto(url, wait_until='domcontentloaded', timeout=30000)
|
||||
|
||||
print(f" Risposta HTTP: {response.status}")
|
||||
|
||||
if response.status != 200:
|
||||
print(f"Errore HTTP {response.status}")
|
||||
break
|
||||
|
||||
print(" Attesa caricamento contenuto...")
|
||||
await asyncio.sleep(3)
|
||||
|
||||
stats = await process_page(page, page_num)
|
||||
|
||||
if sum(stats.values()) == 0:
|
||||
consecutive_empty += 1
|
||||
print(f"Nessun articolo target trovato ({consecutive_empty}/3)")
|
||||
|
||||
if consecutive_empty >= 3:
|
||||
print(f"\nNessun articolo target per 3 pagine consecutive. Fine dello scraping.")
|
||||
break
|
||||
else:
|
||||
consecutive_empty = 0
|
||||
|
||||
for key in total_stats:
|
||||
total_stats[key] += stats[key]
|
||||
|
||||
print("\n Ricerca pagina successiva...")
|
||||
|
||||
current_url = page.url
|
||||
if current_url != url:
|
||||
await page.goto(url, wait_until='domcontentloaded')
|
||||
await asyncio.sleep(2)
|
||||
|
||||
next_page_num = page_num + 1
|
||||
next_link = page.locator(f'a[href*="?page={next_page_num}"]').first
|
||||
|
||||
has_next = await next_link.count() > 0
|
||||
|
||||
if has_next:
|
||||
print(f" Pagina successiva trovata: page={next_page_num}")
|
||||
else:
|
||||
print(f"\nUltima pagina raggiunta ({page_num}). Fine dello scraping.")
|
||||
break
|
||||
|
||||
page_num += 1
|
||||
print(f" Passaggio alla pagina {page_num}")
|
||||
|
||||
await asyncio.sleep(3)
|
||||
|
||||
except Exception as e:
|
||||
print(f"\nErrore durante l'elaborazione della pagina {page_num}: {e}")
|
||||
traceback.print_exc()
|
||||
break
|
||||
|
||||
print(f"\n{'='*60}")
|
||||
print(f"Chiusura browser...")
|
||||
await browser.close()
|
||||
|
||||
print(f"\n{'='*60}")
|
||||
print(f"Scraping completato!")
|
||||
print(f"File scaricati: {total_stats['downloaded']}")
|
||||
print(f"File saltati (già esistenti): {total_stats['skipped']}")
|
||||
print(f"Articoli senza PDF: {total_stats['no_pdf']}")
|
||||
print(f"Errori: {total_stats['error']}")
|
||||
print(f"File salvati in: {DOWNLOAD_DIR.absolute()}")
|
||||
print(f"{'='*60}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user