Merge branch '111-mcp-tools' into dev

src/app.py

@@ -74,6 +74,8 @@ cache.init_app(
     },
 )
 
+_mcp_enabled = os.getenv("DASH_MCP_ENABLED") == "true"
+
 app: Dash = Dash(
     server=server,
     # name="src" (et non "src.app") pour que use_pages enregistre les pages sous
@@ -87,6 +89,7 @@ app: Dash = Dash(
     use_pages=True,
     suppress_callback_exceptions=True,
     compress=True,
+    enable_mcp=_mcp_enabled,
     meta_tags=META_TAGS,
 )
 
@@ -108,6 +111,19 @@ from src.api import init_api  # noqa: E402  # inline: src.db.conn must be ready
 
 init_api(app.server)
 
+# Serveur MCP (issue #111, scope A) : n'expose QUE les fonctions @mcp_enabled,
+# jamais les callbacks/layout/pages d'UI. Activé via DASH_MCP_ENABLED=true.
+if _mcp_enabled:
+    from dash.mcp import configure_mcp_server  # noqa: E402
+
+    configure_mcp_server(
+        include_layout=False,
+        include_callbacks=False,
+        include_pages=False,
+        include_clientside_callbacks=False,
+    )
+    import src.mcp.tools  # noqa: E402,F401  # l'import enregistre les @mcp_enabled
+
 from src.subscriptions.setup import init_subscriptions  # noqa: E402
 
 init_subscriptions(app.server)

src/mcp/queries.py

@@ -0,0 +1,222 @@
+# src/mcp/queries.py
+import re
+
+from src.api.filters import FilterError, build_where
+from src.db import aggregate_marches, count_marches, query_marches
+from src.db import schema as duckdb_schema
+from src.mcp.serialization import to_json_records
+from src.utils.data import DF_ACHETEURS, DF_TITULAIRES
+from src.utils.search import search_org
+
+PAGE_SIZE = 50
+TOP_N = 10
+ORG_FRAMES = {"acheteur": DF_ACHETEURS, "titulaire": DF_TITULAIRES}
+
+# Colonnes renvoyées par rechercher_marches (sortie ciblée, pas SELECT *).
+MARCHES_COLUMNS = [
+    "uid",
+    "objet",
+    "montant",
+    "dateNotification",
+    "codeCPV",
+    "acheteur_id",
+    "acheteur_nom",
+    "acheteur_departement_code",
+    "titulaire_id",
+    "titulaire_nom",
+]
+
+# (param nommé, colonne decp, opérateur du moteur de filtres API).
+# `greater` = >=, `less` = <=, `contains` = LIKE %v%, `exact` = =.
+_NAMED_FILTERS = [
+    ("acheteur_id", "acheteur_id", "exact"),
+    ("titulaire_id", "titulaire_id", "exact"),
+    ("cpv", "codeCPV", "contains"),
+    ("objet_contient", "objet", "contains"),
+    ("montant_min", "montant", "greater"),
+    ("montant_max", "montant", "less"),
+    ("date_min", "dateNotification", "greater"),
+    ("date_max", "dateNotification", "less"),
+    ("departement", "acheteur_departement_code", "exact"),
+]
+
+
+def build_where_args(
+    named: dict, filtres_avances: dict | None
+) -> list[tuple[str, str]]:
+    """Traduit les paramètres nommés + filtres avancés en tuples (col__op, valeur)."""
+    args: list[tuple[str, str]] = []
+    for param, col, op in _NAMED_FILTERS:
+        value = named.get(param)
+        if value is not None:
+            args.append((f"{col}__{op}", str(value)))
+    if filtres_avances:
+        for key, value in filtres_avances.items():
+            args.append((key, str(value)))
+    return args
+
+
+def search_marches(
+    *,
+    acheteur_id: str | None = None,
+    titulaire_id: str | None = None,
+    cpv: str | None = None,
+    objet_contient: str | None = None,
+    montant_min: float | None = None,
+    montant_max: float | None = None,
+    date_min: str | None = None,
+    date_max: str | None = None,
+    departement: str | None = None,
+    page: int = 1,
+    filtres_avances: dict | None = None,
+) -> dict:
+    """Recherche paginée de marchés. Même sémantique de filtres que l'API REST."""
+    named = {
+        "acheteur_id": acheteur_id,
+        "titulaire_id": titulaire_id,
+        "cpv": cpv,
+        "objet_contient": objet_contient,
+        "montant_min": montant_min,
+        "montant_max": montant_max,
+        "date_min": date_min,
+        "date_max": date_max,
+        "departement": departement,
+    }
+    args = build_where_args(named, filtres_avances)
+    try:
+        where_sql, params, order_sql = build_where(args, duckdb_schema)
+    except FilterError as e:
+        return {"error": str(e), "champ": e.field}
+
+    page = max(1, int(page))
+    offset = (page - 1) * PAGE_SIZE
+    order_by = order_sql or '"dateNotification" DESC, "uid" DESC'
+    df = query_marches(
+        where_sql,
+        params,
+        columns=MARCHES_COLUMNS,
+        order_by=order_by,
+        limit=PAGE_SIZE,
+        offset=offset,
+    )
+    total = count_marches(where_sql, params)
+    return {
+        "meta": {"page": page, "page_size": PAGE_SIZE, "total": total},
+        "marches": to_json_records(df),
+    }
+
+
+def _extract_plain_text(html_str: str) -> str:
+    """Extract plain text from HTML link, e.g. '<a...>123</a>' -> '123'."""
+    match = re.search(r">([^<]+)<", html_str)
+    return match.group(1) if match else html_str
+
+
+def search_organisations(
+    query: str, org_type: str = "acheteur", limite: int = 20
+) -> list[dict]:
+    """Recherche des organisations par nom, résout un nom -> id."""
+    if org_type not in ORG_FRAMES:
+        raise ValueError(
+            f"type invalide: {org_type!r} (attendu 'acheteur' ou 'titulaire')"
+        )
+    df = search_org(ORG_FRAMES[org_type], query, org_type, track=False).head(limite)
+    return [
+        {
+            "id": _extract_plain_text(r[f"{org_type}_id"]),
+            "nom": _extract_plain_text(r[f"{org_type}_nom"]),
+            "departement": r.get("Département"),
+        }
+        for r in df.to_dicts()
+    ]
+
+
+def _org_identite(org_type: str, org_id: str) -> dict:
+    df = query_marches(
+        where_sql=f'"{org_type}_id" = ?',
+        params=[org_id],
+        columns=[
+            f"{org_type}_id",
+            f"{org_type}_nom",
+            f"{org_type}_departement_nom",
+            f"{org_type}_commune_nom",
+        ],
+        limit=1,
+    )
+    if df.height == 0:
+        return {"id": org_id, "nom": None, "departement": None, "commune": None}
+    r = df.row(0, named=True)
+    return {
+        "id": org_id,
+        "nom": r[f"{org_type}_nom"],
+        "departement": r[f"{org_type}_departement_nom"],
+        "commune": r[f"{org_type}_commune_nom"],
+    }
+
+
+def compute_org_stats(org_type: str, org_id: str) -> dict:
+    """Statistiques agrégées d'un acheteur ou titulaire."""
+    if org_type not in ORG_FRAMES:
+        raise ValueError(
+            f"type invalide: {org_type!r} (attendu 'acheteur' ou 'titulaire')"
+        )
+    other = "titulaire" if org_type == "acheteur" else "acheteur"
+    where_sql = f'"{org_type}_id" = ?'
+    params = [org_id]
+    identite = _org_identite(org_type, org_id)
+
+    totals = aggregate_marches(
+        select_sql='COUNT("uid") AS nb, COALESCE(SUM("montant"), 0) AS montant_total',
+        where_sql=where_sql,
+        params=params,
+    )
+    nb = int(totals["nb"][0])
+    if nb == 0:
+        return {
+            "identite": identite,
+            "nb_marches": 0,
+            "montant_total": 0.0,
+            "repartition_annuelle": [],
+            f"top_{other}s": [],
+            "top_cpv": [],
+        }
+
+    annuelle = aggregate_marches(
+        select_sql=(
+            "CAST(date_part('year', \"dateNotification\") AS INTEGER) AS annee, "
+            'COUNT("uid") AS nb_marches, '
+            'COALESCE(SUM("montant"), 0) AS montant_total'
+        ),
+        where_sql=where_sql,
+        params=params,
+        group_by="date_part('year', \"dateNotification\")",
+        order_by="annee",
+    )
+    top_other = aggregate_marches(
+        select_sql=(
+            f'"{other}_id" AS id, any_value("{other}_nom") AS nom, '
+            'COUNT("uid") AS nb_marches, '
+            'COALESCE(SUM("montant"), 0) AS montant_total'
+        ),
+        where_sql=where_sql,
+        params=params,
+        group_by=f'"{other}_id"',
+        order_by="nb_marches DESC",
+        limit=TOP_N,
+    )
+    top_cpv = aggregate_marches(
+        select_sql='"codeCPV" AS cpv, COUNT("uid") AS nb_marches',
+        where_sql=where_sql,
+        params=params,
+        group_by='"codeCPV"',
+        order_by="nb_marches DESC",
+        limit=TOP_N,
+    )
+    return {
+        "identite": identite,
+        "nb_marches": nb,
+        "montant_total": float(totals["montant_total"][0]),
+        "repartition_annuelle": to_json_records(annuelle),
+        f"top_{other}s": to_json_records(top_other),
+        "top_cpv": to_json_records(top_cpv),
+    }

src/mcp/serialization.py

@@ -0,0 +1,21 @@
+import datetime
+from typing import Any
+
+import polars as pl
+
+
+def to_json_records(df: pl.DataFrame) -> list[dict[str, Any]]:
+    """Convertit un DataFrame Polars en liste de dicts JSON-sérialisables.
+
+    Les dates/datetimes deviennent des chaînes ISO 8601 ; les valeurs nulles
+    restent None. Utilisé par tous les tools MCP pour produire une sortie propre.
+    """
+    return [
+        {key: _jsonify(value) for key, value in row.items()} for row in df.to_dicts()
+    ]
+
+
+def _jsonify(value: Any) -> Any:
+    if isinstance(value, (datetime.date, datetime.datetime)):
+        return value.isoformat()
+    return value

src/mcp/tools.py

@@ -0,0 +1,86 @@
+from dash.mcp import mcp_enabled
+
+from src.mcp import queries
+from src.utils.tracking import track_mcp_tool
+
+
+@mcp_enabled(name="rechercher_organisations", expose_docstring=True)
+def rechercher_organisations(
+    query: str,
+    type: str = "acheteur",
+    limite: int = 20,  # noqa: A002
+) -> list[dict]:
+    """Recherche des acheteurs ou titulaires publics par nom.
+
+    Utiliser en premier pour résoudre un nom d'organisation vers son
+    identifiant, à passer ensuite à stats_acheteur / stats_titulaire.
+
+    query: texte libre (nom d'organisation).
+    type: "acheteur" ou "titulaire".
+    Retourne une liste de {id, nom, departement}.
+    """
+    track_mcp_tool("rechercher_organisations", query=query)
+    return queries.search_organisations(query, type, limite)
+
+
+@mcp_enabled(name="stats_acheteur", expose_docstring=True)
+def stats_acheteur(acheteur_id: str) -> dict:
+    """Statistiques agrégées d'un acheteur public (par identifiant).
+
+    Retourne nombre de marchés, montant total, répartition annuelle,
+    principaux titulaires et principaux codes CPV.
+    """
+    track_mcp_tool("stats_acheteur")
+    return queries.compute_org_stats("acheteur", acheteur_id)
+
+
+@mcp_enabled(name="stats_titulaire", expose_docstring=True)
+def stats_titulaire(titulaire_id: str) -> dict:
+    """Statistiques agrégées d'un titulaire (entreprise) par identifiant.
+
+    Retourne nombre de marchés remportés, montant total, répartition
+    annuelle, principaux acheteurs et principaux codes CPV.
+    """
+    track_mcp_tool("stats_titulaire")
+    return queries.compute_org_stats("titulaire", titulaire_id)
+
+
+@mcp_enabled(name="rechercher_marches", expose_docstring=True)
+def rechercher_marches(
+    acheteur_id: str | None = None,
+    titulaire_id: str | None = None,
+    cpv: str | None = None,
+    objet_contient: str | None = None,
+    montant_min: float | None = None,
+    montant_max: float | None = None,
+    date_min: str | None = None,
+    date_max: str | None = None,
+    departement: str | None = None,
+    page: int = 1,
+    filtres_avances: dict | None = None,
+) -> dict:
+    """Recherche paginée de marchés publics (DECP).
+
+    Filtres nommés : acheteur_id, titulaire_id, cpv (code CPV, correspondance
+    partielle), objet_contient (texte de l'objet), montant_min, montant_max,
+    date_min / date_max (format YYYY-MM-DD, sur dateNotification),
+    departement (code département de l'acheteur).
+    filtres_avances : dict optionnel {"colonne__operateur": valeur} pour les
+    besoins pointus (mêmes colonnes/opérateurs que l'API REST colibre).
+    page : numéro de page (50 résultats par page).
+    Retourne {meta: {page, page_size, total}, marches: [...]}.
+    """
+    track_mcp_tool("rechercher_marches", query=objet_contient)
+    return queries.search_marches(
+        acheteur_id=acheteur_id,
+        titulaire_id=titulaire_id,
+        cpv=cpv,
+        objet_contient=objet_contient,
+        montant_min=montant_min,
+        montant_max=montant_max,
+        date_min=date_min,
+        date_max=date_max,
+        departement=departement,
+        page=page,
+        filtres_avances=filtres_avances,
+    )

src/utils/search.py

@@ -5,7 +5,9 @@ from src.utils.table import add_links
 from src.utils.tracking import track_search
 
 
-def search_org(dff: pl.DataFrame, query: str, org_type: str) -> pl.DataFrame:
+def search_org(
+    dff: pl.DataFrame, query: str, org_type: str, track: bool = True
+) -> pl.DataFrame:
     """
     Search in either 'acheteur' or 'titulaire' DataFrame.
 
@@ -18,7 +20,8 @@ def search_org(dff: pl.DataFrame, query: str, org_type: str) -> pl.DataFrame:
         return dff.select(pl.lit(False).alias("matches"))
 
     # Enregistrement des recherche dans Matomo
-    track_search(query, "home_page_search")
+    if track:
+        track_search(query, "home_page_search")
 
     # Normalize query
     normalized_query = unidecode(query.strip()).upper()

src/utils/tracking.py

@@ -28,3 +28,35 @@ def track_search(query, category):
             url=f"https://{os.getenv('MATOMO_DOMAIN')}/matomo.php",
             params=params,
         ).raise_for_status()
+
+
+def track_mcp_tool(tool_name: str, query: str | None = None) -> None:
+    """Enregistre un appel d'outil MCP dans Matomo (best-effort, prod uniquement).
+
+    `action_name="MCP / <tool>"`, `dimension1=<tool>`. Si l'outil porte une
+    requête texte, elle est envoyée en `search`. Nécessite un Custom Dimension
+    slot 1 (scope Action) configuré côté Matomo — sinon `dimension1` est ignoré.
+    Ne lève jamais : une panne Matomo ne doit pas casser l'appel du tool.
+    """
+    if DEVELOPMENT or not os.getenv("MATOMO_DOMAIN"):
+        return
+    params = {
+        "idsite": os.getenv("MATOMO_ID_SITE"),
+        "url": "https://colibre.fr/_mcp",
+        "rec": "1",
+        "action_name": f"MCP / {tool_name}",
+        "dimension1": tool_name,
+        "rand": uuid.uuid4().hex,
+        "apiv": "1",
+        "h": localtime().tm_hour,
+        "m": localtime().tm_min,
+        "s": localtime().tm_sec,
+        "token_auth": os.getenv("MATOMO_TOKEN"),
+    }
+    if query:
+        params["search"] = query
+        params["search_cat"] = "mcp"
+    try:
+        post(url=f"https://{os.getenv('MATOMO_DOMAIN')}/matomo.php", params=params)
+    except Exception:
+        pass