dataforge/fn_registry
1
0
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases Wiki Activity

chore: auto-commit (286 archivos)

Browse Source 
- .claude/agents/fn-orquestador/SKILL.md
- .claude/commands/fn_claude.md
- .claude/rules/INDEX.md
- .claude/rules/cpp_apps.md
- .claude/rules/ids_naming.md
- CHANGELOG.md
- apps/dag_engine/README.md
- apps/dag_engine/api.go
- apps/dag_engine/dags_migrated/example.yaml
- apps/dag_engine/dags_migrated/example_lineage_tracking.yaml
- ...

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Egutierrez
2026-05-16 16:33:22 +02:00
parent 0b9af8f1bb
commit a03675113a
281 changed files with 12596 additions and 19526 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/functions/core/chunk_ax_tree.md
+57
View File
@@ -0,0 +1,57 @@
---
name: chunk_ax_tree
kind: function
lang: py
domain: core
version: "1.0.0"
purity: pure
signature: "def chunk_ax_tree(nodes: list[dict], max_chars: int = 25000) -> list[list[dict]]"
description: "Divide una lista de AXNode en chunks de hasta max_chars (serializado JSON). Hace DFS desde raíz y añade nodo 'context' con path-from-root al inicio de cada chunk no inicial."
tags: [navegator, accessibility, chunking, pure, ax-tree, llm]
uses_functions: [trim_ax_tree_py_core]
uses_types: []
returns: []
returns_optional: false
error_type: ""
imports: [json]
params:
- name: nodes
desc: "Lista de AXNode en formato CDP (con nodeId, childIds, role, name). Usar después de trim_ax_tree para reducir ruido."
- name: max_chars
desc: "Límite de caracteres al serializar el chunk a JSON. Default 25000 (~6k tokens)."
output: "Lista de chunks. Cada chunk es lista de AXNode. Los chunks no iniciales arrancan con un nodo sintético role='context' que contiene el path desde la raíz."
tested: true
tests:
- "lista vacia devuelve lista vacia"
- "nodos que caben en un chunk devuelven un solo chunk"
- "nodos que no caben generan multiples chunks"
- "cada chunk serializado no supera max_chars mas overhead"
test_file_path: "python/functions/core/chunk_ax_tree_test.py"
file_path: "python/functions/core/chunk_ax_tree.py"
---
## Ejemplo
```python
from core.trim_ax_tree import trim_ax_tree
from core.chunk_ax_tree import chunk_ax_tree
# Obtener AX tree de CDP, luego:
trimmed = trim_ax_tree(raw_nodes)
chunks = chunk_ax_tree(trimmed, max_chars=25000)
for i, chunk in enumerate(chunks):
print(f"Chunk {i}: {len(chunk)} nodos")
# Enviar chunk a claude_cli_prompt o a la API de Claude
```
## Cuando usarla
Cuando el AX tree de una página compleja supera el context window del modelo (>25k chars). Chunkearlo permite procesar secciones independientes manteniendo el path-from-root como contexto para que el LLM entienda la estructura jerárquica.
## Gotchas
- El nodo `context` sintético (role="context") al inicio de cada chunk no es un nodo CDP real — filtrarlo si se necesita reconstruir el árbol.
- `max_chars` es límite aproximado: el nodo de contexto puede hacer que el chunk supere ligeramente el límite. Se recomienda un 10-20% de margen.
- Usar `trim_ax_tree` antes para reducir el número de nodos y mejorar la utilidad de cada chunk.
- Árboles muy profundos con pocos nodos por nivel generan chunks con mucho overhead de contexto.
python/functions/core/chunk_ax_tree.py
+91
View File
@@ -0,0 +1,91 @@
"""Divide una lista de AXNode en chunks de tamaño máximo para enviar a LLMs."""
import json
def chunk_ax_tree(nodes: list[dict], max_chars: int = 25000) -> list[list[dict]]:
"""Splittea AX tree en chunks de hasta max_chars cuando se serializa a JSON.
Estrategia DFS desde el nodo raíz:
- Acumula nodos en el chunk actual.
- Cuando añadir el siguiente subárbol superaría max_chars, cierra el chunk
y abre uno nuevo.
- Cada chunk incluye al inicio los nodos "context" (path desde root hasta
el primer nodo del chunk) para que el LLM tenga orientación estructural.
Args:
nodes: Lista de AXNode en formato CDP (con nodeId, childIds, role, name).
max_chars: Límite de caracteres al serializar el chunk a JSON. Default 25000.
Returns:
Lista de chunks. Cada chunk es una lista de AXNode. El primer elemento de
cada chunk (salvo el primero) es un nodo sintético con role="context" que
contiene el path desde la raíz.
"""
if not nodes:
return []
# Construir lookup y encontrar raíz
by_id: dict[str, dict] = {n["nodeId"]: n for n in nodes}
all_children: set[str] = set()
for n in nodes:
for cid in n.get("childIds", []):
all_children.add(cid)
roots = [n for n in nodes if n["nodeId"] not in all_children]
if not roots:
# Fallback: usar el primer nodo como raíz
roots = [nodes[0]]
# DFS para obtener orden de visita + path desde raíz
visit_order: list[tuple[dict, list[dict]]] = [] # (nodo, ancestors)
def dfs(node: dict, ancestors: list[dict]) -> None:
visit_order.append((node, list(ancestors)))
for cid in node.get("childIds", []):
child = by_id.get(cid)
if child:
dfs(child, ancestors + [node])
for root in roots:
dfs(root, [])
# Agrupar en chunks respetando max_chars
chunks: list[list[dict]] = []
current_chunk: list[dict] = []
current_chars: int = 2 # "[]"
for node, ancestors in visit_order:
node_json = json.dumps(node, ensure_ascii=False)
node_chars = len(node_json) + 2 # ", " separator
if current_chunk and current_chars + node_chars > max_chars:
chunks.append(current_chunk)
# Nuevo chunk con contexto (path desde root)
current_chunk = []
current_chars = 2
if ancestors:
context_node = {
"nodeId": "context",
"role": {"value": "context"},
"name": {"value": "path-from-root"},
"childIds": [],
"context_path": [
{
"nodeId": a["nodeId"],
"role": a.get("role", {}).get("value", "") if isinstance(a.get("role"), dict) else str(a.get("role", "")),
"name": a.get("name", {}).get("value", "") if isinstance(a.get("name"), dict) else str(a.get("name", "")),
}
for a in ancestors
],
}
ctx_chars = len(json.dumps(context_node, ensure_ascii=False)) + 2
current_chunk.append(context_node)
current_chars += ctx_chars
current_chunk.append(node)
current_chars += node_chars
if current_chunk:
chunks.append(current_chunk)
return chunks
python/functions/core/chunk_ax_tree_test.py
+75
View File
@@ -0,0 +1,75 @@
"""Tests para chunk_ax_tree."""
import json
import sys
import os
sys.path.insert(0, os.path.dirname(__file__))
from chunk_ax_tree import chunk_ax_tree
def _node(node_id: str, child_ids=None) -> dict:
return {
"nodeId": node_id,
"role": {"value": "generic"},
"name": {"value": f"node-{node_id}"},
"childIds": child_ids or [],
}
def _make_linear_tree(n: int) -> list[dict]:
"""Cadena lineal: 0->1->2->...->n-1"""
nodes = []
for i in range(n):
child = [str(i + 1)] if i < n - 1 else []
nodes.append(_node(str(i), child))
return nodes
def test_lista_vacia_devuelve_lista_vacia():
result = chunk_ax_tree([])
assert result == [], f"Expected [], got {result}"
def test_nodos_que_caben_en_un_chunk_devuelven_un_solo_chunk():
nodes = _make_linear_tree(3)
result = chunk_ax_tree(nodes, max_chars=10000)
assert len(result) == 1, f"Expected 1 chunk, got {len(result)}"
# Todos los nodos deben estar presentes
ids_in_chunk = [n["nodeId"] for n in result[0]]
for node in nodes:
assert node["nodeId"] in ids_in_chunk, f"{node['nodeId']} missing from chunk"
def test_nodos_que_no_caben_generan_multiples_chunks():
# 100 nodos, cada uno ~80 chars serializado -> ~8000 chars total
# con max_chars=200 debe generar varios chunks
nodes = [_node(str(i)) for i in range(100)]
# Hacer árbol plano: raíz con todos como hijos
nodes[0]["childIds"] = [str(i) for i in range(1, 100)]
result = chunk_ax_tree(nodes, max_chars=200)
assert len(result) > 1, f"Expected >1 chunks, got {len(result)}"
def test_cada_chunk_serializado_no_supera_max_chars_mas_overhead():
nodes = [_node(str(i)) for i in range(100)]
nodes[0]["childIds"] = [str(i) for i in range(1, 100)]
max_chars = 500
result = chunk_ax_tree(nodes, max_chars=max_chars)
overhead_factor = 1.5 # permitir 50% de overhead por nodo contexto
for i, chunk in enumerate(result):
serialized = json.dumps(chunk, ensure_ascii=False)
limit = max_chars * overhead_factor
assert len(serialized) < limit, (
f"Chunk {i} tiene {len(serialized)} chars, supera {limit} "
f"(max_chars={max_chars}, overhead_factor={overhead_factor})"
)
if __name__ == "__main__":
test_lista_vacia_devuelve_lista_vacia()
test_nodos_que_caben_en_un_chunk_devuelven_un_solo_chunk()
test_nodos_que_no_caben_generan_multiples_chunks()
test_cada_chunk_serializado_no_supera_max_chars_mas_overhead()
print("All tests passed.")
python/functions/core/coerce_types.md
+1 -1
View File
@@ -7,7 +7,7 @@ version: "1.0.0"
purity: pure
signature: "def coerce_types(data: dict, schema: dict[str, str]) -> tuple[dict, list[str]]"
description: "Convierte valores de un dict a los tipos esperados segun un schema declarativo. Soporta int, float, str, bool, datetime, list[str]. Util para normalizar datos de CSV, JSON o query params. Nunca muta el original. Coerciones imposibles generan warning y mantienen el valor original."
tags: [coercion, types, normalization, pure, core, csv, json, pendiente-usar]
tags: [coercion, types, normalization, pure, core, csv, json, pendiente-usar, transformer]
uses_functions: []
uses_types: []
returns: []
python/functions/core/csv_to_parquet_duckdb.md
+1 -1
View File
@@ -7,7 +7,7 @@ version: "1.0.0"
purity: impure
signature: "csv_to_parquet_duckdb(csv_path: str | Path, parquet_path: str | Path, column_casts: dict[str, str] | None = None, overwrite: bool = False) -> bool"
description: "Convierte un CSV a Parquet usando DuckDB read_csv_auto. Si overwrite=False y el parquet ya existe no hace nada. column_casts permite sobreescribir tipos inferidos por columna. Retorna True si escribió."
tags: [csv, parquet, duckdb, etl, core, pendiente-usar]
tags: [csv, parquet, duckdb, etl, core, pendiente-usar, transformer]
uses_functions: []
uses_types: []
returns: []
python/functions/core/from_csv.md
+1 -1
View File
@@ -7,7 +7,7 @@ version: "1.0.0"
purity: pure
signature: "from_csv(text: str, delimiter: str = ',', has_header: bool = True) -> list[dict]"
description: "Parser CSV a datos tabulares. Complemento de to_csv. Soporta campos entre comillas con escaping RFC 4180. Si has_header=False, genera keys col_0, col_1, etc."
tags: [csv, parser, import, tabular, format, pendiente-usar]
tags: [csv, parser, import, tabular, format, pendiente-usar, extractor]
uses_functions: []
uses_types: []
returns: []
python/functions/core/infer_json_rows_schema.md
+53
View File
@@ -0,0 +1,53 @@
---
name: infer_json_rows_schema
kind: function
lang: py
domain: core
version: "1.0.0"
purity: pure
signature: "def infer_json_rows_schema(data: object) -> dict"
description: "Dado un JSON deserializado, localiza el primer array de objetos con >=1 elementos y devuelve su schema: root_path, fields [{name, type, sample}], count."
tags: [navegator, json, schema, scraping, pure]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: ""
imports: []
params:
- name: data
desc: "JSON deserializado: dict, list, o cualquier valor Python resultante de json.loads()."
output: "dict con {root_path: str, fields: [{name, type, sample}], count: int}. Si no encuentra array de objetos devuelve root_path='' y fields=[]."
tested: true
tests:
- "lista de dicts directa retorna root_path dolar"
- "dict con clave results retorna root_path dolar punto results"
- "input vacio retorna schema vacio"
test_file_path: "python/functions/core/infer_json_rows_schema_test.py"
file_path: "python/functions/core/infer_json_rows_schema.py"
---
## Ejemplo
```python
import sys
sys.path.insert(0, "python/functions")
from core.infer_json_rows_schema import infer_json_rows_schema
schema = infer_json_rows_schema([{"id": 1, "name": "foo"}])
# {"root_path": "$", "fields": [{"name": "id", "type": "int", "sample": "1"}, ...], "count": 1}
schema2 = infer_json_rows_schema({"results": [{"a": "x"}]})
# {"root_path": "$.results", "fields": [...], "count": 1}
```
## Cuando usarla
Cuando recibes JSON de una API o scraping y necesitas descubrir automaticamente que campos contiene y sus tipos, sin conocer la estructura de antemano. Paso previo a generar un recipe YAML o mapear campos a un schema destino.
## Notas
- Prefiere claves `results, items, data, records, rows, entries` en DFS.
- Tipos: string|int|float|bool|null|array|object.
- `sample` truncado a 80 chars de la primera fila.
- Funcion pura: sin I/O, sin estado.
python/functions/core/infer_json_rows_schema.py
+79
View File
@@ -0,0 +1,79 @@
"""Infiere schema de un JSON deserializado localizando el primer array de objetos."""
def _type(value) -> str:
if value is None:
return "null"
if isinstance(value, bool):
return "bool"
if isinstance(value, int):
return "int"
if isinstance(value, float):
return "float"
if isinstance(value, str):
return "string"
if isinstance(value, list):
return "array"
if isinstance(value, dict):
return "object"
return "string"
def _truncate(value, max_chars: int = 80) -> str:
s = str(value)
return s[:max_chars] if len(s) > max_chars else s
_PREFERRED_KEYS = ("results", "items", "data", "records", "rows", "entries")
def _find_array_of_dicts(obj, path: str = "$"):
"""DFS buscando el primer array de dicts con >=1 elementos."""
if isinstance(obj, list) and len(obj) >= 1 and isinstance(obj[0], dict):
return path, obj
if isinstance(obj, dict):
# Primero probar claves preferidas en orden
for key in _PREFERRED_KEYS:
if key in obj:
result = _find_array_of_dicts(obj[key], f"{path}.{key}")
if result is not None:
return result
# Luego resto de claves
for key, val in obj.items():
if key in _PREFERRED_KEYS:
continue
result = _find_array_of_dicts(val, f"{path}.{key}")
if result is not None:
return result
return None
def infer_json_rows_schema(data: object) -> dict:
"""Dado un JSON deserializado, localiza el primer array de objetos con >=1 elementos.
Args:
data: JSON deserializado (dict, list, u otro valor Python).
Returns:
Schema con root_path, fields y count. Si no encuentra array, devuelve
root_path vacío y fields vacío.
"""
result = _find_array_of_dicts(data)
if result is None:
return {"root_path": "", "fields": [], "count": 0}
root_path, rows = result
first_row = rows[0]
fields = []
for name, value in first_row.items():
fields.append({
"name": name,
"type": _type(value),
"sample": _truncate(value),
})
return {
"root_path": root_path,
"fields": fields,
"count": len(rows),
}
python/functions/core/infer_json_rows_schema_test.py
+62
View File
@@ -0,0 +1,62 @@
"""Tests para infer_json_rows_schema."""
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
from core.infer_json_rows_schema import infer_json_rows_schema
def test_lista_de_dicts_directa_retorna_root_path_dolar():
data = [{"a": 1, "b": "x"}]
result = infer_json_rows_schema(data)
assert result["root_path"] == "$"
assert len(result["fields"]) == 2
fields_by_name = {f["name"]: f for f in result["fields"]}
assert fields_by_name["a"]["type"] == "int"
assert fields_by_name["b"]["type"] == "string"
assert result["count"] == 1
def test_dict_con_clave_results_retorna_root_path_dolar_punto_results():
data = {"results": [{"id": 42, "label": "foo"}]}
result = infer_json_rows_schema(data)
assert result["root_path"] == "$.results"
assert result["count"] == 1
fields_by_name = {f["name"]: f for f in result["fields"]}
assert fields_by_name["id"]["type"] == "int"
assert fields_by_name["label"]["type"] == "string"
def test_input_vacio_retorna_schema_vacio():
assert infer_json_rows_schema({}) == {"root_path": "", "fields": [], "count": 0}
assert infer_json_rows_schema([]) == {"root_path": "", "fields": [], "count": 0}
def test_tipos_varios():
data = [{"s": "hello", "i": 1, "f": 1.5, "b": True, "n": None, "arr": [1], "obj": {"x": 1}}]
result = infer_json_rows_schema(data)
by_name = {f["name"]: f["type"] for f in result["fields"]}
assert by_name["s"] == "string"
assert by_name["i"] == "int"
assert by_name["f"] == "float"
assert by_name["b"] == "bool"
assert by_name["n"] == "null"
assert by_name["arr"] == "array"
assert by_name["obj"] == "object"
def test_sample_truncado():
long_val = "x" * 100
data = [{"field": long_val}]
result = infer_json_rows_schema(data)
assert len(result["fields"][0]["sample"]) == 80
if __name__ == "__main__":
test_lista_de_dicts_directa_retorna_root_path_dolar()
test_dict_con_clave_results_retorna_root_path_dolar_punto_results()
test_input_vacio_retorna_schema_vacio()
test_tipos_varios()
test_sample_truncado()
print("All tests passed.")
python/functions/core/trim_ax_tree.md
+55
View File
@@ -0,0 +1,55 @@
---
name: trim_ax_tree
kind: function
lang: py
domain: core
version: "1.0.0"
purity: pure
signature: "def trim_ax_tree(nodes: list[dict]) -> list[dict]"
description: "Compacta lista de AXNode CDP descartando nodos ignorados, roles genéricos sin contenido y StaticText vacíos. Colapsa nodos con un único hijo del mismo role."
tags: [navegator, accessibility, dom, scraping, pure, ax-tree]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: ""
imports: []
params:
- name: nodes
desc: "Lista de AXNode en formato CDP (Accessibility.getFullAXTree). Cada nodo es un dict con nodeId, ignored, role, name, childIds."
output: "Lista reducida de AXNode con misma estructura. Nodos descartados y colapsados segun las reglas de trimming."
tested: true
tests:
- "descarta nodo generic sin name ni childIds"
- "mantiene nodo button con name"
- "descarta nodo ignored"
- "descarta StaticText con name vacio"
- "colapsa nodo con un hijo del mismo role"
test_file_path: "python/functions/core/trim_ax_tree_test.py"
file_path: "python/functions/core/trim_ax_tree.py"
---
## Ejemplo
```python
from core.trim_ax_tree import trim_ax_tree
nodes = [
{"nodeId": "1", "ignored": False, "role": {"value": "RootWebArea"}, "name": {"value": ""}, "childIds": ["2", "3"]},
{"nodeId": "2", "ignored": False, "role": {"value": "generic"}, "name": {"value": ""}, "childIds": []},
{"nodeId": "3", "ignored": False, "role": {"value": "button"}, "name": {"value": "Enviar"}, "childIds": []},
]
result = trim_ax_tree(nodes)
# result: nodo "2" descartado, quedan "1" y "3"
print([n["nodeId"] for n in result]) # ["1", "3"]
```
## Cuando usarla
Antes de pasar un AX tree a un LLM o de chunkear con `chunk_ax_tree` — reduce el numero de nodos eliminando ruido (wrappers genéricos, texto vacío, nodos ignorados) y mejora la relacion señal/ruido del contexto enviado al modelo.
## Gotchas
- Funcion pura: no modifica los dicts originales (crea copias al fusionar).
- El colapso iterativo puede cambiar la profundidad del arbol — parentIds de nodos externos no se actualizan.
- Los nodos cuyo nodeId ya no existe en el resultado pero aparecen como childIds de otros nodos quedan como referencias colgantes — filtrar antes de usar el arbol para navegacion.
python/functions/core/trim_ax_tree.py
+93
View File
@@ -0,0 +1,93 @@
"""Compacta una lista de AXNode CDP descartando nodos irrelevantes."""
def trim_ax_tree(nodes: list[dict]) -> list[dict]:
"""Compacta lista de AXNode (formato CDP Accessibility.getFullAXTree).
Descarta:
- Nodos con ignored=true.
- role 'generic' o 'none' sin name no-vacio Y sin childIds.
- role 'StaticText' con name vacio.
Colapsa: si un nodo tiene exactamente 1 hijo y el mismo role, fusiona
el hijo en el padre (hereda childIds del hijo, descarta el nodo hijo).
Preserva: nodeId, parentId, childIds, role.value, name.value, value.value.
Args:
nodes: Lista de AXNode en formato CDP. Cada nodo es un dict con
campos 'nodeId', 'ignored', 'role', 'name', 'childIds', etc.
Returns:
Lista reducida de AXNode con la misma estructura de campos.
"""
if not nodes:
return []
def _role(node: dict) -> str:
r = node.get("role", {})
if isinstance(r, dict):
return r.get("value", "")
return str(r)
def _name(node: dict) -> str:
n = node.get("name", {})
if isinstance(n, dict):
return n.get("value", "")
return str(n) if n else ""
def _child_ids(node: dict) -> list[str]:
return node.get("childIds", [])
# Paso 1: descartar nodos ignorados y roles descartables
def _should_discard(node: dict) -> bool:
if node.get("ignored", False):
return True
role = _role(node)
name = _name(node)
child_ids = _child_ids(node)
if role in ("generic", "none") and not name and not child_ids:
return True
if role == "StaticText" and not name:
return True
return False
kept = [n for n in nodes if not _should_discard(n)]
# Construir lookup por nodeId para el paso de colapso
by_id: dict[str, dict] = {n["nodeId"]: n for n in kept}
# Paso 2: colapso de nodo con 1 solo hijo del mismo role
# Iteramos hasta que no haya mas fusiones (convergencia)
changed = True
while changed:
changed = False
new_kept = []
removed: set[str] = set()
for node in by_id.values():
if node["nodeId"] in removed:
continue
child_ids = _child_ids(node)
if len(child_ids) == 1:
child_id = child_ids[0]
child = by_id.get(child_id)
if child and _role(child) == _role(node):
# Fusionar: heredar childIds del hijo
merged = dict(node)
merged["childIds"] = _child_ids(child)
by_id[node["nodeId"]] = merged
removed.add(child_id)
changed = True
if changed:
by_id = {k: v for k, v in by_id.items() if k not in removed}
# Preservar orden original
original_order = [n["nodeId"] for n in nodes]
result = []
seen: set[str] = set()
for nid in original_order:
if nid in by_id and nid not in seen:
result.append(by_id[nid])
seen.add(nid)
return result
python/functions/core/trim_ax_tree_test.py
+87
View File
@@ -0,0 +1,87 @@
"""Tests para trim_ax_tree."""
import sys
import os
sys.path.insert(0, os.path.dirname(__file__))
from trim_ax_tree import trim_ax_tree
def _node(node_id: str, role: str, name: str = "", child_ids=None, ignored: bool = False) -> dict:
return {
"nodeId": node_id,
"ignored": ignored,
"role": {"value": role},
"name": {"value": name},
"childIds": child_ids or [],
}
def test_descarta_nodo_generic_sin_name_ni_childIds():
nodes = [
_node("1", "RootWebArea", "", ["2"]),
_node("2", "generic", "", []),
]
result = trim_ax_tree(nodes)
ids = [n["nodeId"] for n in result]
assert "2" not in ids, f"generic sin name/childIds debe descartarse, got {ids}"
def test_mantiene_nodo_button_con_name():
nodes = [
_node("1", "RootWebArea", "", ["2"]),
_node("2", "button", "Click me", []),
]
result = trim_ax_tree(nodes)
ids = [n["nodeId"] for n in result]
assert "2" in ids, f"button con name debe mantenerse, got {ids}"
def test_descarta_nodo_ignored():
nodes = [
_node("1", "RootWebArea", "", ["2"]),
_node("2", "button", "Visible", [], ignored=True),
]
result = trim_ax_tree(nodes)
ids = [n["nodeId"] for n in result]
assert "2" not in ids, f"nodo ignored debe descartarse, got {ids}"
def test_descarta_StaticText_con_name_vacio():
nodes = [
_node("1", "RootWebArea", "", ["2", "3"]),
_node("2", "StaticText", "", []),
_node("3", "StaticText", "Hola", []),
]
result = trim_ax_tree(nodes)
ids = [n["nodeId"] for n in result]
assert "2" not in ids, f"StaticText vacio debe descartarse, got {ids}"
assert "3" in ids, f"StaticText con name debe mantenerse, got {ids}"
def test_colapsa_nodo_con_un_hijo_del_mismo_role():
nodes = [
_node("1", "RootWebArea", "", ["2"]),
_node("2", "group", "", ["3"]),
_node("3", "group", "", ["4"]),
_node("4", "button", "OK", []),
]
result = trim_ax_tree(nodes)
ids = [n["nodeId"] for n in result]
# Nodo 2 y 3 tienen mismo role y uno tiene 1 solo hijo — deben colapsar
assert "4" in ids, f"button debe estar presente, got {ids}"
# Tras colapso, el nodo raiz group debe tener childIds apuntando a button
root_group = next((n for n in result if n["nodeId"] in ("2", "3")), None)
if root_group:
assert "4" in root_group.get("childIds", []), \
f"grupo colapsado debe tener button como hijo, got {root_group}"
if __name__ == "__main__":
test_descarta_nodo_generic_sin_name_ni_childIds()
test_mantiene_nodo_button_con_name()
test_descarta_nodo_ignored()
test_descarta_StaticText_con_name_vacio()
test_colapsa_nodo_con_un_hijo_del_mismo_role()
print("All tests passed.")
python/functions/core/validate_recipe_yaml.md
+61
View File
@@ -0,0 +1,61 @@
---
name: validate_recipe_yaml
kind: function
lang: py
domain: core
version: "1.0.0"
purity: pure
signature: "def validate_recipe_yaml(yaml_text: str) -> dict"
description: "Parsea y valida recipe YAML del navegator. Comprueba name (snake_case), url_pattern (regex valido), steps (js o wait_selector), output.schema y output.sink."
tags: [navegator, recipe, validation, yaml, pure, validator]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: ""
imports: [re, yaml]
params:
- name: yaml_text
desc: "Texto completo de la recipe en formato YAML."
output: "dict {valid: bool, errors: [str], warnings: [str], parsed: dict}. valid=True solo si errors esta vacio."
tested: true
tests:
- "recipe valida minima retorna valid true"
- "name ausente produce error"
- "url_pattern invalido produce error"
- "step sin js ni wait_selector produce error"
- "sink invalido produce error"
test_file_path: "python/functions/core/validate_recipe_yaml_test.py"
file_path: "python/functions/core/validate_recipe_yaml.py"
---
## Ejemplo
```python
import sys
sys.path.insert(0, "python/functions")
from core.validate_recipe_yaml import validate_recipe_yaml
yaml_text = """
name: product_list
url_pattern: "https://shop\\.example\\.com/.*"
steps:
- wait_selector: ".product-card"
- js: "Array.from(document.querySelectorAll('.product-card')).map(e => e.innerText)"
output:
sink: stdout
"""
result = validate_recipe_yaml(yaml_text)
# {"valid": True, "errors": [], "warnings": [], "parsed": {...}}
```
## Cuando usarla
Antes de ejecutar una recipe con `cdp_extract_recipe`: validar el YAML en seco sin necesitar Chrome. Tambien util en un editor de recipes para dar feedback al usuario en tiempo real.
## Gotchas
- Requiere `pyyaml` en el entorno; si falta devuelve `valid=False` con mensaje claro (no lanza excepcion).
- Funcion pura: no accede a red ni a disco.
- `url_pattern` se compila con `re.compile` — patrones validos en Python pero invalidos en otros motores no se detectan aqui.
python/functions/core/validate_recipe_yaml.py
+124
View File
@@ -0,0 +1,124 @@
"""Parsea y valida recipe YAML del navegator."""
import re
def validate_recipe_yaml(yaml_text: str) -> dict:
"""Parsea + valida recipe YAML del navegator.
Args:
yaml_text: Texto YAML de la recipe.
Returns:
{valid: bool, errors: [str], warnings: [str], parsed: dict}
"""
try:
import yaml
except ImportError:
return {
"valid": False,
"errors": ["pyyaml no disponible: instalar con 'pip install pyyaml'"],
"warnings": [],
"parsed": {},
}
errors = []
warnings = []
parsed = {}
try:
parsed = yaml.safe_load(yaml_text) or {}
except yaml.YAMLError as e:
return {
"valid": False,
"errors": [f"YAML invalido: {e}"],
"warnings": [],
"parsed": {},
}
if not isinstance(parsed, dict):
return {
"valid": False,
"errors": ["El YAML debe ser un objeto (dict) en el nivel raiz."],
"warnings": [],
"parsed": parsed,
}
# name obligatorio, snake_case
name = parsed.get("name")
if not name:
errors.append("Campo 'name' obligatorio.")
elif not isinstance(name, str):
errors.append("Campo 'name' debe ser string.")
elif not re.fullmatch(r"[a-z][a-z0-9_]*", name):
errors.append(f"Campo 'name' debe ser snake_case (solo [a-z0-9_], empieza con letra): '{name}'.")
# url_pattern obligatorio, regex valido
url_pattern = parsed.get("url_pattern")
if not url_pattern:
errors.append("Campo 'url_pattern' obligatorio.")
elif not isinstance(url_pattern, str):
errors.append("Campo 'url_pattern' debe ser string.")
else:
try:
re.compile(url_pattern)
except re.error as e:
errors.append(f"Campo 'url_pattern' no es un regex valido: {e}")
# steps lista no vacia, cada step requiere js O wait_selector
steps = parsed.get("steps")
if not steps:
errors.append("Campo 'steps' obligatorio y no puede estar vacio.")
elif not isinstance(steps, list):
errors.append("Campo 'steps' debe ser una lista.")
else:
for i, step in enumerate(steps):
if not isinstance(step, dict):
errors.append(f"Step {i}: debe ser un objeto.")
continue
has_js = "js" in step
has_wait = "wait_selector" in step
if not has_js and not has_wait:
errors.append(f"Step {i}: requiere 'js' o 'wait_selector'.")
if has_js and has_wait:
warnings.append(f"Step {i}: tiene tanto 'js' como 'wait_selector'; se usara 'js'.")
# output opcional pero validado si presente
output = parsed.get("output")
if output is not None:
if not isinstance(output, dict):
errors.append("Campo 'output' debe ser un objeto.")
else:
schema = output.get("schema")
if schema is not None:
if not isinstance(schema, list):
errors.append("Campo 'output.schema' debe ser una lista.")
else:
valid_types = {"string", "int", "float", "bool", "date"}
for j, item in enumerate(schema):
if not isinstance(item, dict):
errors.append(f"output.schema[{j}]: debe ser un objeto.")
continue
if "field" not in item:
errors.append(f"output.schema[{j}]: falta campo 'field'.")
if "type" not in item:
errors.append(f"output.schema[{j}]: falta campo 'type'.")
elif item["type"] not in valid_types:
warnings.append(
f"output.schema[{j}]: tipo '{item['type']}' no reconocido "
f"(esperado: {sorted(valid_types)})."
)
sink = output.get("sink")
valid_sinks = {"data_factory.runs", "stdout", "json_file"}
if sink is not None and sink not in valid_sinks:
errors.append(
f"Campo 'output.sink' debe ser uno de {sorted(valid_sinks)}, got '{sink}'."
)
return {
"valid": len(errors) == 0,
"errors": errors,
"warnings": warnings,
"parsed": parsed,
}
python/functions/core/validate_recipe_yaml_test.py
+115
View File
@@ -0,0 +1,115 @@
"""Tests para validate_recipe_yaml."""
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
from core.validate_recipe_yaml import validate_recipe_yaml
VALID_YAML = """
name: product_list
url_pattern: 'https://shop\\.example\\.com/.*'
steps:
- wait_selector: ".product-card"
- js: "document.querySelector('.price').innerText"
output:
sink: stdout
"""
def test_recipe_valida_minima_retorna_valid_true():
result = validate_recipe_yaml(VALID_YAML)
assert result["valid"] is True
assert result["errors"] == []
assert isinstance(result["parsed"], dict)
assert result["parsed"]["name"] == "product_list"
def test_name_ausente_produce_error():
yaml_text = """
url_pattern: "https://example.com"
steps:
- js: "1+1"
"""
result = validate_recipe_yaml(yaml_text)
assert result["valid"] is False
assert any("name" in e for e in result["errors"])
def test_url_pattern_invalido_produce_error():
yaml_text = """
name: my_recipe
url_pattern: "[invalid("
steps:
- js: "1"
"""
result = validate_recipe_yaml(yaml_text)
assert result["valid"] is False
assert any("url_pattern" in e for e in result["errors"])
def test_step_sin_js_ni_wait_selector_produce_error():
yaml_text = """
name: bad_steps
url_pattern: "https://example.com"
steps:
- description: "solo descripcion sin accion"
"""
result = validate_recipe_yaml(yaml_text)
assert result["valid"] is False
assert any("js" in e or "wait_selector" in e for e in result["errors"])
def test_sink_invalido_produce_error():
yaml_text = """
name: bad_sink
url_pattern: "https://example.com"
steps:
- js: "1"
output:
sink: invalid_sink
"""
result = validate_recipe_yaml(yaml_text)
assert result["valid"] is False
assert any("sink" in e for e in result["errors"])
def test_name_no_snake_case_produce_error():
yaml_text = """
name: MyRecipe
url_pattern: "https://example.com"
steps:
- js: "1"
"""
result = validate_recipe_yaml(yaml_text)
assert result["valid"] is False
assert any("snake_case" in e for e in result["errors"])
def test_output_schema_valido():
yaml_text = """
name: with_schema
url_pattern: "https://example.com"
steps:
- js: "1"
output:
sink: stdout
schema:
- field: price
type: float
- field: name
type: string
"""
result = validate_recipe_yaml(yaml_text)
assert result["valid"] is True
if __name__ == "__main__":
test_recipe_valida_minima_retorna_valid_true()
test_name_ausente_produce_error()
test_url_pattern_invalido_produce_error()
test_step_sin_js_ni_wait_selector_produce_error()
test_sink_invalido_produce_error()
test_name_no_snake_case_produce_error()
test_output_schema_valido()
print("All tests passed.")