mirror of
https://github.com/guezoloic/millesima_projetS6.git
synced 2026-03-28 19:13:42 +00:00
Compare commits
39 Commits
exo1
...
exo9_Chahr
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
5afb6e38fe | ||
|
|
f31de22693 | ||
|
|
73c6221080 | ||
|
|
99dd71989d | ||
| d62145e250 | |||
| 829c303e78 | |||
| b584f9a301 | |||
| 547c7ec4c1 | |||
| 0aa765d6a0 | |||
| 8a357abe86 | |||
|
|
2f5af5aabf | ||
| a33b484dea | |||
| dd430b9861 | |||
| 011bb6a689 | |||
| 96dbaaaaf6 | |||
| ed86e588f7 | |||
|
|
0182bbbf20 | ||
|
|
cd1e266f25 | ||
|
|
2aa99453a0 | ||
| 9f1ff1ef7b | |||
|
|
bfc39db652 | ||
|
|
717fce6ca4 | ||
| 9914e8af41 | |||
|
|
5785d571b2 | ||
| ae66e94d6c | |||
| 2bc5d57a31 | |||
| 8f21e48b28 | |||
|
|
9e0cb9737e | ||
| c62a5e6a76 | |||
|
|
8cae082344 | ||
| 9da0159869 | |||
| 5c22777c2d | |||
| 0d78b1aec3 | |||
| 168ccf88dc | |||
| 6992a0ca16 | |||
| 74482af7f0 | |||
| 76017e3ea3 | |||
|
|
683582a253 | ||
|
|
e2317de748 |
2
.gitignore
vendored
2
.gitignore
vendored
@@ -205,3 +205,5 @@ cython_debug/
|
||||
marimo/_static/
|
||||
marimo/_lsp/
|
||||
__marimo__/
|
||||
|
||||
*.csv
|
||||
48
cleaning.py
Normal file
48
cleaning.py
Normal file
@@ -0,0 +1,48 @@
|
||||
#!/usr/bin/env python3
|
||||
from pandas import DataFrame, to_numeric
|
||||
|
||||
|
||||
def display_info(df: DataFrame) -> None:
|
||||
print(df.all())
|
||||
print(df.info())
|
||||
print("\nNombre de valeurs manquantes par colonne :")
|
||||
print(df.isna().sum())
|
||||
|
||||
|
||||
def drop_empty_appellation(df: DataFrame) -> DataFrame:
|
||||
|
||||
return df.dropna(subset=["Appellation"])
|
||||
|
||||
|
||||
def mean_score(df: DataFrame, col: str) -> DataFrame:
|
||||
"""
|
||||
Calcule la moyenne d'une colonne de score par appellation.
|
||||
- Convertit les valeurs en numériques, en remplaçant les non-convertibles par NaN
|
||||
- Calcule la moyenne par appellation
|
||||
- Remplace les NaN résultants par 0
|
||||
|
||||
"""
|
||||
tmp = df[["Appellation", col]].copy()
|
||||
|
||||
tmp[col] = to_numeric(tmp[col], errors="coerce")
|
||||
|
||||
# moyenne par appellation
|
||||
means = tmp.groupby("Appellation", as_index=False)[col].mean()
|
||||
|
||||
means[col] = means[col].fillna(0)
|
||||
|
||||
means = means.rename(columns={col: f"mean_{col}"})
|
||||
|
||||
return means
|
||||
|
||||
|
||||
def mean_robert(df: DataFrame) -> DataFrame:
|
||||
return mean_score(df, "Robert")
|
||||
|
||||
|
||||
def mean_robinson(df: DataFrame) -> DataFrame:
|
||||
return mean_score(df, "Robinson")
|
||||
|
||||
|
||||
def mean_suckling(df: DataFrame) -> DataFrame:
|
||||
return mean_score(df, "Suckling")
|
||||
166
main.py
Normal file → Executable file
166
main.py
Normal file → Executable file
@@ -1,132 +1,60 @@
|
||||
from sys import stderr
|
||||
from typing import cast
|
||||
from requests import Response, Session
|
||||
from bs4 import BeautifulSoup, Tag
|
||||
from json import JSONDecodeError, loads
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from os import getcwd
|
||||
from os.path import normpath, join
|
||||
from sys import argv
|
||||
from pandas import read_csv, DataFrame
|
||||
|
||||
from cleaning import (display_info,
|
||||
drop_empty_appellation,
|
||||
mean_robert,
|
||||
mean_robinson,
|
||||
mean_suckling)
|
||||
|
||||
|
||||
class Scraper:
|
||||
"""
|
||||
Scraper est une classe qui permet de gerer
|
||||
de façon dynamique des requetes uniquement
|
||||
sur le serveur https de Millesima
|
||||
"""
|
||||
def load_csv(filename: str) -> DataFrame:
|
||||
path: str = normpath(join(getcwd(), filename))
|
||||
return read_csv(path)
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""
|
||||
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: Session = Session()
|
||||
# Système de cache pour éviter de solliciter le serveur inutilement
|
||||
self._latest_request: tuple[(str, Response | None)] = ("", None)
|
||||
|
||||
def _request(self, subdir: str) -> Response:
|
||||
"""
|
||||
Effectue une requête GET sur le serveur Millesima.
|
||||
def save_csv(df: DataFrame, out_filename: str) -> None:
|
||||
df.to_csv(out_filename, index=False)
|
||||
|
||||
Args:
|
||||
subdir (str): Le sous-répertoire ou chemin de l'URL (ex: "/vins").
|
||||
|
||||
Returns:
|
||||
Response: L'objet réponse de la requête.
|
||||
def main() -> None:
|
||||
if len(argv) != 2:
|
||||
raise ValueError(f"Usage: {argv[0]} <filename.csv>")
|
||||
|
||||
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
|
||||
df = load_csv(argv[1])
|
||||
|
||||
def getresponse(self, subdir: str = "") -> Response:
|
||||
"""
|
||||
Récupère la réponse d'une page, en utilisant le cache si possible.
|
||||
print("=== Avant nettoyage ===")
|
||||
display_info(df)
|
||||
|
||||
Args:
|
||||
subdir (str, optional): Le chemin de la page.
|
||||
df = drop_empty_appellation(df)
|
||||
save_csv(df, "donnee_clean.csv")
|
||||
|
||||
Returns:
|
||||
Response: L'objet réponse (cache ou nouvelle requête).
|
||||
print("\n=== Après nettoyage d'appellations manquantes ===")
|
||||
display_info(df)
|
||||
|
||||
#la moyenne des notes des vins pour chaque appellation
|
||||
robert_means = mean_robert(df)
|
||||
save_csv(robert_means, "mean_robert_by_appellation.csv")
|
||||
print("\n=== moyenne Robert par appellation ===")
|
||||
print(robert_means.head(10))
|
||||
|
||||
Raise:
|
||||
HTTPError: Si le serveur renvoie un code d'erreur (4xx, 5xx).
|
||||
"""
|
||||
rq_subdir, rq_response = self._latest_request
|
||||
robinson_means = mean_robinson(df)
|
||||
save_csv(robinson_means, "mean_robinson_by_appellation.csv")
|
||||
print("\n===: moyennes Robinson par appellation ===")
|
||||
print(robinson_means.head(10))
|
||||
|
||||
suckling_means = mean_suckling(df)
|
||||
save_csv(suckling_means, "mean_suckling_by_appellation.csv")
|
||||
print("\n===: moyennes Suckling par appellation ===")
|
||||
print(suckling_means.head(10))
|
||||
|
||||
if rq_response is None or subdir != rq_subdir:
|
||||
request: Response = self._request(subdir)
|
||||
self._latest_request = (subdir, request)
|
||||
return request
|
||||
|
||||
return rq_response
|
||||
|
||||
def getsoup(self, subdir: str = "") -> BeautifulSoup:
|
||||
"""
|
||||
Récupère le contenu HTML d'une page et le transforme en objet BeautifulSoup.
|
||||
|
||||
Args:
|
||||
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).
|
||||
"""
|
||||
markup: str = self.getresponse(subdir).text
|
||||
return BeautifulSoup(markup, features="html.parser")
|
||||
|
||||
def getjsondata(
|
||||
self, subdir: str = "", id: str = "__NEXT_DATA__"
|
||||
) -> dict[str, object]:
|
||||
"""
|
||||
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)
|
||||
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 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 {}
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
main()
|
||||
except Exception as e:
|
||||
print(f"ERREUR: {e}")
|
||||
@@ -1,4 +1,6 @@
|
||||
requests>=2.32.5
|
||||
requests-mock>=1.12.1
|
||||
beautifulsoup4>=4.14.3
|
||||
|
||||
requests==2.32.5
|
||||
requests-mock==1.12.1
|
||||
beautifulsoup4==4.14.3
|
||||
pytest==8.4.2
|
||||
requests-mock==1.12.1
|
||||
pandas==2.3.3
|
||||
Binary file not shown.
431
scraper.py
Executable file
431
scraper.py
Executable file
@@ -0,0 +1,431 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from sys import argv
|
||||
from typing import cast
|
||||
from requests import HTTPError, Response, Session
|
||||
from requests.exceptions import Timeout, ConnectionError
|
||||
import time
|
||||
from bs4 import BeautifulSoup, Tag
|
||||
from collections import OrderedDict
|
||||
from json import JSONDecodeError, loads
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
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:
|
||||
"""_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 prix(self) -> float | None:
|
||||
"""
|
||||
Retourne le prix unitaire d'une bouteille (75cl).
|
||||
|
||||
Si aucun prix n'est disponible, retourne None.
|
||||
"""
|
||||
|
||||
content = self._getcontent()
|
||||
if content is None:
|
||||
return None
|
||||
|
||||
items = content.get("items")
|
||||
|
||||
# Vérifie que items existe et n'est pas vide
|
||||
if not isinstance(items, list) or len(items) == 0:
|
||||
return None
|
||||
|
||||
prix_calcule: float | None = None
|
||||
|
||||
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 not isinstance(p, (int, float)) or not nbunit or not equivbtl:
|
||||
continue
|
||||
|
||||
nb = float(nbunit)
|
||||
eq = float(equivbtl)
|
||||
|
||||
if nb <= 0 or eq <= 0:
|
||||
continue
|
||||
|
||||
if nb == 1 and eq == 1:
|
||||
return float(p)
|
||||
|
||||
prix_calcule = round(float(p) / (nb * eq), 2)
|
||||
|
||||
return prix_calcule
|
||||
|
||||
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 and val[1] != "":
|
||||
val[0] = str(round((float(val[0]) + float(val[1])) / 2, 1))
|
||||
|
||||
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
|
||||
|
||||
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()
|
||||
prix = self.prix()
|
||||
|
||||
return f"{appellation},{parker},{robinson},{suckling},{prix}"
|
||||
|
||||
|
||||
class Scraper:
|
||||
"""
|
||||
Scraper est une classe qui permet de gerer
|
||||
de façon dynamique des requetes uniquement
|
||||
sur le serveur https de Millesima
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""
|
||||
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: Session = Session()
|
||||
self._session.headers.update({
|
||||
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
||||
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
||||
"Chrome/122.0.0.0 Safari/537.36",
|
||||
"Accept-Language": "fr-FR,fr;q=0.9,en;q=0.8",
|
||||
})
|
||||
# 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) -> 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("/")
|
||||
|
||||
last_exc: Exception | None = None
|
||||
for attempt in range(1, 4):
|
||||
try:
|
||||
response: Response = self._session.get(url=target_url, timeout=30)
|
||||
response.raise_for_status()
|
||||
return response
|
||||
except (Timeout, ConnectionError) as e:
|
||||
last_exc = e
|
||||
print(f"Timeout/ConnectionError ({attempt}/3) sur {target_url}: {e}")
|
||||
time.sleep(2 * attempt) # 2s, 4s, 6s
|
||||
|
||||
# après 3 essais, on abandonne
|
||||
raise last_exc if last_exc else RuntimeError("Request failed")
|
||||
|
||||
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).
|
||||
"""
|
||||
|
||||
# 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
|
||||
|
||||
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.
|
||||
|
||||
Args:
|
||||
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).
|
||||
"""
|
||||
|
||||
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): 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)
|
||||
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))
|
||||
|
||||
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 _save_progress(self, page: int, i: int, last_link: str) -> None:
|
||||
Path("progress.txt").write_text(f"{page},{i},{last_link}", encoding="utf-8")
|
||||
|
||||
|
||||
def _load_progress(self) -> tuple[int, int, str | None]:
|
||||
p = Path("progress.txt")
|
||||
if not p.exists():
|
||||
return (1, 0, None)
|
||||
|
||||
try:
|
||||
parts = p.read_text(encoding="utf-8").strip().split(",", 2)
|
||||
|
||||
page = int(parts[0])
|
||||
i = int(parts[1])
|
||||
|
||||
last_link = parts[2] if len(parts) == 3 and parts[2] != "" else None
|
||||
return (page, i, last_link)
|
||||
|
||||
except Exception:
|
||||
return (1, 0, None)
|
||||
|
||||
def getvins(self, subdir: str, filename: str):
|
||||
"""_summary_
|
||||
|
||||
Args:
|
||||
subdir (str): _description_
|
||||
filename (str): _description_
|
||||
"""
|
||||
start_page, start_i, last_link = self._load_progress()
|
||||
print(f"__INFO__ Reprise à page={start_page}, index={start_i}, last_link={last_link}")
|
||||
|
||||
with open(filename, "a", encoding="utf-8") as f:
|
||||
cache: set[str] = set[str]()
|
||||
|
||||
if f.tell() == 0:
|
||||
_ = f.write("Appellation,Robert,Robinson,Suckling,Prix\n")
|
||||
|
||||
page = start_page - 1
|
||||
|
||||
while True:
|
||||
page += 1
|
||||
products_list = self._geturlproductslist(f"{subdir}?page={page}")
|
||||
|
||||
if not products_list:
|
||||
break
|
||||
|
||||
products_list_length = len(products_list)
|
||||
start_at = start_i if page == start_page else 0
|
||||
|
||||
for i in range(start_at, products_list_length):
|
||||
product = products_list[i]
|
||||
if not isinstance(product, dict):
|
||||
continue
|
||||
|
||||
link = product.get("seoKeyword")
|
||||
if not link:
|
||||
continue
|
||||
|
||||
# pour eviter les doublons :
|
||||
if (page == start_page) and (last_link is not None) and (link == last_link):
|
||||
self._save_progress(page, + 1, link)
|
||||
continue
|
||||
|
||||
self._save_progress(page, i + 1, link)
|
||||
|
||||
if link in cache:
|
||||
continue
|
||||
|
||||
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()
|
||||
|
||||
Path("progress.txt").unlink(missing_ok=True)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
if len(argv) != 2:
|
||||
raise ValueError(f"{argv[0]} <sous-url>")
|
||||
scraper: Scraper = Scraper()
|
||||
scraper.getvins(argv[1], "donnee.csv")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
main()
|
||||
except Exception as e:
|
||||
print(f"ERREUR: {e}")
|
||||
107
test_main.py
107
test_main.py
@@ -1,107 +0,0 @@
|
||||
from json import dumps
|
||||
from bs4 import Tag
|
||||
import pytest
|
||||
from requests_mock import Mocker
|
||||
from main import Scraper
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def mock_site():
|
||||
with Mocker() as m:
|
||||
m.get(
|
||||
"https://www.millesima.fr/",
|
||||
text="<html><body><h1>MILLESIMA</h1></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",
|
||||
"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",
|
||||
"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,
|
||||
}
|
||||
],
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
html_product = f"""
|
||||
<html>
|
||||
<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):
|
||||
h1: Tag | None = scraper.getsoup().find("h1")
|
||||
|
||||
assert isinstance(h1, Tag)
|
||||
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 len(jsondata["items"]) > 0
|
||||
assert jsondata["items"][0]["offerPrice"] == 390
|
||||
326
test_scraper.py
Normal file
326
test_scraper.py
Normal file
@@ -0,0 +1,326 @@
|
||||
from json import dumps
|
||||
from unittest.mock import patch, mock_open
|
||||
import pytest
|
||||
from requests_mock import Mocker
|
||||
from scraper import Scraper
|
||||
|
||||
|
||||
@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": 842,
|
||||
"offerPrice": 842,
|
||||
"seoKeyword": "vin-de-charazade1867.html",
|
||||
"shortdesc": "Une bouteille du meilleur vin du monde?",
|
||||
"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,
|
||||
},
|
||||
"equivbtl": {
|
||||
"valueId": "1",
|
||||
"name": "equivbtl",
|
||||
"value": "1",
|
||||
"isSpirit": False,
|
||||
},
|
||||
"nbunit": {
|
||||
"valueId": "1",
|
||||
"name": "nbunit",
|
||||
"value": "1",
|
||||
"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": "Madame-Loïk",
|
||||
"url": "Madame-loik.html",
|
||||
"isSpirit": False,
|
||||
"groupIdentifier": "appellation_433",
|
||||
},
|
||||
"note_rp": {
|
||||
"valueId": "91",
|
||||
"name": "Peter Parker",
|
||||
"value": "91",
|
||||
"isSpirit": False,
|
||||
},
|
||||
"note_jr": {
|
||||
"valueId": "17+",
|
||||
"name": "J. Robinson",
|
||||
"value": "17+",
|
||||
"isSpirit": False,
|
||||
},
|
||||
"note_js": {
|
||||
"valueId": "93-94.5",
|
||||
"name": "J. cherazade",
|
||||
"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,
|
||||
)
|
||||
|
||||
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
|
||||
|
||||
|
||||
@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() == "Madame-Loïk"
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
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"
|
||||
|
||||
|
||||
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