Skip to content

Scraper

Sommaire


Classe 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.

Source code in src/scraper.py
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
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()

            products: list[dict[str, Any]] = cast(
                list[dict[str, Any]], data.get("products")
            )

            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 _initstate(self, reset: bool) -> tuple[int, set[str]]:
        """
        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.

        Args:
            reset (bool): pouvoir le reset ou pas

        Returns:
            tuple[int, set[str]]: le contenu de la page et du cache
        """
        if not reset:
            #
            serializable: tuple[int, set[str]] | None = loadstate()
            if isinstance(serializable, tuple):
                return serializable
        return 1, set()

    def _ensuretitle(self, f: TextIOWrapper, title: str) -> None:
        """
        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.

        Args:
            f (TextIOWrapper): buffer stream fichier
            title (str): titre du csv
        """
        _ = f.seek(0, SEEK_SET)
        if not (f.read(len(title)) == title):
            _ = f.write(title)
        else:
            _ = f.seek(0, SEEK_END)

    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: page où commence le scraper
        # cache: tout les pages déjà parcourir
        page, cache = self._initstate(reset)

        try:
            with open(filename, mode) as f:
                self._ensuretitle(f, title)
                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="{l_bar} {bar:20} {r_bar}"
                    )
                    for product in pbar:
                        keyword: str = cast(
                            str, product.get("seoKeyword", "Inconnu")[:40]
                        )
                        pbar.set_description(
                            f"Page: {page:<3} | Product: {keyword:<40}"
                        )
                        self._writevins(cache, product, f)
                    page += 1
                    # va créer un fichier au début et l'override
                    # tout les 5 pages au cas où SIGHUP ou autre
                    if page % 5 == 0 and not reset:
                        savestate((page, cache))
        except (Exception, HTTPError, KeyboardInterrupt, JSONDecodeError):
            if not reset:
                savestate((page, cache))

__init__()

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
Source code in src/scraper.py
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
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
    ]()

getvins(subdir, filename, reset=False)

Scrape toutes les pages d'une catégorie et sauvegarde en CSV.

Parameters:

Name Type Description Default
subdir str

La catégorie (ex: '/vins-rouges').

required
filename str

Nom du fichier de sortie (ex: 'vins.csv').

required
reset bool

(Optionnel) pour réinitialiser le processus.

False
Source code in src/scraper.py
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
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: page où commence le scraper
    # cache: tout les pages déjà parcourir
    page, cache = self._initstate(reset)

    try:
        with open(filename, mode) as f:
            self._ensuretitle(f, title)
            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="{l_bar} {bar:20} {r_bar}"
                )
                for product in pbar:
                    keyword: str = cast(
                        str, product.get("seoKeyword", "Inconnu")[:40]
                    )
                    pbar.set_description(
                        f"Page: {page:<3} | Product: {keyword:<40}"
                    )
                    self._writevins(cache, product, f)
                page += 1
                # va créer un fichier au début et l'override
                # tout les 5 pages au cas où SIGHUP ou autre
                if page % 5 == 0 and not reset:
                    savestate((page, cache))
    except (Exception, HTTPError, KeyboardInterrupt, JSONDecodeError):
        if not reset:
            savestate((page, cache))

getjsondata(subdir, id='__NEXT_DATA__')

Extrait les données JSON contenues dans la balise NEXT_DATA du site.

Parameters:

Name Type Description Default
subdir str

Le chemin de la page.

required
id str

L'identifiant de la balise script.

'__NEXT_DATA__'

Raises:

Type Description
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:

Name Type Description
_ScraperData _ScraperData

Instance contenant les données extraites.

Source code in src/scraper.py
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
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))

getresponse(subdir='', use_cache=True)

Récupère la réponse d'une page, en utilisant le cache si possible.

Parameters:

Name Type Description Default
subdir str

Le chemin de la page.

''
use_cache bool

Utilise la donnée deja sauvegarder ou écrase la donnée utilisé avec la nouvelle

True

Returns:

Name Type Description
Response Response

L'objet réponse (cache ou nouvelle requête).

Raises:

Type Description
HTTPError

Si le serveur renvoie un code d'erreur (4xx, 5xx).

Source code in src/scraper.py
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
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

getsoup(subdir, use_cache=True)

Récupère le contenu HTML d'une page et le transforme en objet BeautifulSoup.

Parameters:

Name Type Description Default
subdir str

Le chemin de la page.

required

Returns:

Name Type Description
BeautifulSoup BeautifulSoup

L'objet parsé pour extraction de données.

Raises:

Type Description
HTTPError

Si le serveur renvoie un code d'erreur (4xx, 5xx).

Source code in src/scraper.py
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
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

Classe _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.

Source code in src/scraper.py
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
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()
        prix = self.prix()

        return f"{appellation},{parker},{robinson},{suckling},{prix}"

__init__(data)

Initialise le conteneur avec un dictionnaire de données.

Parameters:

Name Type Description Default
data dict[str, object]

Le dictionnaire JSON brut extrait de la page.

required
Source code in src/scraper.py
61
62
63
64
65
66
67
68
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

getdata()

Retourne le dictionnaire de données complet.

Source code in src/scraper.py
201
202
203
def getdata(self) -> dict[str, object]:
    """Retourne le dictionnaire de données complet."""
    return self._data

appellation()

Extrait le nom de l'appellation du vin.

Returns:

Type Description
str | None

str | None: Le nom (ex: 'Pauillac') ou None.

Source code in src/scraper.py
147
148
149
150
151
152
153
154
155
156
157
158
159
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

parker()

Note Robert Parker.

Source code in src/scraper.py
189
190
191
def parker(self) -> str | None:
    """Note Robert Parker."""
    return self._getcritiques("note_rp")

robinson()

Note Jancis Robinson.

Source code in src/scraper.py
193
194
195
def robinson(self) -> str | None:
    """Note Jancis Robinson."""
    return self._getcritiques("note_jr")

suckling()

Note James Suckling.

Source code in src/scraper.py
197
198
199
def suckling(self) -> str | None:
    """Note James Suckling."""
    return self._getcritiques("note_js")

prix()

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:

Type Description
float | None

float | None: Le prix calculé arrondi à 2 décimales, ou None.

Source code in src/scraper.py
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
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

informations()

Agrège les données clés pour l'export CSV.

Returns:

Name Type Description
str str

Ligne formatée : "Appellation,Parker,Robinson,Suckling,Prix".

Source code in src/scraper.py
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
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()
    prix = self.prix()

    return f"{appellation},{parker},{robinson},{suckling},{prix}"