3 Commits

Author SHA1 Message Date
DAHMANI chahrazad
0a3561ffaa Fix comments for clarity in main.py 2026-02-07 02:25:31 +01:00
Chahrazad650
6366fcd8dd ajout fonction prix 2026-02-07 02:21:30 +01:00
Chahrazad650
5797b72cbc juste un test 2026-02-05 18:28:07 +01:00
10 changed files with 135 additions and 987 deletions

2
.gitignore vendored
View File

@@ -205,5 +205,3 @@ cython_debug/
marimo/_static/
marimo/_lsp/
__marimo__/
*.csv

View File

@@ -1,106 +0,0 @@
#!/usr/bin/env python3
from pandas import DataFrame, to_numeric
import pandas as pd
SCORE_COLS = ["Robert", "Robinson", "Suckling"]
def display_info(df: DataFrame, name: str = "DataFrame") -> None:
"""
Affiche un résumé du DataFrame
-la taille
-types des colonnes
-valeurs manquantes
-statistiques numériques
"""
print(f"\n===== {name} =====")
print(f"Shape : {df.shape[0]} lignes × {df.shape[1]} colonnes")
print("\nTypes des colonnes :")
print(df.dtypes)
print("\nValeurs manquantes :")
print(df.isna().sum())
print("\nStatistiques numériques :")
print(df.describe().round(2))
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")
def fill_missing_scores(df: DataFrame) -> DataFrame:
"""
Remplacer les notes manquantes par la moyenne
des vins de la même appellation.
"""
df_copy = df.copy()
df_copy["Appellation"] = df_copy["Appellation"].astype(str).str.strip()
for score in SCORE_COLS:
df_copy[score] = to_numeric(df_copy[score], errors="coerce")
temp_cols: list[str] = []
for score in SCORE_COLS:
mean_df = mean_score(df_copy, score)
mean_name = f"mean_{score}"
temp_cols.append(mean_name)
df_copy = df_copy.merge(mean_df, on="Appellation", how="left")
df_copy[score] = df_copy[score].fillna(df_copy[mean_name])
df_copy = df_copy.drop(columns=temp_cols)
return df_copy
def encode_appellation(df: DataFrame, column: str = "Appellation") -> DataFrame:
"""
Remplace la colonne 'Appellation' par des colonnes indicatrices
"""
df_copy = df.copy()
appellations = df_copy[column].astype(str).str.strip()
appellation_dummies = pd.get_dummies(appellations)
df_copy = df_copy.drop(columns=[column])
return df_copy.join(appellation_dummies)

150
main.py Executable file → Normal file
View File

@@ -1,64 +1,110 @@
#!/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,
fill_missing_scores,
encode_appellation)
from sys import stderr
from typing import cast, Any, Dict, Optional
from requests import Response, Session
from bs4 import BeautifulSoup, Tag
from json import JSONDecodeError, loads
def load_csv(filename: str) -> DataFrame:
path: str = normpath(join(getcwd(), filename))
return read_csv(path)
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:
self._url: str = "https://www.millesima.fr/"
self._session: Session = Session()
self._latest_request: tuple[(str, Response | None)] = ("", None)
def save_csv(df: DataFrame, out_filename: str) -> None:
df.to_csv(out_filename, index=False)
def _request(self, subdir: str) -> Response:
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 = "") -> Response:
rq_subdir, rq_response = self._latest_request
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 main() -> None:
if len(argv) != 2:
raise ValueError(f"Usage: {argv[0]} <filename.csv>")
def getsoup(self, subdir: str = "") -> BeautifulSoup:
markup: str = self.getresponse(subdir).text
return BeautifulSoup(markup, features="html.parser")
df = load_csv(argv[1])
def getjsondata(self, subdir: str = "", id: str = "__NEXT_DATA__") -> dict[str, object]:
soup: BeautifulSoup = self.getsoup(subdir)
script: Tag | None = soup.find("script", id=id)
display_info(df, "Avant le nettoyage")
if isinstance(script, Tag) and script.string:
try:
current_data: object = loads(script.string)
keys: list[str] = ["props", "pageProps", "initialReduxState", "product", "content"]
for key in keys:
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}")
df = drop_empty_appellation(df)
save_csv(df, "donnee_clean.csv")
display_info(df, "Après nettoyage d'appellations manquantes")
#la moyenne des notes des vins pour chaque appellation
robert_means = mean_robert(df)
save_csv(robert_means, "mean_robert_by_appellation.csv")
display_info(robert_means, "Moyennes Robert par appellation")
if isinstance(current_data, dict):
return cast(dict[str, object], current_data)
robinson_means = mean_robinson(df)
save_csv(robinson_means, "mean_robinson_by_appellation.csv")
display_info(robinson_means, "Moyennes Robinson par appellation")
suckling_means = mean_suckling(df)
save_csv(suckling_means, "mean_suckling_by_appellation.csv")
display_info(suckling_means, "Moyennes Suckling par appellation")
df_missing_scores = fill_missing_scores(df)
save_csv(df_missing_scores, "donnee_filled.csv")
display_info(df_missing_scores, "Après remplissage des notes manquantes par la moyenne de l'appellation")
df_ready = encode_appellation(df_missing_scores)
save_csv(df_ready, "donnee_ready.csv")
display_info(df_ready, "Après remplacer la colonne 'Appellation' par des colonnes indicatrices")
except (JSONDecodeError, ValueError) as e:
print(f"Erreur lors de l'extraction JSON : {e}", file=stderr)
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)
if __name__ == "__main__":
try:
main()
except Exception as e:
print(f"ERREUR: {e}")
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.")

