fn_registry/python/functions/datascience/build_eda_render_ctx.py

"""build_eda_render_ctx — constructor del `ctx` de datos crudos del motor AutomaticEDA.

Funcion impura (lee de la base de datos) del grupo de capacidad `eda`. Dado un
``db_path`` + ``table`` (DuckDB o PostgreSQL) y el ``TableProfile`` AGREGADO ya
calculado por ``profile_table``, produce el dict ``ctx`` que los renderers
(``render_automatic_eda_pdf`` / ``render_automatic_eda_pptx`` ->
``build_document(profile, ctx)``) pasan a los capitulos que necesitan DATOS
CRUDOS no presentes en el perfil agregado: modelos (``project_clusters_2d`` en
vivo), timeseries, geospatial y agregacion (groupby/pivot push-down).

NO trae tablas enteras a RAM: muestrea con ``LIMIT sample`` y, para la serie
temporal, delega el push-down en ``extract_timeseries_raw`` (una sola query
ordenada). El lector read-only ``query_fn(sql) -> dict`` se construye igual que
en ``profile_table`` (un closure sobre ``duckdb_query_readonly`` / ``pg_query``)
y nunca abre conexiones fuera de esos wrappers.

Estilo dict-no-throw del grupo `eda`: la funcion NUNCA lanza. Si una pieza falla
(query, deteccion, render de una clave), esa clave se degrada a ausente / lista
vacia y el resto del ctx se construye igual. Ante un fallo global devuelve al
menos ``{**base_ctx, "db_path": db_path, "table": table}``.

Claves de DATOS que produce (las consumen los capitulos):
  - ``head_rows``      : [ {col: valor, ...}, ... ] primeras filas CRUDAS de la
                         tabla (``SELECT * LIMIT head_n``), una entrada por fila.
                         La lee el capitulo OVERVIEW para mostrar df.head real en
                         lugar del placeholder "df.head no disponible".
  - ``raw_numeric``    : {col: [float|None, ...]} muestra cruda de las columnas
                         numericas, ALINEADA POR FILA (una entrada por fila aunque
                         sea None). La leen modelos (clustering 2D en vivo) y
                         geospatial (lat/lon salen de aqui).
  - ``timeseries_raw`` : {time_col, t: [iso...], series: {col: [float|None, ...]}}.
                         La lee el capitulo TIMESERIES.
  - ``geo_points``     : {lats: [...], lons: [...]} listas alineadas (ya floats).
                         La lee el capitulo GEOSPATIAL.
  - ``db_path``, ``table`` : los usa el capitulo AGREGACION para el groupby/pivot
                         push-down via DuckDB.

Las claves de PRESENTACION que traiga ``base_ctx`` (dataset_name, source_origin,
...) NO se pisan: esta funcion solo AÑADE las claves de datos sobre una copia.
"""


def _to_float(value):
    """Convierte un valor a float de forma defensiva. None si no es convertible.

    Un bool es subclase de int en Python pero nunca es un valor numerico de
    serie/coordenada valido, asi que se trata como None (mismo criterio que
    extract_timeseries_raw / detect_latlon_columns).
    """
    if value is None or isinstance(value, bool):
        return None
    if isinstance(value, (int, float)):
        return float(value)
    s = str(value).strip()
    if not s:
        return None
    try:
        return float(s)
    except (TypeError, ValueError):
        return None


