guezoloic/archived-millesima-projetS6
1
0
Fork 0
mirror of https://github.com/guezoloic/millesima_projetS6.git synced 2026-03-28 11:03:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

ajout: correction d'erreur, changement de main dans cleaning

Browse Source
This commit is contained in:
Loïc GUEZO
2026-03-06 21:02:52 +01:00
parent acf4ddd881
commit f4ded6d8b5
4 changed files with 75 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
src/cleaning.py
View File

@@ -1,7 +1,14 @@
#!/usr/bin/env python3
from typing import cast, override
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:
@@ -18,7 +25,6 @@ class Cleaning:
def getVins(self) -> DataFrame:
return self._vins.copy(deep=True)
@override
def __str__(self) -> str:
"""
Affiche un résumé du DataFrame
@@ -34,7 +40,7 @@ class Cleaning:
f"Statistiques numériques :\n{self._vins.describe().round(2)}\n\n"
)
def drop_empty_appellation(self) -> Cleaning:
def drop_empty_appellation(self) -> "Cleaning":
self._vins = self._vins.dropna(subset=["Appellation"])
return self
@@ -61,7 +67,7 @@ class Cleaning:
def _mean_suckling(self) -> DataFrame:
return self._mean_score("Suckling")
def fill_missing_scores(self) -> Cleaning:
def fill_missing_scores(self) -> "Cleaning":
"""
Remplacer les notes manquantes par la moyenne
des vins de la même appellation.
@@ -69,14 +75,14 @@ class Cleaning:
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:
def encode_appellation(self, column: str = "Appellation") -> "Cleaning":
"""
Remplace la colonne 'Appellation' par des colonnes indicatrices
"""
@@ -84,4 +90,20 @@ class Cleaning:
appellation_dummies = get_dummies(appellations)
self._vins = self._vins.drop(columns=[column])
self._vins = self._vins.join(appellation_dummies)
return self
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}")

58
src/main.py
View File

@@ -1,58 +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
from cleaning import *
def load_csv(filename: str) -> DataFrame:
path: str = normpath(join(getcwd(), filename))
return read_csv(path)
def save_csv(df: DataFrame, out_filename: str) -> None:
df.to_csv(out_filename, index=False)
def main() -> None:
if len(argv) != 2:
raise ValueError(f"Usage: {argv[0]} <filename.csv>")
df = load_csv(argv[1])
display_info(df, "Avant le nettoyage")
df = drop_empty_appellation(df)
save_csv(df, "donnee_clean.csv")
display_info(df, "Après nettoyage d'appellations manquantes")
#la moyenne des notes des vins pour chaque appellation
robert_means = mean_robert(df)
save_csv(robert_means, "mean_robert_by_appellation.csv")
display_info(robert_means, "Moyennes Robert par appellation")
robinson_means = mean_robinson(df)
save_csv(robinson_means, "mean_robinson_by_appellation.csv")
display_info(robinson_means, "Moyennes Robinson par appellation")
suckling_means = mean_suckling(df)
save_csv(suckling_means, "mean_suckling_by_appellation.csv")
display_info(suckling_means, "Moyennes Suckling par appellation")
df_missing_scores = fill_missing_scores(df)
save_csv(df_missing_scores, "donnee_filled.csv")
display_info(df_missing_scores, "Après remplissage des notes manquantes par la moyenne de l'appellation")
df_ready = encode_appellation(df_missing_scores)
save_csv(df_ready, "donnee_ready.csv")
display_info(df_ready, "Après remplacer la colonne 'Appellation' par des colonnes indicatrices")
if __name__ == "__main__":
try:
main()
except Exception as e:
print(f"ERREUR: {e}")

74
src/scraper.py
View File

@@ -1,7 +1,7 @@
#!/usr/bin/env python3
from collections import OrderedDict
from io import SEEK_END, SEEK_SET, BufferedWriter
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
@@ -407,6 +407,44 @@ class Scraper:
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.
@@ -420,35 +458,13 @@ class Scraper:
mode: Literal["w", "a+"] = "w" if reset else "a+"
# titre
title: str = "Appellation,Robert,Robinson,Suckling,Prix\n"
# page du début
page: int = 1
# le set qui sert de cache
cache: set[str] = set[str]()
# page: page où commence le scraper
# cache: tout les pages déjà parcourir
page, cache = self._initstate(reset)
custom_format = "{l_bar} {bar:20} {r_bar}"
if not reset:
# appelle la fonction pour load le cache, si il existe
# pas, il utilise les variables de base sinon il override
# toute les variables pour continuer et pas recommencer le
# processus en entier.
serializable: tuple[int, set[str]] | None = loadstate()
if isinstance(serializable, tuple):
# override la page et le cache
page, cache = serializable
try:
with open(filename, mode) as f:
# check si le titre est bien présent au début du buffer
# sinon il l'ecrit, petit bug potentiel, a+ ecrit tout le
# temps a la fin du buffer, si on a ecrit des choses avant
# le titre sera apres ces données mais on part du principe
# que personne va toucher le fichier.
_ = f.seek(0, SEEK_SET)
if not (f.read(len(title)) == title):
_ = f.write(title)
else:
_ = f.seek(0, SEEK_END)
self._ensuretitle(f, title)
while True:
products_list: list[dict[str, Any]] | None = (
self._geturlproductslist(f"{subdir}?page={page}")
@@ -457,7 +473,7 @@ class Scraper:
break
pbar: tqdm[dict[str, Any]] = tqdm(
products_list, bar_format=custom_format
products_list, bar_format="{l_bar} {bar:20} {r_bar}"
)
for product in pbar:
keyword: str = cast(
@@ -469,7 +485,7 @@ class Scraper:
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
# 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):

3
tests/test_cleaning.py
View File

@@ -1,5 +1,4 @@
import pytest
from pandas import DataFrame
from unittest.mock import patch, mock_open
from cleaning import Cleaning
@@ -49,7 +48,7 @@ def test_fill_missing_scores(cleaning_raw: Cleaning):
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()