dag ariflow

Author: maxcorbeau

dags/enrich/config/dbt.py

@@ -19,3 +19,5 @@ class DBT:
     MARTS_ENRICH_AE_CLOSED_NOT_REPLACED: str = (
         "marts_enrich_acteurs_closed_suggest_not_replaced"
     )
+    MARTS_ENRICH_ACTEURS_VILLES_TYPO: str = "marts_enrich_acteurs_villes_suggest_typo"
+    MARTS_ENRICH_ACTEURS_VILLES_NEW: str = "marts_enrich_acteurs_villes_suggest_new"

dags/enrich/config/models.py

@@ -93,7 +93,12 @@ class EnrichActeursRGPDConfig(EnrichBaseConfig):
     )
 
 
+class EnrichActeursVillesConfig(EnrichBaseConfig):
+    pass
+
+
 DAG_ID_TO_CONFIG_MODEL = {
     "enrich_acteurs_closed": EnrichActeursClosedConfig,
     "enrich_acteurs_rgpd": EnrichActeursRGPDConfig,
+    "enrich_acteurs_villes": EnrichActeursVillesConfig,
 }

dags/enrich/config/tasks.py

@@ -27,3 +27,7 @@ class TASKS:
         "enrich_acteurs_closed_suggestions_not_replaced"
     )
     ENRICH_DBT_MODELS_REFRESH: str = "enrich_dbt_models_refresh"
+
+    # Villes
+    ENRICH_ACTEURS_VILLES_TYPO: str = "enrich_acteurs_villes_typo"
+    ENRICH_ACTEURS_VILLES_NEW: str = "enrich_acteurs_villes_new"

dags/enrich/dags/enrich_acteurs_villes.py

@@ -0,0 +1,62 @@
+"""DAG to anonymize QFDMO acteurs for RGPD"""
+
+from airflow import DAG
+from enrich.config import (
+    COHORTS,
+    DBT,
+    TASKS,
+    EnrichActeursVillesConfig,
+)
+from enrich.tasks.airflow_logic.enrich_config_create_task import (
+    enrich_config_create_task,
+)
+from enrich.tasks.airflow_logic.enrich_dbt_model_suggest_task import (
+    enrich_dbt_model_suggest_task,
+)
+from enrich.tasks.airflow_logic.enrich_dbt_models_refresh_task import (
+    enrich_dbt_models_refresh_task,
+)
+from shared.config import CATCHUPS, SCHEDULES, START_DATES, config_to_airflow_params
+
+with DAG(
+    dag_id="enrich_acteurs_villes",
+    dag_display_name="🌆 Enrichir - Acteurs Villes",
+    default_args={
+        "owner": "airflow",
+        "depends_on_past": False,
+        "email_on_failure": False,
+        "email_on_retry": False,
+        "retries": 0,
+    },
+    description=("Un DAG pour anonymiser les acteurs vs. RGPD"),
+    tags=["annuaire", "entreprises", "ae", "rgpd", "acteurs", "juridique"],
+    schedule=SCHEDULES.NONE,
+    catchup=CATCHUPS.AWLAYS_FALSE,
+    start_date=START_DATES.YESTERDAY,
+    params=config_to_airflow_params(
+        EnrichActeursVillesConfig(
+            dbt_models_refresh=True,
+            dbt_models_refresh_command=(
+                "dbt build --select tag:marts,tag:enrich,tag:villes"
+            ),
+        )
+    ),
+) as dag:
+    # Instantiation
+    config = enrich_config_create_task(dag)
+    dbt_refresh = enrich_dbt_models_refresh_task(dag)
+    suggest_typo = enrich_dbt_model_suggest_task(
+        dag,
+        task_id=TASKS.ENRICH_ACTEURS_VILLES_TYPO,
+        cohort=COHORTS.ACTEURS_VILLES_TYPO,
+        dbt_model_name=DBT.MARTS_ENRICH_ACTEURS_VILLES_TYPO,
+    )
+    suggest_new = enrich_dbt_model_suggest_task(
+        dag,
+        task_id=TASKS.ENRICH_ACTEURS_VILLES_NEW,
+        cohort=COHORTS.ACTEURS_VILLES_NEW,
+        dbt_model_name=DBT.MARTS_ENRICH_ACTEURS_VILLES_NEW,
+    )
+    config >> dbt_refresh  # type: ignore
+    dbt_refresh >> suggest_typo  # type: ignore
+    dbt_refresh >> suggest_new  # type: ignore

