140 lines
4.8 KiB
Python
140 lines
4.8 KiB
Python
"""Evalua una expresion JavaScript en una pestana de Chrome via Chrome DevTools Protocol.
|
|
|
|
Primitiva de transport CDP: localiza un target (pestana) por substring de su URL,
|
|
abre el WebSocket de depuracion, ejecuta `Runtime.evaluate` y devuelve el valor
|
|
serializado. Base reutilizable para automatizar el navegador diario sin abrir
|
|
ventana nueva ni darle foco.
|
|
"""
|
|
|
|
import json
|
|
import urllib.request
|
|
|
|
import websocket
|
|
|
|
|
|
def cdp_eval(
|
|
expression: str,
|
|
*,
|
|
port: int = 9222,
|
|
target_url_substr: str = "",
|
|
await_promise: bool = False,
|
|
timeout_s: float = 10.0,
|
|
) -> dict:
|
|
"""Evalua una expresion JS en una pestana de Chrome elegida por substring de URL.
|
|
|
|
Args:
|
|
expression: Expresion JavaScript a evaluar en el contexto de la pagina.
|
|
port: Puerto de remote debugging de Chrome. Default 9222.
|
|
target_url_substr: Substring que debe contener la URL del target. Si "",
|
|
usa el primer target de tipo "page".
|
|
await_promise: Si True, espera a que se resuelva una Promise antes de
|
|
devolver el valor (`awaitPromise` de CDP).
|
|
timeout_s: Timeout (segundos) para la conexion WebSocket.
|
|
|
|
Returns:
|
|
dict con claves:
|
|
ok: bool — True si la evaluacion produjo un valor sin excepcion.
|
|
value: valor Python serializable devuelto por la expresion, o None.
|
|
error: str — mensaje de error (vacio si ok).
|
|
target_url: str — URL del target usado (vacio si ninguno).
|
|
"""
|
|
# 1. Listar targets via HTTP.
|
|
try:
|
|
with urllib.request.urlopen(
|
|
f"http://127.0.0.1:{port}/json", timeout=5
|
|
) as resp:
|
|
targets = json.loads(resp.read().decode())
|
|
except Exception as e: # noqa: BLE001 — red/HTTP/JSON, no relanzar
|
|
return {"ok": False, "value": None, "error": str(e), "target_url": ""}
|
|
|
|
# 2. Elegir el primer target type=="page" cuya url contenga el substring.
|
|
chosen = None
|
|
for t in targets:
|
|
if t.get("type") != "page":
|
|
continue
|
|
url = t.get("url", "")
|
|
if target_url_substr == "" or target_url_substr in url:
|
|
chosen = t
|
|
break
|
|
|
|
if chosen is None:
|
|
return {
|
|
"ok": False,
|
|
"value": None,
|
|
"error": f"no target matching {target_url_substr}",
|
|
"target_url": "",
|
|
}
|
|
|
|
target_url = chosen.get("url", "")
|
|
ws_url = chosen.get("webSocketDebuggerUrl", "")
|
|
|
|
# 3-4. Abrir WS, enviar Runtime.evaluate, drenar eventos hasta id==1.
|
|
try:
|
|
ws = websocket.create_connection(ws_url, timeout=timeout_s)
|
|
except Exception as e: # noqa: BLE001 — conexion WS
|
|
return {"ok": False, "value": None, "error": str(e), "target_url": ""}
|
|
|
|
try:
|
|
ws.send(json.dumps({
|
|
"id": 1,
|
|
"method": "Runtime.evaluate",
|
|
"params": {
|
|
"expression": expression,
|
|
"returnByValue": True,
|
|
"awaitPromise": await_promise,
|
|
},
|
|
}))
|
|
|
|
msg = None
|
|
# Drenar eventos intermedios hasta encontrar la respuesta con id==1.
|
|
while True:
|
|
raw = ws.recv()
|
|
if not raw:
|
|
break
|
|
try:
|
|
parsed = json.loads(raw)
|
|
except Exception: # noqa: BLE001 — frame no-JSON, ignorar
|
|
continue
|
|
if parsed.get("id") == 1:
|
|
msg = parsed
|
|
break
|
|
except Exception as e: # noqa: BLE001 — fallo de transport durante recv/send
|
|
return {"ok": False, "value": None, "error": str(e), "target_url": target_url}
|
|
finally:
|
|
try:
|
|
ws.close()
|
|
except Exception: # noqa: BLE001 — cierre best-effort
|
|
pass
|
|
|
|
if msg is None:
|
|
return {
|
|
"ok": False,
|
|
"value": None,
|
|
"error": "no response for evaluate (id=1)",
|
|
"target_url": target_url,
|
|
}
|
|
|
|
result = msg.get("result", {})
|
|
|
|
# 5. Si hubo excepcion en la evaluacion, devolverla como error.
|
|
exc = result.get("exceptionDetails")
|
|
if exc:
|
|
text = exc.get("text", "evaluation exception")
|
|
exception = exc.get("exception", {})
|
|
detail = exception.get("description") or exception.get("value")
|
|
error = f"{text}: {detail}" if detail else text
|
|
return {"ok": False, "value": None, "error": error, "target_url": target_url}
|
|
|
|
# 6. Extraer el valor serializado por returnByValue.
|
|
value = result.get("result", {}).get("value")
|
|
return {"ok": True, "value": value, "error": "", "target_url": target_url}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
|
|
expr = sys.argv[1] if len(sys.argv) > 1 else "document.title"
|
|
substr = sys.argv[2] if len(sys.argv) > 2 else ""
|
|
out = cdp_eval(expr, port=9222, target_url_substr=substr)
|
|
print(json.dumps(out, ensure_ascii=False, indent=2))
|