ajout: ajout prefixe get_dummies

.github/workflows/python-app.yml

@@ -36,10 +36,3 @@ jobs:
 
       - name: Test with pytest
         run: 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

pyproject.toml

@@ -1,7 +1,12 @@
 [project]
 name = "projet-millesima-s6"
 version = "0.1.0"
-dependencies = ["requests==2.32.5", "beautifulsoup4==4.14.3", "pandas==2.3.3"]
+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"]

src/cleaning.py

@@ -0,0 +1,109 @@
+#!/usr/bin/env python3
+
+from os import getcwd
+from os.path import normpath, join
+from typing import cast
+from pandas import DataFrame, read_csv, to_numeric, get_dummies
+from sys import argv
+
+
+def path_filename(filename: str) -> str:
+    return normpath(join(getcwd(), filename))
+
+
+class Cleaning:
+    def __init__(self, filename) -> None:
+        self._vins: DataFrame = read_csv(filename)
+        # créer la liste de tout les scores
+        self.SCORE_COLS: list[str] = [
+            c for c in self._vins.columns if c not in ["Appellation", "Prix"]
+        ]
+        # transforme tout les colonnes score en numérique
+        for col in self.SCORE_COLS:
+            self._vins[col] = to_numeric(self._vins[col], errors="coerce")
+
+    def getVins(self) -> DataFrame:
+        return self._vins.copy(deep=True)
+
+    def __str__(self) -> str:
+        """
+        Affiche un résumé du DataFrame
+            - la taille
+            - types des colonnes
+            - valeurs manquantes
+            - statistiques numériques
+        """
+        return (
+            f"Shape : {self._vins.shape[0]} lignes x {self._vins.shape[1]} colonnes\n\n"
+            f"Types des colonnes :\n{self._vins.dtypes}\n\n"
+            f"Valeurs manquantes :\n{self._vins.isna().sum()}\n\n"
+            f"Statistiques numériques :\n{self._vins.describe().round(2)}\n\n"
+        )
+
+    def drop_empty_appellation(self) -> "Cleaning":
+        self._vins = self._vins.dropna(subset=["Appellation"])
+        return self
+
+    def _mean_score(self, col: str) -> DataFrame:
+        """
+        Calcule la moyenne d'une colonne de score par appellation.
+            - Convertit les valeurs en numériques, en remplaçant les non-convertibles par NaN
+            - Calcule la moyenne par appellation
+            - Remplace les NaN résultants par 0
+
+        """
+        means = self._vins.groupby("Appellation", as_index=False)[col].mean()
+        means = means.rename(
+            columns={col: f"mean_{col}"}
+        )  # pyright: ignore[reportCallIssue]
+        return cast(DataFrame, means.fillna(0))
+
+    def _mean_robert(self) -> DataFrame:
+        return self._mean_score("Robert")
+
+    def _mean_robinson(self) -> DataFrame:
+        return self._mean_score("Robinson")
+
+    def _mean_suckling(self) -> DataFrame:
+        return self._mean_score("Suckling")
+
+    def fill_missing_scores(self) -> "Cleaning":
+        """
+        Remplacer les notes manquantes par la moyenne
+        des vins de la même appellation.
+        """
+        for element in self.SCORE_COLS:
+            means = self._mean_score(element)
+            self._vins = self._vins.merge(means, on="Appellation", how="left")
+
+            mean_col = f"mean_{element}"
+            self._vins[element] = self._vins[element].fillna(self._vins[mean_col])
+
+            self._vins = self._vins.drop(columns=["mean_" + element])
+        return self
+
+    def encode_appellation(self, column: str = "Appellation") -> "Cleaning":
+        """
+        Remplace la colonne 'Appellation' par des colonnes indicatrices
+        """
+        appellations = self._vins[column].astype(str).str.strip()
+        appellation_dummies = get_dummies(appellations, prefix="App")
+        self._vins = self._vins.drop(columns=[column])
+        self._vins = self._vins.join(appellation_dummies)
+        return self
+
+
+def main() -> None:
+    if len(argv) != 2:
+        raise ValueError(f"Usage: {argv[0]} <filename.csv>")
+
+    filename = argv[1]
+    cleaning: Cleaning = Cleaning(filename)
+    _ = cleaning.drop_empty_appellation().fill_missing_scores().encode_appellation()
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception as e:
+        print(f"ERREUR: {e}")

src/main.py

