1️⃣ Suggestions: migrer vers 1 seul template (#1594)

* Suggestions: migrer vers 1 seul template

* fix clustering context & rgpd tests

* supprimer template clustering

Author: maxcorbeau

dags/cluster/tasks/business_logic/cluster_acteurs_suggestions/context.py

@@ -1,23 +1,28 @@
 """Generates the data stored in Suggestion.contexte field"""
 
+import logging
+
 import pandas as pd
 from cluster.config.constants import COL_PARENT_ID_BEFORE
 from utils import logging_utils as log
 
+logger = logging.getLogger(__name__)
+
 
 def suggestion_context_generate(
     df_cluster: pd.DataFrame,
     cluster_fields_exact: list[str],
     cluster_fields_fuzzy: list[str],
-) -> dict:
+) -> dict | None:
     """Generates a dict for use in Suggestion.contexte field"""
     from data.models.change import COL_CHANGE_MODEL_NAME
     from data.models.changes import ChangeActeurCreateAsParent
 
-    clusters_cnt = df_cluster["cluster_id"].nunique()
-    if clusters_cnt != 1:
-        msg = f"We create contexte for 1 cluster at a time, got {clusters_cnt}"
+    cluster_ids = df_cluster["cluster_id"].unique()
+    if len(cluster_ids) != 1:
+        msg = f"We create contexte for 1 cluster at a time, got {cluster_ids=}"
         raise ValueError(msg)
