"""Asigna archivos a un 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 `` 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 `` 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 `` 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))