3 Commits

12 changed files with 268 additions and 675 deletions

View File

@@ -5,41 +5,35 @@ name: Python application
on: on:
push: push:
branches: ["main"] branches: [ "main" ]
pull_request: pull_request:
branches: ["main"] branches: [ "main" ]
permissions: permissions:
contents: write contents: read
jobs: jobs:
build: build:
runs-on: ubuntu-latest runs-on: ubuntu-latest
steps: steps:
- uses: actions/checkout@v4 - uses: actions/checkout@v4
- name: Set up Python 3.10
- name: Set up Python 3.10 uses: actions/setup-python@v3
uses: actions/setup-python@v4 with:
with: python-version: "3.10"
python-version: "3.10" - name: Install dependencies
run: |
- name: install dependencies python -m pip install --upgrade pip
run: | pip install flake8 pytest
python -m pip install --upgrade pip if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
pip install ".[test,doc]" - name: Lint with flake8
run: |
- name: Lint with flake8 # stop the build if there are Python syntax errors or undefined names
run: | flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
- name: Test with pytest
- name: Test with pytest run: |
run: pytest pytest
- name: Deploy Doc
if: github.event_name == 'push' && github.ref == 'refs/heads/main'
run: |
git config user.name github-actions
git config user.email github-actions@github.com
mkdocs gh-deploy --force

2
.gitignore vendored
View File

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

View File

@@ -1 +0,0 @@
# Millesima

View File

@@ -1,3 +0,0 @@
# Scraper
::: scraper.Scraper

View File

@@ -1,4 +0,0 @@
# _ScraperData
::: scraper._ScraperData

224
main.py Normal file
View File

