8742cb25be
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
145 lines
5.3 KiB
Python
145 lines
5.3 KiB
Python
"""Pulsa una tecla nombrada sobre el elemento enfocado de una pestana de Chrome via CDP.
|
|
|
|
Primitiva de input CDP: localiza un target (pestana) por substring de su URL, abre el
|
|
WebSocket de depuracion y emite el par de eventos `Input.dispatchKeyEvent`
|
|
(rawKeyDown + keyUp) de una tecla con nombre canonico (Enter, Escape, Backspace, Tab,
|
|
ArrowDown, ...) con modificadores opcionales. Base reutilizable para enviar mensajes
|
|
(Enter en el composer de WhatsApp), cerrar overlays (Escape), navegar resultados
|
|
(ArrowDown), borrar (Backspace) o combos (Ctrl+A con modifiers).
|
|
"""
|
|
|
|
import json
|
|
import urllib.request
|
|
|
|
import websocket
|
|
|
|
# Mapa de teclas soportadas -> {key, code, windowsVirtualKeyCode}.
|
|
# vk = windowsVirtualKeyCode == nativeVirtualKeyCode (codigos VK de Windows).
|
|
_KEY_MAP = {
|
|
"Enter": {"key": "Enter", "code": "Enter", "vk": 13},
|
|
"Escape": {"key": "Escape", "code": "Escape", "vk": 27},
|
|
"Backspace": {"key": "Backspace", "code": "Backspace", "vk": 8},
|
|
"Tab": {"key": "Tab", "code": "Tab", "vk": 9},
|
|
"Delete": {"key": "Delete", "code": "Delete", "vk": 46},
|
|
"ArrowDown": {"key": "ArrowDown", "code": "ArrowDown", "vk": 40},
|
|
"ArrowUp": {"key": "ArrowUp", "code": "ArrowUp", "vk": 38},
|
|
"ArrowLeft": {"key": "ArrowLeft", "code": "ArrowLeft", "vk": 37},
|
|
"ArrowRight": {"key": "ArrowRight", "code": "ArrowRight", "vk": 39},
|
|
"Home": {"key": "Home", "code": "Home", "vk": 36},
|
|
"End": {"key": "End", "code": "End", "vk": 35},
|
|
}
|
|
|
|
|
|
def cdp_press_key(
|
|
key: str,
|
|
*,
|
|
port: int = 9222,
|
|
target_url_substr: str = "",
|
|
modifiers: int = 0,
|
|
timeout_s: float = 10.0,
|
|
) -> dict:
|
|
"""Pulsa una tecla nombrada sobre el elemento enfocado de una pestana de Chrome.
|
|
|
|
Args:
|
|
key: Nombre canonico de la tecla. Soportadas: Enter, Escape, Backspace,
|
|
Tab, Delete, ArrowDown, ArrowUp, ArrowLeft, ArrowRight, Home, End.
|
|
port: Puerto de remote debugging de Chrome. Default 9222.
|
|
target_url_substr: Substring que debe contener la URL del target (pestana).
|
|
Si "", usa el primer target de tipo "page".
|
|
modifiers: Bitmask de modificadores CDP (Alt=1, Ctrl=2, Meta/Cmd=4,
|
|
Shift=8; combinables con OR). Default 0 (sin modificadores).
|
|
timeout_s: Timeout (segundos) para la conexion WebSocket. Default 10.0.
|
|
|
|
Returns:
|
|
dict con claves:
|
|
ok: bool — True si los eventos de tecla se emitieron sin error.
|
|
error: str — mensaje de error (vacio si ok).
|
|
|
|
Nunca lanza: errores de red/conexion/transport se devuelven en "error"
|
|
con ok=False.
|
|
"""
|
|
# 1. Validar la tecla contra el mapa interno (antes de tocar la red).
|
|
entry = _KEY_MAP.get(key)
|
|
if entry is None:
|
|
return {"ok": False, "error": f"unsupported key: {key}"}
|
|
|
|
k = entry["key"]
|
|
c = entry["code"]
|
|
vk = entry["vk"]
|
|
|
|
# 2. 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, "error": str(e)}
|
|
|
|
# 3. 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, "error": f"no target matching {target_url_substr}"}
|
|
|
|
ws_url = chosen.get("webSocketDebuggerUrl", "")
|
|
|
|
# 4. Abrir WS y emitir rawKeyDown + keyUp.
|
|
try:
|
|
ws = websocket.create_connection(ws_url, timeout=timeout_s)
|
|
except Exception as e: # noqa: BLE001 — conexion WS
|
|
return {"ok": False, "error": str(e)}
|
|
|
|
try:
|
|
for msg_id, ev_type in ((1, "rawKeyDown"), (2, "keyUp")):
|
|
ws.send(json.dumps({
|
|
"id": msg_id,
|
|
"method": "Input.dispatchKeyEvent",
|
|
"params": {
|
|
"type": ev_type,
|
|
"key": k,
|
|
"code": c,
|
|
"windowsVirtualKeyCode": vk,
|
|
"nativeVirtualKeyCode": vk,
|
|
"modifiers": modifiers,
|
|
},
|
|
}))
|
|
|
|
# Drenar respuestas hasta ver el id==2 (o agotar el stream). Ignoramos
|
|
# eventos del server y frames no-JSON.
|
|
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") == 2:
|
|
break
|
|
except Exception as e: # noqa: BLE001 — fallo de transport durante send/recv
|
|
return {"ok": False, "error": str(e)}
|
|
finally:
|
|
try:
|
|
ws.close()
|
|
except Exception: # noqa: BLE001 — cierre best-effort
|
|
pass
|
|
|
|
return {"ok": True, "error": ""}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
|
|
key_arg = sys.argv[1] if len(sys.argv) > 1 else "Enter"
|
|
substr = sys.argv[2] if len(sys.argv) > 2 else ""
|
|
out = cdp_press_key(key_arg, port=9222, target_url_substr=substr)
|
|
print(json.dumps(out, ensure_ascii=False, indent=2))
|