fix(airflow): fiabilise l'import horaire de la Mock API

etl/airflow/dags/mock_api_import.py

@@ -1,7 +1,7 @@
 """DAG d'import périodique des données de l'API Mock EnerVision (issue #15).
 
 Orchestre le pipeline existant `app.etl.mock_api_import` sans dupliquer sa logique ETL.
-Chaque exécution traite l'intervalle horaire Airflow précédent.
+Chaque exécution traite l'heure précédant son déclenchement.
 
 Le pipeline backend reste responsable de la validation, de la normalisation, du suivi de la
 qualité, de l'idempotence et du chargement dans PostgreSQL/TimescaleDB.
@@ -13,12 +13,14 @@ from datetime import datetime, timedelta
 
 from airflow.providers.standard.operators.bash import BashOperator
 from airflow.sdk import DAG
+from airflow.timetables.trigger import CronTriggerTimetable
 
 # Le backend possède son propre environnement uv dans l'image Airflow (ADR 0008).
 COMMANDE_BACKEND = "cd /opt/backend && env -u VIRTUAL_ENV uv run --no-sync python -m"
 
-# L'API et le pipeline backend plafonnent une réponse à 1 000 lectures par site.
-LIMITE_LECTURES = 60
+# Le pipeline backend et l'API acceptent au maximum 1 000 lectures par site.
+# Cette marge évite de perdre silencieusement une lecture si une heure en contient plus de 60.
+LIMITE_LECTURES = 1000
 
 # Deux reprises donnent trois tentatives au total. Même dans le pire cas, l'exécution reste
 # inférieure au pas horaire du DAG.
@@ -26,10 +28,19 @@ NOMBRE_REPRISES = 2
 DELAI_ENTRE_REPRISES = timedelta(minutes=2)
 PLAFOND_PAR_TENTATIVE = timedelta(minutes=10)
 
+# L'intervalle est déclaré explicitement pour ne pas dépendre de la valeur du paramètre Airflow
+# `create_cron_data_intervals`. Le déclenchement à :45 laisse quinze minutes avant `ml_score`,
+# exécuté à l'heure pile, puis avant `alertes`, exécuté à :15.
+PLANIFICATION = CronTriggerTimetable(
+    "45 * * * *",
+    timezone="UTC",
+    interval=timedelta(hours=1),
+)
+
 with DAG(
     dag_id="mock_api_import",
     description="Importe chaque heure les données de l'API Mock dans site et reading.",
-    schedule="@hourly",
+    schedule=PLANIFICATION,
     start_date=datetime(2026, 1, 1),
     catchup=False,
     # Deux exécutions simultanées pourraient demander et traiter le même intervalle.
@@ -40,8 +51,8 @@ with DAG(
         task_id="import_mock_api",
         bash_command=(
             f"{COMMANDE_BACKEND} app.etl.mock_api_import "
-            '--start-time "{{ data_interval_start.isoformat() }}" '
-            '--end-time "{{ data_interval_end.isoformat() }}" '
+            "--start-time \"{{ data_interval_start.strftime('%Y-%m-%dT%H:%M:%S') }}\" "
+            "--end-time \"{{ data_interval_end.strftime('%Y-%m-%dT%H:%M:%S') }}\" "
             f"--limit {LIMITE_LECTURES}"
         ),
         retries=NOMBRE_REPRISES,

etl/airflow/tests/test_dags.py

@@ -1,12 +1,13 @@
 """Tests d'integrite des DAGs : s'importent sans erreur, structure attendue. Pas d'execution
 reelle des taches (ca reclamerait le conteneur avec `uv`/`enervision_ml`), juste la definition."""
 
-from datetime import timedelta
+from datetime import datetime, timedelta
 from pathlib import Path
 
 import pytest
 from airflow.dag_processing.dagbag import DagBag
 from airflow.sdk import BaseOperator
+from airflow.timetables.trigger import CronTriggerTimetable
 
 DAGS_FOLDER = Path(__file__).resolve().parent.parent / "dags"
 
@@ -59,8 +60,17 @@ def test_historical_import_has_no_schedule(dagbag: DagBag) -> None:
     assert dagbag.dags["historical_import"].schedule is None
 
 
-def test_mock_api_import_runs_every_hour(dagbag: DagBag) -> None:
-    assert dagbag.dags["mock_api_import"].timetable.expression == "0 * * * *"
+def test_mock_api_import_uses_an_explicit_hourly_interval(dagbag: DagBag) -> None:
+    timetable = dagbag.dags["mock_api_import"].timetable
+
+    assert isinstance(timetable, CronTriggerTimetable)
+    assert timetable.serialize()["expression"] == "45 * * * *"
+
+    manual_interval = timetable.infer_manual_data_interval(
+        run_after=datetime.fromisoformat("2026-09-22T12:30:00+00:00"),
+    )
+
+    assert manual_interval.end - manual_interval.start == timedelta(hours=1)
 
 
 def test_ml_train_task_calls_the_training_module(dagbag: DagBag) -> None:
@@ -104,9 +114,9 @@ def test_mock_api_import_calls_the_existing_backend_module(dagbag: DagBag) -> No
 def test_mock_api_import_uses_the_airflow_data_interval(dagbag: DagBag) -> None:
     commande = dagbag.dags["mock_api_import"].get_task("import_mock_api").bash_command
 
-    assert '--start-time "{{ data_interval_start.isoformat() }}"' in commande
-    assert '--end-time "{{ data_interval_end.isoformat() }}"' in commande
-    assert "--limit 60" in commande
+    assert "--start-time \"{{ data_interval_start.strftime('%Y-%m-%dT%H:%M:%S') }}\"" in commande
+    assert "--end-time \"{{ data_interval_end.strftime('%Y-%m-%dT%H:%M:%S') }}\"" in commande
+    assert "--limit 1000" in commande
 
 
 @pytest.mark.parametrize("task_id", ["detection", "recommandations"])