mirror of
https://github.com/guezoloic/millesima_projetS6.git
synced 2026-03-28 19:13:42 +00:00
Compare commits
16 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 9914e8af41 | |||
|
|
5785d571b2 | ||
| ae66e94d6c | |||
| 2bc5d57a31 | |||
| 8f21e48b28 | |||
|
|
9e0cb9737e | ||
| c62a5e6a76 | |||
| 9da0159869 | |||
| 5c22777c2d | |||
| 0d78b1aec3 | |||
| 168ccf88dc | |||
| 6992a0ca16 | |||
| 74482af7f0 | |||
| 76017e3ea3 | |||
|
|
683582a253 | ||
|
|
e2317de748 |
174
main.py
174
main.py
@@ -1,8 +1,90 @@
|
||||
from sys import stderr
|
||||
from typing import cast
|
||||
from requests import Response, Session
|
||||
from bs4 import BeautifulSoup, Tag
|
||||
from json import JSONDecodeError, loads
|
||||
from collections import OrderedDict
|
||||
from json import loads
|
||||
|
||||
|
||||
class _ScraperData:
|
||||
def __init__(self, data: dict[str, object]) -> None:
|
||||
self._data: dict[str, object] = data
|
||||
|
||||
def _getcontent(self) -> dict[str, object] | None:
|
||||
"""_summary_
|
||||
|
||||
Returns:
|
||||
dict[str, object]: _description_
|
||||
"""
|
||||
current_data: dict[str, object] = self._data
|
||||
for key in ["initialReduxState", "product", "content"]:
|
||||
new_data: object | None = current_data.get(key)
|
||||
if new_data is None:
|
||||
return None
|
||||
current_data: dict[str, object] = cast(dict[str, object], new_data)
|
||||
|
||||
return current_data
|
||||
|
||||
def _getattributes(self) -> dict[str, object] | None:
|
||||
"""_summary_
|
||||
|
||||
Returns:
|
||||
dict[str, object]: _description_
|
||||
"""
|
||||
current_data: object = self._getcontent()
|
||||
if current_data is None:
|
||||
return None
|
||||
return cast(dict[str, object], current_data.get("attributes"))
|
||||
|
||||
def appellation(self) -> str | None:
|
||||
"""_summary_
|
||||
|
||||
Returns:
|
||||
str: _description_
|
||||
"""
|
||||
attrs: dict[str, object] | None = self._getattributes()
|
||||
|
||||
if attrs is not None:
|
||||
app_dict: object | None = attrs.get("appellation")
|
||||
if isinstance(app_dict, dict):
|
||||
return cast(str, app_dict.get("value"))
|
||||
return None
|
||||
|
||||
def _getcritiques(self, name: str) -> str | None:
|
||||
"""_summary_
|
||||
|
||||
Args:
|
||||
name (str): _description_
|
||||
|
||||
Returns:
|
||||
str | None: _description_
|
||||
"""
|
||||
|
||||
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(name)
|
||||
)
|
||||
if not app_dict:
|
||||
return None
|
||||
|
||||
val = cast(str, app_dict.get("value")).rstrip("+").split("-")
|
||||
if len(val) > 1:
|
||||
val[0] = str((int(val[0]) + int(val[1])) / 2)
|
||||
|
||||
return val[0]
|
||||
return None
|
||||
|
||||
def parker(self) -> str | None:
|
||||
return self._getcritiques("note_rp")
|
||||
|
||||
def robinson(self) -> str | None:
|
||||
return self._getcritiques("note_jr")
|
||||
|
||||
def suckling(self) -> str | None:
|
||||
return self._getcritiques("note_js")
|
||||
|
||||
def getdata(self) -> dict[str, object]:
|
||||
return self._data
|
||||
|
||||
|
||||
class Scraper:
|
||||
@@ -21,7 +103,10 @@ class Scraper:
|
||||
# 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
|
||||
self._latest_soups: OrderedDict[str, BeautifulSoup] = OrderedDict[
|
||||
str, BeautifulSoup
|
||||
]()
|
||||
|
||||
def _request(self, subdir: str) -> Response:
|
||||
"""
|
||||
@@ -41,12 +126,14 @@ class Scraper:
|
||||
response.raise_for_status()
|
||||
return response
|
||||
|
||||
def getresponse(self, subdir: str = "") -> Response:
|
||||
def getresponse(self, subdir: str = "", use_cache: bool = True) -> 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.
|
||||
use_cache (bool, optional): Utilise la donnée deja sauvegarder ou
|
||||
écrase la donnée utilisé avec la nouvelle
|
||||
|
||||
Returns:
|
||||
Response: L'objet réponse (cache ou nouvelle requête).
|
||||
@@ -54,16 +141,24 @@ class Scraper:
|
||||
Raise:
|
||||
HTTPError: Si le serveur renvoie un code d'erreur (4xx, 5xx).
|
||||
"""
|
||||
rq_subdir, rq_response = self._latest_request
|
||||
|
||||
if rq_response is None or subdir != rq_subdir:
|
||||
request: Response = self._request(subdir)
|
||||
# si dans le cache, latest_request existe
|
||||
if use_cache and self._latest_request is not None:
|
||||
rq_subdir, rq_response = self._latest_request
|
||||
|
||||
# si c'est la meme requete et que use_cache est true,
|
||||
# on renvoie celle enregistrer
|
||||
if subdir == rq_subdir:
|
||||
return rq_response
|
||||
|
||||
request: Response = self._request(subdir)
|
||||
# on recrée la structure pour le systeme de cache si activer
|
||||
if use_cache:
|
||||
self._latest_request = (subdir, request)
|
||||
return request
|
||||
|
||||
return rq_response
|
||||
return request
|
||||
|
||||
def getsoup(self, subdir: str = "") -> BeautifulSoup:
|
||||
def getsoup(self, subdir: str, use_cache: bool = True) -> BeautifulSoup:
|
||||
"""
|
||||
Récupère le contenu HTML d'une page et le transforme en objet BeautifulSoup.
|
||||
|
||||
@@ -76,25 +171,35 @@ class Scraper:
|
||||
Raise:
|
||||
HTTPError: Si le serveur renvoie un code d'erreur (4xx, 5xx).
|
||||
"""
|
||||
markup: str = self.getresponse(subdir).text
|
||||
return BeautifulSoup(markup, features="html.parser")
|
||||
|
||||
def getjsondata(
|
||||
self, subdir: str = "", id: str = "__NEXT_DATA__"
|
||||
) -> dict[str, object]:
|
||||
if use_cache and subdir in self._latest_soups:
|
||||
return self._latest_soups[subdir]
|
||||
|
||||
markup: str = self.getresponse(subdir).text
|
||||
soup: BeautifulSoup = BeautifulSoup(markup, features="html.parser")
|
||||
|
||||
if use_cache:
|
||||
self._latest_soups[subdir] = soup
|
||||
|
||||
if len(self._latest_soups) > 10:
|
||||
_ = self._latest_soups.popitem(last=False)
|
||||
|
||||
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.
|
||||
subdir (str): 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.)
|
||||
ValueError: Soulevée manuellement si l'une des clés attendues (props, pageProps, etc.)
|
||||
est absente de la structure JSON.
|
||||
|
||||
Returns:
|
||||
@@ -104,29 +209,16 @@ class Scraper:
|
||||
soup: BeautifulSoup = self.getsoup(subdir)
|
||||
script: Tag | None = soup.find("script", id=id)
|
||||
|
||||
if isinstance(script, Tag) and script.string:
|
||||
try:
|
||||
current_data: object = loads(script.string)
|
||||
# tout le chemin à parcourir pour arriver au données
|
||||
# (plein d'information inutile)
|
||||
keys: list[str] = [
|
||||
"props",
|
||||
"pageProps",
|
||||
"initialReduxState",
|
||||
"product",
|
||||
"content",
|
||||
]
|
||||
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:
|
||||
current_data = current_data[key]
|
||||
else:
|
||||
raise ValueError(f"Clé manquante dans le JSON : {key}")
|
||||
if script is None or not script.string:
|
||||
raise ValueError(f"le script id={id} est introuvable")
|
||||
|
||||
if isinstance(current_data, dict):
|
||||
return cast(dict[str, object], current_data)
|
||||
current_data: object = cast(object, loads(script.string))
|
||||
|
||||
for key in ["props", "pageProps"]:
|
||||
if isinstance(current_data, dict) and key in current_data:
|
||||
current_data = cast(object, current_data[key])
|
||||
continue
|
||||
raise ValueError(f"Clé manquante dans le JSON : {key}")
|
||||
|
||||
return _ScraperData(cast(dict[str, object], current_data))
|
||||
|
||||
except (JSONDecodeError, ValueError) as e:
|
||||
print(f"Erreur lors de l'extraction JSON : {e}", file=stderr)
|
||||
return {}
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
requests>=2.32.5
|
||||
requests-mock>=1.12.1
|
||||
beautifulsoup4>=4.14.3
|
||||
|
||||
beautifulsoup4>=4.14.3
|
||||
159
test_main.py
159
test_main.py
@@ -10,7 +10,47 @@ def mock_site():
|
||||
with Mocker() as m:
|
||||
m.get(
|
||||
"https://www.millesima.fr/",
|
||||
text="<html><body><h1>MILLESIMA</h1></body></html>",
|
||||
text=f"""
|
||||
<html>
|
||||
<body>
|
||||
<script id="__NEXT_DATA__" type="application/json">
|
||||
{dumps({
|
||||
"props": {
|
||||
"pageProps": {
|
||||
"initialReduxState": {
|
||||
"product": {
|
||||
"content": {
|
||||
"items": [],
|
||||
"attributes": {}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
})}
|
||||
</script>
|
||||
</body>
|
||||
</html>
|
||||
""",
|
||||
)
|
||||
|
||||
m.get(
|
||||
"https://www.millesima.fr/poubelle",
|
||||
text=f"""
|
||||
<html>
|
||||
<body>
|
||||
<h1>POUBELLE</h1>
|
||||
<script id="__NEXT_DATA__" type="application/json">
|
||||
{dumps({
|
||||
"props": {
|
||||
"pageProps": {
|
||||
}
|
||||
}
|
||||
})}
|
||||
</script>
|
||||
</body>
|
||||
</html>
|
||||
""",
|
||||
)
|
||||
|
||||
json_data = {
|
||||
@@ -24,11 +64,8 @@ def mock_site():
|
||||
"productName": "Nino Negri : 5 Stelle Sfursat 2022",
|
||||
"productNameForSearch": "Nino Negri : 5 Stelle Sfursat 2022",
|
||||
"storeId": "11652",
|
||||
"longdesc": "<h2>Caractéristiques et conseils de dégustation du 5 Stelle Sfursat 2022 de Nino Negri</h2><p><strong>Dégustation</strong></p><p><em>Robe</em></p><p>La robe dévoile une couleur grenat d'intensité moyenne.</p><p><em>Nez</em></p><p>Le nez révèle des arômes singuliers de fruits mûrs accompagnés de notes d'épices douces.</p><p><em>Bouche</em></p><p>En bouche, ce vin séduit par son équilibre remarquable, sa richesse et son caractère corsé. La dégustation dévoile une concentration intense et vigoureuse, portée par un fond aristocratique de mûre bien mûre et d'épices. La finale se distingue par sa longueur et sa persistance.</p><p><strong>Accords mets et vins</strong></p><p>Ce vin de caractère accompagne parfaitement les viandes rouges braisées, le gibier en sauce ou encore les fromages affinés à pâte dure.</p><p><strong>Service et garde</strong></p><p>Le 5 Stelle Sfursat 2022 gagnera à être servi à une température comprise entre 16 et 18°C.</p><h2>Un Sforzato di Valtellina d'exception élaboré par la Maison Nino Negri</h2><p><strong>La propriété</strong></p><p>Fondée en 1897 par Nino Negri à Chiuro en Valteline, cette Maison lombarde représente aujourd'hui la plus importante cave de la région. Propriété du Gruppo Italiano Vini depuis 1986, elle cultive 38 hectares de vignobles en terrasses sur des pentes alpines aux sols granitiques et calcaires. Sous la houlette de l'œnologue Danilo Drocco, <a href=\"/producteur-nino-negri.html\">Nino Negri</a> perpétue l'excellence du nebbiolo valtelin, notamment à travers son emblématique Sforzato élaboré selon la méthode traditionnelle d'appassimento.</p><p><strong>Le vignoble</strong></p><p>Le 5 Stelle Sfursat est issu de l'appellation <a href=\"/sforzato-di-valtellina.html\">Sforzato di Valtellina</a> DOCG, territoire d'exception où le nebbiolo s'épanouit sur des terrasses alpines escarpées. Les raisins proviennent de vignobles implantés sur des pentes granitiques et calcaires, bénéficiant d'une exposition optimale permettant une maturation idéale du nebbiolo.</p><p><strong>Vinification et élevage</strong></p><p>Le 5 Stelle Sfursat 2022 est produit uniquement lors des saisons les plus favorables. Les raisins sont récoltés manuellement et disposés en couche unique dans des caisses de 4 kg. Ils sont ensuite soumis à un séchage naturel dans un grenier pendant environ trois mois avant la vinification, selon la méthode traditionnelle de l'appassimento. Ce processus permet aux baies de perdre près de 30 % de leur poids, concentrant ainsi les arômes et les sucres naturels.</p><p><strong>Cépage</strong></p><p>Ce <a href=\"/lombardie.html\">vin de Lombardie</a> est un 100 % nebbiolo</p>",
|
||||
"image": "J4131_2022NM_c.png",
|
||||
"seoKeyword": "nino-negri-5-stelle-sfursat-2022.html",
|
||||
"title": "Nino Negri : 5 Stelle Sfursat 2022",
|
||||
"metaDesc": "Nino Negri : 5 Stelle Sfursat 2022 : Vente en ligne, Grand vin d'origine garantie en provenance directe de la propriété - ✅ Qualité de stockage",
|
||||
"items": [
|
||||
{
|
||||
"_id": "J4131/22/C/CC/6-11652",
|
||||
@@ -44,7 +81,7 @@ def mock_site():
|
||||
"name": "En promotion",
|
||||
"value": "Non",
|
||||
"sequence": 80,
|
||||
"displayable": "false",
|
||||
"displayable": "False",
|
||||
"type": "CHECKBOX",
|
||||
"isSpirit": False,
|
||||
},
|
||||
@@ -66,7 +103,47 @@ def mock_site():
|
||||
"stockOrigin": "EUR",
|
||||
"isPrevSale": False,
|
||||
}
|
||||
],
|
||||
],
|
||||
"attributes": {
|
||||
"equivbtl": {
|
||||
"valueId": "1",
|
||||
"name": "equivbtl",
|
||||
"value": "1",
|
||||
"isSpirit": False,
|
||||
},
|
||||
"nbunit": {
|
||||
"valueId": "6",
|
||||
"name": "nbunit",
|
||||
"value": "6",
|
||||
"isSpirit": False,
|
||||
},
|
||||
"appellation": {
|
||||
"valueId": "433",
|
||||
"name": "Appellation",
|
||||
"value": "Sforzato di Valtellina",
|
||||
"url": "sforzato-di-valtellina.html",
|
||||
"isSpirit": False,
|
||||
"groupIdentifier": "appellation_433",
|
||||
},
|
||||
"note_rp": {
|
||||
"valueId": "91",
|
||||
"name": "Parker",
|
||||
"value": "91",
|
||||
"isSpirit": False,
|
||||
},
|
||||
"note_jr": {
|
||||
"valueId": "17+",
|
||||
"name": "J. Robinson",
|
||||
"value": "17+",
|
||||
"isSpirit": False,
|
||||
},
|
||||
"note_js": {
|
||||
"valueId": "93-94",
|
||||
"name": "J. Suckling",
|
||||
"value": "93-94",
|
||||
"isSpirit": False,
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -76,13 +153,18 @@ def mock_site():
|
||||
|
||||
html_product = f"""
|
||||
<html>
|
||||
<script id="__NEXT_DATA__" type="application/json">
|
||||
{dumps(json_data)}
|
||||
</script>
|
||||
</body>
|
||||
<body>
|
||||
<h1>MILLESIMA</h1>
|
||||
<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)
|
||||
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
|
||||
yield m
|
||||
@@ -94,14 +176,51 @@ def scraper() -> Scraper:
|
||||
|
||||
|
||||
def test_soup(scraper: Scraper):
|
||||
h1: Tag | None = scraper.getsoup().find("h1")
|
||||
|
||||
assert isinstance(h1, Tag)
|
||||
assert h1.text == "MILLESIMA"
|
||||
vide = scraper.getsoup("")
|
||||
poubelle = scraper.getsoup("poubelle")
|
||||
contenu = scraper.getsoup("nino-negri-5-stelle-sfursat-2022.html")
|
||||
assert vide.find("h1") is None
|
||||
assert str(poubelle.find("h1")) == "<h1>POUBELLE</h1>"
|
||||
assert str(contenu.find("h1")) == "<h1>MILLESIMA</h1>"
|
||||
|
||||
|
||||
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 len(jsondata["items"]) > 0
|
||||
assert jsondata["items"][0]["offerPrice"] == 390
|
||||
def test_appellation(scraper: Scraper):
|
||||
vide = scraper.getjsondata("")
|
||||
poubelle = scraper.getjsondata("poubelle")
|
||||
contenu = scraper.getjsondata("nino-negri-5-stelle-sfursat-2022.html")
|
||||
assert vide.appellation() is None
|
||||
assert poubelle.appellation() is None
|
||||
assert contenu.appellation() == "Sforzato di Valtellina"
|
||||
|
||||
|
||||
def test_fonctionprivee(scraper: Scraper):
|
||||
vide = scraper.getjsondata("")
|
||||
poubelle = scraper.getjsondata("poubelle")
|
||||
contenu = scraper.getjsondata("nino-negri-5-stelle-sfursat-2022.html")
|
||||
assert vide._getattributes() is not None
|
||||
assert vide._getattributes() == {}
|
||||
assert vide._getcontent() is not None
|
||||
assert vide._getcontent() == {"items": [], "attributes": {}}
|
||||
assert poubelle._getattributes() is None
|
||||
assert poubelle._getcontent() is None
|
||||
assert contenu._getcontent() is not None
|
||||
assert contenu._getattributes() is not None
|
||||
|
||||
|
||||
|
||||
def test_critiques(scraper: Scraper):
|
||||
vide = scraper.getjsondata("")
|
||||
poubelle = scraper.getjsondata("poubelle")
|
||||
contenu = scraper.getjsondata("nino-negri-5-stelle-sfursat-2022.html")
|
||||
assert vide.parker() is None
|
||||
assert vide.robinson() is None
|
||||
assert vide.suckling() is None
|
||||
assert vide._getcritiques("test_ts") is None
|
||||
assert poubelle.parker() is None
|
||||
assert poubelle.robinson() is None
|
||||
assert poubelle.suckling() is None
|
||||
assert poubelle._getcritiques("test_ts") is None
|
||||
assert contenu.parker() == "91"
|
||||
assert contenu.robinson() == "17"
|
||||
assert contenu.suckling() == "93.5"
|
||||
assert contenu._getcritiques("test_ts") is None
|
||||
|
||||
Reference in New Issue
Block a user