mirror of
https://github.com/guezoloic/millesima_projetS6.git
synced 2026-03-31 04:11:34 +00:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0a3561ffaa | ||
|
|
6366fcd8dd | ||
|
|
5797b72cbc |
188
main.py
188
main.py
@@ -1,36 +1,10 @@
|
|||||||
from sys import stderr
|
from sys import stderr
|
||||||
from typing import cast
|
from typing import cast, Any, Dict, Optional
|
||||||
from requests import Response, Session
|
from requests import Response, Session
|
||||||
from bs4 import BeautifulSoup, Tag
|
from bs4 import BeautifulSoup, Tag
|
||||||
from json import JSONDecodeError, loads
|
from json import JSONDecodeError, loads
|
||||||
|
|
||||||
|
|
||||||
class ScraperData:
|
|
||||||
def __init__(self, data: dict[str, object]) -> None:
|
|
||||||
if not data:
|
|
||||||
raise ValueError("Données insuffisantes pour créer un ScraperData.")
|
|
||||||
self._data: dict[str, object] = data
|
|
||||||
|
|
||||||
def _getattributes(self) -> dict[str, object] | None:
|
|
||||||
current_data: object = self._data.get("attributes")
|
|
||||||
if isinstance(current_data, dict):
|
|
||||||
return cast(dict[str, object], current_data)
|
|
||||||
return None
|
|
||||||
|
|
||||||
def appellation(self) -> str | None:
|
|
||||||
current_value: dict[str, object] | None = self._getattributes()
|
|
||||||
if current_value is not None:
|
|
||||||
app_dict: dict[str, object] = cast(
|
|
||||||
dict[str, object], current_value.get("appellation")
|
|
||||||
)
|
|
||||||
if app_dict:
|
|
||||||
return cast(str, app_dict.get("value"))
|
|
||||||
return None
|
|
||||||
|
|
||||||
def getdata(self) -> dict[str, object]:
|
|
||||||
return self._data
|
|
||||||
|
|
||||||
|
|
||||||
class Scraper:
|
class Scraper:
|
||||||
"""
|
"""
|
||||||
Scraper est une classe qui permet de gerer
|
Scraper est une classe qui permet de gerer
|
||||||
@@ -39,128 +13,98 @@ class Scraper:
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self) -> None:
|
def __init__(self) -> None:
|
||||||
"""
|
|
||||||
Initialise la session de scraping.
|
|
||||||
"""
|
|
||||||
self._url: str = "https://www.millesima.fr/"
|
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()
|
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)
|
||||||
self._latest_soup: tuple[(str, BeautifulSoup | 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 not None and subdir == rq_subdir:
|
request: Response = self._request(subdir)
|
||||||
return rq_response
|
self._latest_request = (subdir, request)
|
||||||
|
return request
|
||||||
request: Response = self._request(subdir)
|
return rq_response
|
||||||
self._latest_request = (subdir, request)
|
|
||||||
return request
|
|
||||||
|
|
||||||
def getsoup(self, subdir: str = "") -> BeautifulSoup:
|
def getsoup(self, subdir: str = "") -> BeautifulSoup:
|
||||||
"""
|
markup: str = self.getresponse(subdir).text
|
||||||
Récupère le contenu HTML d'une page et le transforme en objet BeautifulSoup.
|
return BeautifulSoup(markup, features="html.parser")
|
||||||
|
|
||||||
Args:
|
def getjsondata(self, subdir: str = "", id: str = "__NEXT_DATA__") -> dict[str, object]:
|
||||||
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).
|
|
||||||
"""
|
|
||||||
rq_subdir, rq_soup = self._latest_soup
|
|
||||||
|
|
||||||
if rq_soup is not None and subdir == rq_subdir:
|
|
||||||
return rq_soup
|
|
||||||
|
|
||||||
soup: BeautifulSoup = BeautifulSoup(
|
|
||||||
markup=self.getresponse(subdir).text, features="html.parser"
|
|
||||||
)
|
|
||||||
|
|
||||||
self._latest_soup = (subdir, soup)
|
|
||||||
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 <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
|
keys: list[str] = ["props", "pageProps", "initialReduxState", "product", "content"]
|
||||||
# (plein d'information inutile)
|
|
||||||
keys: list[str] = [
|
|
||||||
"props",
|
|
||||||
"pageProps",
|
|
||||||
"initialReduxState",
|
|
||||||
"product",
|
|
||||||
"content",
|
|
||||||
]
|
|
||||||
for key in keys:
|
for key in keys:
|
||||||
# si current_data est bien un dictionnaire et que la clé
|
|
||||||
# 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: object = current_data[key]
|
current_data = current_data[key]
|
||||||
else:
|
else:
|
||||||
raise ValueError(f"Clé manquante dans le JSON : {key}")
|
raise ValueError(f"Clé manquante dans le JSON : {key}")
|
||||||
|
|
||||||
if isinstance(current_data, dict):
|
if isinstance(current_data, dict):
|
||||||
return ScraperData(data=cast(dict[str, object], current_data))
|
return cast(dict[str, object], current_data)
|
||||||
|
|
||||||
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 ScraperData({})
|
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
BIN
projet.pdf
Normal file
Binary file not shown.
@@ -1,4 +1,2 @@
|
|||||||
requests>=2.32.5
|
requests>=2.32.5
|
||||||
requests-mock>=1.12.1
|
beautifulsoup4>=4.14.3
|
||||||
beautifulsoup4>=4.14.3
|
|
||||||
|
|
||||||
142
test_main.py
142
test_main.py
@@ -1,125 +1,35 @@
|
|||||||
from json import dumps
|
import json
|
||||||
from bs4 import Tag
|
from main import Scraper
|
||||||
import pytest
|
|
||||||
from requests_mock import Mocker
|
|
||||||
from main import Scraper, ScraperData
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(autouse=True)
|
def test_json():
|
||||||
def mock_site():
|
scraper = Scraper()
|
||||||
with Mocker() as m:
|
|
||||||
m.get(
|
|
||||||
"https://www.millesima.fr/",
|
|
||||||
text="<html><body><h1>MILLESIMA</h1></body></html>",
|
|
||||||
)
|
|
||||||
|
|
||||||
json_data = {
|
data = scraper.getjsondata("/chateau-gloria-2016.html")
|
||||||
"props": {
|
|
||||||
"pageProps": {
|
|
||||||
"initialReduxState": {
|
|
||||||
"product": {
|
|
||||||
"content": {
|
|
||||||
"_id": "J4131/22-11652",
|
|
||||||
"partnumber": "J4131/22",
|
|
||||||
"productName": "Nino Negri : 5 Stelle Sfursat 2022",
|
|
||||||
"productNameForSearch": "Nino Negri : 5 Stelle Sfursat 2022",
|
|
||||||
"storeId": "11652",
|
|
||||||
"seoKeyword": "nino-negri-5-stelle-sfursat-2022.html",
|
|
||||||
"title": "Nino Negri : 5 Stelle Sfursat 2022",
|
|
||||||
"items": [
|
|
||||||
{
|
|
||||||
"_id": "J4131/22/C/CC/6-11652",
|
|
||||||
"partnumber": "J4131/22/C/CC/6",
|
|
||||||
"taxRate": "H",
|
|
||||||
"listPrice": 390,
|
|
||||||
"offerPrice": 390,
|
|
||||||
"seoKeyword": "nino-negri-5-stelle-sfursat-2022-c-cc-6.html",
|
|
||||||
"shortdesc": "Un carton de 6 Bouteilles (75cl)",
|
|
||||||
"attributes": {
|
|
||||||
"promotion_o_n": {
|
|
||||||
"valueId": "0",
|
|
||||||
"name": "En promotion",
|
|
||||||
"value": "Non",
|
|
||||||
"sequence": 80,
|
|
||||||
"displayable": "false",
|
|
||||||
"type": "CHECKBOX",
|
|
||||||
"isSpirit": False,
|
|
||||||
},
|
|
||||||
"in_stock": {
|
|
||||||
"valueId": "L",
|
|
||||||
"name": "En stock",
|
|
||||||
"value": "Livrable",
|
|
||||||
"sequence": 65,
|
|
||||||
"displayable": "true",
|
|
||||||
"type": "CHECKBOX",
|
|
||||||
"isSpirit": False,
|
|
||||||
},
|
|
||||||
},
|
|
||||||
"stock": 12,
|
|
||||||
"availability": "2026-02-05",
|
|
||||||
"isCustomizable": False,
|
|
||||||
"gtin_cond": "",
|
|
||||||
"gtin_unit": "",
|
|
||||||
"stockOrigin": "EUR",
|
|
||||||
"isPrevSale": False,
|
|
||||||
}
|
|
||||||
],
|
|
||||||
"attributes": {
|
|
||||||
"appellation": {
|
|
||||||
"valueId": "433",
|
|
||||||
"name": "Appellation",
|
|
||||||
"value": "Sforzato di Valtellina",
|
|
||||||
"url": "sforzato-di-valtellina.html",
|
|
||||||
"isSpirit": False,
|
|
||||||
"groupIdentifier": "appellation_433",
|
|
||||||
},
|
|
||||||
},
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
html_product = f"""
|
print("JSON récupéré :")
|
||||||
<html>
|
print(json.dumps(data, indent=4, ensure_ascii=False))
|
||||||
<script id="__NEXT_DATA__" type="application/json">
|
|
||||||
{dumps(json_data)}
|
|
||||||
</script>
|
|
||||||
</body>
|
|
||||||
</html>
|
|
||||||
"""
|
|
||||||
m.get(
|
|
||||||
"https://www.millesima.fr/nino-negri-5-stelle-sfursat-2022.html",
|
|
||||||
text=html_product,
|
|
||||||
)
|
|
||||||
|
|
||||||
# on return m sans fermer le server qui simule la page
|
assert isinstance(data, dict)
|
||||||
yield m
|
assert "items" in data
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
def test_prix():
|
||||||
def scraper() -> Scraper:
|
scraper = Scraper()
|
||||||
return 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)")
|
||||||
|
|
||||||
|
|
||||||
def test_soup(scraper: Scraper):
|
if __name__ == "__main__":
|
||||||
h1: Tag | None = scraper.getsoup().find("h1")
|
test_json()
|
||||||
|
test_prix()
|
||||||
assert isinstance(h1, Tag)
|
print("\nTous les tests terminés")
|
||||||
assert h1.text == "MILLESIMA"
|
|
||||||
|
|
||||||
|
|
||||||
# def test_getProductName(scraper: Scraper):
|
|
||||||
# jsondata = scraper.getjsondata("nino-negri-5-stelle-sfursat-2022.html")
|
|
||||||
# assert jsondata["productName"] == "Nino Negri : 5 Stelle Sfursat 2022"
|
|
||||||
# assert isinstance(jsondata["items"], list)
|
|
||||||
# assert len(jsondata["items"]) > 0
|
|
||||||
# assert jsondata["items"][0]["offerPrice"] == 390
|
|
||||||
|
|
||||||
|
|
||||||
def test_appellation(scraper: Scraper):
|
|
||||||
appellation: ScraperData = scraper.getjsondata(
|
|
||||||
"nino-negri-5-stelle-sfursat-2022.html"
|
|
||||||
)
|
|
||||||
assert appellation.appellation() == "Sforzato di Valtellina"
|
|
||||||
|
|||||||
Reference in New Issue
Block a user