@@ -1,20 +0,0 @@
-#!/usr/bin/env python3
-
-from os import getcwd
-from os.path import normpath, join
-from sys import argv
-from pandas import read_csv, DataFrame
-
-def main() -> None:
-    if len(argv) != 2:
-        raise ValueError(f"{argv[0]} <filename.csv>")
-
-    path: str = normpath(join(getcwd(), argv[1]))
-    db: DataFrame = read_csv(path)
-    print(db.all())
-
-if __name__ == "__main__":
-    try:
-        main()
-    except Exception as e:
-        print(f"ERREUR: {e}")

src/scraper.py

@@ -1,11 +1,53 @@
 #!/usr/bin/env python3
 
-from sys import argv
-from typing import cast
-from requests import HTTPError, Response, Session
-from bs4 import BeautifulSoup, Tag
 from collections import OrderedDict
+from io import SEEK_END, SEEK_SET, BufferedWriter, TextIOWrapper
 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:
@@ -110,7 +152,6 @@ class _ScraperData:
             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):
@@ -174,6 +215,7 @@ class _ScraperData:
         robinson = self.robinson()
         suckling = self.suckling()
         prix = self.prix()
+        prix = self.prix()
 
         return f"{appellation},{parker},{robinson},{suckling},{prix}"
 
@@ -189,7 +231,7 @@ class Scraper:
     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
@@ -328,7 +370,7 @@ class Scraper:
 
         return _ScraperData(cast(dict[str, object], current_data))
 
-    def _geturlproductslist(self, subdir: str) -> list[str] | None:
+    def _geturlproductslist(self, subdir: str) -> list[dict[str, Any]] | None:
         """
         Récupère la liste des produits d'une page de catégorie.
         """
@@ -336,65 +378,129 @@ class Scraper:
             data: dict[str, object] = self.getjsondata(subdir).getdata()
 
             for element in ["initialReduxState", "categ", "content"]:
-                data: dict[str, object] = cast(dict[str, object], data.get(element))
+                data = cast(dict[str, object], data.get(element))
-                if not isinstance(data, dict):
-                    return None
 
-            products: list[str] = cast(list[str], data.get("products"))
+            products: list[dict[str, Any]] = cast(
-            if isinstance(products, list):
+                list[dict[str, Any]], data.get("products")
-                return products
+            )
+
+            return products
 
         except (JSONDecodeError, HTTPError):
             return None
 
-    def getvins(self, subdir: str, filename: str, reset: bool) -> 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_
         """
-        Scrape récursivement toutes les pages d'une catégorie et sauvegarde en CSV.
+        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.
         """
-        with open(filename, "w") as f:
+        # mode d'écriture fichier
-            cache: set[str] = set[str]()
+        mode: Literal["w", "a+"] = "w" if reset else "a+"
-            page = 0
+        # titre
-            _ = f.write("Appellation,Robert,Robinson,Suckling,Prix\n")
+        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)
 
-            while True:
+        try:
-                page += 1
+            with open(filename, mode) as f:
-                products_list: list[str] | None = self._geturlproductslist(
+                self._ensuretitle(f, title)
-                    f"{subdir}?page={page}"
+                while True:
-                )
+                    products_list: list[dict[str, Any]] | None = (
+                        self._geturlproductslist(f"{subdir}?page={page}")
+                    )
+                    if not products_list:
+                        break
 
-                if not products_list:
+                    pbar: tqdm[dict[str, Any]] = tqdm(
-                    break
+                        products_list, bar_format="{l_bar} {bar:20} {r_bar}"
-
+                    )
-                products_list_length = len(products_list)
+                    for product in pbar:
-                for i, product in enumerate(products_list):
+                        keyword: str = cast(
-                    if not isinstance(product, dict):
+                            str, product.get("seoKeyword", "Inconnu")[:40]
-                        continue
+                        )
-
+                        pbar.set_description(
-                    link = product.get("seoKeyword")
+                            f"Page: {page:<3} | Product: {keyword:<40}"
-
+                        )
-                    if link and link not in cache:
+                        self._writevins(cache, product, f)
-                        try:
+                    page += 1
-                            infos = self.getjsondata(link).informations()
+                    # va créer un fichier au début et l'override
-                            _ = f.write(infos + "\n")
+                    # tout les 5 pages au cas où SIGHUP ou autre
-                            print(
+                    if page % 5 == 0 and not reset:
-                                f"page: {page} | {i + 1}/{products_list_length} {link}"
+                        savestate((page, cache))
-                            )
+        except (Exception, HTTPError, KeyboardInterrupt, JSONDecodeError):
-                            cache.add(link)
+            if not reset:
-                        except (JSONDecodeError, HTTPError) as e:
+                savestate((page, cache))
-                            print(f"Erreur sur le produit {link}: {e}")
-                f.flush()
 
 
 def main() -> None:
