Merge pull request #1460 from MTES-MCT/feat-insee-data

Ingest du dossier complet insee

Author: alexisig

airflow/dags/ingest_dossier_complet.py

@@ -0,0 +1,74 @@
+import os
+
+import pandas as pd
+from include.container import DomainContainer as Container
+from include.container import InfraContainer
+from pendulum import datetime
+
+from airflow.decorators import dag, task
+
+URL = "https://www.insee.fr/fr/statistiques/fichier/5359146/dossier_complet.zip"
+
+
+@dag(
+    start_date=datetime(2024, 1, 1),
+    schedule="@once",
+    catchup=False,
+    doc_md=__doc__,
+    max_active_runs=1,
+    default_args={"owner": "Alexis Athlani", "retries": 3},
+    tags=["INSEE"],
+)
+def ingest_dossier_complet():
+    bucket_name = InfraContainer().bucket_name()
+    s3_key = "insee/dossier_complet.csv"
+
+    @task.python
+    def download() -> str:
+        return (
+            Container()
+            .remote_zip_to_s3_file_handler()
+            .download_zip_extract_and_upload_to_s3(
+                url=URL,
+                s3_key=s3_key,
+                s3_bucket=bucket_name,
+                target_extension=".csv",
+            )
+        )
+
+    @task.python
+    def ingest() -> int | None:
+        """Unpivot le CSV (1900+ colonnes) en format EAV (codgeo, variable, value)
+        pour contourner la limite de 1600 colonnes PostgreSQL."""
+        s3_path = f"{bucket_name}/{s3_key}"
+        tmp_localpath = "/tmp/dossier_complet.csv"
+
+        InfraContainer().s3().get_file(s3_path, tmp_localpath)
+
+        engine = InfraContainer().sqlalchemy_dbt_conn()
+        table_name = "insee_dossier_complet"
+        chunk_size = 5000
+        total_rows = 0
+
+        for i, chunk in enumerate(pd.read_csv(tmp_localpath, sep=";", dtype=str, chunksize=chunk_size)):
+            melted = chunk.melt(
+                id_vars=["CODGEO"],
+                var_name="variable",
+                value_name="value",
+            )
+            melted = melted.dropna(subset=["value"])
+            row_count = melted.to_sql(
+                name=table_name,
+                con=engine,
+                if_exists="replace" if i == 0 else "append",
+                index=False,
+            )
+            total_rows += row_count if row_count else len(melted)
+
+        os.remove(tmp_localpath)
+        return total_rows
+
+    download() >> ingest()
+
+
+ingest_dossier_complet()

airflow/dags/update_app.py

@@ -154,6 +154,28 @@ def copy_table_from_dw_to_app(
                 "copy_public_data_landfriche",
                 "copy_public_data_landfrichegeojson",
                 "copy_public_data_landcarroyagebounds",
+                "copy_public_data_dc_population",
+                "copy_public_data_dc_menages",
+                "copy_public_data_dc_logement",
+                "copy_public_data_dc_mobilite_residentielle",
+                "copy_public_data_dc_categories_socioprofessionnelles",
+                "copy_public_data_dc_situation_conjugale",
+                "copy_public_data_dc_scolarisation_diplomes",
+                "copy_public_data_dc_emploi_statut",
+                "copy_public_data_dc_deplacements_domicile_travail",
+                "copy_public_data_dc_activite_chomage",
+                "copy_public_data_dc_emplois_lieu_travail",
+                "copy_public_data_dc_logement_confort",
+                "copy_public_data_dc_historique",
+                "copy_public_data_dc_revenus_pauvrete",
+                "copy_public_data_dc_salaires",
+                "copy_public_data_dc_etablissements",
+                "copy_public_data_dc_creations_entreprises",
+                "copy_public_data_dc_creations_etablissements",
+                "copy_public_data_dc_unites_legales_actives",
+                "copy_public_data_dc_tourisme",
+                "copy_public_data_dc_electeurs",
+                "copy_public_data_dc_equipements_bpe",
             ],
             type="array",
         ),
