feat(browser): auto-commit con 6 cambios

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-06-26 23:09:32 +02:00
parent f5387aa30e
commit 4c4eec4b1d
6 changed files with 862 additions and 0 deletions
@@ -0,0 +1,82 @@
---
name: cdp_set_file_input
kind: function
lang: py
domain: browser
version: "1.0.0"
purity: impure
signature: "def cdp_set_file_input(selector: str, file_paths, *, port: int = 9222, target_url_substr: str = '', timeout_s: float = 10.0) -> dict"
description: "Asigna uno o varios archivos a un <input type=file> de una pestana de un Chrome con remote debugging, via CDP, SIN abrir el dialogo nativo del sistema operativo. Localiza el target por substring de URL, abre el WebSocket y ejecuta DOM.enable -> DOM.getDocument -> DOM.querySelector(selector) -> DOM.setFileInputFiles con las rutas ABSOLUTAS. Es el unico metodo robusto para subir archivos por CDP: el navegador no permite escribir el value de un file input desde JS (seguridad) y simular drag&drop es fragil; setFileInputFiles inyecta los File y dispara el evento change que la SPA escucha. Base de whatsapp_send_image y de cualquier flujo de subida de archivos sobre el navegador diario sin robar el foco al usuario."
tags: [cdp, browser, automation, upload, file-input, python, navegator]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports: ["json", "os", "urllib.request", "websocket"]
params_schema:
params:
- name: selector
desc: "Selector CSS del <input type=file> destino. Debe resolver a UN elemento (se usa el primer match). El input puede estar oculto (display:none); CDP lo localiza igual."
- name: file_paths
desc: "Ruta (str) o lista de rutas a asignar. Se expanden (~) y se convierten a rutas ABSOLUTAS; cada una debe existir en disco o se aborta con ok=False antes de tocar la red."
- name: port
desc: "Puerto de remote debugging de Chrome. Default 9222."
- name: target_url_substr
desc: "Substring que debe contener la URL del target (pestana). Si vacio, usa el primer target de tipo 'page'."
- name: timeout_s
desc: "Timeout en segundos para la conexion WebSocket. Default 10.0."
output: "dict {ok: bool, error: str, node_id: int (nodeId CDP del input localizado, 0 si no se encontro), selector: str (eco), files: list[str] (rutas absolutas asignadas)}. ok=True solo si el input se localizo y setFileInputFiles no devolvio error. Nunca lanza: errores de archivo/red/conexion/transport se devuelven en 'error' con ok=False."
tested: true
tests: ["test_golden_asigna_archivos_al_input", "test_edge_archivo_inexistente_ok_false_sin_red", "test_edge_selector_no_encontrado_ok_false", "test_error_create_connection_lanza_ok_false"]
test_file_path: "python/functions/browser/cdp_set_file_input_test.py"
file_path: "python/functions/browser/cdp_set_file_input.py"
---
## Ejemplo
```python
import sys, os
sys.path.insert(0, os.path.join("python", "functions"))
from browser.cdp_set_file_input import cdp_set_file_input
# Requiere un Chrome lanzado con --remote-debugging-port=9222.
# Adjuntar una imagen al input de subida de una pestana (WhatsApp Web, un formulario, etc).
# El input "vivo" suele exponerse tras pulsar el boton "Adjuntar"/"Subir" (haz el click antes).
res = cdp_set_file_input(
'input[type="file"][multiple]',
"/home/enmanuel/ComfyUI/output/item_icon_potion_00001_.png",
target_url_substr="whatsapp",
)
print(res["ok"], res["node_id"], res["error"])
# -> True 120
```
O directo por CLI: `python3 python/functions/browser/cdp_set_file_input.py 'input[type="file"]' /ruta/abs.png whatsapp`.
## Cuando usarla
Cuando necesites **subir/adjuntar un archivo** a una pestana abierta sin que aparezca el
dialogo nativo de archivos del sistema operativo (que CDP no puede operar). Es la primitiva
de subida sobre la que se construye `whatsapp_send_image_py_browser` y cualquier
automatizacion de formularios con `<input type=file>`. El patron tipico: primero un click
real (`cdp_click_xy_py_browser`) en el boton que expone el input vivo, luego esta funcion
con el selector del input y la ruta absoluta del archivo.
## Gotchas
- **El input debe existir en el DOM al llamar.** Muchas SPAs (WhatsApp Web) solo crean/activan
el `<input type=file>` "vivo" DESPUES de pulsar el boton de adjuntar. Haz el click real
primero; si asignas sobre un input persistente/decoy, `setFileInputFiles` puede devolver ok
pero la SPA no reacciona (no aparece el preview).
- **Rutas ABSOLUTAS obligatorias.** `setFileInputFiles` exige rutas absolutas; la funcion ya
convierte (`os.path.abspath` + expanduser), pero el archivo debe existir o aborta con ok=False
antes de abrir la conexion.
- **El selector debe resolver al input correcto.** Si hay varios `<input type=file>` (uno por
tipo: fotos, documento...), afina el selector (`[multiple]`, `[accept*="image"]`). Si no
matchea ninguno, devuelve `ok=False` con "no element matches selector".
- **Dispara `change`, no `input`.** `DOM.setFileInputFiles` emite el evento `change` nativo
(que la mayoria de uploads escuchan). Si una SPA solo escucha otro evento, no bastara.
- Requiere un Chrome lanzado con `--remote-debugging-port=9222` (o el puerto que pases). Sin
remote debugging, `GET /json` falla y devuelve `ok=False`.
- Nunca lanza: errores de archivo/red/WS/transport se reportan en `error` con `ok=False`.
@@ -0,0 +1,163 @@
"""Asigna archivos a un <input type=file> de una pestana de Chrome via Chrome DevTools Protocol.
Primitiva de input CDP para subir archivos SIN abrir el dialogo nativo del sistema
operativo: localiza un target (pestana) por substring de su URL, abre el WebSocket de
depuracion y ejecuta la secuencia `DOM.enable` -> `DOM.getDocument` -> `DOM.querySelector`
(localiza el input por selector CSS) -> `DOM.setFileInputFiles` (asigna las rutas
absolutas al input).
Es el unico metodo robusto para adjuntar archivos por CDP: el navegador no permite
escribir el `value` de un `<input type=file>` desde JavaScript (seguridad), y simular
drag&drop es fragil. `DOM.setFileInputFiles` inyecta los `File` directamente y dispara el
evento `change` que la SPA escucha. Base de `whatsapp_send_image` y de cualquier flujo de
subida de archivos sobre el navegador diario sin robar el foco al usuario.
"""
import json
import os
import urllib.request
import websocket
def _call(ws, msg_id: int, method: str, params: dict) -> dict:
"""Envia un comando CDP y drena eventos hasta la respuesta con el mismo id."""
ws.send(json.dumps({"id": msg_id, "method": method, "params": params}))
while True:
raw = ws.recv()
if not raw:
return {}
try:
parsed = json.loads(raw)
except Exception: # noqa: BLE001 — frame no-JSON, ignorar
continue
if parsed.get("id") == msg_id:
return parsed
def cdp_set_file_input(
selector: str,
file_paths,
*,
port: int = 9222,
target_url_substr: str = "",
timeout_s: float = 10.0,
) -> dict:
"""Asigna uno o varios archivos a un `<input type=file>` de una pestana de Chrome.
Localiza el target `page` por substring de URL, abre el WebSocket CDP y ejecuta
`DOM.enable` -> `DOM.getDocument` -> `DOM.querySelector(selector)` ->
`DOM.setFileInputFiles`. Equivale a que el usuario elija esos archivos en el dialogo
nativo, pero sin abrirlo: ideal para automatizar subidas (adjuntar imagen en WhatsApp,
subir un fichero a un formulario) sobre una pestana ya abierta.
Args:
selector: Selector CSS del `<input type=file>` destino. Debe resolver a UN
elemento (se usa el primer match). El input puede estar oculto.
file_paths: Ruta (str) o lista de rutas a asignar. Se expanden (`~`) y se
convierten a rutas ABSOLUTAS; cada una debe existir en disco.
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".
timeout_s: Timeout (segundos) para la conexion WebSocket. Default 10.0.
Returns:
dict con claves:
ok: bool — True si el input se localizo y los archivos se asignaron sin error.
error: str — mensaje de error (vacio si ok).
node_id: int — nodeId CDP del input localizado (0 si no se encontro).
selector: str — eco del selector usado.
files: list[str] — rutas absolutas que se intentaron asignar.
Nunca lanza: errores de archivo, red, conexion o transport se devuelven en
"error" con ok=False.
"""
# 1. Normalizar y validar las rutas (antes de tocar la red).
if isinstance(file_paths, str):
raw_paths = [file_paths]
else:
raw_paths = list(file_paths)
abs_paths = [os.path.abspath(os.path.expanduser(p)) for p in raw_paths]
if not abs_paths:
return {"ok": False, "error": "no file paths provided",
"node_id": 0, "selector": selector, "files": []}
missing = [p for p in abs_paths if not os.path.isfile(p)]
if missing:
return {"ok": False, "error": f"file(s) not found: {missing}",
"node_id": 0, "selector": selector, "files": abs_paths}
# 2. Listar targets via HTTP y elegir el primer page que matchee.
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),
"node_id": 0, "selector": selector, "files": abs_paths}
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}",
"node_id": 0, "selector": selector, "files": abs_paths}
ws_url = chosen.get("webSocketDebuggerUrl", "")
# 3. Abrir WS y correr la secuencia DOM.
try:
ws = websocket.create_connection(ws_url, timeout=timeout_s)
except Exception as e: # noqa: BLE001 — conexion WS
return {"ok": False, "error": str(e),
"node_id": 0, "selector": selector, "files": abs_paths}
try:
_call(ws, 1, "DOM.enable", {})
doc = _call(ws, 2, "DOM.getDocument", {"depth": 0})
root_id = doc.get("result", {}).get("root", {}).get("nodeId", 0)
if not root_id:
return {"ok": False, "error": "DOM.getDocument did not return a root nodeId",
"node_id": 0, "selector": selector, "files": abs_paths}
qs = _call(ws, 3, "DOM.querySelector",
{"nodeId": root_id, "selector": selector})
node_id = qs.get("result", {}).get("nodeId", 0)
if not node_id:
return {"ok": False, "error": f"no element matches selector: {selector}",
"node_id": 0, "selector": selector, "files": abs_paths}
sf = _call(ws, 4, "DOM.setFileInputFiles",
{"files": abs_paths, "nodeId": node_id})
err = sf.get("error")
if err:
return {"ok": False, "error": json.dumps(err),
"node_id": node_id, "selector": selector, "files": abs_paths}
except Exception as e: # noqa: BLE001 — fallo de transport durante send/recv
return {"ok": False, "error": str(e),
"node_id": 0, "selector": selector, "files": abs_paths}
finally:
try:
ws.close()
except Exception: # noqa: BLE001 — cierre best-effort
pass
return {"ok": True, "error": "", "node_id": node_id,
"selector": selector, "files": abs_paths}
if __name__ == "__main__":
import sys
sel = sys.argv[1] if len(sys.argv) > 1 else 'input[type="file"]'
path = sys.argv[2] if len(sys.argv) > 2 else ""
substr = sys.argv[3] if len(sys.argv) > 3 else ""
out = cdp_set_file_input(sel, path, port=9222, target_url_substr=substr)
print(json.dumps(out, ensure_ascii=False, indent=2))
@@ -0,0 +1,144 @@
"""Tests para cdp_set_file_input — mockean urlopen + create_connection.
Mockean la capa de red de CDP: urllib.request.urlopen (lista de targets) y
websocket.create_connection (un fake que responde a cada comando DOM por su id, con un
nodeId de raiz, un nodeId de input y una respuesta vacia para setFileInputFiles). Asi NO
hace falta Chrome. Para la validacion de existencia de archivo se usan rutas reales
(__file__) y rutas inexistentes.
"""
import json
import os
import sys
from contextlib import contextmanager
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from browser import cdp_set_file_input as mod # noqa: E402
from browser.cdp_set_file_input import cdp_set_file_input # noqa: E402
_THIS = os.path.abspath(__file__)
class _FakeResp:
def __init__(self, payload):
self._payload = payload
def __enter__(self):
return self
def __exit__(self, *exc):
return False
def read(self):
return json.dumps(self._payload).encode()
class _FakeWS:
"""WebSocket falso que responde a cada comando DOM por method."""
def __init__(self, *, query_node_id=4242, set_error=None):
self.sent = []
self._inbox = []
self.closed = False
self.query_node_id = query_node_id
self.set_error = set_error
def send(self, raw):
msg = json.loads(raw)
self.sent.append(msg)
mid = msg["id"]
method = msg["method"]
if method == "DOM.getDocument":
resp = {"id": mid, "result": {"root": {"nodeId": 1}}}
elif method == "DOM.querySelector":
resp = {"id": mid, "result": {"nodeId": self.query_node_id}}
elif method == "DOM.setFileInputFiles":
if self.set_error is not None:
resp = {"id": mid, "error": self.set_error}
else:
resp = {"id": mid, "result": {}}
else: # DOM.enable y demas
resp = {"id": mid, "result": {}}
self._inbox.append(json.dumps(resp))
def recv(self):
if self._inbox:
return self._inbox.pop(0)
return ""
def close(self):
self.closed = True
@contextmanager
def _patch(targets, ws_obj=None, create_conn_exc=None):
orig_urlopen = mod.urllib.request.urlopen
orig_create = mod.websocket.create_connection
def fake_urlopen(url, timeout=5):
return _FakeResp(targets)
def fake_create(ws_url, timeout=10):
if create_conn_exc is not None:
raise create_conn_exc
return ws_obj
mod.urllib.request.urlopen = fake_urlopen
mod.websocket.create_connection = fake_create
try:
yield
finally:
mod.urllib.request.urlopen = orig_urlopen
mod.websocket.create_connection = orig_create
_TARGETS = [
{"type": "page", "url": "https://web.whatsapp.com/", "webSocketDebuggerUrl": "ws://x/1"},
]
def test_golden_asigna_archivos_al_input():
"""setFileInputFiles recibe la ruta ABSOLUTA y el nodeId del input localizado."""
ws = _FakeWS(query_node_id=777)
with _patch(_TARGETS, ws_obj=ws):
res = cdp_set_file_input('input[type="file"]', _THIS, target_url_substr="whatsapp")
assert res["ok"] is True
assert res["error"] == ""
assert res["node_id"] == 777
assert res["files"] == [_THIS]
# El ultimo comando es setFileInputFiles con la ruta absoluta y el nodeId del input.
setcmd = [m for m in ws.sent if m["method"] == "DOM.setFileInputFiles"][0]
assert setcmd["params"]["files"] == [_THIS]
assert setcmd["params"]["nodeId"] == 777
assert ws.closed is True
def test_edge_archivo_inexistente_ok_false_sin_red():
"""Una ruta inexistente devuelve ok=False antes de abrir cualquier conexion."""
res = cdp_set_file_input('input[type="file"]', "/no/existe/imagen.png",
target_url_substr="whatsapp")
assert res["ok"] is False
assert "not found" in res["error"]
assert res["node_id"] == 0
def test_edge_selector_no_encontrado_ok_false():
"""Si querySelector devuelve nodeId 0, no se asignan archivos y ok=False."""
ws = _FakeWS(query_node_id=0)
with _patch(_TARGETS, ws_obj=ws):
res = cdp_set_file_input('input[type="file"]', _THIS, target_url_substr="whatsapp")
assert res["ok"] is False
assert "no element matches selector" in res["error"]
# NO se llamo a setFileInputFiles.
assert all(m["method"] != "DOM.setFileInputFiles" for m in ws.sent)
def test_error_create_connection_lanza_ok_false():
"""Si create_connection lanza, se captura y devuelve ok=False sin relanzar."""
with _patch(_TARGETS, create_conn_exc=ConnectionRefusedError("ws down")):
res = cdp_set_file_input('input[type="file"]', _THIS, target_url_substr="whatsapp")
assert res["ok"] is False
assert "ws down" in res["error"]
assert res["files"] == [_THIS]
@@ -0,0 +1,94 @@
---
name: whatsapp_send_image
kind: function
lang: py
domain: browser
version: "1.0.0"
purity: impure
signature: "def whatsapp_send_image(name: str, image_path: str, *, caption: str = '', port: int = 9222, target_url_substr: str = 'whatsapp', open_first: bool = True) -> dict"
description: "Envia una imagen (con caption opcional) a un chat de WhatsApp Web en una pestana ya logueada del navegador diario via CDP, sin abrir ventana nueva ni darle foco. Abre el chat por nombre exacto (whatsapp_open_chat) y verifica el destinatario (salvaguarda anti-envio-equivocado), hace click real en 'Adjuntar' para exponer el <input type=file> vivo, asigna la imagen con cdp_set_file_input (DOM.setFileInputFiles), espera el preview de la bandeja inline, teclea el caption opcional con teclado CDP real, y hace click en el boton enviar (icono wds-ic-send-filled) verificando que la bandeja se cerro. Accion con efecto: envia la imagen DE VERDAD, no reversible."
tags: [whatsapp, cdp, browser, automation, image, upload, python, navegator]
uses_functions: [whatsapp_open_chat_py_browser, cdp_eval_py_browser, cdp_click_xy_py_browser, cdp_type_chars_py_browser, cdp_set_file_input_py_browser]
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports: ["os", "sys", "time", "json"]
params_schema:
params:
- name: name
desc: "Nombre EXACTO del chat o grupo destinatario tal y como aparece en la lista lateral. Se usa para abrir el chat y como salvaguarda de que el composer apunta al destinatario correcto antes de adjuntar."
- name: image_path
desc: "Ruta de la imagen a enviar. Se expande (~) y se convierte a ruta ABSOLUTA; debe existir en disco o aborta con error sin abrir el chat."
- name: caption
desc: "Texto opcional que acompana la imagen. Se teclea en el composer con teclado CDP real caracter a caracter; '' (default) envia la imagen sin caption."
- name: port
desc: "Puerto de remote debugging de Chrome. Default 9222."
- name: target_url_substr
desc: "Substring que debe contener la URL del target (pestana). Default 'whatsapp'."
- name: open_first
desc: "Si True (default), abre el chat por su nombre antes de adjuntar. Si False, asume el chat ya abierto pero verifica el aria-label del composer contra name (aborta si no coincide)."
output: "dict {ok: bool (alias de sent), sent: bool, recipient: str, image: str (ruta absoluta), caption: str, error: str (motivo del fallo, vacio si sent=True)}. sent=True solo si la imagen se adjunto, el caption (si lo hay) se verifico y se pulso enviar dejando la bandeja vacia. Nunca lanza: los fallos se reportan en 'sent'/'ok' + 'error'."
tested: true
tests: ["test_golden_adjunta_caption_y_envia", "test_envia_sin_caption_no_teclea", "test_edge_imagen_no_existe_error_sin_abrir", "test_edge_open_fallido_error_sin_adjuntar", "test_seguridad_open_first_false_label_no_coincide_aborta", "test_error_set_file_input_falla_no_envia"]
test_file_path: "python/functions/browser/whatsapp_send_image_test.py"
file_path: "python/functions/browser/whatsapp_send_image.py"
---
## Ejemplo
```python
import sys, os
sys.path.insert(0, os.path.join("python", "functions"))
from browser.whatsapp_send_image import whatsapp_send_image
# Requiere WhatsApp Web abierto y logueado (y DESBLOQUEADO si tiene app-lock) en un
# Chrome lanzado con --remote-debugging-port=9222.
res = whatsapp_send_image(
"NOTAS WASAP",
"/home/enmanuel/ComfyUI/output/item_icon_potion_00001_.png",
caption="item icon: potion",
)
print(res)
# -> {"ok": True, "sent": True, "recipient": "NOTAS WASAP",
# "image": ".../item_icon_potion_00001_.png", "caption": "item icon: potion", "error": ""}
```
O directo por CLI: `python3 python/functions/browser/whatsapp_send_image.py "NOTAS WASAP" /ruta/abs.png "mi caption"`.
## Cuando usarla
Cuando necesites **enviar una imagen (foto, captura, asset generado) a un contacto o grupo
por su nombre exacto** en WhatsApp Web, sin abrir ventana nueva ni robar el foco al usuario.
Es la version "imagen" de `whatsapp_send_message_py_browser`: usala cuando ya tienes el
nombre exacto del destinatario y la ruta de un archivo de imagen en disco. Para texto plano,
usa `whatsapp_send_message`; para leer/confirmar lo enviado, `whatsapp_read_chat`.
## Gotchas
- **Accion con efecto: envia la imagen DE VERDAD.** No es reversible. Verifica que `name` es
EXACTO antes de llamar (la salvaguarda abre el chat y comprueba el composer, pero el nombre
debe coincidir con el `title` de la lista lateral).
- **App-lock de WhatsApp Web.** Si la cuenta tiene el "bloqueo de la aplicacion" activo, el DOM
de chats no se renderiza (solo la pantalla de password) y la funcion fallara al abrir el chat.
Hay que desbloquearlo primero (teclear el password en `input[type=password]` + boton
"Desbloquear"). Sintoma: `whatsapp_open_chat` devuelve `opened: False` y la lista lateral sale
vacia aunque la sesion siga logueada.
- **El input vivo solo existe tras pulsar "Adjuntar".** Por eso la funcion hace el click real en
el boton "Adjuntar" antes de `setFileInputFiles`; asignar al input persistente decoy no abre el
preview. La WhatsApp Web actual usa una **bandeja de medios INLINE** sobre el composer (no un
drawer a pantalla completa): el caption se escribe en el MISMO composer del footer.
- **Selector de aria-label en espanol.** El preview se detecta por `[aria-label="Quitar archivo
adjunto"]` y el boton de adjuntar por `[aria-label="Adjuntar"]`: dependen del idioma de la UI
(espanol). En otro locale habria que ajustar los aria-labels.
- **Las imagenes se ACUMULAN en la bandeja.** Cada `setFileInputFiles` anade una miniatura; si un
envio queda a medias, la siguiente llamada podria sumar a las pendientes. La funcion verifica que
la bandeja queda vacia tras enviar (adjuntos=0) para confirmar; si no se cierra, devuelve
`sent=False` con "envio incierto".
- **Salvaguarda anti-destinatario-equivocado**: con `open_first=True` abre y verifica el chat; con
`open_first=False` lee el aria-label del composer y aborta si no contiene `name`.
- **Caption verificado**: tras teclear, re-lee el `innerText` del composer y solo envia si coincide
EXACTAMENTE con `caption`; si no, devuelve `sent=False` sin enviar.
- **Funciona con la ventana minimizada o sin foco**: CDP opera la pestana sin traerla a primer plano.
- **Viola los ToS de WhatsApp**: automatizar la web tiene riesgo de ban del numero personal. Usar
con cautela y bajo tu responsabilidad.
@@ -0,0 +1,208 @@
"""Envia una imagen (con caption opcional) a un chat de WhatsApp Web via Chrome DevTools Protocol.
Compone `whatsapp_open_chat` (abrir + verificar destinatario) con primitivas CDP del
registry (`cdp_eval`, `cdp_click_xy`, `cdp_set_file_input`, `cdp_type_chars`) para adjuntar
y enviar una imagen a un contacto/grupo SIN abrir ventana nueva ni darle foco al sistema.
Flujo (modelo de bandeja de medios INLINE de la WhatsApp Web actual), con salvaguarda
anti-envio-al-contacto-equivocado:
1. Abre el chat por su nombre exacto (`open_first=True`). Si no abre, aborta. Con
`open_first=False`, asume el chat abierto pero VERIFICA que el aria-label del composer
contiene el nombre; si no, aborta por seguridad.
2. Hace click de raton real en el boton "Adjuntar" del footer: esto expone el
`<input type=file>` "vivo" que escucha la SPA (antes de pulsarlo el input persistente
no dispara el preview).
3. Asigna la imagen al input via `cdp_set_file_input` (`DOM.setFileInputFiles`): la
imagen aparece como miniatura en la bandeja inline sobre el composer.
4. Espera a que el preview aparezca (boton "Quitar archivo adjunto" presente).
5. Si `caption` no esta vacio, enfoca el composer del footer y teclea el texto con
teclado CDP real (`cdp_type_chars`); verifica el `innerText` antes de enviar.
6. Click de raton real en el boton enviar (icono `wds-ic-send-filled`) y verifica que la
bandeja se cerro (sin adjuntos) para confirmar el envio.
Validado contra WhatsApp Web real. Accion CON EFECTO REAL E IRREVERSIBLE: envia la imagen
de verdad.
"""
import json
import os
import sys
import time
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from browser.cdp_eval import cdp_eval
from browser.cdp_click_xy import cdp_click_xy
from browser.cdp_type_chars import cdp_type_chars
from browser.cdp_set_file_input import cdp_set_file_input
from browser.whatsapp_open_chat import whatsapp_open_chat
def _center(expr: str, port: int, substr: str):
"""Evalua una expresion que devuelve JSON {x,y} (o null) y la parsea a dict/None."""
r = cdp_eval(expr, port=port, target_url_substr=substr)
val = r.get("value")
if not val:
return None
try:
return json.loads(val)
except Exception: # noqa: BLE001 — value no-JSON
return None
def _attachment_count(port: int, substr: str) -> int:
"""Numero de adjuntos en la bandeja inline (botones 'Quitar archivo adjunto')."""
r = cdp_eval(
'/*PREVIEW*/document.querySelectorAll(\'[aria-label="Quitar archivo adjunto"]\').length',
port=port, target_url_substr=substr,
)
v = r.get("value")
return v if isinstance(v, int) else 0
def whatsapp_send_image(
name: str,
image_path: str,
*,
caption: str = "",
port: int = 9222,
target_url_substr: str = "whatsapp",
open_first: bool = True,
) -> dict:
"""Envia una imagen (con caption opcional) a un chat de WhatsApp Web ya logueado.
Accion CON EFECTO: envia la imagen DE VERDAD (no reversible). Verifica `name`.
Args:
name: Nombre EXACTO del chat/grupo destinatario, tal y como aparece en la lista
lateral. Se usa para abrir el chat y como salvaguarda de que el composer apunta
al destinatario correcto antes de adjuntar.
image_path: Ruta de la imagen a enviar. Se expande (`~`) y se convierte a ruta
ABSOLUTA; debe existir en disco.
caption: Texto opcional que acompana la imagen. Se teclea en el composer con teclado
CDP real; "" (default) envia la imagen sin caption.
port: Puerto de remote debugging de Chrome. Default 9222.
target_url_substr: Substring que debe contener la URL del target (pestana). Default
"whatsapp".
open_first: Si True (default), abre el chat por su nombre antes de adjuntar. Si
False, asume el chat ya abierto pero verifica el aria-label del composer contra
`name` (aborta si no coincide).
Returns:
dict con claves:
ok: bool — alias de sent (True si la imagen se envio).
sent: bool — True si la imagen se adjunto, (caption escrito) y se envio.
recipient: str — el nombre solicitado.
image: str — ruta absoluta de la imagen.
caption: str — caption solicitado.
error: str — motivo del fallo (vacio si sent=True).
Nunca lanza: los fallos se reportan en "sent"/"ok" + "error".
"""
S = target_url_substr
abs_img = os.path.abspath(os.path.expanduser(image_path))
def fail(error: str) -> dict:
return {"ok": False, "sent": False, "recipient": name,
"image": abs_img, "caption": caption, "error": error}
# 0. La imagen debe existir.
if not os.path.isfile(abs_img):
return fail(f"imagen no encontrada: {abs_img}")
# 1. Abrir + verificar destinatario correcto (salvaguarda anti-equivocacion).
if open_first:
o = whatsapp_open_chat(name, port=port, target_url_substr=S)
if not o.get("opened"):
return fail(o.get("reason", "no se pudo abrir el chat"))
else:
chk = cdp_eval(
'/*LABEL*/var b=document.querySelector(\'footer div[contenteditable="true"]\'); '
"b?b.getAttribute('aria-label'):null",
port=port, target_url_substr=S,
)
if name not in (chk.get("value") or ""):
return fail("el chat abierto no coincide con el destinatario; abortado por seguridad")
# 2. Click real en "Adjuntar" para exponer el <input type=file> vivo.
adj = _center(
'/*ADJUNTAR*/(() => {const e=document.querySelector(\'button[aria-label="Adjuntar"]\');'
"if(!e)return null;const b=e.getBoundingClientRect();"
"return JSON.stringify({x:Math.round(b.x+b.width/2),y:Math.round(b.y+b.height/2)});})()",
port, S,
)
if not adj:
return fail("boton 'Adjuntar' no encontrado en el footer")
cdp_click_xy(adj["x"], adj["y"], port=port, target_url_substr=S)
time.sleep(0.8)
# 3. Asignar la imagen al input multiple (el que se activa tras Adjuntar);
# fallback al primer input file si el selector con [multiple] no resuelve.
r = cdp_set_file_input('input[type="file"][multiple]', abs_img,
port=port, target_url_substr=S)
if not r.get("ok"):
r = cdp_set_file_input('input[type="file"]', abs_img,
port=port, target_url_substr=S)
if not r.get("ok"):
return fail("no se pudo adjuntar la imagen: " + r.get("error", ""))
# 4. Esperar a que el preview/bandeja aparezca (adjunto presente).
attached = False
for _ in range(15):
time.sleep(0.2)
if _attachment_count(port, S) > 0:
attached = True
break
if not attached:
return fail("el preview no aparecio tras adjuntar la imagen")
# 5. Caption opcional: enfocar el composer y teclear con teclado real.
if caption:
cdp_eval(
'/*FOCUS*/var b=document.querySelector(\'div[contenteditable="true"]\'); '
"if(b){b.focus();}",
port=port, target_url_substr=S,
)
time.sleep(0.25)
cdp_type_chars(caption, port=port, target_url_substr=S, delay_ms=15)
time.sleep(0.3)
chk = cdp_eval(
'/*COMPOSER*/var b=document.querySelector(\'div[contenteditable="true"]\'); '
"b?b.innerText.replace(/\\n/g,''):''",
port=port, target_url_substr=S,
)
composer = chk.get("value") or ""
if composer != caption:
return fail("el composer no contiene el caption esperado (no enviado): "
+ repr(composer))
# 6. Click real en el boton enviar (icono wds-ic-send-filled).
snd = _center(
'/*SEND*/(() => {const e=document.querySelector(\'span[data-icon="wds-ic-send-filled"]\');'
"if(!e)return null;const b=e.getBoundingClientRect();"
"if(b.width===0)return null;"
"return JSON.stringify({x:Math.round(b.x+b.width/2),y:Math.round(b.y+b.height/2)});})()",
port, S,
)
if not snd:
return fail("boton de enviar (wds-ic-send-filled) no encontrado")
cdp_click_xy(snd["x"], snd["y"], port=port, target_url_substr=S)
# 7. Verificar que la bandeja se cerro (sin adjuntos) -> envio confirmado.
for _ in range(15):
time.sleep(0.2)
if _attachment_count(port, S) == 0:
return {"ok": True, "sent": True, "recipient": name,
"image": abs_img, "caption": caption, "error": ""}
return fail("la bandeja no se cerro tras pulsar enviar; envio incierto")
if __name__ == "__main__":
chat = sys.argv[1] if len(sys.argv) > 1 else "NOTAS WASAP"
img = sys.argv[2] if len(sys.argv) > 2 else ""
cap = sys.argv[3] if len(sys.argv) > 3 else ""
out = whatsapp_send_image(chat, img, caption=cap,
port=9222, target_url_substr="whatsapp")
print(json.dumps(out, ensure_ascii=False, indent=2))
@@ -0,0 +1,171 @@
"""Tests para whatsapp_send_image.
whatsapp_send_image compone whatsapp_open_chat con cuatro primitivas CDP (cdp_eval,
cdp_click_xy, cdp_type_chars, cdp_set_file_input) y requiere un Chrome vivo. Aqui se
mockean las cinco con monkeypatch sobre el modulo `browser.whatsapp_send_image` (donde
quedan ligados los nombres por el `from browser.X import Y`), de modo que NO hace falta
Chrome.
El fake de cdp_eval distingue cada expresion por un marcador de comentario JS embebido en
ella (`/*ADJUNTAR*/`, `/*PREVIEW*/`, `/*COMPOSER*/`, `/*SEND*/`, `/*LABEL*/`). El estado
`after_send` (que /*SEND*/ activa) hace que /*PREVIEW*/ devuelva 0 adjuntos tras el envio,
simulando el cierre de la bandeja.
"""
import json
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
import browser.whatsapp_send_image as wsi # noqa: E402
from browser.whatsapp_send_image import whatsapp_send_image # noqa: E402
# Imagen real para pasar la validacion de existencia (este propio archivo de test).
_IMG = os.path.abspath(__file__)
class _Spy:
def __init__(self, ret=None):
self.calls = []
self.ret = ret if ret is not None else {"ok": True}
def __call__(self, *args, **kwargs):
self.calls.append((args, kwargs))
return self.ret
def _make_eval(caption_value="", label="Escribir un mensaje para el grupo NOTAS WASAP",
state=None):
state = state if state is not None else {}
def fake(expr, *, port=9222, target_url_substr=""):
if "/*ADJUNTAR*/" in expr:
return {"ok": True, "value": json.dumps({"x": 575, "y": 589}), "error": ""}
if "/*SEND*/" in expr:
state["after_send"] = True
return {"ok": True, "value": json.dumps({"x": 1136, "y": 578}), "error": ""}
if "/*PREVIEW*/" in expr:
return {"ok": True, "value": 0 if state.get("after_send") else 1, "error": ""}
if "/*COMPOSER*/" in expr:
return {"ok": True, "value": caption_value, "error": ""}
if "/*LABEL*/" in expr:
return {"ok": True, "value": label, "error": ""}
return {"ok": True, "value": None, "error": ""}
return fake
def _patch_common(monkeypatch, *, eval_fn, set_ret={"ok": True}, open_ret=None):
open_spy = _Spy(ret=open_ret if open_ret is not None else {"opened": True, "name": "x"})
click_spy = _Spy(ret={"ok": True})
type_spy = _Spy(ret={"ok": True})
set_spy = _Spy(ret=set_ret)
monkeypatch.setattr(wsi, "whatsapp_open_chat", open_spy)
monkeypatch.setattr(wsi, "cdp_eval", eval_fn)
monkeypatch.setattr(wsi, "cdp_click_xy", click_spy)
monkeypatch.setattr(wsi, "cdp_type_chars", type_spy)
monkeypatch.setattr(wsi, "cdp_set_file_input", set_spy)
monkeypatch.setattr(wsi.time, "sleep", lambda *a, **k: None)
return open_spy, click_spy, type_spy, set_spy
def test_golden_adjunta_caption_y_envia(monkeypatch):
cap = "item icon: potion"
state = {}
open_spy, click_spy, type_spy, set_spy = _patch_common(
monkeypatch, eval_fn=_make_eval(caption_value=cap, state=state))
res = whatsapp_send_image("NOTAS WASAP", _IMG, caption=cap,
port=9222, target_url_substr="whatsapp")
assert res["sent"] is True and res["ok"] is True
assert res["recipient"] == "NOTAS WASAP"
assert res["image"] == _IMG
assert res["caption"] == cap
assert res["error"] == ""
# Se adjunto la imagen (ruta absoluta) por setFileInputFiles.
assert set_spy.calls[0][0][0] == 'input[type="file"][multiple]'
assert set_spy.calls[0][0][1] == _IMG
# Se tecleo el caption una vez.
assert len(type_spy.calls) == 1
assert type_spy.calls[0][0][0] == cap
# Dos clicks reales: Adjuntar y Enviar.
assert len(click_spy.calls) == 2
def test_envia_sin_caption_no_teclea(monkeypatch):
state = {}
_, click_spy, type_spy, set_spy = _patch_common(
monkeypatch, eval_fn=_make_eval(caption_value="", state=state))
res = whatsapp_send_image("NOTAS WASAP", _IMG, caption="",
port=9222, target_url_substr="whatsapp")
assert res["sent"] is True
# Sin caption: NO se teclea nada, pero si se adjunta y se envia.
assert len(type_spy.calls) == 0
assert len(set_spy.calls) >= 1
assert len(click_spy.calls) == 2
def test_edge_imagen_no_existe_error_sin_abrir(monkeypatch):
open_spy, click_spy, type_spy, set_spy = _patch_common(
monkeypatch, eval_fn=_make_eval())
res = whatsapp_send_image("NOTAS WASAP", "/no/existe/foo.png",
port=9222, target_url_substr="whatsapp")
assert res["sent"] is False and res["ok"] is False
assert "no encontrada" in res["error"]
# Ni siquiera se intento abrir el chat ni adjuntar.
assert len(open_spy.calls) == 0
assert len(set_spy.calls) == 0
def test_edge_open_fallido_error_sin_adjuntar(monkeypatch):
open_spy, click_spy, type_spy, set_spy = _patch_common(
monkeypatch, eval_fn=_make_eval(),
open_ret={"opened": False, "name": "x",
"reason": "chat no encontrado en la lista (no cargado o nombre inexacto)"})
res = whatsapp_send_image("Inexistente", _IMG,
port=9222, target_url_substr="whatsapp")
assert res["sent"] is False
assert "no encontrado" in res["error"]
# No se adjunto ni se hizo click cuando el chat no abrio.
assert len(set_spy.calls) == 0
assert len(click_spy.calls) == 0
def test_seguridad_open_first_false_label_no_coincide_aborta(monkeypatch):
open_spy, click_spy, type_spy, set_spy = _patch_common(
monkeypatch,
eval_fn=_make_eval(label="Escribir un mensaje para el grupo OTRO CHAT"))
res = whatsapp_send_image("NOTAS WASAP", _IMG,
port=9222, target_url_substr="whatsapp",
open_first=False)
assert res["sent"] is False
assert "abortado por seguridad" in res["error"]
# SEGURIDAD: no se adjunto ni se clicko nada.
assert len(set_spy.calls) == 0
assert len(click_spy.calls) == 0
def test_error_set_file_input_falla_no_envia(monkeypatch):
state = {}
open_spy, click_spy, type_spy, set_spy = _patch_common(
monkeypatch, eval_fn=_make_eval(state=state),
set_ret={"ok": False, "error": "no element matches selector"})
res = whatsapp_send_image("NOTAS WASAP", _IMG,
port=9222, target_url_substr="whatsapp")
assert res["sent"] is False
assert "no se pudo adjuntar" in res["error"]
# Se intento adjuntar (dos selectores) pero no se llego a enviar (solo el click de Adjuntar).
assert len(set_spy.calls) == 2
assert len(click_spy.calls) == 1