"""Helpers puros para construir KPIs con display=smartscalar en Metabase. El display ``smartscalar`` de Metabase v0.59 muestra un numero grande con comparacion (% variacion + flecha verde/roja) cuando el query devuelve una serie temporal y ``scalar.comparisons`` esta configurado. Para mostrar "total del periodo seleccionado vs mismo periodo del ano anterior" sin que el motor pida un breakout temporal, el truco probado en el dashboard ``Informe Lean`` es producir 2 filas: ``(d_min - 52 semanas, n-1_total)`` y ``(d_min, total_actual)``, con columnas ``periodo`` y ``valor``. Smartscalar lo interpreta como serie de 2 puntos y compara natural con ``previousValue``. """ from __future__ import annotations import uuid _TAG_NAMESPACE = uuid.UUID("c0ffee00-6e75-4b14-9e8e-bea7ed5ca1ab") def metabase_smartscalar_kpi_sql( *, act_expr: str, n1_expr: str, body_sql: str, date_expr: str = "MIN(fecha)", ) -> str: """Envuelve una agregacion en el patron de 2 filas que smartscalar requiere. Recibe las expresiones de agregado para periodo actual y n-1, mas el cuerpo SQL (CTEs + JOINs + WHERE con template-tags) que produce la tabla a agregar, y devuelve la query nativa completa lista para incrustar en una card. La salida tiene siempre el mismo shape: 2 filas con columnas ``periodo`` (DATE) y ``valor`` (numero). Args: act_expr: Expresion SQL para el total del periodo actual. Tipicamente ``SUM(...)`` o ``SAFE_DIVIDE(SUM(a), NULLIF(SUM(b), 0))``. Debe poder evaluarse contra el alias del cuerpo (por ejemplo ``v.venta_n``). n1_expr: Equivalente para el mismo periodo del ano anterior. body_sql: Cuerpo SQL que define las tablas/CTEs y termina con un ``FROM ... [JOIN ...] [WHERE ...]`` referenciable como ``ventas v``, o equivalente. Ejemplo en docstring de salida. date_expr: Expresion para extraer la fecha minima del periodo, usada como ``periodo`` en la fila actual y como ancla para calcular la fecha n-1 con ``DATE_SUB(..., INTERVAL 52 WEEK)``. Default ``MIN(fecha)``. Para queries con alias usar ``MIN(v.fecha)``. Returns: SQL nativo (BigQuery dialect por defecto) con la estructura: WITH agg AS ( SELECT COALESCE({date_expr}, CURRENT_DATE()) AS d_min, {act_expr} AS act, {n1_expr} AS n1 {body_sql} ) SELECT DATE_SUB(d_min, INTERVAL 52 WEEK) AS periodo, n1 AS valor FROM agg UNION ALL SELECT d_min, act FROM agg ORDER BY periodo Example: >>> body = ''' ... FROM `proj.ds.base_margenes_aa` v ... WHERE 1=1 [[AND {{fecha}}]] [[AND {{tipo}}]]''' >>> sql = metabase_smartscalar_kpi_sql( ... act_expr="ROUND(SUM(v.venta_n), 2)", ... n1_expr="ROUND(SUM(v.venta_n1), 2)", ... body_sql=body, ... date_expr="MIN(v.fecha)", ... ) >>> assert "UNION ALL" in sql >>> assert "periodo" in sql and "valor" in sql """ inner_select = ( f" SELECT\n" f" COALESCE({date_expr}, CURRENT_DATE()) AS d_min,\n" f" {act_expr} AS act,\n" f" {n1_expr} AS n1\n" f" {body_sql.strip()}" ) return ( "WITH agg AS (\n" f"{inner_select}\n" ")\n" "SELECT DATE_SUB(d_min, INTERVAL 52 WEEK) AS periodo, n1 AS valor FROM agg\n" "UNION ALL\n" "SELECT d_min, act FROM agg\n" "ORDER BY periodo" ) def metabase_smartscalar_dimension_tag( *, name: str, field_id: int, base_type: str = "type/Text", widget_type: str = "string/=", display_name: str = "", ) -> dict: """Construye un template-tag de tipo ``dimension`` listo para Metabase v0.59. Las cards nativas con filtros (field-filters) requieren un dict por template-tag con la firma exacta que espera el frontend MBQL5. Esta funcion rellena ``id``, ``display-name``, ``type``, ``widget-type`` y ``dimension`` con los flags ``base-type``, ``effective-type`` y ``lib/uuid`` que sin los cuales Metabase silenciosamente descarta el query al guardar (PUT 200 sin persistir el ``dataset_query``). El ``lib/uuid`` se deriva deterministicamente del nombre del tag con ``uuid5`` para que la funcion sea pura: dos llamadas con el mismo ``name`` producen el mismo UUID. Esto es seguro porque ``lib/uuid`` solo necesita ser unico dentro de la misma card. Args: name: Identificador del template-tag (snake_case). Tambien sirve como nombre del filtro en el SQL: ``[[AND {{name}}]]``. field_id: ID numerico del Field de Metabase al que apunta el filtro. Se obtiene con ``GET /api/field/`` o navegando el data model. base_type: Tipo base del Field (ej. ``type/Text``, ``type/Date``, ``type/Integer``, ``type/Decimal``). widget_type: Tipo de widget en el dashboard. Comunes: ``string/=`` (multivalor texto), ``date/all-options`` (fecha), ``number/=``, ``string/contains``. display_name: Etiqueta legible mostrada en la UI cuando alguien ejecuta la card directamente. Vacio = se autogenera title-casing del ``name``. Returns: Dict con la estructura ``{"name", "id", "display-name", "type", "widget-type", "dimension": ["field", {...}, field_id]}``. Example: >>> tag = metabase_smartscalar_dimension_tag( ... name="fecha", field_id=322392, ... base_type="type/Date", widget_type="date/all-options", ... display_name="Fecha", ... ) >>> assert tag["dimension"][2] == 322392 >>> assert tag["widget-type"] == "date/all-options" >>> # UUID determinista: misma llamada produce mismo dict >>> assert tag == metabase_smartscalar_dimension_tag( ... name="fecha", field_id=322392, ... base_type="type/Date", widget_type="date/all-options", ... display_name="Fecha", ... ) """ label = display_name or name.replace("_", " ").title() tag_uuid = str(uuid.uuid5(_TAG_NAMESPACE, name)) return { "name": name, "id": f"{name}_tag", "display-name": label, "type": "dimension", "widget-type": widget_type, "dimension": [ "field", { "base-type": base_type, "effective-type": base_type, "lib/uuid": tag_uuid, }, field_id, ], } def metabase_smartscalar_kpi_payload( *, name: str, database_id: int, sql: str, template_tags: dict | None = None, description: str = "", collection_id: int = 0, currency: bool = False, currency_code: str = "EUR", decimals: int = 0, comparison_label: str = "vs n-1", extra_visualization_settings: dict | None = None, ) -> dict: """Construye el payload para POST /api/card de un KPI smartscalar. Combina el SQL nativo (idealmente generado por ``metabase_smartscalar_kpi_sql``) con la configuracion de visualizacion smartscalar que muestra ``valor`` con comparacion ``previousValue`` y formato numero/divisa. El payload resultante puede pasarse directamente a ``metabase_create_card_raw``. Args: name: Nombre de la card. database_id: ID de la database de Metabase contra la que se ejecuta. sql: Query nativa que devuelve columnas ``periodo`` (DATE) y ``valor`` (numero) con 2 filas (n-1, actual). Generable con ``metabase_smartscalar_kpi_sql``. template_tags: Dict de template-tags para field-filters (``{tag_name: tag_dict}``). Construir con ``metabase_smartscalar_dimension_tag``. None = sin filtros. description: Descripcion mostrada en la card. Vacio se autorrellena. collection_id: ID de la coleccion destino. 0 = root. currency: Si True formatea ``valor`` como moneda con ``currency_code``. currency_code: Codigo ISO de moneda. Default EUR. decimals: Decimales mostrados en ``valor``. comparison_label: Etiqueta del bloque de comparacion. Default "vs n-1". extra_visualization_settings: Settings adicionales a fusionar (top-level override) en ``visualization_settings``. Util para anadir ``scalar.title``, ``card.title.alignment``, etc. Returns: Dict con estructura: { "name": ..., "description": ..., "type": "question", "display": "smartscalar", "dataset_query": {"database": ..., "type": "native", "native": {"query": ..., "template-tags": ...}}, "visualization_settings": { "scalar.field": "valor", "scalar.comparisons": [ {"id": "cmp_n1", "type": "previousValue", "label": ...}], "column_settings": {'["name","valor"]': {...formato...}} }, ["collection_id"]: ... } Example: >>> sql = metabase_smartscalar_kpi_sql( ... act_expr="ROUND(SUM(v.venta_n), 2)", ... n1_expr="ROUND(SUM(v.venta_n1), 2)", ... body_sql="FROM ventas v", ... date_expr="MIN(v.fecha)", ... ) >>> tags = {"fecha": metabase_smartscalar_dimension_tag( ... name="fecha", field_id=322392, ... base_type="type/Date", widget_type="date/all-options", ... )} >>> payload = metabase_smartscalar_kpi_payload( ... name="Venta total", database_id=6, sql=sql, ... template_tags=tags, currency=True, decimals=0, ... ) >>> assert payload["display"] == "smartscalar" >>> assert payload["visualization_settings"]["scalar.field"] == "valor" """ fmt: dict = {"decimals": decimals} if currency: fmt.update( { "number_style": "currency", "currency": currency_code, "currency_in_header": False, } ) viz: dict = { "scalar.field": "valor", "scalar.comparisons": [ {"id": "cmp_n1", "type": "previousValue", "label": comparison_label} ], "column_settings": {'["name","valor"]': fmt}, } if extra_visualization_settings: viz.update(extra_visualization_settings) payload: dict = { "name": name, "description": description or f"KPI {name} con comparacion vs mismo periodo del ano anterior.", "type": "question", "display": "smartscalar", "dataset_query": { "database": database_id, "type": "native", "native": { "query": sql, "template-tags": template_tags or {}, }, }, "visualization_settings": viz, } if collection_id > 0: payload["collection_id"] = collection_id return payload