+    cluster_id = cluster_ids[0]
 
     # Exclude parents-to-be-created as non-existing thus not part of clustering context
     df_cluster = df_cluster[
@@ -33,6 +38,14 @@ def suggestion_context_generate(
     # - TODO: we will need to make the below exclusion conditional on feature activation
     df_cluster = df_cluster[df_cluster[COL_PARENT_ID_BEFORE].isnull()]
 
+    # Can happen when DAG to refresh displayed acteurs wasn't ran:
+    # - we can have data in displayed but in reality underlying acteurs
+    # are not longer ACTIVE nor present (e.g. removed parents)
+    if df_cluster.empty:
+        msg = f"{cluster_id=} vide: bien rafraîchir les acteurs affichés"
+        logger.warning(msg)
+        return None
+
     # Ensuring we have 1 exact group:
     exacts = df_cluster.groupby(cluster_fields_exact)
     groups = list(exacts.groups.keys())
@@ -42,9 +55,10 @@ def suggestion_context_generate(
         log.preview("cluster problématique", df_cluster)
         raise ValueError(msg)
 
-    result = {}
-    result["exact_match"] = dict(zip(cluster_fields_exact, groups[0]))  # type: ignore
     cols = ["identifiant_unique"] + cluster_fields_fuzzy
-    result["fuzzy_details"] = df_cluster[cols].to_dict(orient="records")
+    context = {}
+    context["cluster_id"] = cluster_id
+    context["exact_match"] = dict(zip(cluster_fields_exact, groups[0]))  # type: ignore
+    context["fuzzy_details"] = df_cluster[cols].to_dict(orient="records")
 
-    return result
+    return context

dags/cluster/tasks/business_logic/cluster_acteurs_suggestions/prepare.py

@@ -17,7 +17,6 @@ def cluster_changes_get(cluster: pd.DataFrame) -> list[dict]:
         COL_CHANGE_ENTITY_TYPE,
         COL_CHANGE_MODEL_NAME,
         COL_CHANGE_MODEL_PARAMS,
-        COL_CHANGE_NAMESPACE,
         COL_CHANGE_ORDER,
         COL_CHANGE_REASON,
         SuggestionChange,
@@ -31,13 +30,16 @@ def cluster_changes_get(cluster: pd.DataFrame) -> list[dict]:
     )
     from qfdmo.models.acteur import RevisionActeur
 
+    # Building changes for each acteur in the cluster
     changes = []
     for _, row in cluster.iterrows():
+
+        # Common params
         row = row.to_dict()
         model_name = row[COL_CHANGE_MODEL_NAME]
-        # All acteur-related changes have an "identifiant_unique"
         model_params = {"id": row["identifiant_unique"]}
-        # Then params data depends on the type of changes
+
+        # Params specific to the type of change
         if model_name == ChangeActeurUpdateParentId.name():
             model_params["data"] = {"parent_id": row["parent_id"]}
         elif model_name in [
@@ -46,11 +48,9 @@ def cluster_changes_get(cluster: pd.DataFrame) -> list[dict]:
         ]:
             model_params["data"] = row[COL_PARENT_DATA_NEW]
         elif model_name == ChangeActeurVerifyRevision.name():
-            # no extra params to pass for this one
-            pass
+            pass  # no additional param, we just check presence
         elif model_name == ChangeActeurDeleteAsParent.name():
-            # to delete parents we already have their id
-            pass
+            pass  # no additional param, we just delete using id
         else:
             raise ValueError(f"Unexpected model_name: {model_name}")
 
@@ -60,21 +60,20 @@ def cluster_changes_get(cluster: pd.DataFrame) -> list[dict]:
 
         # Validation
         row[COL_CHANGE_MODEL_PARAMS] = model_params
-        change = {
-            x.replace(COL_CHANGE_NAMESPACE, ""): row[x]
-            for x in [
-                COL_CHANGE_ORDER,
-                COL_CHANGE_REASON,
-                COL_CHANGE_ENTITY_TYPE,
-                COL_CHANGE_MODEL_NAME,
-                COL_CHANGE_MODEL_PARAMS,
-            ]
-        }
         try:
+            change = {
+                "order": row[COL_CHANGE_ORDER],
+                "reason": row[COL_CHANGE_REASON],
+                "entity_type": row[COL_CHANGE_ENTITY_TYPE],
+                "model_name": row[COL_CHANGE_MODEL_NAME],
+                "model_params": row[COL_CHANGE_MODEL_PARAMS],
+            }
             SuggestionChange(**change)
         except Exception as e:
             log.preview("🔴 Broken change", change)
             raise e
+
+        # Appending
         changes.append(change)
     return changes
 
@@ -92,12 +91,10 @@ def cluster_acteurs_suggestions_prepare(
         # whenever we encounter issues
         try:
             changes = cluster_changes_get(cluster)
-            suggestion = {"cluster_id": cluster_id, "changes": changes}
-            cluster_id = suggestion["cluster_id"]
+            title = "📦 Cluster/dédup"
+            suggestion = {"title": title, "cluster_id": cluster_id, "changes": changes}
             df_changes = pd.DataFrame(suggestion["changes"])
-            log.preview_df_as_markdown(
-                f"Suggestion pour cluster_id={cluster_id}", df_changes
-            )
+            log.preview_df_as_markdown(f"Suggestion pour {cluster_id=}", df_changes)
             working.append(suggestion)
         except Exception:
             import traceback

dags/cluster/tasks/business_logic/cluster_acteurs_suggestions/to_db.py

@@ -18,17 +18,11 @@ def cluster_acteurs_suggestions_to_db(
     suggestions: list[dict],
     identifiant_action: str,
     identifiant_execution: str,
+    # For context purposes only
     cluster_fields_exact: list[str],
     cluster_fields_fuzzy: list[str],
 ) -> None:
-    """Writing suggestions to DB
-
-    Args:
-        df_clusters (pd.DataFrame): clusters for metadata purposes
-        suggestions (list[dict]): suggestions, 1 per cluster
-        identifiant_action (str): ex: Airflow dag_id
-        identifiant_execution (str): ex: Airflow run_id
-    """
+    """Writing suggestions to DB"""
 
     logger.info(f"{identifiant_action=}")
     logger.info(f"{identifiant_execution=}")
@@ -50,6 +44,10 @@ def cluster_acteurs_suggestions_to_db(
     cohorte.save()
 
     for sugg_dict in suggestions:
+        # TODO: refactor this DAG to generate suggestions & context
+        # at the same time like we do for latest DAGs (BAN, AE etc..)
+        # which would allow getting all data from context and not mismatch
+        # context & changes
         cluster_id = sugg_dict["cluster_id"]
         logging.info(f"suggestion {cluster_id=}")
         df_cluster = df_clusters[df_clusters["cluster_id"] == cluster_id]
@@ -59,18 +57,20 @@ def cluster_acteurs_suggestions_to_db(
             msg = "Cluster vide = présent en suggestion mais plus dans df_clusters!!!"
             raise ValueError(msg)
 
+        # TODO: do the context + suggestion changes generation at once
+        # just like for latest DAGs (BAN, AE etc..) to avoid mismatch
+        # and hacks like the one below
+        contexte = suggestion_context_generate(
+            df_cluster=df_cluster,
+            cluster_fields_exact=cluster_fields_exact,
+            cluster_fields_fuzzy=cluster_fields_fuzzy,
+        )
+        if contexte is None:
+            continue
         sugg_obj = Suggestion(
             suggestion_cohorte=cohorte,
             statut=SuggestionStatut.AVALIDER,
-            # FIXME:
-            # This is causing serialization issues due to certain values
-            # being of Django type (e.g. ActeurType)
-            # contexte=cluster.to_dict(orient="records"),
-            contexte=suggestion_context_generate(
-                df_cluster=df_cluster,
-                cluster_fields_exact=cluster_fields_exact,
-                cluster_fields_fuzzy=cluster_fields_fuzzy,
-            ),
+            contexte=contexte,
             suggestion=sugg_dict,
         )
         log.preview(f"{cluster_id=} suggestion dict", sugg_dict)

dags/crawl/config/constants.py

@@ -6,6 +6,5 @@
     COLS.URLS_TO_TRY,
 ]
 
-LABEL_SCENARIO = "Scénario"
 LABEL_URL_ORIGINE = "URL d'origine"
 LABEL_URL_PROPOSEE = "URL proposée"

dags/crawl/tasks/business_logic/crawl_urls_suggest.py

@@ -2,7 +2,7 @@
 
 import pandas as pd
 from crawl.config.columns import COLS
-from crawl.config.constants import LABEL_SCENARIO, LABEL_URL_ORIGINE, LABEL_URL_PROPOSEE
+from crawl.config.constants import LABEL_URL_ORIGINE, LABEL_URL_PROPOSEE
 from utils import logging_utils as log
 from utils.dataframes import (
     df_col_assert_get_unique,
@@ -70,14 +70,11 @@ def suggestions_prepare(
         suggestions.append(
             {
                 "contexte": {
-                    LABEL_SCENARIO: row[COLS.COHORT],
                     LABEL_URL_ORIGINE: url_original,
                     LABEL_URL_PROPOSEE: url_proposed,
                 },
                 "suggestion": {
-                    "scenario": row[COLS.COHORT],
-                    "url_original": url_original,
-                    "url_proposed": url_proposed,
+                    "title": row[COLS.COHORT],
                     "changes": changes,
                 },
             }
@@ -86,9 +83,9 @@ def suggestions_prepare(
     logging.info(log.banner_string(f"🏁 Résultat pour {cohorte=}"))
     logger.info(f"Suggestion générées: {len(suggestions)}")
     for s in suggestions:
-        scenario = s["contexte"]["Scénario"]
+        title = s["suggestion"]["title"]
         url = s["contexte"][LABEL_URL_ORIGINE]
-        log.preview(f"{scenario}: {url=}", s)
+        log.preview(f"{title}: {url=}", s)
 
     return suggestions
 
@@ -99,14 +96,7 @@ def crawl_urls_suggestions_to_db(
     identifiant_action: str,
     identifiant_execution: str,
 ) -> None:
-    """Writing suggestions to DB
-
-    Args:
-        df_clusters (pd.DataFrame): clusters for metadata purposes
-        suggestions (list[dict]): suggestions, 1 per cluster
-        identifiant_action (str): ex: Airflow dag_id
-        identifiant_execution (str): ex: Airflow run_id
-    """
+    """Writing suggestions to DB"""
 
     logger.info(f"{identifiant_action=}")
     logger.info(f"{identifiant_execution=}")

dags/enrich/tasks/business_logic/enrich_dbt_model_to_suggestions.py

@@ -11,6 +11,8 @@
 )
 from utils import logging_utils as log
 
+from data.models.changes.acteur_rgpd_anonymize import rgpd_data_get
+
 logger = logging.getLogger(__name__)
 
 
@@ -55,7 +57,12 @@ def changes_prepare_villes(row: dict) -> tuple[list[dict], dict]:
             entity_type="acteur_displayed",
         )
     )
-    contexte = {}  # changes are self-explanatory
+    contexte = {
+        "statut": row[COLS.ACTEUR_STATUT],
+        "adresse": row[COLS.ACTEUR_ADRESSE],
+        "ville": row[COLS.ACTEUR_VILLE],
+        "code_postal": row[COLS.ACTEUR_CODE_POSTAL],
+    }
     return changes, contexte
 
 
@@ -68,6 +75,7 @@ def changes_prepare_rgpd(
     changes = []
     model_params = {
         "id": row[COLS.ACTEUR_ID],
+        "data": rgpd_data_get(),
     }
     changes.append(
         changes_prepare(
@@ -96,7 +104,6 @@ def changes_prepare_closed_not_replaced(
     model_params = {
         "id": row[COLS.ACTEUR_ID],
         "data": {
-            "identifiant_unique": row[COLS.ACTEUR_ID],
             "statut": ActeurStatus.INACTIF,
             "siret": row[COLS.ACTEUR_SIRET],
             "siren": row[COLS.ACTEUR_SIRET][:9],
@@ -134,7 +141,6 @@ def changes_prepare_closed_replaced(
     # Parent
     parent_id = parent_id_generate([str(row[COLS.SUGGEST_SIRET])])
     parent_data = dbt_model_row_to_suggest_data(row)
-    parent_data["identifiant_unique"] = parent_id
     parent_data["source"] = None
     parent_data["statut"] = ActeurStatus.ACTIF
     params_parent = {
@@ -158,7 +164,6 @@ def changes_prepare_closed_replaced(
     child_new_id = f"{row[COLS.ACTEUR_ID]}_{row[COLS.ACTEUR_SIRET]}_{now}"
     params_child_new = params_parent.copy()
     params_child_new["id"] = child_new_id
-    params_child_new["data"]["identifiant_unique"] = child_new_id
     params_child_new["data"]["source"] = row[COLS.ACTEUR_SOURCE_ID]
     params_child_new["data"]["parent"] = parent_id
     params_child_new["data"]["parent_reason"] = (
@@ -180,7 +185,6 @@ def changes_prepare_closed_replaced(
     # Existing Child
     params_child_old = params_child_new.copy()
     params_child_old["id"] = row[COLS.ACTEUR_ID]
-    params_child_old["data"]["identifiant_unique"] = row[COLS.ACTEUR_ID]
     params_child_old["data"]["parent"] = parent_id
     params_child_old["data"]["parent_reason"] = (
         f"SIRET {row[COLS.ACTEUR_SIRET]} "

dags_unit_tests/cluster/tasks/business_logic/test_cluster_acteurs_suggestions_contexte_generate.py

@@ -44,13 +44,7 @@ def working(self, df_cluster) -> dict:
     def test_working_structure(self, working, df_cluster):
         assert isinstance(working, dict)
         assert sorted(list(working.keys())) == sorted(
-            [
-                # By definition there is only 1 exact match
-                # per cluster
-                "exact_match",
-                # And multiple fuzzy entries
-                "fuzzy_details",
-            ]
+            ["cluster_id", "exact_match", "fuzzy_details"]
         )
         # Parent to be created should not be in fuzzy details
         assert len(working["fuzzy_details"]) == len(df_cluster) - 1

dags_unit_tests/cluster/tasks/business_logic/test_cluster_acteurs_suggestions_prepare.py

@@ -142,10 +142,7 @@ def failing(self, suggestions):
     def test_structure_and_type(self, working):
         assert isinstance(working, list)
         assert isinstance(working[0], dict)
-        assert list(working[0].keys()) == ["cluster_id", "changes"]
-        assert isinstance(working[0]["cluster_id"], str)
-        assert isinstance(working[0]["changes"], list)
-        assert isinstance(working[0]["changes"][0], dict)
+        assert list(working[0].keys()) == ["title", "cluster_id", "changes"]
 
     def test_one_suggestion_per_cluster(self, df_clusters, working):
         # 1 suggestion per cluster EXCEPT for failing c4

dags_unit_tests/cluster/tasks/business_logic/test_cluster_acteurs_suggestions_to_db_and_apply.py

@@ -320,7 +320,7 @@ def test_each_suggestion_has_one_change_per_acteur(self, df, suggestions_from_db
         # Going through each suggestion to test its content
         cluster_ids_found = set()
         for s in suggestions_from_db:
-            cluster_id = s.suggestion["cluster_id"]
+            cluster_id = s.contexte["cluster_id"]
             changes = s.suggestion["changes"]
             cluster_ids_found.add(cluster_id)
 

dags_unit_tests/enrich/tasks/test_enrich_suggestions_cities.py

@@ -25,6 +25,8 @@ def df_new(self):
                 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_ADRESSE: ["1 rue", "2 rue", "3 rue"],
+                COLS.ACTEUR_CODE_POSTAL: ["12345", "67890", "54321"],
                 COLS.ACTEUR_VILLE: ["old town 1", "old town 2", "closed"],
                 COLS.ACTEUR_STATUT: ["ACTIF", "ACTIF", "INACTIF"],
             }
@@ -38,6 +40,8 @@ def df_typo(self):
                 COLS.SUGGEST_COHORT: [COHORTS.VILLES_TYPO] * 3,
                 COLS.SUGGEST_VILLE: ["Paris", "Laval", "closed"],
                 COLS.ACTEUR_ID: ["typo1", "typo2", "closed 2"],
+                COLS.ACTEUR_ADRESSE: ["1 rue", "2 rue", "3 rue"],
+                COLS.ACTEUR_CODE_POSTAL: ["12345", "67890", "54321"],
                 COLS.ACTEUR_VILLE: ["Pâris", "Lâval", "closed"],
                 COLS.ACTEUR_STATUT: ["ACTIF", "ACTIF", "INACTIF"],
             }