Binary file not shown.

View File

@@ -1,6 +1,2 @@
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
requests>=2.32.5
beautifulsoup4>=4.14.3

BIN
res/projet1.pdf Normal file

Binary file not shown.

View File

@@ -1,431 +0,0 @@
#!/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}")

View File

@@ -1,64 +0,0 @@
import pandas as pd
import pytest
from pandas import DataFrame
from cleaning import (
SCORE_COLS,
drop_empty_appellation,
mean_score,
fill_missing_scores,
encode_appellation,
)
@pytest.fixture
def df_raw() -> DataFrame:
return pd.DataFrame({
"Appellation": ["Pauillac", "Pauillac ", "Margaux", None, "Pomerol", "Pomerol"],
"Robert": ["95", None, "bad", 90, None, None],
"Robinson": [None, "93", 18, None, None, None],
"Suckling": [96, None, None, None, 91, None],
"Prix": ["10.0", "11.0", "20.0", "30.0", "40.0", "50.0"],
})
def test_drop_empty_appellation(df_raw: DataFrame):
out = drop_empty_appellation(df_raw)
assert out["Appellation"].isna().sum() == 0
assert len(out) == 5
def test_mean_score_zero_when_no_scores(df_raw: DataFrame):
out = drop_empty_appellation(df_raw)
m = mean_score(out, "Robert")
assert list(m.columns) == ["Appellation", "mean_Robert"]
# Pomerol n'a aucune note Robert => moyenne doit être 0
pomerol_mean = m.loc[m["Appellation"].str.strip() == "Pomerol", "mean_Robert"].iloc[0]
assert pomerol_mean == 0
def test_fill_missing_scores(df_raw: DataFrame):
out = drop_empty_appellation(df_raw)
filled = fill_missing_scores(out)
# plus de NaN dans les colonnes de scores
for col in SCORE_COLS:
assert filled[col].isna().sum() == 0
assert filled.loc[1, "Robert"] == 95.0
# pas de colonnes temporaires mean_*
for col in SCORE_COLS:
assert f"mean_{col}" not in filled.columns
def test_encode_appellation(df_raw: DataFrame):
out = drop_empty_appellation(df_raw)
filled = fill_missing_scores(out)
encoded = encode_appellation(filled)
# la colonne texte disparaît
assert "Appellation" not in encoded.columns
assert "Pauillac" in encoded.columns
assert encoded.loc[0, "Pauillac"] == 1

View File

@@ -0,0 +1,35 @@
import json
from main import Scraper
def test_json():
scraper = Scraper()
data = scraper.getjsondata("/chateau-gloria-2016.html")
print("JSON récupéré :")
print(json.dumps(data, indent=4, ensure_ascii=False))
assert isinstance(data, dict)
assert "items" in data
def test_prix():
scraper = 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)")
if __name__ == "__main__":
test_json()
test_prix()
print("\nTous les tests terminés")

View File

@@ -1,326 +0,0 @@
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