scraper_base.py

"""
Web Scraper Base - Versione Semplice
Utilizza requests e BeautifulSoup per scraping base
"""

import requests
from bs4 import BeautifulSoup
import json
import csv
from datetime import datetime
from typing import List, Dict, Optional
import time
from urllib.parse import urljoin


class WebScraperBase:
    """Classe base per web scraping semplice"""
    
    def __init__(self, base_url: str, headers: Optional[Dict] = None):
        """
        Inizializza lo scraper
        
        Args:
            base_url: URL base del sito da scrapare
            headers: Headers HTTP personalizzati (opzionale)
        """
        self.base_url = base_url
        self.headers = headers or {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        }
        self.session = requests.Session()
        self.session.headers.update(self.headers)
    
    def get_page(self, url: str, params: Optional[Dict] = None) -> Optional[BeautifulSoup]:
        """
        Recupera e parsifica una pagina web
        
        Args:
            url: URL della pagina
            params: Parametri query string (opzionale)
            
        Returns:
            Oggetto BeautifulSoup o None in caso di errore
        """
        try:
            response = self.session.get(url, params=params, timeout=10)
            response.raise_for_status()
            return BeautifulSoup(response.content, 'html.parser')
        except requests.RequestException as e:
            print(f"Errore nel recupero della pagina {url}: {e}")
            return None
    
    def _normalize_url(self, href: str) -> str:
        """Converte un URL relativo in assoluto rispetto al base_url"""
        return urljoin(self.base_url, href)

    def extract_links(self, soup: BeautifulSoup, selector: str = 'a') -> List[str]:
        """
        Estrae tutti i link da una pagina
        
        Args:
            soup: Oggetto BeautifulSoup
            selector: Selettore CSS per i link
            
        Returns:
            Lista di URL
        """
        links = []
        for link in soup.select(selector):
            href = link.get('href')
            if href:
                links.append(self._normalize_url(href))
        return links

    def filter_links(
        self,
        soup: BeautifulSoup,
        selector: str = 'a',
        keyword: Optional[str] = None,
        file_extension: Optional[str] = None
    ) -> List[Dict[str, Optional[str]]]:
        """
        Filtra i link che corrispondono a una parola chiave o a una estensione

        Args:
            soup: Oggetto BeautifulSoup
            selector: Selettore CSS per i link
            keyword: Parola chiave da cercare nel testo/URL del link
            file_extension: Estensione file (es. 'pdf', '.csv')

        Returns:
            Lista di dict con link e motivazione del match
        """
        if not keyword and not file_extension:
            return []

        keyword_lower = keyword.lower().strip() if keyword else None
        normalized_ext = file_extension.lower().lstrip('.').strip() if file_extension else None

        matches: List[Dict[str, Optional[str]]] = []

        for link in soup.select(selector):
            href = link.get('href')
            if not href:
                continue

            absolute_href = self._normalize_url(href)
            link_text = link.get_text(strip=True) or None
            reasons = []

            if keyword_lower:
                haystack = f"{absolute_href} {link_text or ''}".lower()
                if keyword_lower in haystack:
                    reasons.append(f"keyword:{keyword_lower}")

            if normalized_ext:
                cleaned_href = absolute_href.lower().split('?', 1)[0].split('#', 1)[0]
                if cleaned_href.endswith(f".{normalized_ext}"):
                    reasons.append(f"ext:.{normalized_ext}")

            if reasons:
                matches.append({
                    'link': absolute_href,
                    'testo_link': link_text,
                    'match': ', '.join(reasons)
                })

        return matches
    
    def extract_text(self, soup: BeautifulSoup, selector: str) -> List[str]:
        """
        Estrae testo da elementi specifici
        
        Args:
            soup: Oggetto BeautifulSoup
            selector: Selettore CSS
            
        Returns:
            Lista di testi estratti
        """
        elements = soup.select(selector)
        return [elem.get_text(strip=True) for elem in elements]
    
    def extract_data(self, soup: BeautifulSoup, selectors: Dict[str, str]) -> List[Dict]:
        """
        Estrae dati strutturati usando selettori multipli
        
        Args:
            soup: Oggetto BeautifulSoup
            selectors: Dizionario {campo: selettore_css}
            
        Returns:
            Lista di dizionari con i dati estratti
        """
        data = []
        # Trova il numero massimo di elementi per qualsiasi selettore
        max_items = 0
        extracted = {}
        
        for field, selector in selectors.items():
            elements = soup.select(selector)
            extracted[field] = elements
            max_items = max(max_items, len(elements))
        
        # Crea dizionari per ogni elemento
        for i in range(max_items):
            item = {}
            for field, elements in extracted.items():
                if i < len(elements):
                    item[field] = elements[i].get_text(strip=True)
                else:
                    item[field] = None
            data.append(item)
        
        return data
    
    def save_to_json(self, data: List[Dict], filename: str):
        """Salva i dati in formato JSON"""
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
        print(f"Dati salvati in {filename}")
    
    def save_to_csv(self, data: List[Dict], filename: str):
        """Salva i dati in formato CSV"""
        if not data:
            print("Nessun dato da salvare")
            return
        
        with open(filename, 'w', newline='', encoding='utf-8') as f:
            writer = csv.DictWriter(f, fieldnames=data[0].keys())
            writer.writeheader()
            writer.writerows(data)
        print(f"Dati salvati in {filename}")
    
    def scrape_with_delay(self, urls: List[str], delay: float = 1.0) -> List[BeautifulSoup]:
        """
        Scrape multipli URL con delay tra le richieste
        
        Args:
            urls: Lista di URL da scrapare
            delay: Secondi di attesa tra le richieste
            
        Returns:
            Lista di oggetti BeautifulSoup
        """
        results = []
        for i, url in enumerate(urls):
            print(f"Scraping {i+1}/{len(urls)}: {url}")
            soup = self.get_page(url)
            if soup:
                results.append(soup)
            time.sleep(delay)
        return results


# Esempio di utilizzo
if __name__ == "__main__":
    # Esempio: scraping di un sito di notizie
    scraper = WebScraperBase("https://example.com")
    
    # Recupera la pagina principale
    soup = scraper.get_page("https://example.com")
    
    if soup:
        # Estrae tutti i titoli
        titles = scraper.extract_text(soup, 'h2.title')
        print(f"Trovati {len(titles)} titoli")
        
        # Estrae dati strutturati
        selectors = {
            'title': 'h2.title',
            'description': 'p.description',
            'date': 'span.date'
        }
        data = scraper.extract_data(soup, selectors)
        
        # Salva i risultati
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        scraper.save_to_json(data, f'scraped_data_{timestamp}.json')
        scraper.save_to_csv(data, f'scraped_data_{timestamp}.csv')