diff --git a/README.md b/README.md new file mode 100644 index 0000000..5f5740e --- /dev/null +++ b/README.md @@ -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 diff --git a/scraper_unrae.py b/scraper_unrae.py new file mode 100644 index 0000000..a707305 --- /dev/null +++ b/scraper_unrae.py @@ -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())