mirror of
https://github.com/guezoloic/millesima-ai-engine.git
synced 2026-03-28 18:03:47 +00:00
DAHMANI chahrazad
2
.gitignore
vendored
2
.gitignore
vendored
@@ -205,3 +205,5 @@ cython_debug/
|
||||
marimo/_static/
|
||||
marimo/_lsp/
|
||||
__marimo__/
|
||||
|
||||
*.csv
|
||||
170
main.py
170
main.py
@@ -1,12 +1,19 @@
|
||||
from typing import cast
|
||||
from requests import Response, Session
|
||||
from requests import HTTPError, Response, Session
|
||||
from bs4 import BeautifulSoup, Tag
|
||||
from collections import OrderedDict
|
||||
from json import loads
|
||||
from json import JSONDecodeError, loads
|
||||
|
||||
|
||||
class _ScraperData:
|
||||
"""_summary_
|
||||
"""
|
||||
def __init__(self, data: dict[str, object]) -> None:
|
||||
"""_summary_
|
||||
|
||||
Args:
|
||||
data (dict[str, object]): _description_
|
||||
"""
|
||||
self._data: dict[str, object] = data
|
||||
|
||||
def _getcontent(self) -> dict[str, object] | None:
|
||||
@@ -35,90 +42,50 @@ class _ScraperData:
|
||||
return None
|
||||
return cast(dict[str, object], current_data.get("attributes"))
|
||||
|
||||
def prix(self) -> float:
|
||||
def prix(self) -> float | None:
|
||||
"""
|
||||
Retourne le prix unitaire d'une bouteille (75cl).
|
||||
|
||||
Le JSON contient plusieurs formats de vente dans content["items"] :
|
||||
- bouteille seule : nbunit = 1 et equivbtl = 1 -> prix direct
|
||||
- caisse de plusieurs bouteilles : nbunit > 1 -> on divise le prix total
|
||||
- formats spéciaux (magnum etc.) : equivbtl > 1 -> même calcul
|
||||
|
||||
Formule générale :
|
||||
prix_unitaire = offerPrice / (nbunit * equivbtl)
|
||||
|
||||
Si aucun prix n'est disponible, retourne None.
|
||||
"""
|
||||
|
||||
content = self._getcontent()
|
||||
|
||||
# si content n'existe pas -> erreur
|
||||
if content is None:
|
||||
raise ValueError("Contenu introuvable")
|
||||
return None
|
||||
|
||||
# On récupère la liste des formats disponibles (bouteille, carton...)
|
||||
items = content.get("items")
|
||||
|
||||
# Vérification que items est bien une liste non vide
|
||||
# Vérifie que items existe et n'est pas vide
|
||||
if not isinstance(items, list) or len(items) == 0:
|
||||
raise ValueError("Aucun prix disponible (items vide)")
|
||||
return None
|
||||
|
||||
prix_calcule: float | None = None
|
||||
|
||||
# --------------------------
|
||||
# CAS 1 : bouteille unitaire
|
||||
# --------------------------
|
||||
# On cherche un format où nbunit=1 et equivbtl=1 ->bouteille standard 75cl
|
||||
for item in items:
|
||||
|
||||
if not isinstance(item, dict):
|
||||
continue
|
||||
|
||||
# On récupère les attributs du format
|
||||
p = item.get("offerPrice")
|
||||
attrs = item.get("attributes", {})
|
||||
|
||||
# On récupère nbunit et equivbtl
|
||||
nbunit = attrs.get("nbunit", {}).get("value")
|
||||
equivbtl = attrs.get("equivbtl", {}).get("value")
|
||||
|
||||
# Si c'est une bouteille unitaire
|
||||
if nbunit == "1" and equivbtl == "1":
|
||||
if not isinstance(p, (int, float)) or not nbunit or not equivbtl:
|
||||
continue
|
||||
|
||||
p = item.get("offerPrice")
|
||||
nb = float(nbunit)
|
||||
eq = float(equivbtl)
|
||||
|
||||
# Vérification que c'est bien un nombre
|
||||
if isinstance(p, (int, float)):
|
||||
if nb <= 0 or eq <= 0:
|
||||
continue
|
||||
|
||||
if nb == 1 and eq == 1:
|
||||
return float(p)
|
||||
|
||||
# --------------------------
|
||||
# CAS 2 : caisse ou autre format
|
||||
# --------------------------
|
||||
# On calcule le prix unitaire à partir du prix total
|
||||
for item in items:
|
||||
prix_calcule = round(float(p) / (nb * eq), 2)
|
||||
|
||||
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")
|
||||
|
||||
# Vérification que toutes les valeurs existent
|
||||
if isinstance(p, (int, float)) and nbunit and equivbtl:
|
||||
|
||||
# Calcul du nombre total de bouteilles équivalentes
|
||||
denom = float(nbunit) * float(equivbtl)
|
||||
|
||||
# Évite division par zéro
|
||||
if denom > 0:
|
||||
|
||||
# Calcul du prix unitaire
|
||||
prix_unitaire = float(p) / denom
|
||||
|
||||
# Arrondi à 2 décimales
|
||||
return round(prix_unitaire, 2)
|
||||
|
||||
# Si aucun prix trouvé
|
||||
raise ValueError("Impossible de trouver le prix unitaire.")
|
||||
return prix_calcule
|
||||
|
||||
def appellation(self) -> str | None:
|
||||
"""_summary_
|
||||
@@ -153,7 +120,7 @@ class _ScraperData:
|
||||
return None
|
||||
|
||||
val = cast(str, app_dict.get("value")).rstrip("+").split("-")
|
||||
if len(val) > 1:
|
||||
if len(val) > 1 and val[1] != "":
|
||||
val[0] = str((int(val[0]) + int(val[1])) / 2)
|
||||
|
||||
return val[0]
|
||||
@@ -171,6 +138,23 @@ class _ScraperData:
|
||||
def getdata(self) -> dict[str, object]:
|
||||
return self._data
|
||||
|
||||
def informations(self) -> str:
|
||||
"""
|
||||
Retourne toutes les informations sous la forme :
|
||||
"Appelation,Parker,J.Robinson,J.Suckling,Prix"
|
||||
"""
|
||||
|
||||
appellation = self.appellation()
|
||||
parker = self.parker()
|
||||
robinson = self.robinson()
|
||||
suckling = self.suckling()
|
||||
try:
|
||||
prix = self.prix()
|
||||
except ValueError:
|
||||
prix = None
|
||||
|
||||
return f"{appellation},{parker},{robinson},{suckling},{prix}"
|
||||
|
||||
|
||||
class Scraper:
|
||||
"""
|
||||
@@ -307,3 +291,67 @@ class Scraper:
|
||||
|
||||
return _ScraperData(cast(dict[str, object], current_data))
|
||||
|
||||
def _geturlproductslist(self, subdir: str):
|
||||
"""_summary_
|
||||
|
||||
Args:
|
||||
subdir (str): _description_
|
||||
|
||||
Returns:
|
||||
_type_: _description_
|
||||
"""
|
||||
try:
|
||||
data: dict[str, object] = self.getjsondata(subdir).getdata()
|
||||
|
||||
for element in ["initialReduxState", "categ", "content"]:
|
||||
data: dict[str, object] = cast(dict[str, object], data.get(element))
|
||||
if not isinstance(data, dict):
|
||||
return None
|
||||
|
||||
products: list[str] = cast(list[str], data.get("products"))
|
||||
if isinstance(products, list):
|
||||
return products
|
||||
|
||||
except (JSONDecodeError, HTTPError):
|
||||
return None
|
||||
|
||||
def getvins(self, subdir: str, filename: str):
|
||||
"""_summary_
|
||||
|
||||
Args:
|
||||
subdir (str): _description_
|
||||
filename (str): _description_
|
||||
"""
|
||||
with open(filename, "a") as f:
|
||||
cache: set[str] = set[str]()
|
||||
page = 0
|
||||
|
||||
while True:
|
||||
page += 1
|
||||
products_list = self._geturlproductslist(f"{subdir}?page={page}")
|
||||
|
||||
if not products_list:
|
||||
break
|
||||
|
||||
products_list_length = len(products_list)
|
||||
for i, product in enumerate(products_list):
|
||||
if not isinstance(product, dict):
|
||||
continue
|
||||
|
||||
link = product.get("seoKeyword")
|
||||
|
||||
if link and link not in cache:
|
||||
try:
|
||||
infos = self.getjsondata(link).informations()
|
||||
_ = f.write(infos + "\n")
|
||||
print(
|
||||
f"page: {page} | {i + 1}/{products_list_length} {link}"
|
||||
)
|
||||
cache.add(link)
|
||||
except (JSONDecodeError, HTTPError) as e:
|
||||
print(f"Erreur sur le produit {link}: {e}")
|
||||
f.flush()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
Scraper().getvins("bordeaux.html", "donnee.csv")
|
||||
|
||||
124
test_main.py
124
test_main.py
@@ -1,5 +1,5 @@
|
||||
from json import dumps
|
||||
from bs4 import Tag
|
||||
from unittest.mock import patch, mock_open
|
||||
import pytest
|
||||
from requests_mock import Mocker
|
||||
from main import Scraper
|
||||
@@ -71,10 +71,10 @@ def mock_site():
|
||||
"_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)",
|
||||
"listPrice": 842,
|
||||
"offerPrice": 842,
|
||||
"seoKeyword": "vin-de-charazade1867.html",
|
||||
"shortdesc": "Une bouteille du meilleur vin du monde?",
|
||||
"attributes": {
|
||||
"promotion_o_n": {
|
||||
"valueId": "0",
|
||||
@@ -101,9 +101,9 @@ def mock_site():
|
||||
"isSpirit": False,
|
||||
},
|
||||
"nbunit": {
|
||||
"valueId": "6",
|
||||
"valueId": "1",
|
||||
"name": "nbunit",
|
||||
"value": "6",
|
||||
"value": "1",
|
||||
"isSpirit": False,
|
||||
},
|
||||
},
|
||||
@@ -120,14 +120,14 @@ def mock_site():
|
||||
"appellation": {
|
||||
"valueId": "433",
|
||||
"name": "Appellation",
|
||||
"value": "Sforzato di Valtellina",
|
||||
"url": "sforzato-di-valtellina.html",
|
||||
"value": "Madame-Loïk",
|
||||
"url": "Madame-loik.html",
|
||||
"isSpirit": False,
|
||||
"groupIdentifier": "appellation_433",
|
||||
},
|
||||
"note_rp": {
|
||||
"valueId": "91",
|
||||
"name": "Parker",
|
||||
"name": "Peter Parker",
|
||||
"value": "91",
|
||||
"isSpirit": False,
|
||||
},
|
||||
@@ -139,7 +139,7 @@ def mock_site():
|
||||
},
|
||||
"note_js": {
|
||||
"valueId": "93-94",
|
||||
"name": "J. Suckling",
|
||||
"name": "J. cherazade",
|
||||
"value": "93-94",
|
||||
"isSpirit": False,
|
||||
},
|
||||
@@ -166,6 +166,79 @@ def mock_site():
|
||||
text=html_product,
|
||||
)
|
||||
|
||||
html_product = f"""
|
||||
<html>
|
||||
<body>
|
||||
<h1>MILLESIMA</h1>
|
||||
<script id="__NEXT_DATA__" type="application/json">
|
||||
{dumps(json_data)}
|
||||
</script>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
list_pleine = f"""
|
||||
<html>
|
||||
<body>
|
||||
<h1>LE WINE</h1>
|
||||
<script id="__NEXT_DATA__" type="application/json">
|
||||
{dumps({
|
||||
"props": {
|
||||
"pageProps": {
|
||||
"initialReduxState": {
|
||||
"categ": {
|
||||
"content": {
|
||||
"products": [
|
||||
{"seoKeyword": "/nino-negri-5-stelle-sfursat-2022.html",},
|
||||
{"seoKeyword": "/poubelle",},
|
||||
{"seoKeyword": "/",}
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
)}
|
||||
</script>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
list_vide = f"""
|
||||
<html>
|
||||
<body>
|
||||
<h1>LE WINE</h1>
|
||||
<script id="__NEXT_DATA__" type="application/json">
|
||||
{dumps({
|
||||
"props": {
|
||||
"pageProps": {
|
||||
"initialReduxState": {
|
||||
"categ": {
|
||||
"content": {
|
||||
"products": [
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
)}
|
||||
</script>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
m.get(
|
||||
"https://www.millesima.fr/wine.html",
|
||||
complete_qs=False,
|
||||
response_list=[
|
||||
{"text": list_pleine},
|
||||
{"text": list_vide},
|
||||
],
|
||||
)
|
||||
|
||||
# on return m sans fermer le server qui simule la page
|
||||
yield m
|
||||
|
||||
@@ -190,7 +263,7 @@ def test_appellation(scraper: Scraper):
|
||||
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"
|
||||
assert contenu.appellation() == "Madame-Loïk"
|
||||
|
||||
|
||||
def test_fonctionprivee(scraper: Scraper):
|
||||
@@ -207,7 +280,6 @@ def test_fonctionprivee(scraper: Scraper):
|
||||
assert contenu._getattributes() is not None
|
||||
|
||||
|
||||
|
||||
def test_critiques(scraper: Scraper):
|
||||
vide = scraper.getjsondata("")
|
||||
poubelle = scraper.getjsondata("poubelle")
|
||||
@@ -225,18 +297,30 @@ def test_critiques(scraper: Scraper):
|
||||
assert contenu.suckling() == "93.5"
|
||||
assert contenu._getcritiques("test_ts") is None
|
||||
|
||||
|
||||
def test_prix(scraper: Scraper):
|
||||
vide = scraper.getjsondata("")
|
||||
poubelle = scraper.getjsondata("poubelle")
|
||||
contenu = scraper.getjsondata("nino-negri-5-stelle-sfursat-2022.html")
|
||||
assert vide.prix() is None
|
||||
assert poubelle.prix() is None
|
||||
assert contenu.prix() == 842.0
|
||||
|
||||
# Cas vide : items == [] -> on ne peut pas calculer -> ValueError
|
||||
with pytest.raises(ValueError):
|
||||
_ = vide.prix()
|
||||
|
||||
# Cas poubelle : JSON incomplet -> _getcontent() None -> ValueError
|
||||
with pytest.raises(ValueError):
|
||||
_ = poubelle.prix()
|
||||
def test_informations(scraper: Scraper):
|
||||
contenu = scraper.getjsondata("nino-negri-5-stelle-sfursat-2022.html")
|
||||
assert contenu.informations() == "Madame-Loïk,91,17,93.5,842.0"
|
||||
vide = scraper.getjsondata("")
|
||||
poubelle = scraper.getjsondata("poubelle")
|
||||
assert vide.informations() == "None,None,None,None,None"
|
||||
assert poubelle.informations() == "None,None,None,None,None"
|
||||
|
||||
assert contenu.prix() == 65.0
|
||||
|
||||
def test_search(scraper: Scraper):
|
||||
m = mock_open()
|
||||
with patch("builtins.open", m):
|
||||
scraper.getvins("wine.html", "fake_file.csv")
|
||||
|
||||
assert m().write.called
|
||||
all_writes = "".join(call.args[0] for call in m().write.call_args_list)
|
||||
assert "Madame-Loïk,91,17,93.5,842.0" in all_writes
|
||||
|
||||
Reference in New Issue
Block a user