fn_registry/scratchpad/exf/build.py

import json, os, urllib.request, sys

MB = os.environ["MB"]; KEY = os.environ["KEY"]

def api(method, path, body=None, timeout=180):
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(MB + path, data=data, method=method,
        headers={"X-API-KEY": KEY, "Content-Type": "application/json"})
    try:
        return json.load(urllib.request.urlopen(req, timeout=timeout))
    except urllib.error.HTTPError as e:
        print(f"HTTP {e.code} on {method} {path}:", e.read().decode()[:1200]); raise

# Bridge documento -> service_request (canal + charged), tal cual 1094 card 11751.
BASE = r"""
WITH vf AS (
  SELECT document_id, LOGICAL_OR(is_pw) is_pw FROM (
    SELECT CAST(document_id AS STRING) document_id, ANY_VALUE(is_precaweb) is_pw
      FROM `autingo-159109.anjana_bi_datamart.VENTAS_aurgi` GROUP BY 1
    UNION ALL
    SELECT CAST(document_id AS STRING), ANY_VALUE(is_precaweb)
      FROM `autingo-159109.anjana_bi_datamart.VENTAS_Motortown` GROUP BY 1
  ) GROUP BY 1
),
lineas AS (
  SELECT
    CAST(s.numeroDocumento AS STRING) AS numdoc,
    CAST(s.idCentro AS STRING)        AS idcentro,
    DATE(s.Fecha)                     AS fecha,
    s.Base_imponible_linea            AS bil
  FROM {{#4494}} s
  WHERE DATE(s.Fecha) >= DATE_SUB(CURRENT_DATE(), INTERVAL 365 DAY)
    [[AND DATE(s.Fecha) >= {{fecha_desde}}]]
    [[AND DATE(s.Fecha) <= {{fecha_hasta}}]]
),
web AS (
  SELECT l.numdoc, l.fecha, l.bil, oc.name AS centro, oc.Companies__name AS ambito
  FROM lineas l
  LEFT JOIN vf ON l.numdoc = vf.document_id
  LEFT JOIN `autingo-159109.rag_datasets.Objeto_Centros` oc
    ON l.idcentro = CAST(oc.nav_id AS STRING)
  WHERE (COALESCE(vf.is_pw, FALSE) OR oc.name IN ('Aurgi Web','MT Web'))
    AND (oc.Companies__name IS NULL OR oc.Companies__name NOT IN ('Aurgi Glass','MotorTown Glass'))
    [[AND oc.name IN ({{centro}})]]
    [[AND oc.Companies__name IN ({{ensena}})]]
),
sr_link AS (
  SELECT CAST(inv.nav_id AS STRING) numdoc, CAST(j.service_request_id AS STRING) sr_id
  FROM `autingo-159109.psql_dcpublic.tpv_orders_invoice` inv
  JOIN `autingo-159109.psql_dcpublic.tpv_precawebs_servicerequestjob` j ON j.order_id = inv.order_id
  WHERE inv.nav_id IS NOT NULL
  UNION DISTINCT
  SELECT CAST(invoice_number AS STRING), CAST(service_request_id AS STRING)
  FROM `autingo-159109.psql_dcpublic.logistic_orders`
  WHERE invoice_number IS NOT NULL AND invoice_number != ''
),
sr_link1 AS (SELECT numdoc, MIN(sr_id) sr_id FROM sr_link GROUP BY 1),
sr AS (
  SELECT CAST(id AS STRING) sr_id, channel_id, charged
  FROM `autingo-159109.psql_dcpublic.service_requests`
),
doc AS (
  SELECT
    w.numdoc,
    ANY_VALUE(w.fecha)      AS fecha,
    SUM(w.bil)              AS venta,
    ANY_VALUE(sl.sr_id)     AS sr_id,
    ANY_VALUE(sr.channel_id) AS channel_id,
    ANY_VALUE(sr.charged)   AS charged
  FROM web w
  LEFT JOIN sr_link1 sl USING (numdoc)
  LEFT JOIN sr ON sr.sr_id = sl.sr_id
  GROUP BY w.numdoc
),
fin AS (
  SELECT
    numdoc, fecha, venta,
    CASE WHEN sr_id IS NULL THEN 'Sin solicitud'
         WHEN channel_id = 1 THEN 'aurgi.com'
         WHEN channel_id = 2 THEN 'motortown.es'
         WHEN channel_id = 3 THEN 'Autingo'
         WHEN channel_id IN (11,13,14,15,6,8) THEN 'Marketplaces'
         WHEN channel_id = 10 THEN 'Talleres Digitales'
         ELSE 'Otros' END AS canal,
    CASE WHEN sr_id IS NULL THEN 'Sin solicitud'
         WHEN charged THEN 'Pago web'
         ELSE 'Pago tienda' END AS forma_pago
  FROM doc
)
"""

