Merge branch 'feature/nouvelles_sources' into dev

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

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(