Merge branch 'feature/145_siret_empty_names' into dev

Author: ColinMaudry

.gitignore

@@ -6,6 +6,7 @@ logs
 .venv
 *.egg-info
 *.parquet
+!tests/data/sirene/*.parquet
 !code_officiel_geographique.parquet
 *.gz
 *.zip

pyproject.toml

@@ -41,6 +41,7 @@ testpaths = [
 ]
 env = [
     "DATASETS_REFERENCE_FILEPATH=tests/data/source_datasets_test.json",
+    "SIRENE_DATA_DIR=tests/data/sirene",
     "PREFECT_API_URL=",
     "DECP_PROCESSING_PUBLISH=",
     "DECP_USE_CACHE=false"

src/config.py

@@ -98,7 +98,15 @@ def make_sirene_data_dir(sirene_data_parent_dir) -> Path:
 
 
 SIRENE_DATA_PARENT_DIR = make_path_from_env("SIRENE_DATA_PARENT_DIR", DATA_DIR)
-SIRENE_DATA_DIR = make_sirene_data_dir(SIRENE_DATA_PARENT_DIR)
+
+# SIRENE_DATA_DIR ne doit être spécifié que pour les tests. Laisser vide dans .env et laisser make_sirene_data_dir
+# le déterminer
+SIRENE_DATA_DIR = os.getenv(
+    "SIRENE_DATA_DIR", make_sirene_data_dir(SIRENE_DATA_PARENT_DIR)
+)
+if isinstance(SIRENE_DATA_DIR, str):
+    SIRENE_DATA_DIR = Path(os.path.join(BASE_DIR, SIRENE_DATA_DIR))
+
 # SIRENE_DATA_DIR on ne le crée que si nécessaire, dans flows.py
 print(f"{'SIRENE_DATA_PARENT_DIR':<40}", SIRENE_DATA_PARENT_DIR)
 print(f"{'SIRENE_DATA_DIR':<40}", SIRENE_DATA_DIR)

src/flows/decp_processing.py

@@ -104,7 +104,6 @@ def decp_processing(enable_cache_removal: bool = True):
     # Preprocessing des données SIRENE si :
     # - le dossier n'existe pas encore (= les données n'ont pas déjà été preprocessed ce mois-ci)
     # - on est au moins le 5 du mois (pour être sûr que les données SIRENE ont été mises à jour sur data.gouv.fr)
-    print(SIRENE_DATA_DIR)
     if not SIRENE_DATA_DIR.exists():
         sirene_preprocess()
 

src/flows/sirene_preprocess.py

@@ -3,8 +3,8 @@
 
 from src.config import SIRENE_DATA_DIR
 from src.flows.get_cog import get_cog
-from src.tasks.get import get_etablissements
-from src.tasks.transform import get_prepare_unites_legales, prepare_etablissements
+from src.tasks.get import get_etablissements, get_unite_legales
+from src.tasks.transform import prepare_etablissements
 from src.tasks.utils import create_sirene_data_dir
 
 
@@ -26,7 +26,7 @@ def sirene_preprocess():
         processed_ul_parquet_path = SIRENE_DATA_DIR / "unites_legales.parquet"
         if not processed_ul_parquet_path.exists():
             print("Téléchargement et préparation des unités légales...")
-            get_prepare_unites_legales(processed_ul_parquet_path)
+            get_unite_legales(processed_ul_parquet_path)
         else:
             print(processed_ul_parquet_path, " existe, skipping.")
 
@@ -35,7 +35,7 @@ def sirene_preprocess():
         if not processed_etab_parquet_path.exists():
             print("Téléchargement et préparation des établissements...")
             lf = get_etablissements()
-            prepare_etablissements(lf, processed_etab_parquet_path)
+            prepare_etablissements(lf).sink_parquet(processed_etab_parquet_path)
         else:
             print(processed_etab_parquet_path, " existe, skipping.")
 

src/tasks/enrich.py

@@ -1,5 +1,6 @@
 import polars as pl
 import polars.selectors as cs
+from polars_ds import haversine
 from prefect import task
 
 from src.config import SIRENE_DATA_DIR
@@ -20,6 +21,27 @@ def add_etablissement_data(
     lf_sirets = lf_sirets.join(
         lf_etablissements, how="inner", left_on=siret_column, right_on="siret"
     )
+
+    # On ne prend pas l'activité des acheteurs
+    if type_siret == "acheteur":
+        lf_sirets = lf_sirets.drop(cs.starts_with("activite_"))
+
+    # Si il y a un etablissement_nom (Enseigne1Etablissement ou denominationUsuelleEtablissement),
+    # on l'ajoute au nom de l'organisme, entre parenthèses
+    lf_sirets = lf_sirets.with_columns(
+        pl.when(pl.col("etablissement_nom").is_not_null())
+        .then(
+            pl.concat_str(
+                pl.col(f"{type_siret}_nom"),
+                pl.lit(" ("),
+                pl.col("etablissement_nom"),
+                pl.lit(")"),
+            )
+        )
+        .otherwise(pl.col(f"{type_siret}_nom"))
+        .alias(f"{type_siret}_nom")
+    ).drop("etablissement_nom")
+
     lf_sirets = lf_sirets.rename(
         {
             "latitude": f"{type_siret}_latitude",
@@ -59,6 +81,7 @@ def enrich_from_sirene(lf: pl.LazyFrame):
     # Récupération des données SIRET/SIREN préparées dans sirene-preprocess()
     lf_etablissements = pl.scan_parquet(SIRENE_DATA_DIR / "etablissements.parquet")
     lf_unites_legales = pl.scan_parquet(SIRENE_DATA_DIR / "unites_legales.parquet")
+
     lf_base = lf.clone()
 
     # DONNÉES SIRENE ACHETEURS
@@ -133,29 +156,16 @@ def enrich_from_sirene(lf: pl.LazyFrame):
 
 
 def calculate_distance(lf: pl.LazyFrame) -> pl.LazyFrame:
-    # Implémentation native de la formule de Haversine
-    # R = 6371  # Rayon de la Terre en km
-
-    # Conversion en radians
-    lat1 = pl.col("acheteur_latitude").radians()
-    lon1 = pl.col("acheteur_longitude").radians()
-    lat2 = pl.col("titulaire_latitude").radians()
-    lon2 = pl.col("titulaire_longitude").radians()
-
-    # Différences
-    dlat = lat2 - lat1
-    dlon = lon2 - lon1
-
-    # Formule de Haversine
-    a = (dlat / 2).sin().pow(2) + lat1.cos() * lat2.cos() * (dlon / 2).sin().pow(2)
-    c = 2 * a.sqrt().arcsin()
-
-    # Distance en km
-    distance = 6371 * c
-
+    # Utilisation de polars_ds.haversine
+    # https://polars-ds-extension.readthedocs.io/en/latest/num.html#polars_ds.exprs.num.haversine
     lf = lf.with_columns(
-        distance.round(1).alias(
-            "distance"
-        )  # Arrondi à 1 décimale comme avant (mode="half_away_from_zero" n'est pas dispo direct mais round standard est ok)
+        haversine(
+            pl.col("acheteur_latitude"),
+            pl.col("acheteur_longitude"),
+            pl.col("titulaire_latitude"),
+            pl.col("titulaire_longitude"),
+        )
+        .round(mode="half_away_from_zero")
+        .alias("distance")
     )
     return lf

src/tasks/get.py

@@ -14,6 +14,7 @@
 from prefect import task
 from prefect.transactions import transaction
 
+from config import SIRENE_UNITES_LEGALES_URL
 from src.config import (
     DECP_PROCESSING_PUBLISH,
     DECP_USE_CACHE,
@@ -35,6 +36,7 @@
     gen_artifact_row,
     stream_replace_bytestring,
 )
+from tasks.transform import prepare_unites_legales
 
 
 @task(retries=3, retry_delay_seconds=3)
@@ -314,10 +316,10 @@ def yield_modifications(row: dict, separator="_") -> Iterator[dict] or None:
         raw_mods = raw_mods["modification"]
     # Couvre le (non-)format dans lequel "modifications" ou "modification" mène
     # directement à un dict contenant les métadonnées liées à une modification.
-    if isinstance(raw_mods, dict):
+    elif isinstance(raw_mods, dict):
         raw_mods = [raw_mods]
-
-    raw_mods = [] if raw_mods is None else raw_mods
+    elif isinstance(raw_mods, str) or raw_mods is None:
+        raw_mods = []
 
     mods = [{}] + raw_mods
     for i, mod in enumerate(mods):
@@ -369,6 +371,8 @@ def get_etablissements() -> pl.LazyFrame:
         "longitude": pl.Float64,
         "activitePrincipaleEtablissement": pl.String,
         "nomenclatureActivitePrincipaleEtablissement": pl.String,
+        "enseigne1Etablissement": pl.String,
+        "denominationUsuelleEtablissement": pl.String,
     }
 
     columns = list(schema.keys())
@@ -387,7 +391,7 @@ def get_etablissements() -> pl.LazyFrame:
             hrefs.append(base_url + href)
 
     # Fonction de traitement pour un fichier
-    def process_file(_href: str):
+    def get_process_file(_href: str):
         print(_href.split("/")[-1])
         try:
             response = http_client.get(
@@ -403,18 +407,12 @@ def process_file(_href: str):
         content = response.content
         lff = pl.scan_csv(content, schema_overrides=schema)
         lff = lff.select(columns)
-        lff = lff.with_columns(
-            [
-                pl.col("codeCommuneEtablissement").str.pad_start(5, "0"),
-                pl.col("siret").str.pad_start(14, "0"),
-            ]
-        )
         return lff
 
     # Traitement en parrallèle avec 8 threads
     lfs = []
     with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:
-        futures = [executor.submit(process_file, href) for href in hrefs]
+        futures = [executor.submit(get_process_file, href) for href in hrefs]
         for future in concurrent.futures.as_completed(futures):
             try:
                 lf = future.result()
@@ -474,3 +472,13 @@ def get_clean(
             # Le fichier parquet est déjà disponible pour ce checksum
             print(f"👍 Ressource déjà en cache : {resource['dataset_code']}")
             return parquet_path.with_suffix(".parquet")
+
+
+@task
+def get_unite_legales(processed_parquet_path):
+    print("Téléchargement des données unité légales et sélection des colonnes...")
+    (
+        pl.scan_parquet(SIRENE_UNITES_LEGALES_URL)
+        .pipe(prepare_unites_legales)
+        .sink_parquet(processed_parquet_path)
+    )

src/tasks/transform.py

@@ -3,9 +3,8 @@
 
 import polars as pl
 import polars.selectors as cs
-from prefect import task
 
-from src.config import DATA_DIR, DIST_DIR, SIRENE_UNITES_LEGALES_URL, DecpFormat
+from src.config import DATA_DIR, DIST_DIR, DecpFormat
 from src.tasks.output import save_to_files
 from src.tasks.utils import check_parquet_file
 
@@ -205,33 +204,82 @@ def extract_unique_titulaires_siret(lf: pl.LazyFrame):
     return lf
 
 
-@task
-def get_prepare_unites_legales(processed_parquet_path):
-    print("Téléchargement des données unité légales et sélection des colonnes...")
-    (
-        pl.scan_parquet(SIRENE_UNITES_LEGALES_URL)
-        .select(["siren", "denominationUniteLegale"])
-        .filter(pl.col("siren").is_not_null())
-        .filter(pl.col("denominationUniteLegale").is_not_null())
-        .unique()
-        .sink_parquet(processed_parquet_path, engine="streaming")
+def prepare_unites_legales(lf: pl.LazyFrame) -> pl.LazyFrame:
+    return (
+        lf.select(
+            [
+                "siren",
+                "denominationUniteLegale",
+                "prenomUsuelUniteLegale",
+                "nomUniteLegale",  # toujours rempli pour personnes physique
+                "nomUsageUniteLegale",  # parfois rempli, a la priorité sur nomUniteLegale
+                "statutDiffusionUniteLegale",  # P = non-diffusible
+            ]
+        )
+        .filter(
+            pl.col("siren").is_not_null()
+        )  # utilisation du fichier Stock, normalement pas de siren null
+        .unique()  # utilisation du fichier Stock, normalement pas de doublons
+        .with_columns(
+            pl.when(pl.col("nomUsageUniteLegale").is_not_null())
+            .then(pl.col("nomUsageUniteLegale"))
+            .otherwise(pl.col("nomUniteLegale"))
+            .alias("nomUniteLegale")
+        )
+        .with_columns(
+            pl.when(pl.col("nomUniteLegale").is_not_null())
+            .then(
+                pl.concat_str(
+                    pl.col("prenomUsuelUniteLegale"),
+                    pl.col("nomUniteLegale"),
+                    separator=" ",
+                )
+            )
+            .otherwise(pl.col("denominationUniteLegale"))
+            .alias("denominationUniteLegale")
+        )
+        .with_columns(
+            pl.when(pl.col("statutDiffusionUniteLegale") == "P")
+            .then(pl.lit("[Données personnelles non-diffusibles]"))
+            .otherwise(pl.col("denominationUniteLegale"))
+            .alias("denominationUniteLegale")
+        )
+        .drop(
+            [
+                "prenomUsuelUniteLegale",
+                "statutDiffusionUniteLegale",
+                "nomUniteLegale",
+                "nomUsageUniteLegale",
+            ]
+        )
     )
 
 
-def prepare_etablissements(lf: pl.LazyFrame, processed_parquet_path: Path) -> None:
-    lf = lf.rename(
+def prepare_etablissements(lff: pl.LazyFrame) -> pl.LazyFrame:
+    lff = lff.with_columns(
+        [
+            pl.col("codeCommuneEtablissement").str.pad_start(5, "0"),
+            pl.col("siret").str.pad_start(14, "0"),
+            # Si enseigne1Etablissement est null, on utilise denominationUsuelleEtablissement
+            pl.coalesce(
+                "enseigne1Etablissement", "denominationUsuelleEtablissement"
+            ).alias("etablissement_nom"),
+        ]
+    )
+    lff = lff.drop("denominationUsuelleEtablissement", "enseigne1Etablissement")
+    lff = lff.rename(
         {
             "codeCommuneEtablissement": "commune_code",
             "activitePrincipaleEtablissement": "activite_code",
             "nomenclatureActivitePrincipaleEtablissement": "activite_nomenclature",
         }
     )
 
-    # Ajout des noms de départements, noms régions,
+    # Ajout des noms de commune, départements, régions
     lf_cog = pl.scan_parquet(DATA_DIR / "code_officiel_geographique.parquet")
-    lf = lf.join(lf_cog, on="commune_code", how="left")
+    lff = lff.join(lf_cog, on="commune_code", how="left")
 
-    lf.sink_parquet(processed_parquet_path, engine="streaming")
+    return lff
 
 
 def sort_columns(lf: pl.LazyFrame, config_columns):

tests/data/decp_test_2019.json

@@ -14,7 +14,7 @@
               {
                 "typeIdentifiant": "SIRET",
                 "denominationSociale": "AMC FOLLIOT",
-                "id": "65265021900023"
+                "id": "12345678900022"
               }
             ]
           ]
@@ -28,7 +28,7 @@
         {
           "typeIdentifiant": "SIRET",
           "denominationSociale": "AMC FOLLIOT",
-          "id": "65265021900023"
+          "id": "12345678900022"
         }
       ],
       "id": "2019_83935401",
@@ -68,7 +68,7 @@
       "titulaires": [
         {
           "typeIdentifiant": "SIRET",
-          "id": "34027049500021",
+          "id": "12345678900023",
           "denominationSociale": "FFF"
         },
         {

tests/data/decp_test_2022.json

@@ -30,7 +30,7 @@
           {
             "titulaire": {
               "typeIdentifiant": "SIRET",
-              "id": "34027049500021"
+              "id": "12345678900023"
             }
           },
           {
@@ -184,7 +184,7 @@
                 {
                   "titulaire": {
                     "typeIdentifiant": "SIRET",
-                    "id": "58211867500054"
+                    "id": "12345678900022"
                   }
                 },
                 {