⚒️ BAN: fix acteur_statut manquant (#1575)

* BAN: fix acteur_statut manquant

* commentaires pour tests

Author: maxcorbeau

dags/enrich/tasks/business_logic/enrich_dbt_model_read.py

@@ -4,7 +4,7 @@
 
 import numpy as np
 import pandas as pd
-from utils import logging_utils as log
+from utils.dataframes import df_filter
 from utils.django import django_setup_full
 
 django_setup_full()
@@ -18,41 +18,20 @@ def enrich_dbt_model_read(
     """Reads necessary QFDMO acteurs and AE entries from DB"""
     from django.db import connection
 
+    logger.info(f"Lecture des données de {dbt_model_name}")
+
     # Execute SQL query and get data
     with connection.cursor() as cursor:
         cursor.execute(f"SELECT * FROM {dbt_model_name}")
         columns = [col[0] for col in cursor.description]
         data = cursor.fetchall()
 
-    # Create DataFrame and preview
+    # Create DF from Django data
+    logger.info("Création du DF")
     df = pd.DataFrame(data, columns=columns, dtype="object").replace({np.nan: None})
-    log.preview_df_as_markdown(f"Données de {dbt_model_name} SANS filtre", df)
-
-    # Filtering if needed
-    filter_applied = False
-    if not df.empty:
-        for filter in filters:
-
-            # Assignment & info
-            filter_applied = True
-            field = filter["field"]
-            operator = filter["operator"]
-            value = filter["value"]
-            logger.info(f"\n🔽 Filtre sur {field=} {operator=} {value=}")
-            logger.info(f"Avant filtre : {df.shape[0]} lignes")
-
-            # Filtering
-            if filter["operator"] == "equals":
-                logger.info(f"Filtre sur {field} EQUALS {value}")
-                df = df[df[field] == value].copy()
-            elif filter["operator"] == "contains":
-                df = df[df[field].str.contains(value, regex=True, case=False)].copy()
-            else:
-                raise NotImplementedError(f"{filter['operator']=} non implémenté")
-
-            logger.info(f"Après filtre : {df.shape[0]} lignes")
-
-    if filter_applied:
-        log.preview_df_as_markdown(f"Données de {dbt_model_name} APRES filtre(s)", df)
+
+    # Filtering
+    logger.info("Filtre sur les données")
+    df = df_filter(df, filters)
 
     return df

dags/utils/dataframes.py

@@ -9,6 +9,39 @@
 logger = logging.getLogger(__name__)
 
 
+def df_filter(df: pd.DataFrame, filters: list[dict]) -> pd.DataFrame:
+    """Filters a dataframe given some filters"""
+    log.preview_df_as_markdown("Données SANS filtre", df)
+    log.preview("Filtres", filters)
+    filter_applied = False
+    if not df.empty:
+        for filter in filters:
+            filter_applied = True
+            field = filter["field"]
+            operator = filter["operator"]
+            value = filter["value"]
+            logger.info(f"\n🔽 Filtre sur {field=} {operator=} {value=}")
+            logger.info(f"Avant filtre : {df.shape[0]} lignes")
+
+            # Filtering
+            if filter["operator"] == "equals":
+                logger.info(f"Filtre sur {field} EQUALS {value}")
+                df = df[df[field] == value].copy()
+            elif filter["operator"] == "contains":
+                df = df[df[field].str.contains(value, regex=True, case=False)].copy()
+            else:
+                raise NotImplementedError(f"{filter['operator']=} non implémenté")
+
+            logger.info(f"Après filtre : {df.shape[0]} lignes")
+
+    if filter_applied:
+        log.preview_df_as_markdown("Données APRES filtre(s)", df)
+    else:
+        logger.info("Aucun filtre appliqué")
+
+    return df
+
+
 def df_sort(
     df: pd.DataFrame, sort_rows: list[str] = [], sort_cols: list[str] = []
 ) -> pd.DataFrame:

dags_unit_tests/enrich/tasks/test_enrich_suggestions_cities.py

@@ -1,6 +1,7 @@
 import pandas as pd
 import pytest
-from enrich.config import COHORTS, COLS
+from enrich.config import COHORTS, COLS, EnrichActeursVillesConfig
+from enrich.tasks.business_logic.enrich_dbt_model_read import df_filter
 from enrich.tasks.business_logic.enrich_dbt_model_to_suggestions import (
     enrich_dbt_model_to_suggestions,
 )
@@ -9,34 +10,55 @@
 @pytest.mark.django_db
 class TestEnrichSuggestionsCities:
 
+    @pytest.fixture
+    def config(self):
+        return EnrichActeursVillesConfig(
+            dry_run=False,
+            filter_equals__acteur_statut="ACTIF",
+        )
+
     @pytest.fixture
     def df_new(self):
         return pd.DataFrame(
             {
-                COLS.SUGGEST_COHORT: [COHORTS.VILLES_NEW] * 2,
-                COLS.SUGGEST_VILLE: ["new town 1", "new town 2"],
-                COLS.ACTEUR_ID: ["new1", "new2"],
-                COLS.ACTEUR_VILLE: ["old town 1", "old town 2"],
+                # last entry is INACTIF to test acteur status filter
+                COLS.SUGGEST_COHORT: [COHORTS.VILLES_NEW] * 3,
+                COLS.SUGGEST_VILLE: ["new town 1", "new town 2", "closed"],
+                COLS.ACTEUR_ID: ["new1", "new2", "closed 1"],
+                COLS.ACTEUR_VILLE: ["old town 1", "old town 2", "closed"],
+                COLS.ACTEUR_STATUT: ["ACTIF", "ACTIF", "INACTIF"],
             }
         )
 
     @pytest.fixture
     def df_typo(self):
         return pd.DataFrame(
             {
-                COLS.SUGGEST_COHORT: [COHORTS.VILLES_TYPO] * 2,
-                COLS.SUGGEST_VILLE: ["Paris", "Laval"],
-                COLS.ACTEUR_ID: ["typo1", "typo2"],
-                COLS.ACTEUR_VILLE: ["Pâris", "Lâval"],
+                # last entry is INACTIF to test acteur status filter
+                COLS.SUGGEST_COHORT: [COHORTS.VILLES_TYPO] * 3,
+                COLS.SUGGEST_VILLE: ["Paris", "Laval", "closed"],
+                COLS.ACTEUR_ID: ["typo1", "typo2", "closed 2"],
+                COLS.ACTEUR_VILLE: ["Pâris", "Lâval", "closed"],
+                COLS.ACTEUR_STATUT: ["ACTIF", "ACTIF", "INACTIF"],
             }
         )
 
     @pytest.fixture
-    def acteurs(self, df_new, df_typo):
+    def df_new_filtered(self, df_new, config):
+        # To test that config works (e.g. filter_equals__acteur_statut)
+        return df_filter(df_new, config.filters)
+
+    @pytest.fixture
+    def df_typo_filtered(self, df_typo, config):
+        # To test that config works (e.g. filter_equals__acteur_statut)
+        return df_filter(df_typo, config.filters)
+
+    @pytest.fixture
+    def acteurs(self, df_new_filtered, df_typo_filtered):
         # Creating acteurs as presence required to apply changes
         from unit_tests.qfdmo.acteur_factory import ActeurFactory
 
-        for _, row in pd.concat([df_new, df_typo]).iterrows():
+        for _, row in pd.concat([df_new_filtered, df_typo_filtered]).iterrows():
             ActeurFactory(
                 identifiant_unique=row[COLS.ACTEUR_ID],
                 ville=row[COLS.ACTEUR_VILLE],

dbt/models/marts/enrich/marts_enrich_acteurs_villes_candidates.sql

@@ -9,6 +9,7 @@ SELECT
   acteurs.identifiant_unique AS acteur_id,
   acteurs.ville AS acteur_ville,
   acteurs.code_postal AS acteur_code_postal,
+  acteurs.statut AS acteur_statut,
   ban.ville_ancienne AS ban_ville_ancienne,
   ban.ville AS ban_ville,
   ban.code_postal AS ban_code_postal,