def build_eda_render_ctx(db_path, table, profile, backend="duckdb", sample=5000, base_ctx=None, head_n=10):
    """Construye el ctx de datos crudos para los renderers de AutomaticEDA.

    Args:
        db_path: ruta al archivo DuckDB, o DSN PostgreSQL si backend="postgres".
            Se guarda tal cual en ctx["db_path"] (el capitulo agregacion lo usa
            para el push-down).
        table: nombre de la tabla. Se escapa con comillas dobles en las queries y
            se guarda en ctx["table"].
        profile: TableProfile agregado producido por profile_table. Solo se lee
            su clave ``columns`` (lista de ColumnProfile dict con name /
            inferred_type / numeric.{min,max} / semantic_type). Lectura
            defensiva: si no es dict o no tiene columns, se trata como [].
        backend: "duckdb" (default) o "postgres". Selecciona el lector read-only
            (duckdb_query_readonly / pg_query). Cualquier otro valor devuelve el
            base_ctx tal cual, sin añadir claves de datos.
        sample: maximo de filas a muestrear (clausula LIMIT) tanto para
            raw_numeric como para timeseries_raw. Default 5000.
        base_ctx: dict opcional con claves de presentacion ya preparadas
            (dataset_name, source_origin, ...). Se parte de una copia y NO se
            pisan sus claves; solo se añaden las de datos. Default None -> {}.
        head_n: numero de filas crudas a muestrear para ``ctx["head_rows"]``
            (df.head del capitulo OVERVIEW). Default 10. <=0 omite la clave.

    Returns:
        El dict ``ctx`` directamente (NO un wrapper {status,...}): se pasa tal
        cual como ``meta={"ctx": <ese dict>}`` a render_automatic_eda_pdf/pptx.
        Nunca lanza. Claves que puede contener: head_rows, raw_numeric,
        timeseries_raw, geo_points (omitidas si no aplican o fallan), y siempre
        db_path + table para backends validos.
    """
    # Copia de base_ctx: nunca mutamos el dict del caller. Las claves de
    # presentacion que ya traiga se conservan; las de datos se añaden encima.
    ctx = dict(base_ctx) if isinstance(base_ctx, dict) else {}

    try:
        # 1) Lector read-only del backend activo, construido EXACTAMENTE como en
        # profile_table (closure sobre el wrapper del registry). Imports perezosos
        # dentro de la funcion: este modulo vive en el paquete `datascience`, asi
        # que importar sus hermanas a nivel de modulo crearia un ciclo al cargar
        # el __init__ del paquete. Lazy import rompe el ciclo y respeta el
        # contrato (imports explicitos, sin `import *`).
        if backend == "duckdb":
            from infra import duckdb_query_readonly

            def query_fn(sql):
                return duckdb_query_readonly(db_path, sql)

        elif backend == "postgres":
            from infra import pg_query

            def query_fn(sql):
                return pg_query(db_path, sql)

        else:
            # Backend desconocido: devolver base_ctx tal cual, sin claves de datos.
            return ctx

        # 7) db_path + table SIEMPRE (para backends validos): el capitulo
        # agregacion los necesita para el groupby/pivot push-down via DuckDB.
        ctx["db_path"] = db_path
        ctx["table"] = table

        # 1.5) head_rows: primeras filas CRUDAS de la tabla (SELECT * LIMIT n)
        # para que el capitulo OVERVIEW muestre df.head real en vez del
        # placeholder. Una sola query, dict-no-throw: si falla, se omite la
        # clave (el capitulo degrada a su nota honesta). No se pisa una clave
        # head_rows que ya viniera en base_ctx (presentacion).
        if head_n and int(head_n) > 0 and "head_rows" not in ctx:
            try:
                hq = query_fn(f'SELECT * FROM "{table}" LIMIT {int(head_n)}')
                if isinstance(hq, dict) and hq.get("status") == "ok":
                    hrows = [
                        dict(r) for r in (hq.get("rows") or [])
                        if isinstance(r, dict)
                    ]
                    if hrows:
                        ctx["head_rows"] = hrows
            except Exception:  # noqa: BLE001 - dict-no-throw: omitir la clave
                pass

        # 2) Columnas del perfil agregado (lectura defensiva).
        cols = profile.get("columns") if isinstance(profile, dict) else None
        cols = cols or []

        # 3) Deteccion temporal/numerica con la funcion PURA del registry.
        from datascience import detect_time_column

        det = detect_time_column(cols)
        time_col = det.get("time_col")
        numeric_cols = det.get("numeric_cols") or []

        # 4) raw_numeric: muestra de las columnas numericas CRUDAS, ALINEADAS POR
        # FILA en UNA sola query. Cada columna queda con una entrada por fila
        # (None si no parsea) para no desalinear filas: project_clusters_2d
        # descarta filas listwise, asi que las listas deben tener igual longitud.
        raw_numeric = {}
        if numeric_cols:
            try:
                cols_sql = ", ".join(f'"{c}"' for c in numeric_cols)
                sql = f'SELECT {cols_sql} FROM "{table}" LIMIT {int(sample)}'
                q = query_fn(sql)
                if isinstance(q, dict) and q.get("status") == "ok":
                    rows = q.get("rows", []) or []
                    raw_numeric = {c: [] for c in numeric_cols}
                    for row in rows:
                        for c in numeric_cols:
                            raw_numeric[c].append(_to_float(row.get(c)))
            except Exception:  # noqa: BLE001 - dict-no-throw: degradar la clave
                raw_numeric = {}
        if raw_numeric:
            ctx["raw_numeric"] = raw_numeric

        # 5) timeseries_raw: SOLO si hay columna temporal y numericas. Se delega
        # el push-down en la funcion impura extract_timeseries_raw (una sola query
        # ordenada cronologicamente). Solo se adjunta si trae filas.
        if time_col and numeric_cols:
            try:
                from datascience import extract_timeseries_raw

                ts = extract_timeseries_raw(
                    query_fn, table, time_col, numeric_cols, max_rows=sample
                )
                if (
                    isinstance(ts, dict)
                    and ts.get("status") == "ok"
                    and (ts.get("n") or 0) > 0
                ):
                    ctx["timeseries_raw"] = {
                        "time_col": ts["time_col"],
                        "t": ts["t"],
                        "series": ts["series"],
                    }
            except Exception:  # noqa: BLE001 - dict-no-throw: omitir la clave
                pass

        # 6) geo_points: detecta el par lat/lon con la funcion PURA del registry.
        # Solo se adjunta si AMBAS columnas estan en raw_numeric (ya floats,
        # alineadas por fila). Si no hay par o no estan, se omite: el capitulo
        # geospatial sabe degradar.
        try:
            from datascience import detect_latlon_columns

            geo = detect_latlon_columns(cols)
            lat_col = geo.get("lat_col")
            lon_col = geo.get("lon_col")
            if lat_col and lon_col and lat_col in raw_numeric and lon_col in raw_numeric:
                ctx["geo_points"] = {
                    "lats": raw_numeric[lat_col],
                    "lons": raw_numeric[lon_col],
                }
        except Exception:  # noqa: BLE001 - dict-no-throw: omitir la clave
            pass

        return ctx
    except Exception:  # noqa: BLE001 - dict-no-throw global: nunca reventar.
        # Fallback minimo: copia de base_ctx + db_path/table para que el capitulo
        # agregacion siga teniendo lo imprescindible.
        out = dict(base_ctx) if isinstance(base_ctx, dict) else {}
        out["db_path"] = db_path
        out["table"] = table
        return out