Merge branch 'release/2.12.0'

Author: ColinMaudry

CHANGELOG.md

@@ -1,3 +1,20 @@
+### 2.12.0 2026-06-05
+
+- Ajout de nouvelles sources de données :
+  - Modula :
+    - <www.proxilegales.fr>
+    - <www.marches-demat.com>
+    - <www.laprovencemarchespublics.com>
+    - <www.corsematinmarchespublics.com>
+    - <www.achatspublicscorse.com>
+    - <www.marchespublics.pro>
+    - <www.antilles-legales.fr>
+    - <www.guyane-legales.fr>
+  - alsacemarchespublics.eu
+  - Eurométropole de Strasbourg
+  - Ville de Strasbourg
+- Meilleur gestion du tampon d'ingestion XML qui ne voyait pas les petits fichiers
+
 ### 2.11.0 2026-06-02
 
 - Utilisation du [géocodage de Géoplateforme (IGN)](https://cartes.gouv.fr/aide/fr/guides-utilisateur/utiliser-les-services-de-la-geoplateforme/geocodage/) au lieu des données SIRENE géocodées d'Etalab, qui ne sont plus publiées

README.md

@@ -1,7 +1,5 @@
 # DECP processing
 
-> version 2.10.2([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.
 
@@ -13,9 +11,9 @@ Ce projet prend sa source dans la complexité de la publication des données par
 
 - code source de l'agrégation [ouvert mais très peu lisible](https://github.com/139bercy/decp-arr2022)
 - documentation incomplète et éparpillée
-  - https://github.com/139bercy/decp-arr2022/blob/main/README.md
-  - https://www.data.gouv.fr/fr/datasets/5cd57bf68b4c4179299eb0e9
-  - https://data.economie.gouv.fr/pages/donnees-essentielles-de-la-commande-publique/
+  - <https://github.com/139bercy/decp-arr2022/blob/main/README.md>
+  - <https://www.data.gouv.fr/fr/datasets/5cd57bf68b4c4179299eb0e9>
+  - <https://data.economie.gouv.fr/pages/donnees-essentielles-de-la-commande-publique/>
 - schéma de données DECP 2 [complexe à utiliser](https://github.com/ColinMaudry/decp-processing/issues/4)
 
 Vous trouverez des informations sur le contexte, le cadre réglementaire et les données de la commande publique sur [le wiki](https://github.com/ColinMaudry/decp-processing/wiki).
@@ -133,7 +131,7 @@ Attention, la version de prefect du client utilisé pour le déploiement et cell
 python src/deployments.py
 ```
 
-4. Le run se lancera tous les jours selon la configuration cron. Si tu souhaites exécuter le run maintenant :
+1. Le run se lancera tous les jours selon la configuration cron. Si tu souhaites exécuter le run maintenant :
 
 ```bash
 prefect deployment run decp-processing

pyproject.toml

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

reference/source_datasets.json

@@ -254,7 +254,7 @@
   },
   {
     "id": "5df410e86f44413a91d34be3",
-    "name": "AIFE_DE_13001977100018",
+    "name": "API DECP (AIFE DUME)",
     "owner_org_name": "Agence pour l'Informatique Financière de l'Etat",
     "code": "aife_dume"
   },
@@ -330,5 +330,59 @@
     "code": "minef",
     "owner_org_name": "MINEF",
     "skip": true
+  },
+  {
+    "id": "68c951f5501fc2243b7996b1",
+    "name": "Données Essentielles - www.proxilegales.fr",
+    "code": "modula_proxilegales",
+    "owner_org_name": "Modula"
+  },
+  {
+    "id": "68c93247fda0dc4d2aa52a86",
+    "name": "Données Essentielles - www.marches-demat.com",
+    "code": "modula_marches-demat",
+    "owner_org_name": "Modula"
+  },
+  {
+    "id": "68c9571b7d6a0e64a5f7582b",
+    "name": "Données Essentielles - www.laprovencemarchespublics.com et www.corsematinmarchespublics.com ",
+    "code": "modula_laprovence_corsematin",
+    "owner_org_name": "Modula"
+  },
+  {
+    "id": "68c955112e0cc622a82626b8",
+    "name": "Données Essentielles - www.achatspublicscorse.com",
+    "code": "modula_achatspublicscorse",
+    "owner_org_name": "Modula"
+  },
+  {
+    "id": "68c953c24aa801f3ce4289ea",
+    "name": "Données Essentielles - www.marchespublics.pro",
+    "code": "modula_marchespublics.pro",
+    "owner_org_name": "Modula"
+  },
+  {
+    "id": "68c959a6fecb75ff0b47f079",
+    "name": "Données Essentielles - www.antilles-legales.fr et www.guyane-legales.fr",
+    "code": "modula_antilles_guyane_legales",
+    "owner_org_name": "Modula"
+  },
+  {
+    "id": "68834dc0f3f93ee2ff6050e2",
+    "name": "Données essentielles de la commande publique - Eurométropole de Strasbourg",
+    "code": "eurometropole_strasbourg",
+    "owner_org_name": "Eurométropole de Strasbourg"
+  },
+  {
+    "id": "68834a9b7285ebd8466fb23e",
+    "name": "Données essentielles de la commande publique - Ville de Strasbourg",
+    "code": "ville_strasbourg",
+    "owner_org_name": "Eurométropole de Strasbourg"
+  },
+  {
+    "id": "67b4575f32c309fd76327c87",
+    "name": "Données Essentielles du profil acheteur Alsace Marchés Publics",
+    "code": "alsace_mp",
+    "owner_org_name": "Alsace marchés publics"
   }
 ]

src/tasks/get.py

@@ -86,6 +86,7 @@ def get_resource(
     output_path.parent.mkdir(exist_ok=True, parents=True)
     url = r["url"]
     file_format = r["format"]
+    logger.debug(f"Récupération de {r['dataset_code']} - {r['ori_filename']}")
     if file_format == "json":
         fields, decp_format = json_stream_to_parquet(url, output_path, r)
     elif file_format == "xml":
@@ -182,8 +183,12 @@ def json_stream_to_parquet(
     http_stream_iter = stream_get(url)
 
     stream_replace_iter = stream_replace_bytestring(
-        http_stream_iter, rb"NaN([,\n])", rb"null\1"
-    )  # Nan => null
+        http_stream_iter, b"\xef\xbb\xbf", b""
+    )  # Strip UTF-8 BOM
+
+    stream_replace_iter = stream_replace_bytestring(
+        stream_replace_iter, rb"NaN([,\n])", rb"null\1"
+    )  # NaN => null
 
     # Le dataset AWS scraping a pas mal de bugs de backslash
     if "/68caf6b135f19236a4f37a32/" in url or "/aws/" in url:
@@ -246,8 +251,6 @@ def json_stream_to_parquet(
 def xml_stream_to_parquet(
     url: str, output_path: Path, fix_chars=False
 ) -> tuple[set, DecpFormat]:
-    """Uniquement utilisé pour les données publiées par l'AIFE."""
-
     decp_format_2022 = DecpFormat("DECP 2022", SCHEMA_MARCHE_2022, "marches.marche")
 
     fields = set()
@@ -263,6 +266,10 @@ def xml_stream_to_parquet(
                 marche = parse_element(elem)
                 new_fields = write_marche_rows(marche, tmp_file, decp_format_2022)
                 fields = fields.union(new_fields)
+        # Vidage du buffer d'écriture sur le disque avant la lecture par son nom :
+        # sans cela, les petits fichiers (< taille du buffer) restent en mémoire et
+        # pl.scan_ndjson lit un fichier vide → 0 ligne (le chemin JSON, lui, fait seek(0)).
+        tmp_file.flush()
         lf = pl.scan_ndjson(tmp_file.name, schema=decp_format_2022.schema)
         sink_to_files(lf, output_path, file_format="parquet")
     return fields, decp_format_2022
@@ -441,9 +448,10 @@ def get_insee_cog_data(url, schema_overrides, columns) -> pl.DataFrame:
 
 def get_clean(
     resource, resources_artifact: list, available_parquet_files: set
-) -> pl.DataFrame or None:
+) -> pl.DataFrame | None:
     logger = get_logger(level=LOG_LEVEL)
 
+    logger.debug(f"get_clean {resource['ori_filename']}")
     with transaction():
         checksum = resource["checksum"]
         parquet_path = RESOURCE_CACHE_DIR / f"{checksum}"

tests/test_get.py

@@ -1,7 +1,11 @@
 import polars as pl
 
 from src.config import SIRET_LATLONG_SCHEMA
-from src.tasks.get import bootstrap_siret_latlong, get_etablissements
+from src.tasks.get import (
+    bootstrap_siret_latlong,
+    get_etablissements,
+    xml_stream_to_parquet,
+)
 
 REQUIRED_GEO_COLUMNS = {
     "numeroVoieEtablissement",
@@ -19,6 +23,35 @@ def test_get_etablissements_includes_geocoding_columns():
     assert not missing, f"Colonnes manquantes : {missing}"
 
 
+def test_xml_stream_to_parquet_small_file_is_not_empty(tmp_path):
+    """Régression : un petit XML (ndjson < taille du buffer d'écriture) doit
+    quand même produire des lignes. Sans flush() avant scan_ndjson, le buffer
+    n'était jamais vidé sur le disque et le parquet ressortait vide (0 ligne)."""
+    xml = (
+        '<?xml version="1.0" encoding="ISO-8859-15"?>\n'
+        "<marches>\n"
+        " <marche>\n"
+        "  <id>20261115446200</id>\n"
+        "  <acheteur><id>21130112200084</id></acheteur>\n"
+        "  <nature>March\xe9</nature>\n"
+        "  <montant>262578.28</montant>\n"
+        "  <dateNotification>2026-02-11</dateNotification>\n"
+        " </marche>\n"
+        "</marches>\n"
+    ).encode("ISO-8859-15")
+    xml_path = tmp_path / "decp_small.xml"
+    xml_path.write_bytes(xml)
+
+    output_path = tmp_path / "out"
+    xml_stream_to_parquet(str(xml_path), output_path, fix_chars=False)
+
+    df = pl.read_parquet(output_path.with_suffix(".parquet"))
+    assert df.height == 1
+    assert df["acheteur_id"].to_list() == ["21130112200084"]
+    # L'ISO-8859-15 doit être décodé correctement
+    assert df["nature"].to_list() == ["Marché"]
+
+
 def test_bootstrap_siret_latlong_produces_extended_schema(tmp_path, monkeypatch):
     fake_decp = tmp_path / "decp_fake.parquet"
     pl.DataFrame(