Merge branch 'hotfix/2.6.4'

Author: ColinMaudry

CHANGELOG.md

@@ -1,15 +1,20 @@
-### 2.6.3 2025-12-16
+#### 2.6.4 2025-12-19
+
+- Tri et numérotation des modifications après la concaténation plutôt que par ressource, pour réduire le nombre de doublons ([#156](https://github.com/ColinMaudry/decp-processing/issues/156))
+- Utilisation du logger de prefect plûtot que `log_prints=True`
+
+#### 2.6.3 2025-12-16
 
 - Téléchargement des ressources plus résilient aux erreurs ([tenacity](https://tenacity.readthedocs.io/en/latest/))
 - Téléchargement des données établissements plus résilient aux erreurs ([tenacity](https://tenacity.readthedocs.io/en/latest/))
 
-### 2.6.2 2025-12-15
+#### 2.6.2 2025-12-15
 
 - Réduction du nombre de tâches prefect pour réduire la charge sur la BDD et la latence
 - Utilisation du multithreading standard de Python plutôt que celui de Prefect
 - Le nom d'établissement n'est ajouté entre parenthèses que s'il est différent de celui de l'unité légale
 
-### 2.6.1 2025-12-14
+#### 2.6.1 2025-12-14
 
 - Séparation des fichiers de référence et des fichiers de données
 - Réorganisation des variables d'environnement

pyproject.toml

@@ -42,11 +42,13 @@ testpaths = [
     "tests",
 ]
 env = [
+    "LOG_LEVEL=DEBUG",
     "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"
+    "DECP_USE_CACHE=false",
+    "MAX_PREFECT_WORKERS=1"
 ]
 addopts = "-p no:warnings"
 

src/config.py

@@ -9,11 +9,14 @@
 import httpx
 from dotenv import find_dotenv, load_dotenv
 from ijson import sendable_list
+from prefect.logging import get_logger
 from prefect.variables import Variable
 
+logger = get_logger(__name__)
+
 dotenv_path = find_dotenv()
 if dotenv_path == "":
-    print("Création du fichier .env à partir de template.env")
+    logger.info("Création du fichier .env à partir de template.env")
     template_dotenv_path = Path(find_dotenv("template.env"))
     dotenv_path = template_dotenv_path.with_name(".env")
     shutil.copyfile(template_dotenv_path, dotenv_path)
@@ -28,6 +31,9 @@ def make_path_from_env(env: str, alternative_path: Path) -> Path:
 
 ALL_CONFIG = {}
 
+# Niveau des logs
+LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
+
 # Nombre maximal de workers utilisables par Prefect. Défaut : 16
 MAX_PREFECT_WORKERS = int(os.getenv("MAX_PREFECT_WORKERS", 4))
 ALL_CONFIG["MAX_PREFECT_WORKERS"] = MAX_PREFECT_WORKERS

src/flows/decp_processing.py

@@ -15,6 +15,7 @@
     DATE_NOW,
     DECP_PROCESSING_PUBLISH,
     DIST_DIR,
+    LOG_LEVEL,
     MAX_PREFECT_WORKERS,
     PREFECT_API_URL,
     RESOURCE_CACHE_DIR,
@@ -32,22 +33,26 @@
     calculate_naf_cpv_matching,
     concat_parquet_files,
     sort_columns,
+    sort_modifications,
 )
 from src.tasks.utils import (
     full_resource_name,
     generate_stats,
+    get_logger,
     print_all_config,
     remove_unused_cache,
 )
 
 
 @flow(log_prints=True)
 def decp_processing(enable_cache_removal: bool = True):
-    print("🚀  Début du flow decp-processing")
+    logger = get_logger(level=LOG_LEVEL)
+
+    logger.info("🚀  Début du flow decp-processing")
 
     print_all_config(ALL_CONFIG)
 
-    print("Liste de toutes les ressources des datasets...")
+    logger.info("Liste de toutes les ressources des datasets...")
     resources: list[dict] = list_resources(TRACKED_DATASETS)
 
     # Initialisation du tableau des artifacts de ressources
@@ -84,10 +89,13 @@ def decp_processing(enable_cache_removal: bool = True):
         )
         del resources_artifact
 
-    print("Fusion des dataframes...")
+    logger.info("Concaténation des dataframes...")
     lf: pl.LazyFrame = concat_parquet_files(parquet_files)
 
-    print("Ajout des données SIRENE...")
+    logger.info("Tri des modifications...")
+    lf = sort_modifications(lf)
+
+    logger.info("Ajout des données SIRENE...")
     # 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)
@@ -104,17 +112,19 @@ def decp_processing(enable_cache_removal: bool = True):
     sink_to_files(lf, DIST_DIR / "decp", file_format="parquet")
     lf: pl.LazyFrame = pl.scan_parquet(DIST_DIR / "decp.parquet")
 
-    print("Ajout de la colonne 'dureeRestanteMois'...")
+    logger.info("Ajout de la colonne 'dureeRestanteMois'...")
     lf = add_duree_restante(lf)
 
-    print("Génération des probabilités NAF/CPV...")
+    logger.info("Génération des probabilités NAF/CPV...")
     calculate_naf_cpv_matching(lf)
     lf = lf.drop(cs.starts_with("activite"))
 
-    print("Génération de l'artefact (statistiques) sur le base df...")
+    logger.info("Génération de l'artefact (statistiques) sur le base df...")
     generate_stats(lf)
 
-    print("Génération du schéma et enregistrement des DECP aux formats CSV, Parquet...")
+    logger.info(
+        "Génération du schéma et enregistrement des DECP aux formats CSV, Parquet..."
+    )
     lf: pl.LazyFrame = sort_columns(lf, BASE_DF_COLUMNS)
     generate_final_schema(lf)
     sink_to_files(lf, DIST_DIR / "decp")
@@ -124,16 +134,16 @@ def decp_processing(enable_cache_removal: bool = True):
     # make_data_tables()
 
     if decp_publish:
-        print("Publication sur data.gouv.fr...")
+        logger.info("Publication sur data.gouv.fr...")
         publish_to_datagouv()
     else:
-        print("Publication sur data.gouv.fr désactivée.")
+        logger.info("Publication sur data.gouv.fr désactivée.")
 
     if enable_cache_removal:
-        print("Suppression des fichiers de cache inutilisés...")
+        logger.info("Suppression des fichiers de cache inutilisés...")
         remove_unused_cache()
 
-    print("☑️  Fin du flow principal decp_processing.")
+    logger.info("☑️  Fin du flow principal decp_processing.")
 
 
 @task(retries=2)
@@ -145,8 +155,9 @@ def process_batch(
     resources_artifact,
     resources_to_process,
 ):
+    logger = get_logger(level=LOG_LEVEL)
     batch = resources_to_process[i : i + batch_size]
-    print(
+    logger.info(
         f"🗃️ Traitement du lot {i // batch_size + 1} / {len(resources_to_process) // batch_size + 1}"
     )
     futures = {}
@@ -164,8 +175,10 @@ def process_batch(
                 parquet_files.append(result)
         except Exception as e:
             resource_name = futures[future]
-            print(f"❌ Erreur de traitement de {resource_name} ({type(e).__name__}):")
-            print(e)
+            logger.error(
+                f"❌ Erreur de traitement de {resource_name} ({type(e).__name__}):"
+            )
+            logger.info(e)
     # Nettoyage explicite
     futures.clear()
 

src/flows/get_cog.py

@@ -1,15 +1,19 @@
 import polars as pl
 from prefect import flow
 
-from src.config import DATA_DIR
+from src.config import DATA_DIR, LOG_LEVEL
 from src.tasks.get import get_insee_cog_data
+from src.tasks.utils import get_logger
 
 
 @flow(log_prints=True)
 def get_cog():
     """Téléchargement et préparation des données du Code Officiel Géographique"""
 
-    print("Téléchargement et préparation des données du Code Officiel Géographique...")
+    logger = get_logger(level=LOG_LEVEL)
+    logger.info(
+        "Téléchargement et préparation des données du Code Officiel Géographique..."
+    )
 
     # # # # # # # # #
     # Communes      #

src/flows/scrap.py

@@ -3,16 +3,25 @@
 
 from prefect import flow
 
-from src.config import DATE_NOW, DIST_DIR, MONTH_NOW, SCRAPING_MODE, SCRAPING_TARGET
+from src.config import (
+    DATE_NOW,
+    DIST_DIR,
+    LOG_LEVEL,
+    MONTH_NOW,
+    SCRAPING_MODE,
+    SCRAPING_TARGET,
+)
 from src.tasks.scrap import scrap_aws_month, scrap_marches_securises_month
+from src.tasks.utils import get_logger
 
 
 @flow(log_prints=True)
 def scrap(target: str = None, mode: str = None, month=None, year=None):
+    logger = get_logger(level=LOG_LEVEL)
     # Remise à zéro du dossier dist
     dist_dir: Path = DIST_DIR / target
     if dist_dir.exists():
-        print(f"Suppression de {dist_dir}...")
+        logger.debug(f"Suppression de {dist_dir}...")
         rmtree(dist_dir)
     else:
         dist_dir.mkdir(parents=True)
@@ -26,7 +35,7 @@ def scrap(target: str = None, mode: str = None, month=None, year=None):
     elif target == "marches-securises.fr":
         scrap_target_month = scrap_marches_securises_month
     else:
-        print("Quel target ?")
+        logger.error("Quel target ?")
         raise ValueError
 
     current_year = DATE_NOW[:4]
@@ -52,4 +61,4 @@ def scrap(target: str = None, mode: str = None, month=None, year=None):
             scrap(target=target, mode="year", year=str(year))
 
     else:
-        print("Mauvaise configuration")
+        logger.error("Mauvaise configuration")

src/flows/sirene_preprocess.py

@@ -1,11 +1,11 @@
 from prefect import flow
 from prefect.transactions import transaction
 
-from src.config import SIRENE_DATA_DIR
+from src.config import LOG_LEVEL, SIRENE_DATA_DIR
 from src.flows.get_cog import get_cog
 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
+from src.tasks.utils import create_sirene_data_dir, get_logger
 
 
 @flow(log_prints=True)
@@ -14,7 +14,9 @@ def sirene_preprocess():
     Pour chaque ressource (unités légales, établissements), un fichier parquet est produit.
     """
 
-    print("🚀  Pré-traitement des données SIRENE")
+    logger = get_logger(level=LOG_LEVEL)
+
+    logger.info("🚀  Pré-traitement des données SIRENE")
     # Soit les tâches de ce flow vont au bout (success), soit le dossier SIRENE_DATA_DIR est supprimé (voir remove_sirene_data_dir())
     with transaction():
         create_sirene_data_dir()
@@ -25,18 +27,18 @@ def sirene_preprocess():
         # préparer les données unités légales
         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...")
+            logger.info("Téléchargement et préparation des unités légales...")
             get_unite_legales(processed_ul_parquet_path)
         else:
-            print(processed_ul_parquet_path, " existe, skipping.")
+            logger.info(processed_ul_parquet_path, " existe, skipping.")
 
         # préparer les données établissements
         processed_etab_parquet_path = SIRENE_DATA_DIR / "etablissements.parquet"
         if not processed_etab_parquet_path.exists():
-            print("Téléchargement et préparation des établissements...")
+            logger.info("Téléchargement et préparation des établissements...")
             lf = get_etablissements()
             prepare_etablissements(lf).sink_parquet(processed_etab_parquet_path)
         else:
-            print(processed_etab_parquet_path, " existe, skipping.")
+            logger.info(processed_etab_parquet_path, " existe, skipping.")
 
-    print("☑️  Fin du flow sirene_preprocess.")
+    logger.info("☑️  Fin du flow sirene_preprocess.")