Merge branch 'release/2.9.0'

Author: ColinMaudry

CHANGELOG.md

@@ -1,3 +1,8 @@
+### 2.9.0 2026-01-29
+
+- Ajout du champ `type` qui indique le type de marché : fournitures, services ou travaux (dérivé du code CPV)
+- `distance` renommé `titulaire_distance` par cohérence, cette donnée étant liée au titulaire, et un marché peut avoir plusieurs titulaires
+
 ### 2.8.0 2026-01-23
 
 - Ajout du champ `titulaire_categorie` (PME, ETI, GE)

README.md

@@ -1,6 +1,6 @@
 # DECP processing
 
-> version 2.8.0 ([notes de version](https://github.com/ColinMaudry/decp-processing/blob/main/CHANGELOG.md))
+> version 2.9.0 ([notes de version](https://github.com/ColinMaudry/decp-processing/blob/main/CHANGELOG.md))
 
 Projet de traitement et de publication de meilleures données sur les marchés publics attribués en France. Vous pouvez consulter, filtrer et télécharger
 ces données sur le site [decp.info](https://decp.info). Enfin la section [À propos](https://decp.info/a-propos) décrit les objectifs du projet et regroupe toutes les informations clés.

pyproject.toml

@@ -1,10 +1,10 @@
 [project]
 name = "decp-processing"
 description = "Traitement des données des marchés publics français."
-version = "2.8.0"
+version = "2.9.0"
 requires-python = ">= 3.10"
 authors = [
-    { name = "Colin Maudry", email = "colin+decp@maudry.com" }
+    { name = "Colin Maudry", email = "colin@colmo.tech" }
 ]
 dependencies = [
     "python-dotenv",

reference/schema_base.json

@@ -70,6 +70,13 @@
       "description": "Montant forfaitaire ou montant maximum estimé hors-taxes, en euros. Ce montant est le montant attribué. Le montant final payé aux titulaires peut évoluer lors de la signature du contrat et de l'exécution du marché.",
       "short_title": "Montant"
     },
+    {
+      "name": "type",
+      "type": "string",
+      "title": "Type",
+      "description": "Type de marché public : fournitures, services ou travaux (dérivé du code CPV).",
+      "short_title": "Type"
+    },
     {
       "type": "string",
       "name": "codeCPV",
@@ -241,8 +248,8 @@
       "short_title": "Id. accord-cadre"
     },
     {
-      "type": "string",
-      "name": "distance",
+      "type": "integer",
+      "name": "titulaire_distance",
       "title": "Distance acheteur-titulaire",
       "description": "Distance en kilomètres entre l'adresse de l'acheteur et celle du titulaire.",
       "short_title": "Distance"

reference/source_datasets.json

@@ -323,5 +323,12 @@
     "name": "Données essentielles de la commande publique (DECP) de Klekoon",
     "code": "scrap_klekoon",
     "owner_org_name": "Colin Maudry"
+  },
+  {
+    "id": "5cd57bf68b4c4179299eb0e9",
+    "name": "Données essentielles de la commande publique - fichiers consolidés (MINEF)",
+    "code": "minef",
+    "owner_org_name": "MINEF",
+    "skip": true
   }
 ]

src/config.py

@@ -218,7 +218,10 @@ def make_sirene_data_dir(sirene_data_parent_dir) -> Path:
     tracked_datasets_complete = json.load(f)
 TRACKED_DATASETS = []
 for dataset in tracked_datasets_complete:
-    if len(SOLO_DATASETS) == 0 or dataset["id"] in SOLO_DATASETS:
+    if (len(SOLO_DATASETS) == 0 and not (dataset.get("skip"))) or dataset[
+        "id"
+    ] in SOLO_DATASETS:
+        logger.debug(f"{dataset['name']} added.")
         TRACKED_DATASETS.append(dataset)
 
 

src/flows/decp_processing.py

@@ -24,12 +24,11 @@
 )
 from src.flows.sirene_preprocess import sirene_preprocess
 from src.tasks.dataset_utils import list_resources
-from src.tasks.enrich import enrich_from_sirene
+from src.tasks.enrich import add_type_marche, enrich_from_sirene
 from src.tasks.get import get_clean
 from src.tasks.output import generate_final_schema, sink_to_files
 from src.tasks.publish import publish_to_datagouv
 from src.tasks.transform import (
-    add_duree_restante,
     calculate_naf_cpv_matching,
     concat_parquet_files,
     sort_columns,
@@ -42,6 +41,7 @@
     print_all_config,
     remove_unused_cache,
 )
+from tasks.enrich import add_duree_restante
 
 
 @flow(log_prints=True)
@@ -119,6 +119,9 @@ def decp_processing(enable_cache_removal: bool = True):
     logger.info("Ajout de la colonne 'dureeRestanteMois'...")
     lf = add_duree_restante(lf)
 
+    logger.info("Ajout du type de marché...")
+    lf = add_type_marche(lf)
+
     logger.info("Génération des probabilités NAF/CPV...")
     calculate_naf_cpv_matching(lf)
     lf = lf.drop(cs.starts_with("activite"))

src/tasks/enrich.py

@@ -1,3 +1,5 @@
+from datetime import datetime
+
 import polars as pl
 import polars.selectors as cs
 
@@ -206,7 +208,7 @@ def calculate_distance(lf: pl.LazyFrame) -> pl.LazyFrame:
         )
         .round(mode="half_away_from_zero")
         .cast(pl.Int16)
-        .alias("distance")
+        .alias("titulaire_distance")
     )
     return lf
 
@@ -215,7 +217,7 @@ def haversine(
     lat1: pl.Expr, lon1: pl.Expr, lat2: pl.Expr, lon2: pl.Expr, R: float = 6371.0
 ) -> pl.Expr:
     """
-    Calcule la distance haversine entre deux points (lat1, lon1) et (lat2, lon2)     en km.
+    Calcule la distance haversine entre deux points (lat1, lon1) et (lat2, lon2) en km.
     Utilise des opérations vectorisées Polars.
     Généré par la LLM Euria, développée et hébergée en Suisse par Infomaniak.
     """
@@ -235,3 +237,44 @@ def haversine(
 
     # Distance
     return R * c
+
+
+def add_type_marche(lf: pl.LazyFrame) -> pl.LazyFrame:
+    cpv_division = pl.col("codeCPV").str.slice(0, 2).cast(pl.Int8, strict=False)
+
+    lf = lf.with_columns(
+        pl.when(cpv_division == 45)
+        .then(pl.lit("Travaux"))
+        .when(cpv_division.is_in(range(1, 45)) | (cpv_division == 48))
+        .then(pl.lit("Fournitures"))
+        .when(cpv_division.is_in(range(50, 99)))
+        .then(pl.lit("Services"))
+        .otherwise(pl.lit("Non catégorisé"))
+        .fill_null(pl.lit("Code CPV invalide"))
+        .alias("type")
+    )
+    return lf
+
+
+def add_duree_restante(lff: pl.LazyFrame):
+    today = datetime.now().date()
+    duree_mois_days_int = pl.col("dureeMois") * 30.5
+    end_date = pl.col("dateNotification") + pl.duration(days=duree_mois_days_int)
+    duree_restante_mois = ((end_date - today).dt.total_days() / 30).round(1)
+
+    # Pas de valeurs négatives.
+    lff = lff.with_columns(
+        pl.when(duree_restante_mois < 0)
+        .then(pl.lit(0))
+        .otherwise(duree_restante_mois)
+        .alias("dureeRestanteMois")
+    )
+
+    # Si dureeRestanteMois > dureeMois, dureeRestanteMois = dureeMois
+    lff = lff.with_columns(
+        pl.when(pl.col("dureeRestanteMois") > pl.col("dureeMois"))
+        .then(pl.col("dureeMois").cast(pl.Float32))
+        .otherwise(pl.col("dureeRestanteMois"))
+        .alias("dureeRestanteMois")
+    )
+    return lff

src/tasks/transform.py

@@ -1,4 +1,3 @@
-from datetime import datetime
 from pathlib import Path
 
 import polars as pl
@@ -194,7 +193,7 @@ def prepare_unites_legales(lf: pl.LazyFrame) -> pl.LazyFrame:
                 "nomUsageUniteLegale",  # parfois rempli, a la priorité sur nomUniteLegale
                 "statutDiffusionUniteLegale",  # P = non-diffusible
                 "categorieEntreprise",  # PME, ETI, GE
-                # "categorieJuridiqueUniteLegale" # 1000, etc.
+                # "categorieJuridiqueUniteLegale",  # 1000, etc.
             ]
         )
         .filter(
@@ -402,30 +401,6 @@ def calculate_naf_cpv_matching(lf_naf_cpv: pl.LazyFrame):
     save_to_files(df_results, DIST_DIR / "probabilites_naf_cpv", "csv")
 
 
-def add_duree_restante(lff: pl.LazyFrame):
-    today = datetime.now().date()
-    duree_mois_days_int = pl.col("dureeMois") * 30.5
-    end_date = pl.col("dateNotification") + pl.duration(days=duree_mois_days_int)
-    duree_restante_mois = ((end_date - today).dt.total_days() / 30).round(1)
-
-    # Pas de valeurs négatives.
-    lff = lff.with_columns(
-        pl.when(duree_restante_mois < 0)
-        .then(pl.lit(0))
-        .otherwise(duree_restante_mois)
-        .alias("dureeRestanteMois")
-    )
-
-    # Si dureeRestanteMois > dureeMois, dureeRestanteMois = dureeMois
-    lff = lff.with_columns(
-        pl.when(pl.col("dureeRestanteMois") > pl.col("dureeMois"))
-        .then(pl.col("dureeMois").cast(pl.Float32))
-        .otherwise(pl.col("dureeRestanteMois"))
-        .alias("dureeRestanteMois")
-    )
-    return lff
-
-
 #
 # ⬇️⬇️⬇️ Fonctions à refactorer avec Polars et le format DECP 2022 ⬇️⬇️⬇️
 #

tests/conftest.py

@@ -0,0 +1,30 @@
+import os
+import subprocess
+import time
+
+import pytest
+from prefect.testing.utilities import prefect_test_harness
+
+
+@pytest.fixture(autouse=True, scope="session")
+def prefect_test_fixture(tmp_path_factory):
+    os.environ["PREFECT_SERVER_EPHEMERAL_STARTUP_TIMEOUT_SECONDS"] = "90"
+
+    with prefect_test_harness():
+        yield
+
+    # Force cleanup: try to stop any lingering Prefect server
+    # (This is a workaround — Prefect doesn't expose a clean stop API in test harness)
+    # Généré par Euria, la LLM d'Infomaniak.
+
+    # Try to kill any remaining prefect server process
+    try:
+        # This is a bit brute-force, but works if no other prefect server is running
+        subprocess.run(
+            ["pkill", "-f", "prefect server start"],
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+        time.sleep(0.5)  # Give time to terminate
+    except Exception:
+        pass