Merge branch 'optimisation' into exo7

Exo1

main.py

@@ -1,8 +1,90 @@
-from sys import stderr
-from typing import cast, Any, Dict, Optional
+from typing import cast
 from requests import Response, Session
 from bs4 import BeautifulSoup, Tag
-from json import JSONDecodeError, loads
+from collections import OrderedDict
+from json import loads
+
+
+class _ScraperData:
+    def __init__(self, data: dict[str, object]) -> None:
+        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 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:
@@ -13,98 +95,130 @@ class Scraper:
     """
 
     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._latest_request: tuple[(str, Response | None)] = ("", None)
+        # 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 = "") -> Response:
-        rq_subdir, rq_response = self._latest_request
-        if rq_response is None or subdir != rq_subdir:
-            request: Response = self._request(subdir)
+    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
-        return rq_response
 
-    def getsoup(self, subdir: str = "") -> BeautifulSoup:
+        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
-        return BeautifulSoup(markup, features="html.parser")
+        soup: BeautifulSoup = BeautifulSoup(markup, features="html.parser")
 
-    def getjsondata(self, subdir: str = "", id: str = "__NEXT_DATA__") -> dict[str, object]:
+        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 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}")
+        if script is None or not script.string:
+            raise ValueError(f"le script id={id} est introuvable")
 
-                if isinstance(current_data, dict):
-                    return cast(dict[str, object], current_data)
+        current_data: object = cast(object, loads(script.string))
 
-            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)
-
-            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):
+        for key in ["props", "pageProps"]:
+            if isinstance(current_data, dict) and key in current_data:
+                current_data = cast(object, current_data[key])
                 continue
-            attrs = item.get("attributes", {})
-            nbunit = attrs.get("nbunit", {}).get("value")
-            equivbtl = attrs.get("equivbtl", {}).get("value")
+            raise ValueError(f"Clé manquante dans le JSON : {key}")
 
-            if nbunit == "1" and equivbtl == "1":
-                p = item.get("offerPrice")
-                if isinstance(p, (int, float)):
-                    return float(p)
+        return _ScraperData(cast(dict[str, object], current_data))
 
-        # 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.")

requirements.txt

@@ -1,2 +1,3 @@
 requests>=2.32.5
+requests-mock>=1.12.1
 beautifulsoup4>=4.14.3

test_main.py

@@ -1,35 +1,226 @@
-import json
+from json import dumps
+from bs4 import Tag
+import pytest
+from requests_mock import Mocker
 from main import Scraper
 
 
-def test_json():
-    scraper = 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>
+            """,
+        )
 
-    data = scraper.getjsondata("/chateau-gloria-2016.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>
+            """,
+        )
 
-    print("JSON récupéré :")
-    print(json.dumps(data, indent=4, ensure_ascii=False))
+        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": 390,
+                                        "offerPrice": 390,
+                                        "seoKeyword": "nino-negri-5-stelle-sfursat-2022-c-cc-6.html",
+                                        "shortdesc": "Un carton de 6 Bouteilles (75cl)",
+                                        "attributes": {
+                                            "promotion_o_n": {
+                                                "valueId": "0",
+                                                "name": "En promotion",
+                                                "value": "Non",
+                                                "sequence": 80,
+                                                "displayable": "False",
+                                                "type": "CHECKBOX",
+                                                "isSpirit": False,
+                                            },
+                                            "in_stock": {
+                                                "valueId": "L",
+                                                "name": "En stock",
+                                                "value": "Livrable",
+                                                "sequence": 65,
+                                                "displayable": "true",
+                                                "type": "CHECKBOX",
+                                                "isSpirit": False,
+                                            },
+                                        },
+                                        "stock": 12,
+                                        "availability": "2026-02-05",
+                                        "isCustomizable": False,
+                                        "gtin_cond": "",
+                                        "gtin_unit": "",
+                                        "stockOrigin": "EUR",
+                                        "isPrevSale": False,
+                                    }
+                                ],
+                                "attributes": {
+                                    "equivbtl": {
+                                        "valueId": "1",
+                                        "name": "equivbtl",
+                                        "value": "1",
+                                        "isSpirit": False,
+                                    },
+                                    "nbunit": {
+                                        "valueId": "6",
+                                        "name": "nbunit",
+                                        "value": "6",
+                                        "isSpirit": False,
+                                    },
+                                    "appellation": {
+                                        "valueId": "433",
+                                        "name": "Appellation",
+                                        "value": "Sforzato di Valtellina",
+                                        "url": "sforzato-di-valtellina.html",
+                                        "isSpirit": False,
+                                        "groupIdentifier": "appellation_433",
+                                    },
+                                    "note_rp": {
+                                        "valueId": "91",
+                                        "name": "Parker",
+                                        "value": "91",
+                                        "isSpirit": False,
+                                    },
+                                    "note_jr": {
+                                        "valueId": "17+",
+                                        "name": "J. Robinson",
+                                        "value": "17+",
+                                        "isSpirit": False,
+                                    },
+                                    "note_js": {
+                                        "valueId": "93-94",
+                                        "name": "J. Suckling",
+                                        "value": "93-94",
+                                        "isSpirit": False,
+                                    },
+                                },
+                            }
+                        }
+                    }
+                }
+            }
+        }
 
-    assert isinstance(data, dict)
-    assert "items" in data
+        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,
+        )
+
+        # on return m sans fermer le server qui simule la page
+        yield m
 
 
-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)")
+@pytest.fixture
+def scraper() -> Scraper:
+    return Scraper()
 
 
-if __name__ == "__main__":
-    test_json()
-    test_prix()
-    print("\nTous les tests terminés")
+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() == "Sforzato di Valtellina"
+
+
+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