Implémentation de build_database avec transforms Polars et tables dérivées

- _load_source_frame reprend la pipeline Polars (sort, filtre donneesActuelles,
  booleans_to_strings, remplacement des noms null)
- build_database utilise write_parquet + read_parquet pour zéro-dépendance
  pyarrow (pyarrow absent du venv) et écrit atomiquement via .tmp + os.replace
- 4 tables dérivées : acheteurs_marches, titulaires_marches,
  acheteurs_departement, titulaires_departement

refs #71

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

src/db.py

@@ -1,28 +1,109 @@
 import logging
 import os
 from pathlib import Path
+from time import sleep
+
+import duckdb
+import polars as pl
+import polars.selectors as cs
+from polars.exceptions import ComputeError
 
 logger = logging.getLogger("decp.info")
 
 
 def should_rebuild(db_path: Path, parquet_path: Path) -> bool:
-    """Decide whether to rebuild the DuckDB database from the source Parquet.
-
-    Rules:
-      - Rebuild if the DuckDB file does not exist.
-      - Otherwise, rebuild only if the source Parquet is newer than the DB,
-        EXCEPT in development mode without REBUILD_DUCKDB=true (dev keeps
-        a stable DB across reloads unless explicitly opted in).
-    """
     db_path = Path(db_path)
     parquet_path = Path(parquet_path)
-
     if not db_path.exists():
         return True
-
     dev = os.getenv("DEVELOPMENT", "False").lower() == "true"
     force = os.getenv("REBUILD_DUCKDB", "False").lower() == "true"
     if dev and not force:
         return False
-
     return parquet_path.stat().st_mtime > db_path.stat().st_mtime
+
+
+def _load_source_frame(parquet_path: Path) -> pl.DataFrame:
+    """Read the source parquet and apply the row-level transforms.
+
+    Kept here (not in utils.py) so src.db has no dependency on utils.
+    Mirrors the behavior previously in utils.get_decp_data().
+    """
+    try:
+        lff: pl.LazyFrame = pl.scan_parquet(str(parquet_path))
+    except ComputeError:
+        logger.info("Lecture du parquet échouée, nouvelle tentative dans 10s...")
+        sleep(10)
+        lff = pl.scan_parquet(str(parquet_path))
+
+    lff = lff.sort(by=["dateNotification", "uid"], descending=True, nulls_last=True)
+    lff = lff.filter(pl.col("donneesActuelles")).drop("donneesActuelles")
+
+    # booleans_to_strings: true → "oui", false → "non"
+    lff = lff.with_columns(
+        pl.col(cs.Boolean)
+        .cast(pl.String)
+        .str.replace("true", "oui")
+        .str.replace("false", "non")
+    )
+
+    for col in ["acheteur_nom", "titulaire_nom"]:
+        lff = lff.with_columns(
+            pl.when(pl.col(col).is_null())
+            .then(pl.lit("[Identifiant non reconnu dans la base INSEE]"))
+            .otherwise(pl.col(col))
+            .name.keep()
+        )
+
+    return lff.collect()
+
+
+def build_database(db_path: Path, parquet_path: Path) -> None:
+    """Build the DuckDB database atomically under an exclusive lock.
+
+    Caller MUST hold the fcntl.flock on the .lock file.
+    """
+    db_path = Path(db_path)
+    parquet_path = Path(parquet_path)
+    tmp_path = db_path.with_suffix(".duckdb.tmp")
+    staging_parquet = db_path.with_suffix(".staging.parquet")
+    if tmp_path.exists():
+        tmp_path.unlink()
+
+    logger.info(f"Construction de la base DuckDB à partir de {parquet_path}...")
+    frame = _load_source_frame(parquet_path)
+
+    # Write transformed frame as parquet so DuckDB can read it natively
+    # (avoids pyarrow dependency for the Polars→DuckDB handoff)
+    frame.write_parquet(str(staging_parquet))
+    try:
+        with duckdb.connect(str(tmp_path)) as w:
+            w.execute(
+                f"CREATE TABLE decp AS SELECT * FROM read_parquet('{staging_parquet}')"
+            )
+            w.execute(
+                "CREATE TABLE acheteurs_marches AS "
+                "SELECT DISTINCT uid, objet, acheteur_id FROM decp "
+                "ORDER BY acheteur_id"
+            )
+            w.execute(
+                "CREATE TABLE titulaires_marches AS "
+                "SELECT DISTINCT uid, objet, titulaire_id FROM decp "
+                "ORDER BY titulaire_id"
+            )
+            w.execute(
+                "CREATE TABLE acheteurs_departement AS "
+                "SELECT DISTINCT acheteur_id, acheteur_nom, acheteur_departement_code "
+                "FROM decp ORDER BY acheteur_nom"
+            )
+            w.execute(
+                "CREATE TABLE titulaires_departement AS "
+                "SELECT DISTINCT titulaire_id, titulaire_nom, titulaire_departement_code "
+                "FROM decp ORDER BY titulaire_nom"
+            )
+    finally:
+        if staging_parquet.exists():
+            staging_parquet.unlink()
+
+    os.replace(tmp_path, db_path)
+    logger.info(f"Base DuckDB construite : {db_path}")

