mirror of
https://github.com/guezoloic/millesima-ai-engine.git
synced 2026-03-28 18:03:47 +00:00
ajout: commentaires et les requirements
This commit is contained in:
106
main.py
106
main.py
@@ -6,19 +6,54 @@ from json import JSONDecodeError, loads
|
|||||||
|
|
||||||
|
|
||||||
class Scraper:
|
class Scraper:
|
||||||
def __init__(self, subdir: str = "") -> None:
|
"""
|
||||||
self._url: str = "https://www.millemisa.fr/"
|
Scraper est une classe qui permet de gerer
|
||||||
self._session: Session = Session()
|
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_request: tuple[(str, Response | None)] = ("", None)
|
||||||
|
|
||||||
def _request(self, subdir: str) -> Response:
|
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("/")
|
target_url: str = self._url + subdir.lstrip("/")
|
||||||
response: Response = self._session.get(url=target_url, timeout=10)
|
response: Response = self._session.get(url=target_url, timeout=10)
|
||||||
response.raise_for_status()
|
response.raise_for_status()
|
||||||
return response
|
return response
|
||||||
|
|
||||||
def getresponse(self, subdir: str = "") -> Response:
|
def getresponse(self, subdir: str = "") -> 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.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Response: L'objet réponse (cache ou nouvelle requête).
|
||||||
|
|
||||||
|
Raise:
|
||||||
|
HTTPError: Si le serveur renvoie un code d'erreur (4xx, 5xx).
|
||||||
|
"""
|
||||||
rq_subdir, rq_response = self._latest_request
|
rq_subdir, rq_response = self._latest_request
|
||||||
|
|
||||||
if rq_response is None or subdir != rq_subdir:
|
if rq_response is None or subdir != rq_subdir:
|
||||||
@@ -29,47 +64,50 @@ class Scraper:
|
|||||||
return rq_response
|
return rq_response
|
||||||
|
|
||||||
def getsoup(self, subdir: str = "") -> BeautifulSoup:
|
def getsoup(self, subdir: str = "") -> 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).
|
||||||
|
"""
|
||||||
markup: str = self.getresponse(subdir).text
|
markup: str = self.getresponse(subdir).text
|
||||||
return BeautifulSoup(markup, features="html.parser")
|
return BeautifulSoup(markup, features="html.parser")
|
||||||
|
|
||||||
# def getjsondata(self, subdir: str = "", id: str = "__NEXT_DATA__") -> dict[str, object]:
|
|
||||||
# soup: BeautifulSoup = self.getsoup(subdir)
|
|
||||||
# # On s'assure que c'est bien un Tag pour avoir accès à .string
|
|
||||||
# script = soup.find("script", id=id)
|
|
||||||
|
|
||||||
# if isinstance(script, Tag) and script.string:
|
|
||||||
# try:
|
|
||||||
# # On commence avec le dictionnaire complet
|
|
||||||
# current_data: Any = loads(script.string)
|
|
||||||
|
|
||||||
# # Parcours de la structure imbriquée
|
|
||||||
# keys = ['props', 'pageProps', 'initialReduxState', 'product', 'content']
|
|
||||||
# for key in keys:
|
|
||||||
# if isinstance(current_data, dict) and key in current_data:
|
|
||||||
# current_data = current_data[key]
|
|
||||||
# else:
|
|
||||||
# # Si une clé manque, on lève une erreur explicite
|
|
||||||
# raise ValueError(f"Clé manquante dans le JSON : {key}")
|
|
||||||
|
|
||||||
# # On garantit à Pyright que le résultat final est un dictionnaire
|
|
||||||
# if isinstance(current_data, dict):
|
|
||||||
# return cast(dict[str, object], current_data)
|
|
||||||
|
|
||||||
# except (decoder.JSONDecodeError, ValueError) as e:
|
|
||||||
# print(f"Erreur lors de l'extraction JSON : {e}", file=stderr)
|
|
||||||
|
|
||||||
# return {}
|
|
||||||
|
|
||||||
def getjsondata(
|
def getjsondata(
|
||||||
self, subdir: str = "", id: str = "__NEXT_DATA__"
|
self, subdir: str = "", id: str = "__NEXT_DATA__"
|
||||||
) -> dict[str, object]:
|
) -> dict[str, object]:
|
||||||
|
"""
|
||||||
|
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 <script> pour l'hydratation côté client.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
subdir (str, optional): Le chemin de la page.
|
||||||
|
id (str, optional): L'identifiant de la balise script (par défaut __NEXT_DATA__).
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
HTTPError: Soulevée par `getresponse` si le serveur renvoie un code d'erreur (4xx, 5xx).
|
||||||
|
JSONDecodeError: Soulevée par `loads` si le contenu de la balise n'est pas un JSON valide.
|
||||||
|
ValueError: Soulevée manuellement si l'une des clés attendues (props, pageProps, etc.)
|
||||||
|
est absente de la structure JSON.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict[str, object]: Un dictionnaire contenant les données utiles
|
||||||
|
ou un dictionnaire vide en cas d'erreur.
|
||||||
|
"""
|
||||||
soup: BeautifulSoup = self.getsoup(subdir)
|
soup: BeautifulSoup = self.getsoup(subdir)
|
||||||
script: Tag | None = soup.find("script", id=id)
|
script: Tag | None = soup.find("script", id=id)
|
||||||
|
|
||||||
if isinstance(script, Tag) and script.string:
|
if isinstance(script, Tag) and script.string:
|
||||||
try:
|
try:
|
||||||
current_data: object = loads(script.string)
|
current_data: object = loads(script.string)
|
||||||
# tout le chemin à parcourir pour arriver au données
|
# tout le chemin à parcourir pour arriver au données
|
||||||
# (plein d'information inutile)
|
# (plein d'information inutile)
|
||||||
keys: list[str] = [
|
keys: list[str] = [
|
||||||
"props",
|
"props",
|
||||||
@@ -79,7 +117,7 @@ class Scraper:
|
|||||||
"content",
|
"content",
|
||||||
]
|
]
|
||||||
for key in keys:
|
for key in keys:
|
||||||
# si current_data est bien un dictionnaire et que la clé
|
# si current_data est bien un dictionnaire et que la clé
|
||||||
# est bien dedans
|
# est bien dedans
|
||||||
if isinstance(current_data, dict) and key in current_data:
|
if isinstance(current_data, dict) and key in current_data:
|
||||||
current_data = current_data[key]
|
current_data = current_data[key]
|
||||||
@@ -91,4 +129,4 @@ class Scraper:
|
|||||||
|
|
||||||
except (JSONDecodeError, ValueError) as e:
|
except (JSONDecodeError, ValueError) as e:
|
||||||
print(f"Erreur lors de l'extraction JSON : {e}", file=stderr)
|
print(f"Erreur lors de l'extraction JSON : {e}", file=stderr)
|
||||||
return {}
|
return {}
|
||||||
|
|||||||
@@ -1,2 +1,4 @@
|
|||||||
requests>=2.32.5
|
requests>=2.32.5
|
||||||
beautifulsoup4>=4.14.3
|
requests-mock>=1.12.1
|
||||||
|
beautifulsoup4>=4.14.3
|
||||||
|
|
||||||
Reference in New Issue
Block a user