from typing import cast from requests import Response, Session from bs4 import BeautifulSoup, Tag from collections import OrderedDict from json import loads class _ScraperData: def __init__(self, data: dict[str, object], scraper: Scraper | None = None) -> None: self._data: dict[str, object] = data self._scraper: Scraper | None = scraper def _getcontent(self) -> dict[str, object] | None: """_summary_ Returns: dict[str, object]: _description_ """ current_data: dict[str, object] = self._data for key in ["initialReduxState", "product", "content"]: new_data: object | None = current_data.get(key) if new_data is None: return None current_data: dict[str, object] = cast(dict[str, object], new_data) return current_data def _getattributes(self) -> dict[str, object] | None: """_summary_ Returns: dict[str, object]: _description_ """ current_data: object = self._getcontent() if current_data is None: return None return cast(dict[str, object], current_data.get("attributes")) def appellation(self) -> str | None: """_summary_ Returns: str: _description_ """ current_value: dict[str, object] | None = self._getattributes() if current_value is None: return None app_dict: dict[str, object] = cast( dict[str, object], current_value.get("appellation") ) return cast(str, app_dict["value"]) def acceder(self, name: str) -> _ScraperData: scraper: Scraper | None = self._scraper if scraper is None: scraper = Scraper() return scraper.getjsondata(name) def _getvin(self, name: str) -> str | None: """_summary_ Args: name (str): _description_ Returns: str | None: _description_ """ current_value: dict[str, object] | None = self._getattributes() if current_value is None: return None app_dict: dict[str, object] = cast( dict[str, object], current_value.get(name) ) val: list[str] = cast(str, app_dict.get("attributes")).rstrip("+").split("-") # dans le cas où 93-94 -> [93, 94] -> 93.5 if len(val) > 1: val[0] = str((int(val[0]) + int(val[1])) / 2) return val[0] def parker(self) -> str | None: return self._getvin("note_rp") def robinson(self) -> str | None: return self._getvin("note_jr") def suckling(self) -> str | None: return self._getvin("note_js") def getdata(self) -> dict[str, object]: return self._data class Scraper: """ Scraper est une classe qui permet de gerer de façon dynamique des requetes uniquement sur le serveur https de Millesima """ def __init__(self) -> None: """ Initialise la session de scraping. """ self._url: str = "https://www.millesima.fr/" # Très utile pour éviter de renvoyer toujours les mêmes handshake # TCP et d'avoir toujours une connexion constante avec le server self._session: Session = Session() # Système de cache pour éviter de solliciter le serveur inutilement self._latest_request: tuple[(str, Response)] | None = None self._latest_soups: OrderedDict[str, BeautifulSoup] = OrderedDict[ str, BeautifulSoup ]() def _request(self, subdir: str) -> Response: """ Effectue une requête GET sur le serveur Millesima. Args: subdir (str): Le sous-répertoire ou chemin de l'URL (ex: "/vins"). Returns: Response: L'objet réponse de la requête. Raise: HTTPError: Si le serveur renvoie un code d'erreur (4xx, 5xx). """ target_url: str = self._url + subdir.lstrip("/") response: Response = self._session.get(url=target_url, timeout=10) response.raise_for_status() return response def getresponse(self, subdir: str = "", use_cache: bool = True) -> Response: """ Récupère la réponse d'une page, en utilisant le cache si possible. Args: subdir (str, optional): Le chemin de la page. use_cache (bool, optional): Utilise la donnée deja sauvegarder ou écrase la donnée utilisé avec la nouvelle Returns: Response: L'objet réponse (cache ou nouvelle requête). Raise: HTTPError: Si le serveur renvoie un code d'erreur (4xx, 5xx). """ # si dans le cache, latest_request existe if use_cache and self._latest_request is not None: rq_subdir, rq_response = self._latest_request # si c'est la meme requete et que use_cache est true, # on renvoie celle enregistrer if subdir == rq_subdir: return rq_response request: Response = self._request(subdir) # on recrée la structure pour le systeme de cache si activer if use_cache: self._latest_request = (subdir, request) return request def getsoup(self, subdir: str, use_cache: bool = True) -> BeautifulSoup: """ Récupère le contenu HTML d'une page et le transforme en objet BeautifulSoup. Args: subdir (str, optional): Le chemin de la page. Returns: BeautifulSoup: L'objet parsé pour extraction de données. Raise: HTTPError: Si le serveur renvoie un code d'erreur (4xx, 5xx). """ if use_cache and subdir in self._latest_soups: return self._latest_soups[subdir] markup: str = self.getresponse(subdir).text soup: BeautifulSoup = BeautifulSoup(markup, features="html.parser") if use_cache: self._latest_soups[subdir] = soup if len(self._latest_soups) > 10: _ = self._latest_soups.popitem(last=False) return soup def getjsondata(self, subdir: str, id: str = "__NEXT_DATA__") -> _ScraperData: """ Extrait les données JSON contenues dans la balise __NEXT_DATA__ du site. Beaucoup de sites modernes (Next.js) stockent leur état initial dans une balise