-    if len(argv) != 2:
+    if len(argv) != 3:
-        raise ValueError(f"{argv[0]} <sous-url>")
+        raise ValueError(f"{argv[0]} <filename> <sous-url>")
+    filename = argv[1]
+    suburl = argv[2]
+
     scraper: Scraper = Scraper()
-    scraper.getvins(argv[1], "donnee.csv", False)
+    scraper.getvins(suburl, filename)
 
 
 if __name__ == "__main__":

tests/test_cleaning.py

@@ -0,0 +1,67 @@
+import pytest
+from unittest.mock import patch, mock_open
+from cleaning import Cleaning
+
+
+@pytest.fixture
+def cleaning_raw() -> Cleaning:
+    """
+    "Appellation": ["Pauillac", "Pauillac ", "Margaux", None  , "Pomerol", "Pomerol"],
+    "Robert":      ["95"      , None       , "bad"    , 90    , None     , None     ],
+    "Robinson":    [None      , "93"       , 18       , None  , None     , None     ],
+    "Suckling":    [96        , None       , None     , None  , 91       , None     ],
+    "Prix":        ["10.0"    , "11.0"     , "20.0"   , "30.0", "40.0"   , "50.0"   ],
+    """
+    csv_content = """Appellation,Robert,Robinson,Suckling,Prix
+Pauillac,95,,96,10.0
+Pauillac ,,93,,11.0
+Margaux,bad,18,,20.0
+,90,,,30.0
+Pomerol,,,91,40.0
+Pomerol,,,,50.0
+"""
+    m = mock_open(read_data=csv_content)
+    with patch("builtins.open", m):
+        return Cleaning("donnee.csv")
+
+
+def test_drop_empty_appellation(cleaning_raw: Cleaning) -> None:
+    out = cleaning_raw.drop_empty_appellation().getVins()
+    assert out["Appellation"].isna().sum() == 0
+    assert len(out) == 5
+
+
+def test_mean_score_zero_when_no_scores(cleaning_raw: Cleaning) -> None:
+    out = cleaning_raw.drop_empty_appellation()
+    m = out._mean_score("Robert")
+    assert list(m.columns) == ["Appellation", "mean_Robert"]
+    pomerol_mean = m.loc[m["Appellation"].str.strip() == "Pomerol", "mean_Robert"].iloc[
+        0
+    ]
+    assert pomerol_mean == 0
+
+
+def test_fill_missing_scores(cleaning_raw: Cleaning):
+    cleaning_raw._vins["Appellation"] = cleaning_raw._vins["Appellation"].str.strip()
+
+    cleaning_raw.drop_empty_appellation()
+    filled = cleaning_raw.fill_missing_scores().getVins()
+    for col in cleaning_raw.SCORE_COLS:
+        assert filled[col].isna().sum() == 0
+
+    pauillac_robert = filled[filled["Appellation"] == "Pauillac"]["Robert"]
+    assert (pauillac_robert == 95.0).all()
+
+
+def test_encode_appellation(cleaning_raw: Cleaning):
+    cleaning_raw._vins["Appellation"] = cleaning_raw._vins["Appellation"].str.strip()
+
+    out = (
+        cleaning_raw.drop_empty_appellation()
+        .fill_missing_scores()
+        .encode_appellation()
+        .getVins()
+    )
+    assert "App_Appellation" not in out.columns
+    assert "App_Pauillac" in out.columns
+    assert int(out.loc[0, "App_Pauillac"]) == 1

tests/test_scraper.py

@@ -153,7 +153,7 @@ def mock_site():
 
         html_product = f"""
         <html>
-            <body>    
+            <body>
                 <h1>MILLESIMA</h1>
                 <script id="__NEXT_DATA__" type="application/json">
                     {dumps(json_data)}
@@ -168,7 +168,7 @@ def mock_site():
 
         html_product = f"""
         <html>
-            <body>    
+            <body>
                 <h1>MILLESIMA</h1>
                 <script id="__NEXT_DATA__" type="application/json">
                     {dumps(json_data)}
@@ -179,7 +179,7 @@ def mock_site():
 
         list_pleine = f"""
             <html>
-                <body>    
+                <body>
                     <h1>LE WINE</h1>
                     <script id="__NEXT_DATA__" type="application/json">
                         {dumps({
@@ -207,7 +207,7 @@ def mock_site():
 
         list_vide = f"""
             <html>
-                <body>    
+                <body>
                     <h1>LE WINE</h1>
                     <script id="__NEXT_DATA__" type="application/json">
                         {dumps({
@@ -319,7 +319,7 @@ def test_informations(scraper: Scraper):
 def test_search(scraper: Scraper):
     m = mock_open()
     with patch("builtins.open", m):
-        scraper.getvins("wine.html", "fake_file.csv", False)
+        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)