mirror of
https://github.com/guezoloic/millesima-ai-engine.git
synced 2026-03-28 18:03:47 +00:00
Compare commits
20 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 9914e8af41 | |||
|
|
5785d571b2 | ||
| ae66e94d6c | |||
| 2bc5d57a31 | |||
| 8f21e48b28 | |||
|
|
9e0cb9737e | ||
| c62a5e6a76 | |||
| 9da0159869 | |||
| 5c22777c2d | |||
| 0d78b1aec3 | |||
| 168ccf88dc | |||
| 6992a0ca16 | |||
| 74482af7f0 | |||
| 76017e3ea3 | |||
|
|
683582a253 | ||
| a81d5be5a9 | |||
| 2327974b6b | |||
| f02a23f032 | |||
|
|
e2317de748 | ||
| 76475d4a2a |
256
main.py
256
main.py
@@ -1,88 +1,224 @@
|
||||
import requests
|
||||
from typing import Any, Dict
|
||||
from bs4 import BeautifulSoup
|
||||
import json
|
||||
from typing import cast
|
||||
from requests import Response, Session
|
||||
from bs4 import BeautifulSoup, Tag
|
||||
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:
|
||||
"""
|
||||
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):
|
||||
def __init__(self) -> None:
|
||||
"""
|
||||
Initialise la session de scraping et récupère la page d'accueil.
|
||||
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: requests.Session = requests.Session()
|
||||
self._url: str = "https://www.millesima.fr/"
|
||||
self._soup = self.getsoup()
|
||||
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_soups: OrderedDict[str, BeautifulSoup] = OrderedDict[
|
||||
str, BeautifulSoup
|
||||
]()
|
||||
|
||||
def _request(
|
||||
self, subdir: str, use_cache: bool = True
|
||||
) -> requests.Response | requests.HTTPError:
|
||||
def _request(self, subdir: str) -> Response:
|
||||
"""
|
||||
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).
|
||||
|
||||
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("/")
|
||||
response: Response = self._session.get(url=target_url, timeout=10)
|
||||
response.raise_for_status()
|
||||
return 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).
|
||||
|
||||
Raise:
|
||||
HTTPError: Si le serveur renvoie un code d'erreur (4xx, 5xx).
|
||||
"""
|
||||
|
||||
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
|
||||
# si dans le cache, latest_request existe
|
||||
if use_cache and self._latest_request is not None:
|
||||
rq_subdir, rq_response = self._latest_request
|
||||
|
||||
self._response: requests.Response = self._session.get(
|
||||
target_url, timeout=10)
|
||||
self._response.raise_for_status()
|
||||
# si c'est la meme requete et que use_cache est true,
|
||||
# on renvoie celle enregistrer
|
||||
if subdir == rq_subdir:
|
||||
return rq_response
|
||||
|
||||
return self._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)
|
||||
|
||||
def getsoup(self, subdir: str = "/") -> BeautifulSoup:
|
||||
return request
|
||||
|
||||
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.
|
||||
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
|
||||
Args:
|
||||
subdir (str, optional): Le chemin de la page.
|
||||
|
||||
def get_json_data(self) -> Dict[str, Any]:
|
||||
Returns:
|
||||
BeautifulSoup: L'objet parsé pour extraction de données.
|
||||
|
||||
Raise:
|
||||
HTTPError: Si le serveur renvoie un code d'erreur (4xx, 5xx).
|
||||
"""
|
||||
Extrait les données JSON contenues dans la balise __NEXT_DATA__ du
|
||||
site.
|
||||
|
||||
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.
|
||||
|
||||
:return Dict[str, Any]: Un dictionnaire contenant les props de la page,
|
||||
ou un dictionnaire vide en cas d'erreur ou
|
||||
d'absence.
|
||||
Args:
|
||||
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.)
|
||||
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.
|
||||
"""
|
||||
script = self._soup.find("script", id="__NEXT_DATA__")
|
||||
if script 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
|
||||
return {}
|
||||
soup: BeautifulSoup = self.getsoup(subdir)
|
||||
script: Tag | None = soup.find("script", id=id)
|
||||
|
||||
if script is None or not script.string:
|
||||
raise ValueError(f"le script id={id} est introuvable")
|
||||
|
||||
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))
|
||||
|
||||
|
||||
@@ -1,2 +1,3 @@
|
||||
requests>=2.32.5
|
||||
requests-mock>=1.12.1
|
||||
beautifulsoup4>=4.14.3
|
||||
228
test_main.py
228
test_main.py
@@ -1,6 +1,226 @@
|
||||
from main import *
|
||||
from json import dumps
|
||||
from bs4 import Tag
|
||||
import pytest
|
||||
from requests_mock import Mocker
|
||||
from main import Scraper
|
||||
|
||||
scraper = Scraper()
|
||||
|
||||
def test_soup():
|
||||
assert scraper.getsoup().find('h1').text[3:12] == "MILLESIMA"
|
||||
@pytest.fixture(autouse=True)
|
||||
def mock_site():
|
||||
with Mocker() as m:
|
||||
m.get(
|
||||
"https://www.millesima.fr/",
|
||||
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 = {
|
||||
"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": {
|
||||
"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,
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
html_product = f"""
|
||||
<html>
|
||||
<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,
|
||||
)
|
||||
|
||||
# on return m sans fermer le server qui simule la page
|
||||
yield m
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def scraper() -> Scraper:
|
||||
return Scraper()
|
||||
|
||||
|
||||
def test_soup(scraper: Scraper):
|
||||
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_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