tests/test_db.py

@@ -1,6 +1,8 @@
+import datetime
 import os
 import time
 
+import polars as pl
 import pytest
 
 from src.db import should_rebuild
@@ -65,3 +67,129 @@ def test_should_rebuild_dev_when_rebuild_forced(parquet_and_db, monkeypatch):
     monkeypatch.setenv("DEVELOPMENT", "true")
     monkeypatch.setenv("REBUILD_DUCKDB", "true")
     assert should_rebuild(db, parquet) is True
+
+
+@pytest.fixture
+def built_db(tmp_path, monkeypatch):
+    """Build a DuckDB from a small Polars frame written as parquet."""
+    parquet_path = tmp_path / "source.parquet"
+    db_path = tmp_path / "decp.duckdb"
+
+    data = pl.DataFrame(
+        [
+            {
+                "uid": "1",
+                "id": "1",
+                "objet": "Travaux",
+                "acheteur_id": "A1",
+                "acheteur_nom": "Mairie",
+                "acheteur_departement_code": "75",
+                "titulaire_id": "T1",
+                "titulaire_nom": "Entreprise",
+                "titulaire_departement_code": "35",
+                "titulaire_typeIdentifiant": "SIRET",
+                "montant": 1000.0,
+                "dateNotification": datetime.date(2025, 1, 1),
+                "donneesActuelles": True,
+                "marcheInnovant": True,
+            },
+            {
+                "uid": "2",
+                "id": "2",
+                "objet": "Études",
+                "acheteur_id": "A1",
+                "acheteur_nom": "Mairie",
+                "acheteur_departement_code": "75",
+                "titulaire_id": "T2",
+                "titulaire_nom": None,
+                "titulaire_departement_code": "75",
+                "titulaire_typeIdentifiant": "SIRET",
+                "montant": 500.0,
+                "dateNotification": datetime.date(2024, 6, 1),
+                "donneesActuelles": True,
+                "marcheInnovant": False,
+            },
+            {
+                "uid": "3",
+                "id": "3",
+                "objet": "Ancien",
+                "acheteur_id": "A2",
+                "acheteur_nom": None,
+                "acheteur_departement_code": "13",
+                "titulaire_id": "T3",
+                "titulaire_nom": "Autre",
+                "titulaire_departement_code": "13",
+                "titulaire_typeIdentifiant": "SIRET",
+                "montant": 100.0,
+                "dateNotification": datetime.date(2023, 1, 1),
+                "donneesActuelles": False,  # must be filtered out
+                "marcheInnovant": False,
+            },
+        ]
+    )
+    data.write_parquet(parquet_path)
+    monkeypatch.setenv("DATA_FILE_PARQUET_PATH", str(parquet_path))
+
+    from src.db import build_database
+
+    build_database(db_path, parquet_path)
+    return db_path
+
+
+def test_build_filters_donnees_actuelles(built_db):
+    import duckdb
+
+    with duckdb.connect(str(built_db), read_only=True) as c:
+        rows = c.execute("SELECT uid FROM decp ORDER BY uid").fetchall()
+    assert [r[0] for r in rows] == ["1", "2"]
+
+
+def test_build_converts_booleans_to_oui_non(built_db):
+    import duckdb
+
+    with duckdb.connect(str(built_db), read_only=True) as c:
+        values = c.execute("SELECT marcheInnovant FROM decp ORDER BY uid").fetchall()
+    assert [v[0] for v in values] == ["oui", "non"]
+
+
+def test_build_replaces_null_org_names(built_db):
+    import duckdb
+
+    with duckdb.connect(str(built_db), read_only=True) as c:
+        titulaire_2 = c.execute(
+            "SELECT titulaire_nom FROM decp WHERE uid = '2'"
+        ).fetchone()
+    assert titulaire_2[0] == "[Identifiant non reconnu dans la base INSEE]"
+
+
+def test_build_creates_derived_tables(built_db):
+    import duckdb
+
+    with duckdb.connect(str(built_db), read_only=True) as c:
+        tables = {r[0] for r in c.execute("SHOW TABLES").fetchall()}
+    assert {
+        "decp",
+        "acheteurs_marches",
+        "titulaires_marches",
+        "acheteurs_departement",
+        "titulaires_departement",
+    } <= tables
+
+
+@pytest.mark.skip(reason="implemented in Task 6")
+def test_query_marches_returns_polars_frame(built_db, monkeypatch):
+    monkeypatch.setenv(
+        "DATA_FILE_PARQUET_PATH", str(built_db.parent / "source.parquet")
+    )
+    # Force src.db to load pointing at this test DB.
+    import importlib
+
+    import src.db
+
+    importlib.reload(src.db)
+    from src.db import query_marches
+
+    frame = query_marches("acheteur_id = ?", ("A1",))
+    assert isinstance(frame, pl.DataFrame)
+    assert frame.height == 2
+    assert set(frame["uid"].to_list()) == {"1", "2"}