| """Construction de requêtes SQL paramétrées (pas de SQL libre LLM).""" |
| from __future__ import annotations |
|
|
| from typing import Any, Literal |
|
|
| from app.sql.executor import execute_readonly |
| from app.sql.sql_validator import assert_whitelisted_identifiers |
|
|
| AGG_SQL = { |
| "sum": "SUM", |
| "max": "MAX", |
| "min": "MIN", |
| "mean": "AVG", |
| "avg": "AVG", |
| "count": "COUNT", |
| } |
|
|
|
|
| def build_sales_query( |
| metric: Literal["chiffre_affaires", "marge"] = "chiffre_affaires", |
| aggregation: str = "sum", |
| filters: dict[str, Any] | None = None, |
| group_by: str | None = None, |
| top_n: int | None = None, |
| ) -> tuple[str, list[Any]]: |
| """Construit un SELECT whitelisté sur la table sales.""" |
| filters = filters or {} |
| agg = AGG_SQL.get(aggregation.lower()) |
| if agg is None: |
| raise ValueError(f"Agrégation inconnue: {aggregation}") |
|
|
| columns = [metric, *filters.keys()] |
| if group_by: |
| columns.append(group_by) |
| assert_whitelisted_identifiers("sales", columns, agg) |
|
|
| where_parts: list[str] = [] |
| params: list[Any] = [] |
| for key, value in filters.items(): |
| where_parts.append(f"LOWER({key}) = LOWER(?)") |
| params.append(value) |
|
|
| where_sql = f" WHERE {' AND '.join(where_parts)}" if where_parts else "" |
|
|
| if group_by: |
| query = ( |
| f"SELECT {group_by} AS label, {agg}({metric}) AS value " |
| f"FROM sales{where_sql} " |
| f"GROUP BY {group_by} " |
| f"ORDER BY value DESC" |
| ) |
| if top_n: |
| query += f" LIMIT {int(top_n)}" |
| else: |
| query = ( |
| f"SELECT {agg}({metric}) AS value, COUNT(*) AS rows_considered " |
| f"FROM sales{where_sql}" |
| ) |
|
|
| return query, params |
|
|
|
|
| def run_sql_query( |
| metric: Literal["chiffre_affaires", "marge"] = "chiffre_affaires", |
| aggregation: str = "sum", |
| filters: dict | None = None, |
| group_by: str | None = None, |
| top_n: int | None = None, |
| ) -> dict: |
| """Point d'entrée métier : construit + exécute + normalise le résultat.""" |
| try: |
| query, params = build_sales_query( |
| metric=metric, |
| aggregation=aggregation, |
| filters=filters, |
| group_by=group_by, |
| top_n=top_n, |
| ) |
| except ValueError as exc: |
| return {"error": str(exc)} |
|
|
| raw = execute_readonly(query, params) |
| if "error" in raw: |
| return raw |
|
|
| rows = raw.get("rows", []) |
| if group_by: |
| ranked = [ |
| (r["label"], float(r["value"]) if r["value"] is not None else 0.0) |
| for r in rows |
| ] |
| return { |
| "metric": metric, |
| "aggregation": aggregation, |
| "group_by": group_by, |
| "filters": filters or {}, |
| "result": ranked, |
| "sql": raw.get("query"), |
| "duration_ms": raw.get("duration_ms"), |
| } |
|
|
| value = rows[0]["value"] if rows else 0 |
| considered = rows[0].get("rows_considered", 0) if rows else 0 |
| return { |
| "metric": metric, |
| "aggregation": aggregation, |
| "filters": filters or {}, |
| "result": float(value) if value is not None else 0.0, |
| "rows_considered": considered, |
| "sql": raw.get("query"), |
| "duration_ms": raw.get("duration_ms"), |
| } |
|
|
