MEP - 8.2.0

.buildpacks

@@ -1,2 +1,3 @@
 https://github.com/Scalingo/apt-buildpack
+https://github.com/Scalingo/nodejs-buildpack
 https://github.com/Scalingo/python-buildpack

.env.example

@@ -36,8 +36,7 @@ EMAIL_SMTP_KEY=ASK_A_MAINTAINER
 EMAIL_HOST_USER=YOUR_EMAIL
 
 MATOMO_ACTIVATE=0
-MATOMO_SCRIPT_NAME=container_mlVvm9TG.js
-MATOMO_TOKEN=TO_FILL
+
 HIGHCHART_SERVER=https://highcharts-export.osc-fr1.scalingo.io
 
 MATTERMOST_WEBHOOK=https://mattermost.incubateur.net/hooks/uak581f8bidyxp5td67rurj5sh

.gitignore

@@ -166,3 +166,7 @@ staticroot/*
 public_data/static/SSCH*
 diff_bfc_*
 .idea
+
+# Webpack bundles
+static/assets/scripts/
+static/assets/styles/*.css.map

.slugignore

@@ -1,2 +1,4 @@
 assets/scripts
 assets/styles
+assets/types
+/node_modules

README.md

@@ -69,8 +69,6 @@ Le site est désormais accessible en local à cette adresse:
 | LOCAL_FILE_DIRECTORY | Emplacement des données locales (utile pour charger des shapefile en local au lieu de S3) | public_data/local_data |
 | MATTERMOST_WEBHOOK | Webhook personnel pour envoyer des messages dans Mattermost | https://mattermost.incubateur.net/hooks/uak581f8bidyxp5td67rurj5sh |
 | MATOMO_ACTIVATE | Détermine si des infos doivent être envoyé à Matomo | 0 |
-| MATOMO_SCRIPT_NAME | | |
-| MATOMO_TOKEN | Token pour envoyer les données à Matomo |  |
 | POSTGRES_DB | Nom de la base de donnée (local uniquement) | postgres |
 | POSTGRES_USER | Username par défaut de la base de donnée (local uniquement) | postgres |
 | POSTGRES_PASSWORD | Password par défaut de la base de donnée (local uniquement) | postgres |

airflow/dags/ingest_admin_express.py

@@ -39,51 +39,46 @@ def get_source_by_name(name: str) -> dict:
             type="string",
             enum=zones,
         ),
-        "refresh_source": Param(
-            default=False,
-            description="Rafraîchir la source",
-            type="boolean",
-        ),
     },
 )
 def ingest_admin_express():
     bucket_name = "airflow-staging"
+    tmp_path = "/tmp/admin_express"
 
     @task.python
     def download_admin_express(**context) -> str:
+        print(context["params"]["zone"])
         url = get_source_by_name(context["params"]["zone"])["url"]
-
+        print(url)
         filename = url.split("/")[-1]
+        print(filename)
         path_on_bucket = f"{bucket_name}/{filename}"
         print(path_on_bucket)
-
-        file_exists = Container().s3().exists(path_on_bucket)
-
-        if file_exists and not context["params"]["refresh_source"]:
-            return path_on_bucket
-
         opener = URLopener()
         opener.addheader("User-Agent", "Mozilla/5.0")
         opener.retrieve(url=url, filename=filename)
-
         Container().s3().put_file(filename, path_on_bucket)
-
+        os.remove(filename)
         return path_on_bucket
 
     @task.python
     def ingest(path_on_bucket, **context) -> str:
-        srid = get_source_by_name(context["params"]["zone"])["srid"]
-        shp_to_table_map = get_source_by_name(context["params"]["zone"])["shapefile_to_table"]
+        source = get_source_by_name(context["params"]["zone"])
+        srid = source["srid"]
+        print(srid)
+        shp_to_table_map = source["shapefile_to_table"]
+        print(shp_to_table_map)
 
         with Container().s3().open(path_on_bucket, "rb") as f:
-            py7zr.SevenZipFile(f, mode="r").extractall()
-            for dirpath, _, filenames in os.walk("."):
+            py7zr.SevenZipFile(f, mode="r").extractall(path=tmp_path)
+            for dirpath, _, filenames in os.walk(tmp_path):
                 for filename in filenames:
                     if filename.endswith(".shp"):
                         table_name = shp_to_table_map.get(filename)
                         if not table_name:
                             continue
                         path = os.path.abspath(os.path.join(dirpath, filename))
+                        print(path)
                         cmd = [
                             "ogr2ogr",
                             "-f",
@@ -113,9 +108,14 @@ def dbt_run(**context):
         dbt_run_cmd = f"dbt build -s {dbt_selector}"
         return 'cd "${AIRFLOW_HOME}/include/sql/sparte" && ' + dbt_run_cmd
 
+    @task.bash
+    def cleanup() -> str:
+        return f"rm -rf {tmp_path}"
+
     path_on_bucket = download_admin_express()
     ingest_result = ingest(path_on_bucket)
     ingest_result >> dbt_run()
+    ingest_result >> cleanup()
 
 
 # Instantiate the DAG

airflow/dags/ingest_cog_changes.py

@@ -0,0 +1,61 @@
+import os
+from urllib.request import URLopener
+
+import pandas as pd
+from airflow.decorators import dag, task
+from airflow.models.param import Param
+from include.container import Container
+from pendulum import datetime
+
+urls = {
+    "2024": "https://www.insee.fr/fr/statistiques/fichier/7766585/v_mvt_commune_2024.csv",
+}
+
+
+@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"],
+    params={
+        "year": Param(
+            default="2024",
+            description="Année des changements",
+            type="string",
+            enum=list(urls.keys()),
+        ),
+    },
+)
+def ingest_cog_changes():
+    bucket_name = "airflow-staging"
+    tmp_localpath = "/tmp/cog_changes"
+
+    @task.python
+    def download_change_file(**context) -> str:
+        year = context["params"]["year"]
+        filename = urls[year].split("/")[-1]
+        path_on_bucket = f"{bucket_name}/{filename}"
+        opener = URLopener()
+        opener.addheader("User-Agent", "Mozilla/5.0")
+        opener.retrieve(url=urls[year], filename=filename)
+        Container().s3().put_file(filename, path_on_bucket)
+        os.remove(filename)
+        return path_on_bucket
+
+    @task.python
+    def ingest(path_on_bucket, **context) -> int | None:
+        Container().s3().get_file(path_on_bucket, tmp_localpath)
+        df = pd.read_csv(tmp_localpath)
+        table_name = f"insee_cog_changes_{context['params']['year']}"
+        row_count = df.to_sql(table_name, con=Container().sqlalchemy_dbt_conn(), if_exists="replace")
+        os.remove(tmp_localpath)
+        return row_count
+
+    path_on_bucket = download_change_file()
+    ingest(path_on_bucket)
+
+
+ingest_cog_changes()

airflow/dags/ingest_majic.py

@@ -0,0 +1,101 @@
+import shutil
+import subprocess
+from typing import List
+from zipfile import ZipFile
+
+from airflow.decorators import dag, task
+from airflow.operators.python import PythonOperator
+from include.container import Container
+from include.majic.sources import sources
+from include.pools import DBT_POOL
+from include.utils import (
+    get_dbt_command_from_directory,
+    get_first_shapefile_path_in_dir,
+)
+from pendulum import datetime
+
+BUCKET_NAME = "airflow-staging"
+TMP_PATH = "/tmp/majic"
+
+
+def load_shapefile_to_postgres(source: dict):
+    shapefile_on_s3 = source["shapefile_on_s3"]
+    srid = source["srid"]
+    table_name = source["table_name"]
+    path_on_bucket = f"{BUCKET_NAME}/{shapefile_on_s3}"
+
+    with Container().s3().open(path_on_bucket, "rb") as f:
+        zip_file = ZipFile(f)
+        tmp_path = f"{TMP_PATH}/{table_name}"
+        zip_file.extractall(tmp_path)
+        shapefile_path = get_first_shapefile_path_in_dir(tmp_path)
+        cmd = [
+            "ogr2ogr",
+            "-f",
+            '"PostgreSQL"',
+            f'"{Container().gdal_dbt_conn().encode()}"',
+            "-overwrite",
+            "-lco",
+            "GEOMETRY_NAME=geom",
+            "-a_srs",
+            f"EPSG:{srid}",
+            "-nlt",
+            "MULTIPOLYGON",
+            "-nlt",
+            "PROMOTE_TO_MULTI",
+            "-nln",
+            table_name,
+            shapefile_path,
+            "--config",
+            "PG_USE_COPY",
+            "YES",
+        ]
+        subprocess.run(" ".join(cmd), shell=True, check=True)
+
+    shutil.rmtree(TMP_PATH)
+
+
+@dag(
+    start_date=datetime(2024, 1, 1),
+    schedule="@once",
+    catchup=False,
+    max_active_runs=1,
+    default_args={"owner": "Alexis Athlani", "retries": 3},
+    tags=["Majic", "Cerema"],
+)
+def ingest_majic():
+    # Les fichiers de consommation d'espace sont sur le dropbox du Cerema
+    # Ils ne sont pas acessibles programmatiquement, donc il faut les télécharger manuellement
+    # et les mettre dans le bucket airflow-staging, dans le dossier majic
+
+    ingest_tasks: List[PythonOperator] = []
+
+    for source in sources:
+        ingest_task = PythonOperator(
+            task_id=f"ingest_{source['table_name']}",
+            python_callable=load_shapefile_to_postgres,
+            op_kwargs={"source": source},
+        )
+        ingest_tasks.append(ingest_task)
+
+    @task.bash(pool=DBT_POOL)
+    def dbt_build() -> str:
+        return get_dbt_command_from_directory(cmd="dbt build -s +consommation.sql+")
+
+    @task.bash()
+    def cleanup() -> str:
+        return f"rm -rf {TMP_PATH}"
+
+    for i in range(len(ingest_tasks) - 1):
+        ingest_tasks[i] >> ingest_tasks[i + 1]
+
+    build = dbt_build()
+    cleanup_task = cleanup()
+
+    # Build the models and clean up the temporary files
+    # after all the ingest tasks have completed
+    ingest_tasks[-1].set_downstream(build)
+    build.set_downstream(cleanup_task)
+
+
+ingest_majic()

airflow/dags/ingest_scots.py

@@ -0,0 +1,87 @@
+import os
+
+import pandas as pd
+import requests
+from airflow.decorators import dag, task
+from include.container import Container
+from include.pools import DBT_POOL
+from pendulum import datetime
+
+SCOT_ENDPOINT = "https://api-sudocuh.datahub.din.developpement-durable.gouv.fr/sudocuh/enquetes/ref/scot/liste/CSV?annee_cog=2024"  # noqa: E501 (line too long)
+SCOT_COMMUNES_ENDPOINT = "https://api-sudocuh.datahub.din.developpement-durable.gouv.fr/sudocuh/enquetes/ref/scot/communes/CSV?annee_cog=2024"  # noqa: E501 (line too long)
+
+
+@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=["SUDOCUH"],
+)
+def ingest_scots():
+    bucket_name = "airflow-staging"
+    scot_filename = "scot.csv"
+    scot_communes_filename = "scot_communes.csv"
+    tmp_localpath_scot = f"/tmp/{scot_filename}"
+    tmp_localpath_scot_communes = f"/tmp/{scot_communes_filename}"
+
+    @task.python
+    def download_scots() -> str:
+        request = requests.get(SCOT_ENDPOINT)
+
+        with open(tmp_localpath_scot, "wb") as f:
+            f.write(request.content)
+
+        path_on_bucket = f"{bucket_name}/{scot_filename}"
+
+        Container().s3().put_file(tmp_localpath_scot, path_on_bucket)
+        os.remove(tmp_localpath_scot)
+        return path_on_bucket
+
+    @task.python
+    def download_scot_communes() -> str:
+        request = requests.get(SCOT_COMMUNES_ENDPOINT)
+
+        with open(tmp_localpath_scot_communes, "wb") as f:
+            f.write(request.content)
+
+        path_on_bucket = f"{bucket_name}/{scot_communes_filename}"
+
+        Container().s3().put_file(tmp_localpath_scot_communes, path_on_bucket)
+        os.remove(tmp_localpath_scot_communes)
+        return path_on_bucket
+
+    @task.python
+    def ingest_scots(path_on_bucket) -> int | None:
+        Container().s3().get_file(path_on_bucket, tmp_localpath_scot)
+        df = pd.read_csv(tmp_localpath_scot, sep=";")
+        table_name = "sudocuh_scot"
+        row_count = df.to_sql(table_name, con=Container().sqlalchemy_dbt_conn(), if_exists="replace")
+        os.remove(tmp_localpath_scot)
+        return row_count
+
+    @task.python
+    def ingest_scot_communes(path_on_bucket) -> int | None:
+        Container().s3().get_file(path_on_bucket, tmp_localpath_scot_communes)
+        df = pd.read_csv(tmp_localpath_scot_communes, sep=";")
+        table_name = "sudocuh_scot_communes"
+        row_count = df.to_sql(table_name, con=Container().sqlalchemy_dbt_conn(), if_exists="replace")
+        os.remove(tmp_localpath_scot_communes)
+        return row_count
+
+    @task.bash(retries=0, trigger_rule="all_success", pool=DBT_POOL)
+    def dbt_build(**context):
+        return 'cd "${AIRFLOW_HOME}/include/sql/sparte" && dbt build -s sudocuh'
+
+    scots_path = download_scots()
+    scot_communes_path = download_scot_communes()
+
+    ingest_scots_result = ingest_scots(scots_path)
+    ingest_scot_communes_result = ingest_scot_communes(scot_communes_path)
+
+    ingest_scots_result >> ingest_scot_communes_result >> dbt_build()
+
+
+ingest_scots()