3 Commits

Author SHA1 Message Date
DAHMANI chahrazad
0a3561ffaa Fix comments for clarity in main.py 2026-02-07 02:25:31 +01:00
Chahrazad650
6366fcd8dd ajout fonction prix 2026-02-07 02:21:30 +01:00
Chahrazad650
5797b72cbc juste un test 2026-02-05 18:28:07 +01:00
3 changed files with 127 additions and 76 deletions

166
main.py
View File

@@ -1,88 +1,110 @@
import requests from sys import stderr
from typing import Any, Dict from typing import cast, Any, Dict, Optional
from bs4 import BeautifulSoup from requests import Response, Session
import json from bs4 import BeautifulSoup, Tag
from json import JSONDecodeError, loads
class Scraper: class Scraper:
""" """
Scraper est une classe qui permet de gerer Scraper est une classe qui permet de gerer
de façon dynamique des requetes uniquement de façon dynamique des requetes uniquement
sur le serveur https de Millesina sur le serveur https de Millesima
""" """
def __init__(self): def __init__(self) -> None:
"""
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()
self._url: str = "https://www.millesima.fr/" 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( def _request(self, subdir: str) -> Response:
self, subdir: str, use_cache: bool = True target_url: str = self._url + subdir.lstrip("/")
) -> requests.Response | requests.HTTPError: response: Response = self._session.get(url=target_url, timeout=10)
""" response.raise_for_status()
Effectue une requête GET sur le serveur Millesima. return response
: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).
"""
target_url: str = f"{self._url}{subdir.lstrip('/')}" def getresponse(self, subdir: str = "") -> Response:
# Éviter un max possible de faire des requetes au servers même rq_subdir, rq_response = self._latest_request
# en ayant un tunnel tcp avec le paramètre `use_cache` que si if rq_response is None or subdir != rq_subdir:
# activer, va comparer l'url avec l'url précédant request: Response = self._request(subdir)
if use_cache and hasattr(self, "_response") \ self._latest_request = (subdir, request)
and self._response is not None: return request
if self._response.url == target_url: return rq_response
return self._response
self._response: requests.Response = self._session.get( def getsoup(self, subdir: str = "") -> BeautifulSoup:
target_url, timeout=10) markup: str = self.getresponse(subdir).text
self._response.raise_for_status() 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: if isinstance(script, Tag) and script.string:
"""
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 <script> pour l'hydratation côté client.
:return Dict[str, Any]: Un dictionnaire contenant les props de la page,
ou un dictionnaire vide en cas d'erreur ou
d'absence.
"""
script = self._soup.find("script", id="__NEXT_DATA__")
if script and script.string:
try: try:
data: dict[str, Any] = json.loads(script.string) current_data: object = loads(script.string)
for element in ['props', 'pageProps', 'initialReduxState', keys: list[str] = ["props", "pageProps", "initialReduxState", "product", "content"]
'product', 'content']: for key in keys:
data.get(element) if isinstance(current_data, dict) and key in current_data:
return data current_data = current_data[key]
except json.decoder.JSONDecodeError: else:
pass raise ValueError(f"Clé manquante dans le JSON : {key}")
if isinstance(current_data, dict):
return cast(dict[str, object], current_data)
except (JSONDecodeError, ValueError) as e:
print(f"Erreur lors de l'extraction JSON : {e}", file=stderr)
return {} return {}
def prix(self, subdir: str) -> float:
"""
Retourne le prix d'une bouteille (75cl).
Les données récupérées depuis le site contiennent plusieurs formats
de vente dans la liste "items" :
- bouteille seule si nbunit=1 et equivbtl=1
-> prix direct (format vendu à l'unité).
- caisse de plusieurs bouteilles si nbunit=1
-> prix direct (format vendu à l'unité).
- formats spéciaux (magnum, impériale, etc.)sinon
-> calcul du prix unitaire : offerPrice / (nbunit * equivbtl)
Chaque item possède notamment :
- offerPrice : prix total du format proposé
- nbunit : nombre d'unités dans le format
- equivbtl : équivalent en nombre de bouteilles standard (75cl)
"""
data = self.getjsondata(subdir)
items = data.get("items")
if not isinstance(items, list) or len(items) == 0:
raise ValueError("Aucun prix disponible (items vide).")
# 1) bouteille 75cl (nbunit=1 et equivbtl=1)
for item in items:
if not isinstance(item, dict):
continue
attrs = item.get("attributes", {})
nbunit = attrs.get("nbunit", {}).get("value")
equivbtl = attrs.get("equivbtl", {}).get("value")
if nbunit == "1" and equivbtl == "1":
p = item.get("offerPrice")
if isinstance(p, (int, float)):
return float(p)
# 2) calcul depuis caisse
for item in items:
if not isinstance(item, dict):
continue
p = item.get("offerPrice")
attrs = item.get("attributes", {})
nbunit = attrs.get("nbunit", {}).get("value")
equivbtl = attrs.get("equivbtl", {}).get("value")
if isinstance(p, (int, float)) and nbunit and equivbtl:
denom = float(nbunit) * float(equivbtl)
if denom > 0:
return round(float(p) / denom, 2)
raise ValueError("Impossible de trouver le prix unitaire.")

BIN
projet.pdf Normal file

Binary file not shown.

View File

@@ -1,6 +1,35 @@
from main import * import json
from main import Scraper
scraper = Scraper()
def test_soup(): def test_json():
assert scraper.getsoup().find('h1').text[3:12] == "MILLESIMA" scraper = Scraper()
data = scraper.getjsondata("/chateau-gloria-2016.html")
print("JSON récupéré :")
print(json.dumps(data, indent=4, ensure_ascii=False))
assert isinstance(data, dict)
assert "items" in data
def test_prix():
scraper = Scraper()
try:
p = scraper.prix("/chateau-saint-pierre-2011.html")
print("Prix unitaire =", p)
assert isinstance(p, float)
assert p > 0
except ValueError:
# le vin n'est pas disponible à la vente
print("OK : aucun prix (vin indisponible, items vide)")
if __name__ == "__main__":
test_json()
test_prix()
print("\nTous les tests terminés")