@@ -721,6 +743,203 @@ def copy_public_data_landcarroyagebounds(**context):
                 ["land_id", "land_type", "start_year", "end_year", "destination"],
             ],
         )
+    @task.python
+    def copy_public_data_dc_population(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_dc_population",
+            to_table="public.public_data_dc_population",
+            environment=context["params"]["environment"],
+            btree_index_columns=[["land_id", "land_type"]],
+        )
+
+    @task.python
+    def copy_public_data_dc_menages(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_dc_menages",
+            to_table="public.public_data_dc_menages",
+            environment=context["params"]["environment"],
+            btree_index_columns=[["land_id", "land_type"]],
+        )
+
+    @task.python
+    def copy_public_data_dc_logement(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_dc_logement",
+            to_table="public.public_data_dc_logement",
+            environment=context["params"]["environment"],
+            btree_index_columns=[["land_id", "land_type"]],
+        )
+
+    @task.python
+    def copy_public_data_dc_mobilite_residentielle(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_dc_mobilite_residentielle",
+            to_table="public.public_data_dc_mobilite_residentielle",
+            environment=context["params"]["environment"],
+            btree_index_columns=[["land_id", "land_type"]],
+        )
+
+    @task.python
+    def copy_public_data_dc_categories_socioprofessionnelles(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_dc_categories_socioprofessionnelles",
+            to_table="public.public_data_dc_categories_socioprofessionnelles",
+            environment=context["params"]["environment"],
+            btree_index_columns=[["land_id", "land_type"]],
+        )
+
+    @task.python
+    def copy_public_data_dc_situation_conjugale(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_dc_situation_conjugale",
+            to_table="public.public_data_dc_situation_conjugale",
+            environment=context["params"]["environment"],
+            btree_index_columns=[["land_id", "land_type"]],
+        )
+
+    @task.python
+    def copy_public_data_dc_scolarisation_diplomes(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_dc_scolarisation_diplomes",
+            to_table="public.public_data_dc_scolarisation_diplomes",
+            environment=context["params"]["environment"],
+            btree_index_columns=[["land_id", "land_type"]],
+        )
+
+    @task.python
+    def copy_public_data_dc_emploi_statut(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_dc_emploi_statut",
+            to_table="public.public_data_dc_emploi_statut",
+            environment=context["params"]["environment"],
+            btree_index_columns=[["land_id", "land_type"]],
+        )
+
+    @task.python
+    def copy_public_data_dc_deplacements_domicile_travail(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_dc_deplacements_domicile_travail",
+            to_table="public.public_data_dc_deplacements_domicile_travail",
+            environment=context["params"]["environment"],
+            btree_index_columns=[["land_id", "land_type"]],
+        )
+
+    @task.python
+    def copy_public_data_dc_activite_chomage(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_dc_activite_chomage",
+            to_table="public.public_data_dc_activite_chomage",
+            environment=context["params"]["environment"],
+            btree_index_columns=[["land_id", "land_type"]],
+        )
+
+    @task.python
+    def copy_public_data_dc_emplois_lieu_travail(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_dc_emplois_lieu_travail",
+            to_table="public.public_data_dc_emplois_lieu_travail",
+            environment=context["params"]["environment"],
+            btree_index_columns=[["land_id", "land_type"]],
+        )
+
+    @task.python
+    def copy_public_data_dc_logement_confort(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_dc_logement_confort",
+            to_table="public.public_data_dc_logement_confort",
+            environment=context["params"]["environment"],
+            btree_index_columns=[["land_id", "land_type"]],
+        )
+
+    @task.python
+    def copy_public_data_dc_historique(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_dc_historique",
+            to_table="public.public_data_dc_historique",
+            environment=context["params"]["environment"],
+            btree_index_columns=[["land_id", "land_type"]],
+        )
+
+    @task.python
+    def copy_public_data_dc_revenus_pauvrete(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_dc_revenus_pauvrete",
+            to_table="public.public_data_dc_revenus_pauvrete",
+            environment=context["params"]["environment"],
+            btree_index_columns=[["land_id", "land_type"]],
+        )
+
+    @task.python
+    def copy_public_data_dc_salaires(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_dc_salaires",
+            to_table="public.public_data_dc_salaires",
+            environment=context["params"]["environment"],
+            btree_index_columns=[["land_id", "land_type"]],
+        )
+
+    @task.python
+    def copy_public_data_dc_etablissements(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_dc_etablissements",
+            to_table="public.public_data_dc_etablissements",
+            environment=context["params"]["environment"],
+            btree_index_columns=[["land_id", "land_type"]],
+        )
+
+    @task.python
+    def copy_public_data_dc_creations_entreprises(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_dc_creations_entreprises",
+            to_table="public.public_data_dc_creations_entreprises",
+            environment=context["params"]["environment"],
+            btree_index_columns=[["land_id", "land_type"]],
+        )
+
+    @task.python
+    def copy_public_data_dc_creations_etablissements(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_dc_creations_etablissements",
+            to_table="public.public_data_dc_creations_etablissements",
+            environment=context["params"]["environment"],
+            btree_index_columns=[["land_id", "land_type"]],
+        )
+
+    @task.python
+    def copy_public_data_dc_unites_legales_actives(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_dc_unites_legales_actives",
+            to_table="public.public_data_dc_unites_legales_actives",
+            environment=context["params"]["environment"],
+            btree_index_columns=[["land_id", "land_type"]],
+        )
+
+    @task.python
+    def copy_public_data_dc_tourisme(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_dc_tourisme",
+            to_table="public.public_data_dc_tourisme",
+            environment=context["params"]["environment"],
+            btree_index_columns=[["land_id", "land_type"]],
+        )
+
+    @task.python
+    def copy_public_data_dc_electeurs(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_dc_electeurs",
+            to_table="public.public_data_dc_electeurs",
+            environment=context["params"]["environment"],
+            btree_index_columns=[["land_id", "land_type"]],
+        )
+
+    @task.python
+    def copy_public_data_dc_equipements_bpe(**context):
+        return copy_table_from_dw_to_app(
+            from_table="public_for_app.for_app_dc_equipements_bpe",
+            to_table="public.public_data_dc_equipements_bpe",
+            environment=context["params"]["environment"],
+            btree_index_columns=[["land_id", "land_type"]],
+        )
 
     @task.branch
     def copy_public_data_branch(**context):
