🌆 [ENRICH] BAN: villes anciennes -> nouvelles

dags/enrich/config/__init__.py

@@ -5,6 +5,7 @@
     DAG_ID_TO_CONFIG_MODEL,
     EnrichActeursClosedConfig,
     EnrichActeursRGPDConfig,
+    EnrichActeursVillesConfig,
     EnrichDbtModelsRefreshConfig,
 )
 from .tasks import TASKS  # noqa: F401

dags/enrich/config/cohorts.py

@@ -11,3 +11,5 @@ class COHORTS:
     )
     CLOSED_REP_SAME_SIREN = "🚪 Acteurs Fermés: 🟢 remplacés par SIRET du même SIREN"
     RGPD = "Anonymisation RGPD"
+    VILLES_TYPO = "🌆 Changement de ville: 🟢 variation d'ortographe"
+    VILLES_NEW = "🌆 Changement de ville: 🟡 ancienne -> nouvelle"

dags/enrich/config/columns.py

@@ -13,9 +13,7 @@ class COLS:
     # Acteurs
     ACTEUR_ID: str = "acteur_id"
     ACTEUR_TYPE_ID: str = "acteur_type_id"
-    ACTEUR_TYPE_CODE: str = "acteur_type_code"
     ACTEUR_SOURCE_ID: str = "acteur_source_id"
-    ACTEUR_SOURCE_CODE: str = "acteur_source_code"
     ACTEUR_SIRET: str = "acteur_siret"
     ACTEUR_NOM: str = "acteur_nom"
     ACTEUR_NOMS_ORIGINE: str = "acteur_noms_origine"
@@ -43,6 +41,18 @@ class COLS:
     SUGGEST_VILLE: str = "suggest_ville"
     SUGGEST_NAF: str = "suggest_naf"
 
+    # Suggestions
+    SUGGEST_COHORT_CODE: str = "suggestion_cohorte_code"
+    SUGGEST_COHORT: str = "suggest_cohort"
+
+    # Replacements
+    SUGGEST_SIRET: str = "suggest_siret"
+    SUGGEST_NOM: str = "suggest_nom"
+    SUGGEST_ADRESSE: str = "suggest_adresse"
+    SUGGEST_CODE_POSTAL: str = "suggest_code_postal"
+    SUGGEST_VILLE: str = "suggest_ville"
+    SUGGEST_NAF: str = "suggest_naf"
+
     # Suggestions
     SUGGEST_COHORT: str = f"{SUGGEST_PREFIX}_cohort"
     SUGGEST_SIRET: str = f"{SUGGEST_PREFIX}_siret"

dags/enrich/config/dbt.py

@@ -19,4 +19,5 @@ class DBT:
     MARTS_ENRICH_AE_CLOSED_NOT_REPLACED: str = (
         "marts_enrich_acteurs_closed_suggest_not_replaced"
     )
-    MARTS_ENRICH_AE_RGPD: str = "marts_enrich_ae_rgpd"
+    MARTS_ENRICH_VILLES_TYPO: str = "marts_enrich_acteurs_villes_suggest_typo"
+    MARTS_ENRICH_VILLES_NEW: str = "marts_enrich_acteurs_villes_suggest_new"

dags/enrich/config/models.py

@@ -100,8 +100,13 @@ class EnrichDbtModelsRefreshConfig(BaseModel):
     )
 
 
+class EnrichActeursVillesConfig(EnrichBaseConfig):
+    pass
+
+
 DAG_ID_TO_CONFIG_MODEL = {
     "enrich_acteurs_closed": EnrichActeursClosedConfig,
     "enrich_acteurs_rgpd": EnrichActeursRGPDConfig,
     "enrich_dbt_models_refresh": EnrichDbtModelsRefreshConfig,
+    "enrich_acteurs_villes": EnrichActeursVillesConfig,
 }

dags/enrich/config/tasks.py

@@ -28,3 +28,7 @@ class TASKS:
     )
     ENRICH_DBT_MODELS_REFRESH: str = "enrich_dbt_models_refresh"
     READ_AE_RGPD: str = "enrich_ae_rgpd_read"
+
+    # Villes
+    ENRICH_VILLES_TYPO: str = "enrich_acteurs_villes_typo"
+    ENRICH_VILLES_NEW: str = "enrich_acteurs_villes_new"

dags/enrich/dags/enrich_acteurs_villes.py

