8e4bc8ad94
Nucleo propio: oye con whisper.cpp, piensa con un modelo de Ollama, habla con Piper, y hace RAG sobre los apuntes del usuario. 100% local, sin cuentas ni claves. Escrito bajo una restriccion dura, 4 GB de VRAM: el cerebro y whisper comparten tarjeta y solo caben porque estan dimensionados para ello. El RAG usa embeddings estaticos con busqueda hibrida; la voz clonada se sirve de una cache de frases. Incluye instalador (install.sh), requisitos, y documentacion del stack, del manejo de root y de las acciones. Los apuntes indexados y el diario NO se incluyen: son privados y el .gitignore los bloquea.
114 lines
5.2 KiB
Python
114 lines
5.2 KiB
Python
"""Lo que JARVIS puede HACER, y el unico sitio por donde pasa.
|
|
|
|
`Manos` es la puerta: el catalogo entra por aqui, el cerebro entra por aqui y
|
|
el panel entrara por aqui. Tener un solo punto importa porque es donde se
|
|
decide que se ejecuta y que no, y esa decision no puede estar repartida.
|
|
|
|
Las cuatro piezas de debajo vienen tal cual de la version sobre Newelle
|
|
—`acciones.py`, `tareas.py`, `sudo_root.py`, `seleccion.py`— y se comprobo al
|
|
copiarlas que 374 frases dan exactamente el mismo resultado en los dos sitios.
|
|
"""
|
|
import os
|
|
import shlex
|
|
import subprocess
|
|
|
|
from .acciones import Catalogo, ejecuta, formatea
|
|
|
|
AQUI = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
CATALOGO = os.path.join(AQUI, "datos", "acciones.json")
|
|
PROPIO = os.path.expanduser("~/.config/jarvis/acciones.json")
|
|
|
|
# Lo que no se ejecuta por mucho que lo pida el modelo. No es una lista de
|
|
# seguridad completa —eso es imposible— sino un tope contra el accidente: el
|
|
# modelo es servicial y a veces se inventa una limpieza que nadie pidio.
|
|
PROHIBIDO = (
|
|
"rm -rf /", "mkfs", "dd if=", ":(){", "shutdown", "reboot", "halt",
|
|
"> /dev/sd", "chmod -R 777 /", "chown -R", "userdel", "passwd",
|
|
)
|
|
|
|
TOPE_SALIDA = 4000
|
|
|
|
|
|
class Manos:
|
|
def __init__(self, catalogo=None, al_ejecutar=None):
|
|
self.catalogo = catalogo or Catalogo([PROPIO, CATALOGO])
|
|
self.al_ejecutar = al_ejecutar
|
|
# Root para la sesion: cuando el usuario desbloquea el candado con su
|
|
# contraseña, esa misma queda aqui y el cerebro puede hacer sudo hasta
|
|
# que caduque. Sin desbloquear, no hay root: el candado es la puerta.
|
|
from manos.sudo_root import GestorRoot
|
|
self.root = GestorRoot([])
|
|
|
|
# ------------------------------------------------------------- catalogo
|
|
|
|
def busca(self, texto: str):
|
|
return self.catalogo.busca(texto)
|
|
|
|
def haz(self, accion, argumento="") -> tuple:
|
|
"""Ejecuta una accion ya emparejada. Devuelve (lo que se dice, crudo)."""
|
|
dicho, salida = ejecuta(accion, [], argumento)
|
|
if salida:
|
|
dicho = formatea(accion.respuesta, salida, argumento)
|
|
return dicho, salida
|
|
|
|
def por_orden(self, orden: str, al_ejecutar=None) -> str:
|
|
"""Para el cerebro: empareja contra el catalogo y ejecuta.
|
|
|
|
Devuelve texto para el modelo, no para el usuario: si no encaja, se le
|
|
dice claramente para que pruebe con el shell en vez de insistir.
|
|
"""
|
|
if not orden.strip():
|
|
return "no me has dicho ninguna orden"
|
|
# El modelo a veces cuela un comando aqui de todas formas. Se detecta y
|
|
# se le devuelve el motivo, que es mas util que un "no encontrado".
|
|
if any(c in orden for c in "|><;&$") or orden.split()[0] in ("df", "ls", "ps", "cat"):
|
|
return ("eso es un comando de shell, no una orden del catalogo. "
|
|
"Usa la herramienta comando_de_shell.")
|
|
accion, argumento = self.catalogo.busca(orden)
|
|
if accion is None:
|
|
return (f"'{orden}' no esta en el catalogo. Si hace falta, usa "
|
|
"comando_de_shell.")
|
|
if al_ejecutar or self.al_ejecutar:
|
|
(al_ejecutar or self.al_ejecutar)(accion.id, argumento)
|
|
dicho, salida = self.haz(accion, argumento)
|
|
return (salida or dicho)[:TOPE_SALIDA]
|
|
|
|
# ---------------------------------------------------------------- shell
|
|
|
|
def por_shell(self, comando: str, al_ejecutar=None) -> str:
|
|
if not comando.strip():
|
|
return "no me has dicho ningun comando"
|
|
bajo = comando.lower()
|
|
for veneno in PROHIBIDO:
|
|
if veneno in bajo:
|
|
return (f"no ejecuto eso: contiene '{veneno}'. Si de verdad hace "
|
|
"falta, que lo pida el usuario expresamente.")
|
|
if al_ejecutar or self.al_ejecutar:
|
|
(al_ejecutar or self.al_ejecutar)("shell", comando)
|
|
|
|
from manos.sudo_root import necesita_root
|
|
# Los comandos con sudo van por GestorRoot, que da la contraseña de la
|
|
# sesion por stdin. Sin ella, `bash -c "sudo ..."` se colgaria pidiendo
|
|
# una contraseña que nadie teclea. Y si el candado no se ha abierto, no
|
|
# hay contraseña: el root queda cerrado hasta que el usuario desbloquee.
|
|
if necesita_root(comando):
|
|
if not self.root.sesion_activa():
|
|
return ("eso necesita root y el candado no esta abierto. "
|
|
"Desbloquee con su contraseña y lo repito.")
|
|
try:
|
|
r = self.root.ejecuta(comando, timeout=60)
|
|
except Exception as e:
|
|
return f"no se pudo ejecutar como root: {e}"
|
|
else:
|
|
try:
|
|
r = subprocess.run(["bash", "-c", comando], capture_output=True,
|
|
text=True, timeout=60)
|
|
except subprocess.TimeoutExpired:
|
|
return "el comando tardo mas de un minuto y lo he cortado"
|
|
except OSError as e:
|
|
return f"no se pudo ejecutar: {e}"
|
|
|
|
salida = (r.stdout or r.stderr or "").strip()
|
|
if not salida:
|
|
return "hecho, sin salida" if r.returncode == 0 else f"fallo (codigo {r.returncode})"
|
|
return salida[:TOPE_SALIDA]
|