@@ -775,6 +994,28 @@ def copy_public_data_branch(**context):
         copy_public_data_landfriche(),
         copy_public_data_landfrichegeojson(),
         copy_public_data_landcarroyagebounds(),
+        copy_public_data_dc_population(),
+        copy_public_data_dc_menages(),
+        copy_public_data_dc_logement(),
+        copy_public_data_dc_mobilite_residentielle(),
+        copy_public_data_dc_categories_socioprofessionnelles(),
+        copy_public_data_dc_situation_conjugale(),
+        copy_public_data_dc_scolarisation_diplomes(),
+        copy_public_data_dc_emploi_statut(),
+        copy_public_data_dc_deplacements_domicile_travail(),
+        copy_public_data_dc_activite_chomage(),
+        copy_public_data_dc_emplois_lieu_travail(),
+        copy_public_data_dc_logement_confort(),
+        copy_public_data_dc_historique(),
+        copy_public_data_dc_revenus_pauvrete(),
+        copy_public_data_dc_salaires(),
+        copy_public_data_dc_etablissements(),
+        copy_public_data_dc_creations_entreprises(),
+        copy_public_data_dc_creations_etablissements(),
+        copy_public_data_dc_unites_legales_actives(),
+        copy_public_data_dc_tourisme(),
+        copy_public_data_dc_electeurs(),
+        copy_public_data_dc_equipements_bpe(),
     ]