dag ariflow

Author: maxcorbeau

dags/enrich/config/__init__.py

@@ -1,7 +1,11 @@
 from .cohorts import COHORTS  # noqa: F401
 from .columns import COLS  # noqa: F401
 from .dbt import DBT  # noqa: F401
-from .models import DAG_ID_TO_CONFIG_MODEL, EnrichActeursClosedConfig  # noqa: F401
+from .models import (  # noqa: F401
+    DAG_ID_TO_CONFIG_MODEL,
+    EnrichActeursClosedConfig,
+    EnrichActeursVillesConfig,
+)
 from .paths import DIR_SQL_READ  # noqa: F401
 from .tasks import TASKS  # noqa: F401
 from .xcoms import XCOMS, xcom_pull  # noqa: F401

dags/enrich/config/cohorts.py

@@ -10,3 +10,5 @@ class COHORTS:
     CLOSED_NOT_REPLACED = f"{CLOSED} 🔴 non remplacés"
     CLOSED_REP_OTHER_SIREN = f"{CLOSED} 🟡 remplacés par SIRET d'un autre SIREN"
     CLOSED_REP_SAME_SIREN = f"{CLOSED} 🟢 remplacés par SIRET du même SIREN"
+    ACTEURS_VILLES_TYPO = "🌆 Changement de ville: 🟢 variation d'ortographe"
+    ACTEURS_VILLES_NEW = "🌆 Changement de ville: 🟡 ancienne -> nouvelle"

dags/enrich/config/dbt.py

@@ -15,3 +15,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

@@ -82,6 +82,11 @@ class EnrichActeursClosedConfig(EnrichBaseConfig):
     )
 
 
+class EnrichActeursVillesConfig(EnrichBaseConfig):
+    pass
+
+
 DAG_ID_TO_CONFIG_MODEL = {
     "enrich_acteurs_closed": EnrichActeursClosedConfig,
+    "enrich_acteurs_villes": EnrichActeursVillesConfig,
 }

dags/enrich/config/tasks.py

@@ -24,3 +24,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

@@ -32,6 +32,29 @@ def changes_prepare(
     ).model_dump()
 
 
+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]:
@@ -186,6 +209,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.ACTEURS_VILLES_TYPO: SuggestionAction.ENRICH_ACTEURS_VILLES_TYPO,
+        COHORTS.ACTEURS_VILLES_NEW: SuggestionAction.ENRICH_ACTEURS_VILLES_NEW,
     }
 
     # Validation

data/models/suggestion.py

@@ -55,6 +55,14 @@ class SuggestionCohorteStatut(models.TextChoices):
 class SuggestionAction(models.TextChoices):
     CRAWL_URLS = SUGGESTION_CRAWL_URLS, "🔗 URLs scannées"
     ENRICH_ACTEURS_CLOSED = "ENRICH_ACTEURS_CLOSED", "🚪 Acteurs fermés"
+    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,