CARDS = {
  "total": {
    "name": "Venta web total (facturacion NAV / modelo 4494)",
    "sql": BASE + "SELECT ROUND(SUM(venta),0) AS venta_web_eur, COUNT(DISTINCT numdoc) AS documentos FROM fin",
    "display": "scalar",
  },
  "canal": {
    "name": "Venta web por canal",
    "sql": BASE + "SELECT canal, ROUND(SUM(venta),0) AS venta_eur, COUNT(DISTINCT numdoc) AS documentos FROM fin GROUP BY canal ORDER BY venta_eur DESC",
    "display": "bar",
  },
  "pago": {
    "name": "Venta web por forma de pago",
    "sql": BASE + "SELECT forma_pago, ROUND(SUM(venta),0) AS venta_eur, COUNT(DISTINCT numdoc) AS documentos FROM fin GROUP BY forma_pago ORDER BY venta_eur DESC",
    "display": "row",
  },
  "matriz": {
    "name": "Venta web: matriz canal x forma de pago",
    "sql": BASE + "SELECT canal, forma_pago, ROUND(SUM(venta),0) AS venta_eur, COUNT(DISTINCT numdoc) AS documentos FROM fin GROUP BY canal, forma_pago ORDER BY venta_eur DESC",
    "display": "table",
  },
  "evolutivo": {
    "name": "Venta web mensual por canal",
    "sql": BASE + "SELECT DATE_TRUNC(fecha, MONTH) AS mes, canal, ROUND(SUM(venta),0) AS venta_eur FROM fin GROUP BY mes, canal ORDER BY mes, venta_eur DESC",
    "display": "bar",
  },
}

TAGS = {
  "#4494": {"type":"card","name":"#4494","id":"card__4494","display-name":"#4494","card-id":4494},
  "fecha_desde": {"type":"date","name":"fecha_desde","id":"tag-fecha-desde","display-name":"Fecha desde"},
  "fecha_hasta": {"type":"date","name":"fecha_hasta","id":"tag-fecha-hasta","display-name":"Fecha hasta"},
  "centro": {"type":"text","name":"centro","id":"tag-centro","display-name":"Centro"},
  "ensena": {"type":"text","name":"ensena","id":"tag-ensena","display-name":"Ensena"},
}

def dq(sql):
    return {"type":"native","database":6,"native":{"query":sql,"template-tags":TAGS}}

def test_query(sql, params=None):
    body = dq(sql)
    body["parameters"] = params or []
    r = api("POST", "/api/dataset", body)
    if r.get("error"):
        print("QUERY ERROR:", r.get("error")); return None
    cols = [c["name"] for c in r["data"]["cols"]]
    rows = r["data"]["rows"]
    return cols, rows

if __name__ == "__main__":
    which = sys.argv[1] if len(sys.argv) > 1 else "all"
    # param YTD 2026 para verificar reconciliacion
    p_ytd = [{"type":"date/single","value":"2026-01-01","target":["variable",["template-tag","fecha_desde"]]}]
    for k, c in CARDS.items():
        if which != "all" and which != k: continue
        print(f"\n===== TEST {k}: {c['name']} =====")
        res = test_query(c["sql"], p_ytd)
        if res:
            cols, rows = res
            print("cols:", cols)
            for row in rows[:15]: print(" ", row)