@@ -0,0 +1,224 @@
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], scraper: 'Scraper | None' = None) -> None:
self._data: dict[str, object] = data
self._scraper: Scraper | None = scraper
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 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()
# 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("/")
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).
"""
# 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))

View File

@@ -1,14 +0,0 @@
site_name: "Projet Millesima S6"
theme:
name: "material"
plugins:
- search
- mkdocstrings
markdown_extensions:
- admonition
- pymdownx.details
- pymdownx.superfences
- pymdownx.tabbed

View File

@@ -1,12 +0,0 @@
[project]
name = "projet-millesima-s6"
version = "0.1.0"
dependencies = ["requests==2.32.5", "beautifulsoup4==4.14.3", "pandas==2.3.3", "tqdm==4.67.3"]
[project.optional-dependencies]
test = ["pytest==8.4.2", "requests-mock==1.12.1", "flake8==7.3.0"]
doc = ["mkdocs<2.0.0", "mkdocs-material==9.6.23", "mkdocstrings[python]"]
[build-system]
requires = ["setuptools", "wheel"]
build-backend = "setuptools.build_meta"

4
requirements.txt Normal file
View File

@@ -0,0 +1,4 @@
requests>=2.32.5
requests-mock>=1.12.1
beautifulsoup4>=4.14.3

Binary file not shown.

View File

@@ -1,484 +0,0 @@
#!/usr/bin/env python3
from collections import OrderedDict
from io import SEEK_END, SEEK_SET, BufferedWriter
from json import JSONDecodeError, loads
from os import makedirs
from os.path import dirname, exists, join, normpath, realpath
from pickle import UnpicklingError, dump, load
from sys import argv
from tqdm.std import tqdm
from typing import Any, Callable, Literal, TypeVar, cast
from bs4 import BeautifulSoup, Tag
from requests import HTTPError, Response, Session
_dir: str = dirname(realpath(__name__))
T = TypeVar("T")
def _getcache(mode: Literal["rb", "wb"], fn: Callable[[Any], T]) -> T | None:
"""_summary_
Returns:
_type_: _description_
"""
cache_dirname = normpath(join(_dir, ".cache"))
save_path = normpath(join(cache_dirname, "save"))
if not exists(cache_dirname):
makedirs(cache_dirname)
try:
with open(save_path, mode) as f:
return fn(f)
except (FileNotFoundError, EOFError, UnpicklingError):
return None
def savestate(data: tuple[int, set[str]]) -> None:
def save(f: BufferedWriter) -> None:
_ = f.seek(0)
_ = f.truncate()
dump(data, f)
f.flush()
_getcache("wb", save)
def loadstate() -> tuple[int, set[str]] | None:
return _getcache("rb", lambda f: load(f))
class _ScraperData:
"""
Conteneur de données spécialisé pour extraire les informations des dictionnaires JSON.
Cette classe agit comme une interface simplifiée au-dessus du dictionnaire brut
renvoyé par la balise __NEXT_DATA__ du site Millesima.
"""
def __init__(self, data: dict[str, object]) -> None:
"""
Initialise le conteneur avec un dictionnaire de données.
Args:
data (dict[str, object]): Le dictionnaire JSON brut extrait de la page.
"""
self._data: dict[str, object] = data
def _getcontent(self) -> dict[str, object] | None:
"""
Navigue dans l'arborescence Redux pour atteindre le contenu du produit.
Returns:
dict[str, object] | None: Le dictionnaire du produit ou None si la structure diffère.
"""
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:
"""
Extrait les attributs techniques (notes, appellations, etc.) du produit.
Returns:
dict[str, object] | None: Les attributs du vin ou None.
"""
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:
"""
Calcule le prix unitaire d'une bouteille (standardisée à 75cl).
Le site vend souvent par caisses (6, 12 bouteilles) ou formats (Magnum).
Cette méthode normalise le prix pour obtenir celui d'une seule unité.
Returns:
float | None: Le prix calculé arrondi à 2 décimales, ou 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:
"""
Extrait le nom de l'appellation du vin.
Returns:
str | None: Le nom (ex: 'Pauillac') ou None.
"""
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:
"""
Méthode générique pour parser les notes des critiques (Parker, Suckling, etc.).
Gère les notes simples ("95") et les plages de notes ("95-97") en faisant la moyenne.
Args:
name (str): La clé de l'attribut dans le JSON (ex: 'note_rp').
Returns:
str | None: La note formatée en chaîne de caractères ou None.
"""
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:
"""Note Robert Parker."""
return self._getcritiques("note_rp")
def robinson(self) -> str | None:
"""Note Jancis Robinson."""
return self._getcritiques("note_jr")
def suckling(self) -> str | None:
"""Note James Suckling."""
return self._getcritiques("note_js")
def getdata(self) -> dict[str, object]:
"""Retourne le dictionnaire de données complet."""
return self._data
def informations(self) -> str:
"""
Agrège les données clés pour l'export CSV.
Returns:
str: Ligne formatée : "Appellation,Parker,Robinson,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:
"""
Client HTTP optimisé pour le scraping de millesima.fr.
Gère la session persistante, les headers de navigation et un cache double
pour optimiser les performances et la discrétion.
"""
def __init__(self) -> None:
"""
Initialise l'infrastructure de navigation:
- créer une session pour éviter de faire un handshake pour chaque requête
- ajout d'un header pour éviter le blocage de l'accès au site
- ajout d'un système de cache
"""
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()
# Crée une "fausse carte d'identité" pour éviter que le site nous
# bloque car on serait des robots
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
# utilise pour _request
self._latest_request: tuple[(str, Response)] | None = None
# utilise pour getsoup
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.
Raises:
HTTPError: Si le serveur renvoie un code d'erreur (4xx, 5xx).
"""
target_url: str = self._url + subdir.lstrip("/")
# envoyer une requête GET sur la page si erreur, renvoie un raise
response: Response = self._session.get(url=target_url, timeout=30)
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).
Raises:
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.
Raises:
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.
Args:
subdir (str): Le chemin de la page.
id (str, optional): L'identifiant de la balise script.
Raises:
HTTPError: Erreur renvoyée par le serveur (4xx, 5xx).
JSONDecodeError: Si le contenu de la balise n'est pas un JSON valide.
ValueError: Si les clés 'props' ou 'pageProps' sont absentes.
Returns:
_ScraperData: Instance contenant les données extraites.
"""
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) -> list[dict[str, Any]] | None:
"""
Récupère la liste des produits d'une page de catégorie.
"""
try:
data: dict[str, object] = self.getjsondata(subdir).getdata()
for element in ["initialReduxState", "categ", "content"]:
data = cast(dict[str, object], data.get(element))
products: list[dict[str, Any]] = cast(
list[dict[str, Any]], data.get("products")
)
if isinstance(products, list):
return products
except (JSONDecodeError, HTTPError):
return None
def _writevins(self, cache: set[str], product: dict[str, Any], f: Any) -> None:
"""_summary_
Args:
cache (set[str]): _description_
product (dict): _description_
f (Any): _description_
"""
if isinstance(product, dict):
link: Any | None = product.get("seoKeyword")
if link and link not in cache:
try:
infos = self.getjsondata(link).informations()
_ = f.write(infos + "\n")
cache.add(link)
except (JSONDecodeError, HTTPError) as e:
print(f"Erreur sur le produit {link}: {e}")
def getvins(self, subdir: str, filename: str, reset: bool = False) -> None:
"""
Scrape toutes les pages d'une catégorie et sauvegarde en CSV.
Args:
subdir (str): La catégorie (ex: '/vins-rouges').
filename (str): Nom du fichier de sortie (ex: 'vins.csv').
reset (bool): (Optionnel) pour réinitialiser le processus.
"""
# mode d'écriture fichier
mode: Literal["w", "a+"] = "w" if reset else "a+"
# titre
title: str = "Appellation,Robert,Robinson,Suckling,Prix\n"
# page du début
page: int = 1
# le set qui sert de cache
cache: set[str] = set[str]()
custom_format = "{l_bar} {bar:20} {r_bar}"
if not reset:
# appelle la fonction pour load le cache, si il existe
# pas, il utilise les variables de base sinon il override
# toute les variables pour continuer et pas recommencer le
# processus en entier.
serializable: tuple[int, set[str]] | None = loadstate()
if isinstance(serializable, tuple):
page, cache = serializable
try:
with open(filename, mode) as f:
# check si le titre est bien présent au début du buffer
# sinon il l'ecrit, petit bug potentiel, a+ ecrit tout le
# temps a la fin du buffer, si on a ecrit des choses avant
# le titre sera apres ces données mais on part du principe
# que personne va toucher le fichier.
_ = f.seek(0, SEEK_SET)
if not (f.read(len(title)) == title):
_ = f.write(title)
else:
_ = f.seek(0, SEEK_END)
while True:
products_list: list[dict[str, Any]] | None = (
self._geturlproductslist(f"{subdir}?page={page}")
)
if not products_list:
break
pbar: tqdm[dict[str, Any]] = tqdm(
products_list, bar_format=custom_format
)
for product in pbar:
keyword = product.get("seoKeyword", "Inconnu")[:40]
pbar.set_description(
f"Page: {page:<3} | Product: {keyword:<40}"
)
self._writevins(cache, product, f)
page += 1
except Exception:
if not reset:
savestate((page, cache))
def main() -> None:
if len(argv) != 3:
raise ValueError(f"{argv[0]} <filename> <sous-url>")
scraper: Scraper = Scraper()
scraper.getvins(argv[2], argv[1])
if __name__ == "__main__":
try:
main()
except Exception as e:
print(f"ERREUR: {e}")

View File

@@ -1,8 +1,8 @@
from json import dumps from json import dumps
from unittest.mock import patch, mock_open from bs4 import Tag
import pytest import pytest
from requests_mock import Mocker from requests_mock import Mocker
from scraper import Scraper from main import Scraper
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
@@ -71,10 +71,10 @@ def mock_site():
"_id": "J4131/22/C/CC/6-11652", "_id": "J4131/22/C/CC/6-11652",
"partnumber": "J4131/22/C/CC/6", "partnumber": "J4131/22/C/CC/6",
"taxRate": "H", "taxRate": "H",
"listPrice": 842, "listPrice": 390,
"offerPrice": 842, "offerPrice": 390,
"seoKeyword": "vin-de-charazade1867.html", "seoKeyword": "nino-negri-5-stelle-sfursat-2022-c-cc-6.html",
"shortdesc": "Une bouteille du meilleur vin du monde?", "shortdesc": "Un carton de 6 Bouteilles (75cl)",
"attributes": { "attributes": {
"promotion_o_n": { "promotion_o_n": {
"valueId": "0", "valueId": "0",
@@ -94,18 +94,6 @@ def mock_site():
"type": "CHECKBOX", "type": "CHECKBOX",
"isSpirit": False, "isSpirit": False,
}, },
"equivbtl": {
"valueId": "1",
"name": "equivbtl",
"value": "1",
"isSpirit": False,
},
"nbunit": {
"valueId": "1",
"name": "nbunit",
"value": "1",
"isSpirit": False,
},
}, },
"stock": 12, "stock": 12,
"availability": "2026-02-05", "availability": "2026-02-05",
@@ -120,14 +108,14 @@ def mock_site():
"appellation": { "appellation": {
"valueId": "433", "valueId": "433",
"name": "Appellation", "name": "Appellation",
"value": "Madame-Loïk", "value": "Sforzato di Valtellina",
"url": "Madame-loik.html", "url": "sforzato-di-valtellina.html",
"isSpirit": False, "isSpirit": False,
"groupIdentifier": "appellation_433", "groupIdentifier": "appellation_433",
}, },
"note_rp": { "note_rp": {
"valueId": "91", "valueId": "91",
"name": "Peter Parker", "name": "Parker",
"value": "91", "value": "91",
"isSpirit": False, "isSpirit": False,
}, },
@@ -138,8 +126,8 @@ def mock_site():
"isSpirit": False, "isSpirit": False,
}, },
"note_js": { "note_js": {
"valueId": "93-94.5", "valueId": "93-94",
"name": "J. cherazade", "name": "J. Suckling",
"value": "93-94", "value": "93-94",
"isSpirit": False, "isSpirit": False,
}, },
@@ -153,7 +141,7 @@ def mock_site():
html_product = f""" html_product = f"""
<html> <html>
<body> <body>
<h1>MILLESIMA</h1> <h1>MILLESIMA</h1>
<script id="__NEXT_DATA__" type="application/json"> <script id="__NEXT_DATA__" type="application/json">
{dumps(json_data)} {dumps(json_data)}
@@ -166,79 +154,6 @@ def mock_site():
text=html_product, 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 # on return m sans fermer le server qui simule la page
yield m yield m
@@ -248,6 +163,7 @@ def scraper() -> Scraper:
return Scraper() return Scraper()
# EXO1
def test_soup(scraper: Scraper): def test_soup(scraper: Scraper):
vide = scraper.getsoup("") vide = scraper.getsoup("")
poubelle = scraper.getsoup("poubelle") poubelle = scraper.getsoup("poubelle")
@@ -257,15 +173,17 @@ def test_soup(scraper: Scraper):
assert str(contenu.find("h1")) == "<h1>MILLESIMA</h1>" assert str(contenu.find("h1")) == "<h1>MILLESIMA</h1>"
# EXO3
def test_appellation(scraper: Scraper): def test_appellation(scraper: Scraper):
vide = scraper.getjsondata("") vide = scraper.getjsondata("")
poubelle = scraper.getjsondata("poubelle") poubelle = scraper.getjsondata("poubelle")
contenu = scraper.getjsondata("nino-negri-5-stelle-sfursat-2022.html") contenu = scraper.getjsondata("nino-negri-5-stelle-sfursat-2022.html")
assert vide.appellation() is None assert vide.appellation() is None
assert poubelle.appellation() is None assert poubelle.appellation() is None
assert contenu.appellation() == "Madame-Loïk" assert contenu.appellation() == "Sforzato di Valtellina"
# test fonctions privée
def test_fonctionprivee(scraper: Scraper): def test_fonctionprivee(scraper: Scraper):
vide = scraper.getjsondata("") vide = scraper.getjsondata("")
poubelle = scraper.getjsondata("poubelle") poubelle = scraper.getjsondata("poubelle")
@@ -280,6 +198,7 @@ def test_fonctionprivee(scraper: Scraper):
assert contenu._getattributes() is not None assert contenu._getattributes() is not None
# EXO4-5
def test_critiques(scraper: Scraper): def test_critiques(scraper: Scraper):
vide = scraper.getjsondata("") vide = scraper.getjsondata("")
poubelle = scraper.getjsondata("poubelle") poubelle = scraper.getjsondata("poubelle")
@@ -296,31 +215,3 @@ def test_critiques(scraper: Scraper):
assert contenu.robinson() == "17" assert contenu.robinson() == "17"
assert contenu.suckling() == "93.5" assert contenu.suckling() == "93.5"
assert contenu._getcritiques("test_ts") is None 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", True)
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