@@ -0,0 +1,63 @@
+"""DAG to suggestion city corrections based on BAN data"""
+
+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 suggérer des corrections de villes"),
+    tags=["annuaire", "entreprises", "ae", "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"
+            ),
+            filter_equals__acteur_statut="ACTIF",
+        )
+    ),
+) 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_VILLES_TYPO,
+        cohort=COHORTS.VILLES_TYPO,
+        dbt_model_name=DBT.MARTS_ENRICH_VILLES_TYPO,
+    )
+    suggest_new = enrich_dbt_model_suggest_task(
+        dag,
+        task_id=TASKS.ENRICH_VILLES_NEW,
+        cohort=COHORTS.VILLES_NEW,
+        dbt_model_name=DBT.MARTS_ENRICH_VILLES_NEW,
+    )
+    config >> dbt_refresh  # type: ignore
+    dbt_refresh >> suggest_typo  # type: ignore
+    dbt_refresh >> suggest_new  # type: ignore

dags/enrich/dags/enrich_dbt_models_refresh.py

@@ -30,10 +30,8 @@
     params=config_to_airflow_params(
         EnrichDbtModelsRefreshConfig(
             dbt_models_refresh_commands=[
-                "dbt build --select +int_ae_unite_legale",
-                "dbt build --select +int_ae_etablissement",
-                "dbt build --select +int_ban_adresses",
-                "dbt build --select int_ban_villes",
+                "dbt build --select +tag:intermediate,tag:ae",
+                "dbt build --select +tag:intermediate,tag:ban",
             ],
         )
     ),

dags/enrich/tasks/business_logic/enrich_dbt_model_to_suggestions.py

@@ -35,6 +35,30 @@ def changes_prepare(
     ).model_dump()
 
 
+def changes_prepare_villes(row: dict) -> tuple[list[dict], 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",
+        )
+    )
+    contexte = {}  # changes are self-explanatory
+    return changes, contexte
+
+
 def changes_prepare_rgpd(
     row: dict,
 ) -> tuple[list[dict], dict]:
@@ -184,6 +208,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.VILLES_TYPO: changes_prepare_villes,
+    COHORTS.VILLES_NEW: changes_prepare_villes,
 }
 
 
@@ -200,11 +226,17 @@ def enrich_dbt_model_to_suggestions(
         SuggestionStatut,
     )
 
+    # TODO: once all suggestions have been migrated to pydantic, we no
+    # longer need SuggestionCohorte.type_action and any of the following
+    # identifiant_execution = cohort AND pydantic models take care of
+    # handling the specifics
     COHORTS_TO_SUGGESTION_ACTION = {
         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.VILLES_TYPO: SuggestionAction.ENRICH_ACTEURS_VILLES_TYPO,
+        COHORTS.VILLES_NEW: SuggestionAction.ENRICH_ACTEURS_VILLES_NEW,
     }
 
     # Validation
