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
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
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
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 présents sur une page donnée.

        Args:
            subdir (str): URL de la page à scraper.

        Returns:
            list[dict[str, Any]] | None: Liste des produits sous forme de dictionnaires,
            ou None en cas d'erreur (JSON invalide ou problème réseau).
        """
        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:
        """
        Extrait les informations d'un produit et les écrit dans le fichier CSV.

        Vérifie également si le produit a déjà été traité afin d'éviter
        les doublons.

        Args:
            cache (set[str]): Ensemble des identifiants déjà traités.
            product (dict[str, Any]): Données du produit.
            f (Any): Flux de fichier ouvert en écriture.
        """
        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, cache_name: str) -> tuple[int, set[str]]:
        """
        Initialise l'état du scraping.

        Charge l'état précédent depuis le cache si disponible,
        sinon initialise une nouvelle session.

        Args:
            reset (bool): Si True, ignore le cache et recommence depuis le début.
            cache_name (str): Nom du fichier de cache.

        Returns:
            tuple[int, set[str]]: Numéro de page de départ et ensemble
            des produits déjà traités.
        """
        if not reset:
            serializable: tuple[int, set[str]] | None = loadstate(cache_name)
            if isinstance(serializable, tuple):
                return serializable
        return 1, set()

    def _ensuretitle(self, f: TextIOWrapper, title: str) -> None:
        """
        Vérifie que l'en-tête CSV est présent au début du fichier.

        Si l'en-tête n'est pas présent, il est ajouté.

        Args:
            f (TextIOWrapper): Flux du fichier CSV.
            title (str): Ligne d'en-tête à vérifier ou écrire.
        """
        _ = 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, filename)

        try:
            with open(filename + ".csv", 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), filename)
        except (Exception, HTTPError, KeyboardInterrupt, JSONDecodeError):
            if not reset:
                savestate((page, cache), filename)

__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
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
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
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
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, filename)

    try:
        with open(filename + ".csv", 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), filename)
    except (Exception, HTTPError, KeyboardInterrupt, JSONDecodeError):
        if not reset:
            savestate((page, cache), filename)

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
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
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
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
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
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
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
 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
221
222
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
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
 97
 98
 99
100
101
102
103
104
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
237
238
239
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
183
184
185
186
187
188
189
190
191
192
193
194
195
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
225
226
227
def parker(self) -> str | None:
    """Note Robert Parker."""
    return self._getcritiques("note_rp")

robinson()

Note Jancis Robinson.

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

suckling()

Note James Suckling.

Source code in src/scraper.py
233
234
235
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
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
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
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
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}"