Export de données : Ajouter filtres td afin d'être ISO avec données rapport

macantine/etl/analysis.py

@@ -190,8 +190,14 @@ def load_dataset(self):
         """
         Load in database
         """
+        logger.info(f"Loading {len(self.df)} objects in db")
         self.warehouse.insert_dataframe(self.df, self.extracted_table_name)
 
+    def _clean_dataset(self):
+        self.df = self.df.loc[:, ~self.df.columns.duplicated()]
+        self.df = utils.filter_dataframe_with_schema_cols(self.df, self.schema)
+        self.df = self.df.drop_duplicates(subset=["id"])
+
 
 class ETL_ANALYSIS_TD(ETL_ANALYSIS):
     """
@@ -204,7 +210,7 @@ class ETL_ANALYSIS_TD(ETL_ANALYSIS):
     def __init__(self):
         self.df = None
         self.years = utils.CAMPAIGN_DATES.keys()
-        self.extracted_table_name = "teledeclarations_extracted"
+        self.extracted_table_name = "teledeclarations"
         self.warehouse = DataWareHouse()
         self.schema = json.load(open("data/schemas/schema_analysis.json"))
 
@@ -225,6 +231,9 @@ def transform_dataset(self):
         # Aggregate columns for complete TD - Must occur before other transformations
         self.df = aggregate(self.df)
 
+        # Add additionnal filters (that couldn't be processed at queryset)
+        self.df = utils.filter_teledeclarations(self.df)
+
         self.compute_miscellaneous_columns()
 
         # Convert types
@@ -238,14 +247,14 @@ def transform_dataset(self):
         self.fill_geo_names(prefix="canteen.")
 
         # Fill campaign participation
-        logger.info("Canteens : Fill campaign participations...")
+        logger.info("TD : Fill campaign participations...")
         for year in utils.CAMPAIGN_DATES.keys():
             campaign_participation = utils.map_canteens_td(year)
             col_name_campaign = f"declaration_{year}"
             self.df[col_name_campaign] = self.df["id"].apply(lambda x: x in campaign_participation)
 
         # Extract the sector names and categories
-        logger.info("Canteens : Extract sectors...")
+        logger.info("TD : Extract sectors...")
         self.df[["secteur", "catégorie"]] = self.df.apply(
             lambda x: utils.format_td_sector_column(x, "canteen.sectors"), axis=1, result_type="expand"
         )
@@ -302,7 +311,7 @@ class ETL_ANALYSIS_CANTEEN(ETL_ANALYSIS):
 
     def __init__(self):
         self.df = None
-        self.extracted_table_name = "canteens_extracted"
+        self.extracted_table_name = "canteens"
         self.warehouse = DataWareHouse()
         self.schema = json.load(open("data/schemas/schema_analysis_cantines.json"))
         # The following mapper is used for renaming columns and for selecting the columns to extract from db
@@ -338,7 +347,9 @@ def transform_dataset(self):
         # Extract the sector names and categories
         logger.info("Canteens : Extract sectors and SPE...")
         self.df = utils.extract_sectors(self.df, extract_spe=True, split_category_and_sector=True, only_one_value=True)
-        self.df = self.df.rename(columns={"categories": "categorie"})
 
+        self.df = self.df.rename(columns={"categories": "categorie"})
         self.df = self.df.rename(columns=self.columns_mapper)
-        self.df = utils.filter_dataframe_with_schema_cols(self.df, self.schema)
+
+        logger.info("Canteens : Clean dataset")
+        self._clean_dataset()

macantine/etl/utils.py

@@ -225,6 +225,33 @@ def map_sectors():
     return sectors_mapper
 
 
+def filter_empty_values(df: pd.DataFrame, col_name) -> pd.DataFrame:
+    """
+    Filtering out the teledeclarations for wich a certain field is empty
+    """
+    return df.dropna(subset=col_name)
+
+
+def filter_aberrant_td(df: pd.DataFrame) -> pd.DataFrame:
+    """
+    Filtering out the teledeclarations that :
+    *  products > 1 million €
+    AND
+    * an avg meal cost > 20 €
+    """
+    mask = (df["teledeclaration.value_total_ht"] > 1000000) & (
+        df["teledeclaration.value_total_ht"] / df["canteen.yearly_meal_count"] > 20
+    )
+    return df[~mask]
+
+
+def filter_teledeclarations(df: pd.DataFrame):
+    df = filter_empty_values(df, col_name="teledeclaration.value_total_ht")
+    df = filter_empty_values(df, col_name="teledeclaration.value_bio_ht")
+    df = filter_aberrant_td(df)
+    return df
+
+
 def fetch_teledeclarations(years: list) -> pd.DataFrame:
     df = pd.DataFrame()
     for year in years:
@@ -238,7 +265,17 @@ def fetch_teledeclarations(years: list) -> pd.DataFrame:
                     ),
                     status=Teledeclaration.TeledeclarationStatus.SUBMITTED,
                     canteen_id__isnull=False,
-                ).values()
+                    canteen_siret__isnull=False,
+                    diagnostic__value_total_ht__isnull=False,
+                    diagnostic__value_bio_ht__isnull=False,
+                )
+                .exclude(
+                    canteen__deletion_date__range=(
+                        CAMPAIGN_DATES[year]["start_date"],
+                        CAMPAIGN_DATES[year]["end_date"],
+                    ),
+                )
+                .values()
             )
             df = pd.concat([df, df_year])
         else:

macantine/tests/test_etl_analysis.py

@@ -54,7 +54,7 @@ def test_extraction_teledeclaration(self):
         """
         Only teledeclarations that occurred during teledeclaration campaigns should be extracted
         """
-        canteen = CanteenFactory.create()
+        canteen = CanteenFactory.create(siret="98648424243607")
         applicant = UserFactory.create()
         with freeze_time("1991-01-14"):  # Faking time to mock creation_date
             diagnostic_1990 = DiagnosticFactory.create(canteen=canteen, year=1990, diagnostic_type=None)

macantine/tests/test_etl_open_data.py

@@ -21,7 +21,7 @@ def test_td_range_years(self, mock):
         """
         Only teledeclarations that occurred during one specific teledeclaration campaign should be extracted
         """
-        canteen = CanteenFactory.create()
+        canteen = CanteenFactory.create(siret="98648424243607")
         applicant = UserFactory.create()
         test_cases = [
             {"name": "Ignore years out of range", "year": 1990, "expected_outcome": 0},
@@ -37,7 +37,7 @@ def test_td_range_years(self, mock):
 
     @freeze_time("2023-05-14")  # Faking time to mock creation_date
     def test_ignore_cancelled_tds(self, mock):
-        canteen = CanteenFactory.create()
+        canteen = CanteenFactory.create(siret="98648424243607")
         applicant = UserFactory.create()
         diagnostic = DiagnosticFactory.create(canteen=canteen, year=2022, diagnostic_type=None)
         teledeclaration = Teledeclaration.create_from_diagnostic(diagnostic, applicant)
@@ -119,10 +119,10 @@ def test_extraction_canteen(self, mock):
         self.assertEqual(etl_canteen.len_dataset(), 0, "There shoud be an empty dataframe")
 
         # Adding data in the db
-        canteen_1 = CanteenFactory.create()
+        canteen_1 = CanteenFactory.create(siret="98648424243607")
         canteen_1.managers.add(UserFactory.create())
 
-        canteen_2 = CanteenFactory.create()  # Another canteen, but without a manager
+        canteen_2 = CanteenFactory.create(siret="98648424243607")  # Another canteen, but without a manager
         canteen_2.managers.clear()
 
         etl_canteen.extract_dataset()