@@ -223,12 +255,12 @@ def enrich_dbt_model_to_suggestions(
 
         try:
             changes, contexte = COHORTS_TO_PREPARE_CHANGES[cohort](row)
-            suggestions.append(
-                {
-                    "contexte": contexte,
-                    "suggestion": {"title": cohort, "changes": changes},
-                }
-            )
+            suggestion = {
+                "contexte": contexte,
+                "suggestion": {"title": cohort, "changes": changes},
+            }
+            log.preview("🔢 Suggestion", suggestion)
+            suggestions.append(suggestion)
 
         # We tolerate some errors
         except Exception as e:

dags_unit_tests/enrich/config/test_enrich_acteurs_closed_config.py

@@ -1,5 +1,5 @@
 import pytest
-from enrich.config.models import EnrichActeursClosedConfig
+from dags.enrich.config.models import EnrichActeursClosedConfig
 
 
 class TestEnrichClosedConfig:

dags_unit_tests/enrich/tasks/test_enrich_suggestions_cities.py

@@ -0,0 +1,100 @@
+import pandas as pd
+import pytest
+from enrich.config import COHORTS, COLS
+from enrich.tasks.business_logic.enrich_dbt_model_to_suggestions import (
+    enrich_dbt_model_to_suggestions,
+)
+
+
+@pytest.mark.django_db
+class TestEnrichSuggestionsCities:
+
+    @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"],
+            }
+        )
+
+    @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"],
+            }
+        )
+
+    @pytest.fixture
+    def acteurs(self, df_new, df_typo):
+        # 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():
+            ActeurFactory(
+                identifiant_unique=row[COLS.ACTEUR_ID],
+                ville=row[COLS.ACTEUR_VILLE],
+            )
+
+    def test_cohort_new(self, acteurs, df_new):
+        from data.models.suggestion import Suggestion, SuggestionCohorte
+        from qfdmo.models import RevisionActeur
+
+        # Write suggestions to DB
+        enrich_dbt_model_to_suggestions(
+            df=df_new,
+            cohort=COHORTS.VILLES_NEW,
+            identifiant_action="test_new",
+            dry_run=False,
+        )
+
+        # Check suggestions have been written to DB
+        cohort = SuggestionCohorte.objects.get(identifiant_action="test_new")
+        suggestions = Suggestion.objects.filter(suggestion_cohorte=cohort)
+        assert len(suggestions) == 2
+
+        # Apply suggestions
+        for suggestion in suggestions:
+            suggestion.apply()
+
+        # Verify changes
+        # 2 revisions should be created but not parent
+        new1 = RevisionActeur.objects.get(pk="new1")
+        assert new1.ville == "new town 1"
+
+        new2 = RevisionActeur.objects.get(pk="new2")
+        assert new2.ville == "new town 2"
+
+    def test_cohort_typo(self, acteurs, df_typo):
+        from data.models.suggestion import Suggestion, SuggestionCohorte
+        from qfdmo.models import RevisionActeur
+
+        # Write suggestions to DB
+        enrich_dbt_model_to_suggestions(
+            df=df_typo,
+            cohort=COHORTS.VILLES_TYPO,
+            identifiant_action="test_typo",
+            dry_run=False,
+        )
+
+        # Check suggestions have been written to DB
+        cohort = SuggestionCohorte.objects.get(identifiant_action="test_typo")
+        suggestions = Suggestion.objects.filter(suggestion_cohorte=cohort)
+        assert len(suggestions) == 2
+
+        # Apply suggestions
+        for suggestion in suggestions:
+            suggestion.apply()
+
+        # Verify changes
+        typo1 = RevisionActeur.objects.get(pk="typo1")
+        assert typo1.ville == "Paris"
+
+        typo2 = RevisionActeur.objects.get(pk="typo2")
+        assert typo2.ville == "Laval"

data/migrations/0013_alter_suggestioncohorte_type_action.py

@@ -0,0 +1,35 @@
+# Generated by Django 5.1.6 on 2025-04-28 05:22
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("data", "0012_alter_suggestioncohorte_type_action"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="suggestioncohorte",
+            name="type_action",
+            field=models.CharField(
+                blank=True,
+                choices=[
+                    ("CRAWL_URLS", "🔗 URLs scannées"),
+                    ("ENRICH_ACTEURS_CLOSED", "🚪 Acteurs fermés"),
+                    ("ENRICH_ACTEURS_RGPD", "🕵 Anonymisation RGPD"),
+                    ("ENRICH_ACTEURS_VILLES_TYPO", "🏙️ Acteurs villes typographiques"),
+                    ("ENRICH_ACTEURS_VILLES_NEW", "🏙️ Acteurs villes nouvelles"),
+                    ("CLUSTERING", "regroupement/déduplication des acteurs"),
+                    ("SOURCE_AJOUT", "ingestion de source de données - nouveau acteur"),
+                    (
+                        "SOURCE_MODIFICATION",
+                        "ingestion de source de données - modification d'acteur existant",
+                    ),
+                    ("SOURCE_SUPRESSION", "ingestion de source de données"),
+                ],
+                max_length=50,
+            ),
+        ),
+    ]

data/models/changes/__init__.py

@@ -10,7 +10,6 @@
 from .sample_model_do_nothing import SampleModelDoNothing
 
 CHANGE_MODELS = {
-    ChangeActeurRgpdAnonymize.name(): ChangeActeurRgpdAnonymize,
     ChangeActeurUpdateData.name(): ChangeActeurUpdateData,
     ChangeActeurCreateAsChild.name(): ChangeActeurCreateAsChild,
     ChangeActeurCreateAsParent.name(): ChangeActeurCreateAsParent,
@@ -20,4 +19,5 @@
     ChangeActeurNothingBase.name(): ChangeActeurNothingBase,
     ChangeActeurKeepAsParent.name(): ChangeActeurKeepAsParent,
     SampleModelDoNothing.name(): SampleModelDoNothing,
+    ChangeActeurRgpdAnonymize.name(): ChangeActeurRgpdAnonymize,
 }