dags/enrich/tasks/business_logic/enrich_dbt_model_to_suggestions.py

@@ -61,6 +61,29 @@ def changes_prepare_rgpd(
     return changes, contexte
 
 
+def changes_prepare_villes(row: dict) -> list[dict]:
+    """Prepare suggestions for villes cohorts"""
+    from data.models.changes import ChangeActeurUpdateData
+
+    changes = []
+    model_params = {
+        "id": row[COLS.ACTEUR_ID],
+        "data": {
+            "ville": row[COLS.SUGGEST_VILLE],
+        },
+    }
+    changes.append(
+        changes_prepare(
+            model=ChangeActeurUpdateData,
+            model_params=model_params,
+            order=1,
+            reason="On fait confiance à la BAN",
+            entity_type="acteur_displayed",
+        )
+    )
+    return changes
+
+
 def changes_prepare_closed_not_replaced(
     row: dict,
 ) -> tuple[list[dict], dict]:
@@ -184,6 +207,8 @@ def changes_prepare_closed_replaced(
     COHORTS.CLOSED_REP_OTHER_SIREN: changes_prepare_closed_replaced,
     COHORTS.CLOSED_REP_SAME_SIREN: changes_prepare_closed_replaced,
     COHORTS.RGPD: changes_prepare_rgpd,
+    COHORTS.ACTEURS_VILLES_TYPO: changes_prepare_villes,
+    COHORTS.ACTEURS_VILLES_NEW: changes_prepare_villes,
 }
 
 
@@ -204,7 +229,8 @@ def enrich_dbt_model_to_suggestions(
         COHORTS.CLOSED_NOT_REPLACED: SuggestionAction.ENRICH_ACTEURS_CLOSED,
         COHORTS.CLOSED_REP_OTHER_SIREN: SuggestionAction.ENRICH_ACTEURS_CLOSED,
         COHORTS.CLOSED_REP_SAME_SIREN: SuggestionAction.ENRICH_ACTEURS_CLOSED,
-        COHORTS.RGPD: SuggestionAction.ENRICH_ACTEURS_RGPD,
+        COHORTS.ACTEURS_VILLES_TYPO: SuggestionAction.ENRICH_ACTEURS_VILLES_TYPO,
+        COHORTS.ACTEURS_VILLES_NEW: SuggestionAction.ENRICH_ACTEURS_VILLES_NEW,
     }
 
     # Validation

data/models/suggestion.py

@@ -56,6 +56,14 @@ class SuggestionAction(models.TextChoices):
     CRAWL_URLS = SUGGESTION_CRAWL_URLS, "🔗 URLs scannées"
     ENRICH_ACTEURS_CLOSED = "ENRICH_ACTEURS_CLOSED", "🚪 Acteurs fermés"
     ENRICH_ACTEURS_RGPD = "ENRICH_ACTEURS_RGPD", "🕵 Anonymisation RGPD"
+    ENRICH_ACTEURS_VILLES_TYPO = (
+        "ENRICH_ACTEURS_VILLES_TYPO",
+        "🏙️ Acteurs villes typographiques",
+    )
+    ENRICH_ACTEURS_VILLES_NEW = (
+        "ENRICH_ACTEURS_VILLES_NEW",
+        "🏙️ Acteurs villes nouvelles",
+    )
     CLUSTERING = SUGGESTION_CLUSTERING, "regroupement/déduplication des acteurs"
     SOURCE_AJOUT = (
         SUGGESTION_SOURCE_AJOUT,