3 Commits
exo3 ... exo2

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 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 <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:
if isinstance(script, Tag) and script.string:
try:
data: dict[str, Any] = json.loads(script.string)
for element in ['props', 'pageProps', 'initialReduxState',
'product', 'content']:
data.get(element)
return data
except json.decoder.JSONDecodeError:
pass
current_data: object = loads(script.string)
keys: list[str] = ["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:
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 {}
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():
assert scraper.getsoup().find('h1').text[3:12] == "MILLESIMA"
def test_json():
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")