diff --git a/main.py b/main.py index d3f16b2..86bc1ea 100644 --- a/main.py +++ b/main.py @@ -1,97 +1,110 @@ -import requests -from typing import Any, Dict -from bs4 import BeautifulSoup -import json +from sys import stderr +from typing import cast, Any, Dict, Optional +from requests import Response, Session +from bs4 import BeautifulSoup, Tag +from json import JSONDecodeError, loads class Scraper: """ Scraper est une classe qui permet de gerer de façon dynamique des requetes uniquement - sur le serveur https de Millesina + sur le serveur https de Millesima """ - def __init__(self): - """ - Initialise la session de scraping et récupère la page d'accueil. - """ - # 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: requests.Session = requests.Session() + def __init__(self) -> None: self._url: str = "https://www.millesima.fr/" - self._soup = self.getsoup() + self._session: Session = Session() + self._latest_request: tuple[(str, Response | None)] = ("", None) - def _request( - self, subdir: str, use_cache: bool = True - ) -> requests.Response | requests.HTTPError: - """ - Effectue une requête GET sur le serveur Millesima. - :param subdir: Le sous-répertoire ou chemin de l'URL (ex: "/vins"). - :param use_cache: Si True, retourne la réponse précédente si l'URL est - identique. - :return: requests.Response: L'objet réponse de la requête. - :rtype: requests.HTTPError: Si le serveur renvoie un code d'erreur - (4xx, 5xx). - """ + def _request(self, subdir: str) -> Response: + target_url: str = self._url + subdir.lstrip("/") + response: Response = self._session.get(url=target_url, timeout=10) + response.raise_for_status() + return response - target_url: str = f"{self._url}{subdir.lstrip('/')}" - # Éviter un max possible de faire des requetes au servers même - # en ayant un tunnel tcp avec le paramètre `use_cache` que si - # activer, va comparer l'url avec l'url précédant - if use_cache and hasattr(self, "_response") \ - and self._response is not None: - if self._response.url == target_url: - return self._response + def getresponse(self, subdir: str = "") -> Response: + rq_subdir, rq_response = self._latest_request + if rq_response is None or subdir != rq_subdir: + request: Response = self._request(subdir) + self._latest_request = (subdir, request) + return request + return rq_response - self._response: requests.Response = self._session.get( - target_url, timeout=10) - self._response.raise_for_status() + def getsoup(self, subdir: str = "") -> BeautifulSoup: + markup: str = self.getresponse(subdir).text + return BeautifulSoup(markup, features="html.parser") - return self._response + def getjsondata(self, subdir: str = "", id: str = "__NEXT_DATA__") -> dict[str, object]: + soup: BeautifulSoup = self.getsoup(subdir) + script: Tag | None = soup.find("script", id=id) - def getsoup(self, subdir: str = "/") -> BeautifulSoup: - """ - Récupère le contenu HTML d'une page et le transforme en objet - BeautifulSoup. - - :param subdir: Le chemin de la page. Si None, retourne la soupe - actuelle. - :return: BeautifulSoup: L'objet parsé pour extraction de données. - :rtype: BeautifulSoup - """ - if subdir is not None: - self._request(subdir) - self._soup = BeautifulSoup(self._response.text, "html.parser") - return self._soup - - def get_json_data(self) -> Dict[str, Any]: - """ - 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