323 lines
13 KiB
Python
323 lines
13 KiB
Python
"""Ejecuta una recipe YAML contra Chrome remoto via CDP."""
|
|
|
|
import json
|
|
import re
|
|
import sys
|
|
import os
|
|
import time
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|
|
|
import urllib.request
|
|
import websocket
|
|
|
|
from core.validate_recipe_yaml import validate_recipe_yaml
|
|
|
|
|
|
def _ws_send_recv(ws, msg_id: int, method: str, params: dict, timeout: float = 10.0) -> dict:
|
|
"""Envia un mensaje CDP y espera respuesta con el mismo id."""
|
|
import threading
|
|
result_holder = {}
|
|
event = threading.Event()
|
|
|
|
original_on_message = ws.on_message
|
|
|
|
def on_message_wrapper(ws_app, message):
|
|
try:
|
|
msg = json.loads(message)
|
|
if msg.get("id") == msg_id:
|
|
result_holder["result"] = msg
|
|
event.set()
|
|
except Exception:
|
|
pass
|
|
if original_on_message:
|
|
original_on_message(ws_app, message)
|
|
|
|
ws.on_message = on_message_wrapper
|
|
ws.send(json.dumps({"id": msg_id, "method": method, "params": params}))
|
|
event.wait(timeout=timeout)
|
|
ws.on_message = original_on_message
|
|
return result_holder.get("result", {})
|
|
|
|
|
|
def _poll_selector(ws, selector: str, timeout_s: float = 10.0) -> bool:
|
|
"""Polling cada 200ms hasta que document.querySelector(selector) no sea null.
|
|
|
|
Drena eventos CDP (paginas con Page.enable emiten loads, frames, etc.) y
|
|
matchea por `id` para evitar leer respuestas ajenas o eventos del server.
|
|
"""
|
|
deadline = time.time() + timeout_s
|
|
msg_id = 1000
|
|
ws.settimeout(0.5)
|
|
while time.time() < deadline:
|
|
ws.send(json.dumps({
|
|
"id": msg_id,
|
|
"method": "Runtime.evaluate",
|
|
"params": {
|
|
"expression": f"!!document.querySelector({json.dumps(selector)})",
|
|
"returnByValue": True,
|
|
}
|
|
}))
|
|
# Leer hasta 30 frames buscando uno con nuestro id; ignorar eventos.
|
|
got_response = False
|
|
for _ in range(30):
|
|
try:
|
|
raw = ws.recv()
|
|
except Exception:
|
|
break
|
|
if not raw:
|
|
break
|
|
try:
|
|
msg = json.loads(raw)
|
|
except Exception:
|
|
continue
|
|
if msg.get("id") == msg_id:
|
|
got_response = True
|
|
val = msg.get("result", {}).get("result", {}).get("value", False)
|
|
if val:
|
|
return True
|
|
break
|
|
msg_id += 1
|
|
if not got_response:
|
|
time.sleep(0.2)
|
|
return False
|
|
|
|
|
|
def cdp_extract_recipe(
|
|
recipe_path: str,
|
|
debug_port: int = 9222,
|
|
tab_id: str | None = None,
|
|
record_run: bool = True,
|
|
) -> dict:
|
|
"""Ejecuta una recipe YAML contra Chrome remoto via CDP.
|
|
|
|
Args:
|
|
recipe_path: Ruta al archivo .yaml de la recipe.
|
|
debug_port: Puerto de depuracion remota de Chrome. Default 9222.
|
|
tab_id: ID del tab a usar. Si None, busca tab cuyo URL matchee url_pattern.
|
|
record_run: Si True y output.sink=='data_factory.runs', llama data_factory_record_run.
|
|
|
|
Returns:
|
|
{status, rows_out, kb_out, duration_ms, error, sample_rows}
|
|
"""
|
|
start_ms = int(time.time() * 1000)
|
|
|
|
# Leer y validar recipe
|
|
try:
|
|
with open(recipe_path, "r", encoding="utf-8") as f:
|
|
yaml_text = f.read()
|
|
except OSError as e:
|
|
return {"status": "error", "rows_out": 0, "kb_out": 0.0,
|
|
"duration_ms": 0, "error": str(e), "sample_rows": []}
|
|
|
|
validation = validate_recipe_yaml(yaml_text)
|
|
if not validation["valid"]:
|
|
return {"status": "error", "rows_out": 0, "kb_out": 0.0,
|
|
"duration_ms": 0, "error": "recipe invalida: " + "; ".join(validation["errors"]),
|
|
"sample_rows": []}
|
|
|
|
recipe = validation["parsed"]
|
|
url_pattern = recipe["url_pattern"]
|
|
steps = recipe["steps"]
|
|
output_cfg = recipe.get("output", {})
|
|
sink = output_cfg.get("sink", "stdout")
|
|
|
|
# Obtener lista de tabs
|
|
try:
|
|
with urllib.request.urlopen(
|
|
f"http://127.0.0.1:{debug_port}/json/list", timeout=5
|
|
) as resp:
|
|
tabs = json.loads(resp.read().decode())
|
|
except Exception as e:
|
|
return {"status": "error", "rows_out": 0, "kb_out": 0.0,
|
|
"duration_ms": 0,
|
|
"error": f"no se pudo conectar a Chrome en port {debug_port}: {e}",
|
|
"sample_rows": []}
|
|
|
|
# Encontrar tab
|
|
ws_url = None
|
|
if tab_id:
|
|
for tab in tabs:
|
|
if tab.get("id") == tab_id:
|
|
ws_url = tab.get("webSocketDebuggerUrl")
|
|
break
|
|
else:
|
|
for tab in tabs:
|
|
tab_url = tab.get("url", "")
|
|
if re.search(url_pattern, tab_url):
|
|
ws_url = tab.get("webSocketDebuggerUrl")
|
|
break
|
|
|
|
if not ws_url:
|
|
return {"status": "error", "rows_out": 0, "kb_out": 0.0,
|
|
"duration_ms": 0,
|
|
"error": f"no tab matching pattern: {url_pattern}",
|
|
"sample_rows": []}
|
|
|
|
# Ejecutar steps
|
|
last_result = None
|
|
try:
|
|
ws = websocket.create_connection(ws_url, timeout=10)
|
|
try:
|
|
for i, step in enumerate(steps):
|
|
if "wait_selector" in step:
|
|
selector = step["wait_selector"]
|
|
found = _poll_selector(ws, selector, timeout_s=10.0)
|
|
if not found:
|
|
raise RuntimeError(f"step {i}: timeout esperando selector '{selector}'")
|
|
elif "js" in step:
|
|
ws.send(json.dumps({
|
|
"id": i + 1,
|
|
"method": "Runtime.evaluate",
|
|
"params": {
|
|
"expression": step["js"],
|
|
"returnByValue": True,
|
|
"awaitPromise": True,
|
|
}
|
|
}))
|
|
raw = ws.recv()
|
|
msg = json.loads(raw)
|
|
result_obj = msg.get("result", {}).get("result", {})
|
|
last_result = result_obj.get("value")
|
|
finally:
|
|
ws.close()
|
|
except Exception as e:
|
|
return {"status": "error", "rows_out": 0, "kb_out": 0.0,
|
|
"duration_ms": int(time.time() * 1000) - start_ms,
|
|
"error": str(e), "sample_rows": []}
|
|
|
|
# Calcular metricas
|
|
rows = last_result if isinstance(last_result, list) else (
|
|
[last_result] if last_result is not None else []
|
|
)
|
|
rows_out = len(rows)
|
|
kb_out = len(json.dumps(rows, ensure_ascii=False).encode()) / 1024
|
|
sample_rows = rows[:5]
|
|
duration_ms = int(time.time() * 1000) - start_ms
|
|
|
|
# Sink
|
|
if sink == "stdout":
|
|
print(json.dumps(rows, ensure_ascii=False, indent=2))
|
|
elif sink == "json_file":
|
|
out_path = output_cfg.get("path", "output.json")
|
|
with open(out_path, "w", encoding="utf-8") as f:
|
|
json.dump(rows, f, ensure_ascii=False, indent=2)
|
|
elif sink == "duckdb":
|
|
duckdb_path = output_cfg.get("duckdb_path", "")
|
|
table_name = output_cfg.get("table", "")
|
|
if not duckdb_path or not table_name:
|
|
# not fatal: rows already returned via sample_rows
|
|
pass
|
|
else:
|
|
import duckdb
|
|
import uuid
|
|
import datetime
|
|
# resolve duckdb_path relative to FN_REGISTRY_ROOT if not absolute
|
|
if not os.path.isabs(duckdb_path):
|
|
duckdb_path = os.path.join(os.environ.get("FN_REGISTRY_ROOT", ""), duckdb_path)
|
|
os.makedirs(os.path.dirname(duckdb_path), exist_ok=True)
|
|
conn = duckdb.connect(duckdb_path)
|
|
try:
|
|
if rows:
|
|
# Detect columns from first row keys (assumes list of dicts).
|
|
if not isinstance(rows[0], dict):
|
|
# Fallback: wrap scalar rows as {"value": v}.
|
|
rows = [{"value": r} for r in rows]
|
|
cols = list(rows[0].keys())
|
|
# Build CREATE TABLE IF NOT EXISTS with VARCHAR for safety
|
|
# plus extracted_at TIMESTAMP and run_id VARCHAR for lineage.
|
|
col_defs = ", ".join(f'"{c}" VARCHAR' for c in cols)
|
|
ddl = (
|
|
f'CREATE TABLE IF NOT EXISTS "{table_name}" ('
|
|
f' run_id VARCHAR, extracted_at TIMESTAMP, {col_defs}'
|
|
f')'
|
|
)
|
|
conn.execute(ddl)
|
|
run_id_str = uuid.uuid4().hex[:16]
|
|
now_iso = datetime.datetime.utcnow().isoformat() + "Z"
|
|
placeholders = ", ".join(["?"] * (len(cols) + 2))
|
|
insert_sql = (
|
|
f'INSERT INTO "{table_name}" '
|
|
f'(run_id, extracted_at, {", ".join(chr(34) + c + chr(34) for c in cols)}) '
|
|
f'VALUES ({placeholders})'
|
|
)
|
|
for r in rows:
|
|
vals = [run_id_str, now_iso] + [str(r.get(c, "")) for c in cols]
|
|
conn.execute(insert_sql, vals)
|
|
# Also record into data_factory.runs with storage info
|
|
registry_root = os.environ.get("FN_REGISTRY_ROOT", "")
|
|
if registry_root and record_run:
|
|
import sqlite3
|
|
df_db = os.path.join(registry_root, "apps", "data_factory", "data_factory.db")
|
|
if os.path.exists(df_db):
|
|
try:
|
|
df_conn = sqlite3.connect(df_db)
|
|
df_conn.execute("PRAGMA foreign_keys = ON")
|
|
trigger = "dag" if os.environ.get("DAGU_ENV") else "manual"
|
|
db_id = output_cfg.get("database_id", recipe.get("name", "unknown") + "_db")
|
|
df_run_id = uuid.uuid4().hex[:16]
|
|
df_conn.execute(
|
|
"INSERT INTO runs(id, node_id, started_at, finished_at, status,"
|
|
" rows_in, rows_out, kb_in, kb_out, duration_ms, trigger, error, notes,"
|
|
" storage_db_id, storage_table)"
|
|
" VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
|
|
(
|
|
df_run_id, recipe.get("name", "unknown"),
|
|
now_iso, now_iso, "success",
|
|
0, rows_out, 0, int(round(kb_out)), duration_ms,
|
|
trigger, "",
|
|
json.dumps({"sample": sample_rows[:2]}, ensure_ascii=False)[:1000],
|
|
db_id, table_name,
|
|
),
|
|
)
|
|
df_conn.commit()
|
|
df_conn.close()
|
|
except Exception:
|
|
pass
|
|
finally:
|
|
conn.close()
|
|
elif sink == "data_factory.runs" and record_run:
|
|
# Escribe DIRECTO a data_factory.db evitando spawn `fn run` (loop infinito
|
|
# si data_factory_record_run re-ejecuta esta misma funcion). Confia en que
|
|
# el node ya existe en `nodes` con id == recipe.name.
|
|
try:
|
|
import sqlite3
|
|
import datetime
|
|
import uuid
|
|
registry_root = os.environ.get("FN_REGISTRY_ROOT", "").strip()
|
|
if not registry_root:
|
|
# No fatal — el dato ya fue extraido / impreso por otro sink
|
|
raise RuntimeError("FN_REGISTRY_ROOT not set; cannot locate data_factory.db")
|
|
db_path = os.path.join(registry_root, "apps", "data_factory", "data_factory.db")
|
|
trigger = "dag" if os.environ.get("DAGU_ENV") else "manual"
|
|
run_id = uuid.uuid4().hex[:16]
|
|
now = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%fZ")
|
|
node_id = recipe.get("name", "unknown")
|
|
conn = sqlite3.connect(db_path)
|
|
conn.execute("PRAGMA foreign_keys = ON")
|
|
conn.execute(
|
|
"INSERT INTO runs(id, node_id, started_at, finished_at, status,"
|
|
" rows_in, rows_out, kb_in, kb_out, duration_ms, trigger, error, notes)"
|
|
" VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)",
|
|
(
|
|
run_id, node_id, now, now, "success",
|
|
0, rows_out, 0, int(round(kb_out)), duration_ms,
|
|
trigger, "",
|
|
json.dumps({"sample": sample_rows[:2]}, ensure_ascii=False)[:1000],
|
|
),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
except Exception:
|
|
# No fatal — el dato ya fue extraido (sample_rows en retorno)
|
|
pass
|
|
|
|
return {
|
|
"status": "ok",
|
|
"rows_out": rows_out,
|
|
"kb_out": round(kb_out, 2),
|
|
"duration_ms": duration_ms,
|
|
"error": "",
|
|
"sample_rows": sample_rows,
|
|
}
|