feat(eda): EDA de carpeta/base multi-tabla -> AutomaticEDA por capitulos (PDF+PPTX+MD)

Pipeline render_automatic_eda_folder: apunta el AutomaticEDA a una CARPETA de
archivos tabulares (CSV/Parquet/JSON) o a una DuckDB existente y emite el informe
de la BASE por capitulos en PDF (A5 movil) + PPTX (16:9) + Markdown. Documento-base
con portada-base, resumen de todas las tablas y relaciones inter-tabla (FK
candidatas por containment + diagrama Mermaid del join graph). Flag per_table_eda
anexa el mini-EDA de cada tabla. Aditivo: render_automatic_eda (tabla unica) intacto.

Funcion nueva load_folder_to_duckdb (infra, grupo eda+duckdb): carga una carpeta a
una DuckDB (temp si no se da path), CREATE TABLE por archivo con read_csv_auto/
read_parquet/read_json_auto. dict-no-throw.

Compone profile_database + los 3 renderers del motor AutomaticEDA + build_document
(per-tabla), sin reimplementar su logica. Tests: golden 3 CSV relacionados (FK
orders.customer_id->customers.id detectada) + edges (carpeta vacia, 1 tabla,
DuckDB existente, path inexistente). fn index sin error.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-06-30 20:34:10 +02:00
parent a1e2e3567c
commit 6a1520f458
7 changed files with 958 additions and 0 deletions
+2
View File
@@ -34,6 +34,7 @@ from .upsert_xlsx_sheet import upsert_xlsx_sheet
from .duckdb_query_readonly import duckdb_query_readonly
from .duckdb_execute import duckdb_execute
from .duckdb_upsert import duckdb_upsert
from .load_folder_to_duckdb import load_folder_to_duckdb
from .imap_connect import imap_connect
from .imap_list_mailboxes import imap_list_mailboxes
from .imap_search import imap_search
@@ -50,6 +51,7 @@ __all__ = [
"upsert_xlsx_sheet",
"duckdb_query_readonly",
"duckdb_execute",
"load_folder_to_duckdb",
"duckdb_upsert",
"pg_insert_rows",
"pg_apply_sql",
@@ -0,0 +1,100 @@
---
name: load_folder_to_duckdb
kind: function
lang: py
domain: infra
version: "1.0.0"
purity: impure
signature: "def load_folder_to_duckdb(folder: str, db_path: str = None, pattern: str = '*.csv,*.parquet,*.json') -> dict"
description: "Escanea el primer nivel de una CARPETA buscando archivos tabulares (CSV/TSV/TXT, Parquet, JSON/NDJSON) y los carga como tablas en una base DuckDB usando los lectores nativos read_csv_auto/read_parquet/read_json_auto. Es la pieza de entrada del EDA a nivel de carpeta (grupo eda). Por cada archivo crea una tabla cuyo nombre se deriva del basename saneado a [0-9a-zA-Z_] en minusculas (prefijo t_ si empieza por digito, sufijos _2/_3 ante colisiones, tabla_<i> si queda vacio). El path se escapa (comilla simple '->'') antes de interpolarlo porque los lectores DuckDB no aceptan el path como parametro posicional. Glob NO recursivo: un glob.glob(os.path.join(folder, g)) por cada patron del CSV, dedup y ordenado. db_path=None genera una DuckDB temporal (mkstemp, se borra el placeholder vacio porque DuckDB rechaza un archivo de 0 bytes) y devuelve su ruta. Un fallo al cargar un archivo concreto no aborta el resto: se registra en errors y se continua. Devuelve siempre un dict sin lanzar (estilo del grupo duckdb): {status:'ok', db_path, tables, errors} en exito (carpeta sin archivos tabulares incluida, tables=[]) y {status:'error', error} cuando la carpeta no existe o falla algo global. Depende del paquete duckdb (1.5.2)."
tags: [eda, duckdb, ingest, etl, folder]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: "error_py_core"
imports: [glob, os, re, tempfile, duckdb]
params:
- name: folder
desc: "ruta a un directorio. Se escanea solo su primer nivel (NO recursivo). Si no existe o no es un directorio devuelve {status:'error'} sin lanzar."
- name: db_path
desc: "ruta del archivo DuckDB destino, abierto en modo read-write (lo crea si no existe). None (default) genera una DuckDB temporal unica con tempfile.mkstemp y devuelve su ruta en el campo db_path del retorno. DuckDB es single-writer: si otro proceso lo tiene abierto en escritura, connect falla con error de lock devuelto en el dict."
- name: pattern
desc: "CSV de globs separados por coma (default '*.csv,*.parquet,*.json'). Cada glob se aplica con glob.glob(os.path.join(folder, g)) sobre el primer nivel de folder; los resultados de todos los globs se deduplican y ordenan. Los globs con ** NO descienden recursivamente (glob.glob sin recursive=True)."
output: "dict. En exito: {status:'ok', db_path:str (ruta DuckDB usada), tables:[{name:str, source_file:str, n_rows:int}], errors:[{name?:str, source_file:str, error:str}]}. La carpeta sin archivos tabulares es un exito con tables=[] y errors=[]. En error (sin lanzar): {status:'error', error:str}."
tested: true
tests:
- "test_carga_dos_csv_como_tablas"
- "test_db_path_none_crea_temporal"
- "test_carpeta_vacia_es_ok_sin_tablas"
- "test_carpeta_inexistente_devuelve_status_error"
test_file_path: "python/functions/infra/load_folder_to_duckdb_test.py"
file_path: "python/functions/infra/load_folder_to_duckdb.py"
---
## Ejemplo
```python
import sys
sys.path.insert(0, "python/functions")
from infra.load_folder_to_duckdb import load_folder_to_duckdb
# Preparar una carpeta de demo con dos CSV.
import os
os.makedirs("/tmp/eda_folder_demo", exist_ok=True)
with open("/tmp/eda_folder_demo/ventas.csv", "w") as f:
f.write("id,total\n1,10.5\n2,20.0\n3,5.25\n")
with open("/tmp/eda_folder_demo/clientes.csv", "w") as f:
f.write("id,nombre\n1,ana\n2,luis\n")
# Cargar todos los tabulares de la carpeta a una DuckDB temporal.
res = load_folder_to_duckdb("/tmp/eda_folder_demo")
print(res["status"]) # ok
print(res["db_path"]) # /tmp/tmpXXXXXXXX.duckdb (temporal)
for t in res["tables"]:
print(t["name"], t["n_rows"]) # ventas 3 / clientes 2
# Persistir en una DuckDB concreta y limitar a CSV.
res2 = load_folder_to_duckdb(
"/tmp/eda_folder_demo",
db_path="/tmp/eda_folder_demo/folder.duckdb",
pattern="*.csv",
)
print(res2["tables"]) # [{'name': 'clientes', ...}, {'name': 'ventas', ...}]
```
## Cuando usarla
Cuando tienes una carpeta de datos sueltos (un dump, un export, varios CSV/Parquet
descargados) y quieres analizarlos juntos con SQL sin montar la ingesta a mano,
archivo por archivo. Es el primer eslabon del EDA a nivel de carpeta (grupo `eda`):
deja una DuckDB con una tabla por archivo, lista para perfilar con
`duckdb_table_schema_py_infra`, consultar con `duckdb_query_readonly_py_infra`, o
correlacionar aguas abajo. Usala antes de cualquier paso de perfilado cuando la
unidad de trabajo es "todos los archivos de este directorio".
## Gotchas
- **Glob NO recursivo**: solo se escanea el primer nivel de `folder`. Archivos en
subdirectorios se ignoran (ni siquiera con `**` en el patron, porque
`glob.glob` se llama sin `recursive=True`). Si necesitas recursion, aplana la
carpeta antes o amplia la funcion.
- **Saneo de nombres de tabla**: el basename se reduce a `[0-9a-zA-Z_]` en
minusculas. `Ventas 2024.csv` -> tabla `ventas_2024`. Dos archivos distintos
pueden sanear al mismo nombre (`a-b.csv` y `a_b.csv`); el segundo se desambigua
con sufijo `_2`, `_3`, ... El mapeo real archivo->tabla esta en `tables[].name`
/ `tables[].source_file`, no lo asumas.
- **`read_json_auto` requiere JSON tabular** (array de objetos u objetos NDJSON
homogeneos). Un JSON anidado o irregular puede fallar la carga de ESA tabla; el
error se registra en `errors` y el resto de archivos siguen cargandose.
- **Extension desconocida = se salta**, no falla: queda anotada en `errors` con
`unsupported extension`. Mapeo de lectores: `.csv/.tsv/.txt`->`read_csv_auto`,
`.parquet/.pq`->`read_parquet`, `.json/.ndjson`->`read_json_auto`.
- **Escritura real en disco (impura)**. DuckDB es single-writer: si otro proceso
tiene `db_path` abierto en escritura, `connect` falla con error de lock devuelto
en el dict. Un `db_path` con un directorio padre inexistente tambien falla.
- **`db_path=None` crea un archivo temporal que NO se borra solo**: la ruta se
devuelve en `db_path` para que el llamador la consuma y la limpie cuando termine.
- **Tipos inferidos por los lectores `_auto`**: los tipos de columna los infiere
DuckDB. Revisa el schema con `duckdb_table_schema_py_infra` si el tipado importa
aguas abajo.
@@ -0,0 +1,175 @@
"""Carga una carpeta de archivos tabulares (CSV/Parquet/JSON) como tablas DuckDB.
Funcion impura: escanea el primer nivel de un directorio buscando archivos que
casen con uno o varios globs, y por cada archivo crea una tabla en una base
DuckDB usando los lectores nativos (`read_csv_auto`, `read_parquet`,
`read_json_auto`). Es la pieza de entrada del EDA a nivel de carpeta (grupo
`eda`): deja una DuckDB con una tabla por archivo, lista para perfilar y
correlacionar aguas abajo.
Devuelve siempre un dict sin lanzar excepciones, siguiendo el estilo del grupo
duckdb del registry: {status:'ok', db_path, tables, errors} en exito (incluida
la carpeta sin archivos tabulares, que es un exito con tables=[]) y
{status:'error', error:str} cuando la carpeta no existe o falla algo global.
El nombre de cada tabla se deriva del basename del archivo, saneado a
`[0-9a-zA-Z_]` en minusculas, prefijado con `t_` si empieza por digito, y
desambiguado con sufijos `_2`, `_3`, ... ante colisiones. El path del archivo se
escapa (comilla simple, `'`->`''`) antes de interpolarlo en el SQL del lector,
ya que los lectores DuckDB no admiten el path como parametro posicional. Un fallo
al cargar un archivo concreto NO aborta el resto: se registra en `errors` y se
continua con los siguientes.
"""
import glob
import os
import re
import tempfile
def _sanitize_table_name(basename_no_ext: str, index: int) -> str:
"""Deriva un identificador de tabla valido desde el basename de un archivo.
Reemplaza todo lo que no sea ``[0-9a-zA-Z_]`` por ``_`` y baja a minusculas.
Si tras el saneo queda vacio, usa ``tabla_<index>``. Si empieza por digito,
prefija ``t_`` para que sea un identificador SQL valido.
"""
name = re.sub(r"[^0-9a-zA-Z_]", "_", basename_no_ext).lower()
if not name:
name = f"tabla_{index}"
if name[0].isdigit():
name = "t_" + name
return name
def _reader_for_extension(ext: str, quoted_path: str):
"""Devuelve la expresion de lector DuckDB para una extension, o None.
El ``quoted_path`` ya viene escapado y entre comillas simples. Extensiones
desconocidas devuelven None para que el llamador salte el archivo.
"""
ext = ext.lower()
if ext in (".csv", ".tsv", ".txt"):
return f"read_csv_auto('{quoted_path}')"
if ext in (".parquet", ".pq"):
return f"read_parquet('{quoted_path}')"
if ext in (".json", ".ndjson"):
return f"read_json_auto('{quoted_path}')"
return None
def load_folder_to_duckdb(
folder: str,
db_path: str = None,
pattern: str = "*.csv,*.parquet,*.json",
) -> dict:
"""Carga los archivos tabulares de una carpeta como tablas en una DuckDB.
Args:
folder: ruta a un directorio. Si no existe o no es un directorio,
devuelve {status:'error', ...} sin lanzar.
db_path: ruta de la DuckDB destino (read-write, se crea si no existe). Si
es None, se genera una base temporal con NamedTemporaryFile y su ruta
se devuelve en el retorno (`db_path`).
pattern: CSV de globs separados por coma (default
"*.csv,*.parquet,*.json"). Cada glob se aplica con
glob.glob(os.path.join(folder, g)) en el primer nivel (NO recursivo);
los resultados se deduplican y ordenan.
Returns:
dict. En exito: {status:'ok', db_path:str, tables:[{name, source_file,
n_rows}], errors:[{name?, source_file, error}]}. La carpeta sin archivos
tabulares es un exito con tables=[] y errors=[]. En error (sin lanzar):
{status:'error', error:str}.
"""
if not isinstance(folder, str) or not os.path.isdir(folder):
return {
"status": "error",
"error": f"folder does not exist or is not a directory: {folder!r}",
}
conn = None
try:
# Resolver la ruta de la DuckDB destino. Si no se da, reservar un nombre
# temporal unico y borrar el archivo vacio que crea mkstemp: DuckDB 1.5.2
# rechaza abrir un archivo de 0 bytes ("not a valid DuckDB database
# file"), por lo que debe crear el archivo el mismo desde cero.
if db_path is None:
fd, tmp_name = tempfile.mkstemp(suffix=".duckdb")
os.close(fd)
os.remove(tmp_name)
db_path = tmp_name
# Resolver los archivos: un glob por cada patron, dedup + orden estable.
globs = [g.strip() for g in pattern.split(",") if g.strip()]
found = set()
for g in globs:
for path in glob.glob(os.path.join(folder, g)):
if os.path.isfile(path):
found.add(path)
files = sorted(found)
conn = __import__("duckdb").connect(db_path)
tables = []
errors = []
used_names = set()
for i, path in enumerate(files):
base = os.path.basename(path)
stem, ext = os.path.splitext(base)
quoted_path = path.replace("'", "''")
reader = _reader_for_extension(ext, quoted_path)
if reader is None:
errors.append(
{
"source_file": path,
"error": f"unsupported extension: {ext!r}",
}
)
continue
name = _sanitize_table_name(stem, i)
# Desambiguar colisiones con sufijos _2, _3, ...
if name in used_names:
suffix = 2
while f"{name}_{suffix}" in used_names:
suffix += 1
name = f"{name}_{suffix}"
quoted_ident = '"' + name.replace('"', '""') + '"'
try:
conn.execute(
f"CREATE TABLE {quoted_ident} AS SELECT * FROM {reader}"
)
n_rows = conn.execute(
f"SELECT count(*) FROM {quoted_ident}"
).fetchone()[0]
used_names.add(name)
tables.append(
{
"name": name,
"source_file": path,
"n_rows": int(n_rows),
}
)
except Exception as e: # noqa: BLE001
errors.append(
{
"name": name,
"source_file": path,
"error": str(e),
}
)
return {
"status": "ok",
"db_path": db_path,
"tables": tables,
"errors": errors,
}
except Exception as e: # noqa: BLE001
return {"status": "error", "error": str(e)}
finally:
if conn is not None:
conn.close()
@@ -0,0 +1,73 @@
"""Tests para load_folder_to_duckdb."""
import os
import sys
sys.path.insert(0, os.path.dirname(__file__))
import duckdb # noqa: E402
from load_folder_to_duckdb import load_folder_to_duckdb # noqa: E402
def _write_csv(path: str, header: str, rows: list[str]) -> None:
with open(path, "w", encoding="utf-8") as f:
f.write(header + "\n")
for r in rows:
f.write(r + "\n")
def test_carga_dos_csv_como_tablas(tmp_path):
_write_csv(
str(tmp_path / "ventas.csv"),
"id,total",
["1,10.5", "2,20.0", "3,5.25"],
)
_write_csv(
str(tmp_path / "clientes.csv"),
"id,nombre",
["1,ana", "2,luis"],
)
db = tmp_path / "out.duckdb"
res = load_folder_to_duckdb(str(tmp_path), str(db))
assert res["status"] == "ok", res
assert res["errors"] == []
assert len(res["tables"]) == 2
assert res["db_path"] == str(db)
assert os.path.exists(str(db))
by_name = {t["name"]: t for t in res["tables"]}
assert by_name["ventas"]["n_rows"] == 3
assert by_name["clientes"]["n_rows"] == 2
# Verificar que las tablas existen realmente en la base.
con = duckdb.connect(str(db), read_only=True)
assert con.execute("SELECT count(*) FROM ventas").fetchone()[0] == 3
assert con.execute("SELECT count(*) FROM clientes").fetchone()[0] == 2
con.close()
def test_db_path_none_crea_temporal(tmp_path):
_write_csv(str(tmp_path / "datos.csv"), "x", ["1", "2"])
res = load_folder_to_duckdb(str(tmp_path))
assert res["status"] == "ok", res
assert res["db_path"]
assert os.path.exists(res["db_path"])
assert len(res["tables"]) == 1
assert res["tables"][0]["n_rows"] == 2
os.remove(res["db_path"])
def test_carpeta_vacia_es_ok_sin_tablas(tmp_path):
db = tmp_path / "out.duckdb"
res = load_folder_to_duckdb(str(tmp_path), str(db))
assert res["status"] == "ok", res
assert res["tables"] == []
assert res["errors"] == []
def test_carpeta_inexistente_devuelve_status_error(tmp_path):
res = load_folder_to_duckdb(str(tmp_path / "no_existe"))
assert res["status"] == "error"
assert "folder" in res["error"]
@@ -0,0 +1,112 @@
---
name: render_automatic_eda_folder
kind: pipeline
lang: py
domain: pipelines
purity: impure
version: "1.0.0"
signature: "def render_automatic_eda_folder(path: str, out_dir: str = \"reports\", basename: str = None, profile_level: str = \"standard\", emit_pdf: bool = True, emit_pptx: bool = True, emit_md: bool = True, per_table_eda: bool = False, min_inclusion: float = 0.9, ctx_extra: dict = None) -> dict"
description: "Informe AutomaticEDA a nivel de BASE one-shot de una CARPETA de archivos tabulares (CSV/Parquet/JSON) o de una DuckDB existente. Carga la carpeta a una DuckDB temporal con load_folder_to_duckdb (o usa la DuckDB dada directa), perfila TODA la base con profile_database (resumen de cada tabla + FK candidatas por containment + join graph con diagrama Mermaid), ENSAMBLA un documento-base por capitulos (portada-base con nombre/n tablas/totales/fecha/fuente, resumen de tablas con una fila por tabla, y relaciones inter-tabla con la tabla de FK candidatas + el diagrama Mermaid) y lo renderiza con el motor AutomaticEDA a PDF (A5 movil), PPTX (16:9) y Markdown autocontenido a la vez. Con per_table_eda=True anexa los capitulos de mini-EDA de cada tabla (build_document por tabla). Es el hermano a nivel de base de render_automatic_eda (que perfila UNA tabla): aqui el informe es de la base y de sus relaciones. Devuelve las rutas de PDF/PPTX/MD, el manifiesto y el DatabaseProfile."
tags: [eda, duckdb, database, profiling, relations, pipeline, dataops, report, pdf, pptx, launcher]
uses_functions:
- load_folder_to_duckdb_py_infra
- profile_database_py_pipelines
- render_automatic_eda_pdf_py_datascience
- render_automatic_eda_pptx_py_datascience
- render_automatic_eda_markdown_py_datascience
uses_types: []
returns: []
returns_optional: false
error_type: error_go_core
imports: []
tested: true
tests:
- "golden: carpeta con 3 CSV relacionados (customers/orders/products) emite PDF+PPTX+MD del documento-base con 3 tablas y la FK orders.customer_id->customers.id"
- "edge: carpeta vacia -> status ok con documento minimo, sin lanzar"
- "edge: 1 sola tabla -> funciona sin relaciones (capitulo relaciones dice 'sin FK')"
test_file_path: "python/functions/pipelines/render_automatic_eda_folder_test.py"
file_path: "python/functions/pipelines/render_automatic_eda_folder.py"
params:
- name: path
desc: "DIRECTORIO con archivos tabulares (CSV/Parquet/JSON) que se cargan a una DuckDB temporal, o una DuckDB ya existente (.duckdb/.ddb/.db) que se perfila directa."
- name: out_dir
desc: "Directorio de salida de los informes (se crea si no existe). Default 'reports'."
- name: basename
desc: "Nombre base de los archivos sin extension. Default 'aeda_base_<nombre>_<timestamp>'."
- name: profile_level
desc: "Preset de coste del perfil por tabla ('lite'/'standard'/'full'); ajusta el sample que profile_database pasa a cada tabla (lite=2000, standard/full=5000)."
- name: emit_pdf
desc: "Emite el PDF A5 movil del documento-base. Default True."
- name: emit_pptx
desc: "Emite el PPTX 16:9 del documento-base. Default True."
- name: emit_md
desc: "Emite el Markdown autocontenido del documento-base. Default True."
- name: per_table_eda
desc: "Si True, anexa al documento-base los capitulos de mini-EDA de cada tabla (Heading 'Tabla: <n>' + build_document por tabla). Default False (solo documento-base: portada + resumen + relaciones)."
- name: min_inclusion
desc: "Umbral de inclusion (0-1) para emitir una FK candidata (se pasa a profile_database). Default 0.9."
- name: ctx_extra
desc: "Dict opcional de claves de presentacion (p.ej. dataset_name, description) que se mezclan en el contexto de la portada-base."
output: "Dict dict-no-throw. En exito: {status:'ok', pdf_path, pptx_path, md_path, manifest_path, n_tables, n_pages, n_slides, md_chars, db_path, db_profile}. En error: {status:'error', error:str}."
---
# render_automatic_eda_folder
EDA de una **carpeta / base multi-tabla** → informe AutomaticEDA por capítulos
en PDF (móvil A5) + PPTX (16:9) + Markdown, en una sola llamada. Es el hermano a
nivel de **base** de `render_automatic_eda` (que perfila una sola tabla): aquí el
documento resume **todas** las tablas y, sobre todo, sus **relaciones**
inter-tabla (FK candidatas por containment + join graph con diagrama Mermaid).
Compone, sin reimplementar su lógica: `load_folder_to_duckdb` (carga la carpeta),
`profile_database` (perfila la base + infiere FK + join graph) y los tres
renderers del motor AutomaticEDA (`render_automatic_eda_pdf`/`_pptx`/`_markdown`),
que aceptan directamente la lista de capítulos del documento-base que este
pipeline ensambla. El pipeline de tabla única (`render_automatic_eda`) queda
intacto: esto es aditivo.
## Ejemplo
```bash
# Carpeta con varios CSV/Parquet/JSON relacionados:
./fn run render_automatic_eda_folder /tmp/eda_folder_demo
# Una DuckDB ya existente (rama directa):
./fn run render_automatic_eda_folder temp/bigdata/taxi.duckdb
```
```python
import sys, os
sys.path.insert(0, os.path.join("python", "functions"))
from pipelines.render_automatic_eda_folder import render_automatic_eda_folder
r = render_automatic_eda_folder("/tmp/eda_folder_demo", out_dir="reports")
# r["status"] == "ok"; r["pdf_path"], r["pptx_path"], r["md_path"]
# r["n_tables"] == 3; r["db_profile"]["fk_candidates"] incluye
# orders.customer_id -> customers.id
```
## Cuando usarla
Cuando quieras un EDA de una **base entera** (una carpeta de exports o una
DuckDB con varias tablas), no de una sola tabla: para ver de un vistazo qué
tablas hay, su tamaño y calidad, y cómo se relacionan (FK candidatas + diagrama),
en el mismo formato rico por capítulos (PDF móvil + PPTX + MD) que el EDA de
tabla. Usa `per_table_eda=True` cuando además quieras el mini-EDA de cada tabla
anexado.
## Gotchas
- Impuro: lee archivos del disco y escribe PDF/PPTX/MD en `out_dir`. En la rama
"carpeta" crea una **DuckDB temporal** (su ruta sale en `db_path`); no se borra
automáticamente (queda para reinspección).
- `path` se interpreta así: directorio → se carga la carpeta; archivo con
extensión `.duckdb`/`.ddb`/`.db` → se usa directo; cualquier otro archivo o un
path inexistente → `{status:'error'}` (no lanza).
- El escaneo de la carpeta es **no recursivo** (solo el primer nivel) y por
defecto cubre `*.csv,*.parquet,*.json` (ver `load_folder_to_duckdb`).
- El diagrama Mermaid se vuelca como **bloque de código**: en el Markdown queda
como diagrama renderizable; en PDF/PPTX se muestra el **texto** del grafo (no
se rasteriza el diagrama a imagen en esta versión).
- Carpeta vacía o con 1 sola tabla: funciona igual; el capítulo de relaciones
dice "sin FK". dict-no-throw en todos los caminos.
@@ -0,0 +1,350 @@
"""render_automatic_eda_folder — EDA de una CARPETA / base multi-tabla one-shot.
Pipeline impuro del grupo de capacidad `eda`, a nivel de BASE. Dada una CARPETA
de archivos tabulares (CSV/Parquet/JSON) o una DuckDB ya existente, produce el
informe AutomaticEDA de la BASE en sus tres formatos a la vez (PDF móvil A5 +
PPTX 16:9 + Markdown autocontenido), con los capítulos POBLADOS, en una sola
llamada. Es el hermano a nivel de base de ``render_automatic_eda`` (que perfila
UNA tabla): aquí el documento por capítulos resume TODAS las tablas y, sobre
todo, sus RELACIONES inter-tabla (FK candidatas + join graph).
Compone funciones del registry SIN reimplementar su lógica:
- load_folder_to_duckdb : carga una carpeta de archivos a una DuckDB temporal
(rama "carpeta"). En la rama "ya es duckdb" se omite.
- profile_database : perfila TODA la base (resumen de cada tabla,
TableProfiles completos, FK candidatas por
containment y join graph con diagrama Mermaid).
- render_automatic_eda_pdf : renderiza el documento-base por capítulos a PDF.
- render_automatic_eda_pptx : renderiza el mismo documento-base a PPTX.
- render_automatic_eda_markdown : serializa el mismo documento-base a Markdown
autocontenido (texto + tablas markdown).
- build_document : (solo con per_table_eda=True) ensambla los capítulos
canónicos de CADA tabla para anexarlos al documento.
La capa propia de este pipeline es ENSAMBLAR EL DOCUMENTO-BASE de capítulos a
partir del ``DatabaseProfile`` que devuelve ``profile_database`` y cablear los
tres renderers del motor AutomaticEDA. El documento-base mínimo tiene tres
capítulos: portada-base (nombre/nº tablas/totales/fecha/fuente), resumen de
tablas (una fila por tabla) y relaciones inter-tabla (FK candidatas + diagrama
Mermaid). Con ``per_table_eda=True`` anexa, por cada tabla, sus capítulos de
mini-EDA.
Estilo dict-no-throw del grupo `eda`: nunca lanza; captura cualquier error y
degrada a ``{"status": "error", "error": str}``.
"""
import os
from datetime import datetime, timezone
from datascience import (
render_automatic_eda_markdown,
render_automatic_eda_pdf,
render_automatic_eda_pptx,
)
from datascience.automatic_eda import build_document
from infra import load_folder_to_duckdb
from pipelines.profile_database import profile_database
# Mapa profile_level -> tamaño de muestra por columna del perfil de cada tabla.
# A nivel de base el coste lo domina el nº de tablas; el preset solo ajusta el
# sample que profile_database pasa a profile_table.
_SAMPLE_BY_LEVEL = {"lite": 2000, "standard": 5000, "full": 5000}
# Extensiones que se consideran "una DuckDB ya hecha" en la rama directa.
_DUCKDB_EXTS = (".duckdb", ".ddb", ".db")
def _fmt_num(v) -> str:
"""Formatea un entero con separador de millar; '' si no es número."""
if isinstance(v, bool) or not isinstance(v, (int, float)):
return ""
try:
return f"{int(v):,}".replace(",", ".")
except Exception: # noqa: BLE001
return str(v)
def _portada_chapter(db_profile: dict, source_path: str, db_path: str,
meta_ctx: dict) -> dict:
"""Capítulo de portada a nivel de base (NO reusa chapters/portada.py, que es
de tabla única): nombre de la base, nº de tablas, totales y procedencia."""
tables = db_profile.get("tables", []) or []
total_rows = sum(
(t.get("n_rows") or 0) for t in tables if isinstance(t.get("n_rows"), (int, float))
)
total_cols = sum(
(t.get("n_cols") or 0) for t in tables if isinstance(t.get("n_cols"), (int, float))
)
base_name = (meta_ctx or {}).get("dataset_name") or os.path.basename(
os.path.normpath(source_path)
) or source_path
rows = [
("Base", base_name),
("Tablas", _fmt_num(db_profile.get("n_tables"))),
("Filas totales", _fmt_num(total_rows)),
("Columnas totales", _fmt_num(total_cols)),
("Relaciones FK", _fmt_num(len(db_profile.get("fk_candidates", []) or []))),
("Fuente", source_path),
("DuckDB", db_path),
("Generado", datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")),
]
blocks = [
{"kind": "heading", "text": f"EDA de la base — {base_name}", "level": 1},
{"kind": "kv_table", "rows": rows, "title": "Resumen de la base"},
]
errs = db_profile.get("errors", []) or []
if errs:
blocks.append({
"kind": "note",
"text": f"{len(errs)} aviso(s) durante el perfilado (ver detalle).",
})
return {"id": "portada_base", "title": "Portada", "version": "1.0.0",
"blocks": blocks}
def _resumen_chapter(db_profile: dict) -> dict:
"""Capítulo con una fila por tabla: filas, columnas, calidad, key_candidates."""
header = ["Tabla", "Filas", "Columnas", "Calidad", "key_candidates"]
rows = []
for t in db_profile.get("tables", []) or []:
keys = ", ".join(t.get("key_candidates") or []) or ""
rows.append([
t.get("table"),
_fmt_num(t.get("n_rows")),
_fmt_num(t.get("n_cols")),
t.get("quality_score"),
keys,
])
if rows:
blocks = [{
"kind": "data_table", "header": header, "rows": rows,
"title": "Tablas de la base",
"note": "Una fila por tabla. Calidad = score agregado del TableProfile.",
}]
else:
blocks = [{"kind": "note",
"text": "La base no contiene tablas perfilables."}]
return {"id": "resumen_tablas", "title": "Resumen de tablas",
"version": "1.0.0", "blocks": blocks}
def _relaciones_chapter(db_profile: dict) -> dict:
"""Capítulo de relaciones inter-tabla: tabla de FK candidatas + diagrama
Mermaid del join graph (vuelca el Mermaid como bloque de código)."""
fks = db_profile.get("fk_candidates", []) or []
blocks = [{
"kind": "heading", "text": "Relaciones inter-tabla", "level": 2,
}]
if fks:
header = ["From", "To", "Inclusión", "Cardinalidad"]
rows = []
for fk in fks:
frm = f"{fk.get('from_table')}.{fk.get('from_col')}"
to = f"{fk.get('to_table')}.{fk.get('to_col')}"
inc = fk.get("inclusion")
inc_s = f"{inc:.3f}" if isinstance(inc, (int, float)) else str(inc)
rows.append([frm, to, inc_s, fk.get("cardinality")])
blocks.append({
"kind": "data_table", "header": header, "rows": rows,
"title": "FK candidatas (por containment de valores)",
"note": "Inclusión = fracción de valores de From contenidos en To.",
})
else:
blocks.append({
"kind": "note",
"text": "Sin relaciones FK candidatas detectadas entre las tablas.",
})
mermaid = (db_profile.get("join_graph") or {}).get("mermaid", "") or ""
if mermaid.strip():
blocks.append({"kind": "heading", "text": "Diagrama (join graph)",
"level": 3})
# El Mermaid se vuelca como bloque de código: en MD queda como diagrama
# renderizable; en PDF/PPTX se muestra el texto del grafo (legible).
blocks.append({"kind": "markdown",
"text": "```mermaid\n" + mermaid.strip() + "\n```"})
return {"id": "relaciones", "title": "Relaciones inter-tabla",
"version": "1.0.0", "blocks": blocks}
def _build_db_document(db_profile: dict, source_path: str, db_path: str,
meta_ctx: dict, per_table_eda: bool) -> list:
"""Ensambla el documento-base por capítulos a partir del DatabaseProfile.
Mínimo: portada-base + resumen de tablas + relaciones. Con per_table_eda
True anexa, por cada tabla, un capítulo separador + los capítulos canónicos
de su mini-EDA (reusando build_document sobre cada TableProfile)."""
chapters = [
_portada_chapter(db_profile, source_path, db_path, meta_ctx),
_resumen_chapter(db_profile),
_relaciones_chapter(db_profile),
]
if per_table_eda:
for prof in db_profile.get("table_profiles", []) or []:
tname = prof.get("table") or "tabla"
chapters.append({
"id": f"tabla_{tname}", "title": f"Tabla: {tname}",
"version": "1.0.0",
"blocks": [{"kind": "heading", "text": f"Tabla: {tname}",
"level": 1}],
})
try:
# build_document devuelve los capítulos canónicos de la tabla.
# ctx None -> los capítulos que necesitan datos crudos degradan,
# pero salen completos los de portada/overview/distrib/calidad.
chapters.extend(build_document(prof, None) or [])
except Exception: # noqa: BLE001 — una tabla mala no rompe el doc.
chapters.append({
"id": f"tabla_{tname}_err", "title": f"Tabla: {tname}",
"version": "1.0.0",
"blocks": [{"kind": "note",
"text": "No se pudo ensamblar el mini-EDA de "
"esta tabla."}],
})
return chapters
def _resolve_db_path(path: str) -> dict:
"""Resuelve el DuckDB a perfilar desde ``path``.
- Directorio -> carga la carpeta con load_folder_to_duckdb (DuckDB temp).
- Archivo .duckdb/.ddb/.db -> se usa directo (rama "ya es duckdb").
- Otro archivo / inexistente -> error.
Devuelve {status, db_path, loaded, n_tables, load_errors}.
"""
if os.path.isdir(path):
lr = load_folder_to_duckdb(path)
if lr.get("status") != "ok":
return {"status": "error",
"error": f"load_folder_to_duckdb falló: {lr.get('error')}"}
return {
"status": "ok",
"db_path": lr.get("db_path"),
"loaded": True,
"n_tables": len(lr.get("tables", []) or []),
"load_errors": lr.get("errors", []) or [],
}
if os.path.isfile(path):
if path.lower().endswith(_DUCKDB_EXTS):
return {"status": "ok", "db_path": path, "loaded": False,
"n_tables": None, "load_errors": []}
return {"status": "error",
"error": f"'{path}' no es un directorio ni una DuckDB "
f"(extensiones {_DUCKDB_EXTS})."}
return {"status": "error", "error": f"path no existe: {path}"}
def render_automatic_eda_folder(
path: str,
out_dir: str = "reports",
basename: str = None,
profile_level: str = "standard",
emit_pdf: bool = True,
emit_pptx: bool = True,
emit_md: bool = True,
per_table_eda: bool = False,
min_inclusion: float = 0.9,
ctx_extra: dict = None,
) -> dict:
"""Perfila una CARPETA (o una DuckDB) y emite el informe AutomaticEDA de la base.
Args:
path: o bien un DIRECTORIO con archivos tabulares (CSV/Parquet/JSON) que
se cargan a una DuckDB temporal, o bien una DuckDB ya existente
(``.duckdb``/``.ddb``/``.db``) que se perfila directa.
out_dir: directorio de salida (se crea si no existe). Default "reports".
basename: nombre base de los archivos sin extensión. Default
"aeda_base_<nombre>_<timestamp>".
profile_level: preset de coste del perfil por tabla ("lite"/"standard"/
"full"); ajusta el ``sample`` que profile_database pasa a cada tabla.
emit_pdf / emit_pptx / emit_md: qué formatos emitir. Default los tres.
per_table_eda: si True, anexa al documento-base los capítulos de mini-EDA
de cada tabla (un Heading "Tabla: <n>" + build_document por tabla).
Default False (solo el documento-base: portada + resumen + relaciones).
min_inclusion: umbral de inclusión para emitir una FK candidata (0-1).
ctx_extra: dict opcional de claves de presentación (p.ej. dataset_name,
description) que se mezclan en el contexto de la portada.
Returns:
dict (nunca lanza). En éxito::
{"status": "ok", "pdf_path": str|None, "pptx_path": str|None,
"md_path": str|None, "manifest_path": str|None,
"n_tables": int, "n_pages": int|None, "n_slides": int|None,
"md_chars": int|None, "db_path": str, "db_profile": <DatabaseProfile>}
En error: {"status": "error", "error": str}.
"""
try:
# 1) Resolver la DuckDB a perfilar (cargar carpeta o usar la dada).
rdb = _resolve_db_path(path)
if rdb.get("status") != "ok":
return {"status": "error", "error": rdb.get("error")}
db_path = rdb.get("db_path")
# 2) Perfilar la base entera (resumen + FK + join graph). Sin report
# propio (write_report/emit_pdf False): este pipeline emite el suyo.
sample = _SAMPLE_BY_LEVEL.get(profile_level, 5000)
pres = profile_database(
db_path, sample=sample, write_report=False,
min_inclusion=min_inclusion, emit_pdf=False,
)
if pres.get("status") != "ok":
return {"status": "error",
"error": f"profile_database falló: {pres.get('error')}"}
db_profile = pres.get("db_profile") or {}
# 3) Ensamblar el documento-base por capítulos.
meta_ctx = dict(ctx_extra or {})
chapters = _build_db_document(
db_profile, path, db_path, meta_ctx, per_table_eda
)
# 4) Render a los tres formatos desde el MISMO documento por capítulos.
os.makedirs(out_dir, exist_ok=True)
ts = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
nm = (meta_ctx.get("dataset_name")
or os.path.basename(os.path.normpath(path)) or "base")
nm = "".join(c if c.isalnum() else "_" for c in str(nm)).strip("_") or "base"
base = basename or f"aeda_base_{nm}_{ts}"
title = f"EDA base — {meta_ctx.get('dataset_name') or nm}"
meta = {"title": title}
pdf_path = pptx_path = md_path = manifest_path = None
n_pages = n_slides = md_chars = None
if emit_pdf:
target = os.path.join(out_dir, base + ".pdf")
rpdf = render_automatic_eda_pdf(chapters, target, meta) or {}
pdf_path = rpdf.get("path")
n_pages = rpdf.get("n_pages")
manifest_path = rpdf.get("manifest_path")
if emit_pptx:
target = os.path.join(out_dir, base + ".pptx")
rpptx = render_automatic_eda_pptx(chapters, target, meta) or {}
pptx_path = rpptx.get("path")
n_slides = rpptx.get("n_slides")
if emit_md:
target = os.path.join(out_dir, base + ".md")
rmd = render_automatic_eda_markdown(chapters, target, meta) or {}
md_path = rmd.get("path")
md_chars = rmd.get("n_chars")
return {
"status": "ok",
"pdf_path": pdf_path,
"pptx_path": pptx_path,
"md_path": md_path,
"manifest_path": manifest_path,
"n_tables": db_profile.get("n_tables"),
"n_pages": n_pages,
"n_slides": n_slides,
"md_chars": md_chars,
"db_path": db_path,
"db_profile": db_profile,
}
except Exception as e: # noqa: BLE001 — dict-no-throw: degradar, nunca lanzar.
return {"status": "error", "error": str(e)}
@@ -0,0 +1,146 @@
"""Tests para render_automatic_eda_folder — EDA de una carpeta / base multi-tabla.
Golden: una carpeta con 3 CSV relacionados (customers/orders/products) produce el
documento-base en PDF + PPTX + MD, con las 3 tablas en el resumen y la FK
orders.customer_id -> customers.id en el capítulo de relaciones. Edges: carpeta
vacía (documento mínimo, sin lanzar), 1 sola tabla (sin relaciones) y la rama
"ya es una DuckDB" sobre un archivo .duckdb existente.
"""
import os
import sys
import duckdb
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
from pipelines.render_automatic_eda_folder import render_automatic_eda_folder
def _write_demo_folder(folder: str) -> None:
"""3 CSV relacionados: orders.customer_id -> customers.id (FK detectable)."""
with open(os.path.join(folder, "customers.csv"), "w", encoding="utf-8") as fh:
fh.write("id,name,city\n")
fh.write("1,Alice,Madrid\n2,Bob,Barcelona\n3,Carol,Valencia\n"
"4,Dave,Sevilla\n5,Eve,Madrid\n")
with open(os.path.join(folder, "orders.csv"), "w", encoding="utf-8") as fh:
fh.write("order_id,customer_id,product_id,total\n")
fh.write("100,1,10,49.90\n101,1,11,12.50\n102,2,10,49.90\n"
"103,3,12,8.00\n104,3,11,12.50\n105,5,10,49.90\n"
"106,2,12,8.00\n")
with open(os.path.join(folder, "products.csv"), "w", encoding="utf-8") as fh:
fh.write("product_id,product_name,price\n")
fh.write("10,Widget,49.90\n11,Gadget,12.50\n12,Gizmo,8.00\n")
def _has_fk(db_profile: dict, from_t: str, from_c: str, to_t: str) -> bool:
for fk in db_profile.get("fk_candidates", []) or []:
if (fk.get("from_table") == from_t and fk.get("from_col") == from_c
and fk.get("to_table") == to_t):
return True
return False
def test_golden_folder_three_csv(tmp_path):
"""Carpeta con 3 CSV relacionados -> PDF+PPTX+MD, 3 tablas, FK detectada."""
folder = tmp_path / "demo"
folder.mkdir()
_write_demo_folder(str(folder))
out = tmp_path / "out"
r = render_automatic_eda_folder(str(folder), out_dir=str(out))
assert r["status"] == "ok", r
assert r["n_tables"] == 3
# Los tres formatos se emitieron y existen en disco.
assert r["pdf_path"] and os.path.exists(r["pdf_path"])
assert r["pptx_path"] and os.path.exists(r["pptx_path"])
assert r["md_path"] and os.path.exists(r["md_path"])
assert (r["n_pages"] or 0) >= 1
assert (r["n_slides"] or 0) >= 1
# La FK orders.customer_id -> customers.id se detecta por containment.
assert _has_fk(r["db_profile"], "orders", "customer_id", "customers"), \
r["db_profile"].get("fk_candidates")
# El Markdown menciona las 3 tablas y la relación.
md = open(r["md_path"], encoding="utf-8").read()
for t in ("customers", "orders", "products"):
assert t in md
assert "customer_id" in md
def test_edge_empty_folder(tmp_path):
"""Carpeta vacía -> status ok con documento mínimo, sin lanzar."""
folder = tmp_path / "empty"
folder.mkdir()
out = tmp_path / "out"
r = render_automatic_eda_folder(str(folder), out_dir=str(out))
assert r["status"] == "ok", r
assert r["n_tables"] == 0
# Aun sin tablas, emite el documento-base mínimo (portada + resumen vacío +
# relaciones "sin FK").
assert r["pdf_path"] and os.path.exists(r["pdf_path"])
assert r["md_path"] and os.path.exists(r["md_path"])
def test_edge_single_table_no_relations(tmp_path):
"""Carpeta con 1 sola tabla -> funciona sin relaciones (capítulo 'sin FK')."""
folder = tmp_path / "single"
folder.mkdir()
with open(folder / "lonely.csv", "w", encoding="utf-8") as fh:
fh.write("a,b\n1,x\n2,y\n3,z\n")
out = tmp_path / "out"
r = render_automatic_eda_folder(str(folder), out_dir=str(out))
assert r["status"] == "ok", r
assert r["n_tables"] == 1
assert not (r["db_profile"].get("fk_candidates") or [])
md = open(r["md_path"], encoding="utf-8").read()
assert "Sin relaciones FK" in md or "sin FK" in md.lower()
def test_accepts_existing_duckdb(tmp_path):
"""Rama 'ya es una DuckDB': un archivo .duckdb existente se perfila directo."""
db = tmp_path / "base.duckdb"
conn = duckdb.connect(str(db))
try:
conn.execute("CREATE TABLE customers (id INTEGER, name VARCHAR)")
conn.execute("INSERT INTO customers VALUES (1,'Ana'),(2,'Luis'),(3,'Eva')")
conn.execute("CREATE TABLE orders (oid INTEGER, customer_id INTEGER)")
conn.execute("INSERT INTO orders VALUES (10,1),(11,2),(12,1),(13,3)")
finally:
conn.close()
out = tmp_path / "out"
r = render_automatic_eda_folder(str(db), out_dir=str(out))
assert r["status"] == "ok", r
assert r["n_tables"] == 2
assert r["db_path"] == str(db)
assert r["pdf_path"] and os.path.exists(r["pdf_path"])
def test_emit_flags_select_formats(tmp_path):
"""emit_pdf/pptx/md controlan qué formatos se emiten."""
folder = tmp_path / "demo"
folder.mkdir()
_write_demo_folder(str(folder))
out = tmp_path / "out"
r = render_automatic_eda_folder(
str(folder), out_dir=str(out),
emit_pdf=True, emit_pptx=False, emit_md=False,
)
assert r["status"] == "ok", r
assert r["pdf_path"] and os.path.exists(r["pdf_path"])
assert r["pptx_path"] is None
assert r["md_path"] is None
def test_path_does_not_exist(tmp_path):
"""Path inexistente -> status error, sin lanzar."""
r = render_automatic_eda_folder(str(tmp_path / "nope"))
assert r["status"] == "error"
assert "no existe" in r["error"].lower()