JARVIS/config/aprender.py

#!/usr/bin/env python3
"""Enseña acciones nuevas a partir de lo que no entendió.

    python3 ~/COFRE/CODERS/JARVIS/config/aprender.py

Cada vez que dices algo que no encaja con ninguna accion, el panel lo anota en
config/jarvis-sin-accion.jsonl. Esto lo lee, agrupa lo repetido y pregunta a
que accion pertenece cada frase; lo que decidas se escribe como alias en
config/jarvis-acciones.json, que el catalogo carga ANTES que el del paquete.

Por que asi y no entrenando un modelo: los fallos del reconocedor son
sistematicos, no aleatorios. "abre el correo" se oye "a ver el correo" siempre,
no una de cada diez veces. Un alias lo arregla del todo, se lee, se borra si
molesta y no cuesta un milisegundo. Reentrenar costaria horas para acertar
"casi siempre".
"""
import json
import os
import sys
from collections import Counter

CONFIG = os.path.expanduser("~/.var/app/io.github.qwersyk.Newelle/config")
REGISTRO = os.path.join(CONFIG, "jarvis-sin-accion.jsonl")
PROPIO = os.path.join(CONFIG, "jarvis-acciones.json")
PAQUETE = os.path.expanduser(
    "~/.local/share/flatpak/app/io.github.qwersyk.Newelle/x86_64/master/"
    "active/files/share/newelle/acciones/acciones.json")


def carga_catalogo():
    acciones = {}
    for ruta in (PAQUETE, PROPIO):
        try:
            with open(ruta) as f:
                for a in json.load(f).get("acciones", []):
                    acciones.setdefault(a["id"], a)
        except (OSError, json.JSONDecodeError):
            pass
    return acciones


def carga_fallos():
    """Las frases sin accion, agrupadas por veces que se han dicho."""
    cuenta = Counter()
    try:
        with open(REGISTRO) as f:
            for linea in f:
                try:
                    texto = json.loads(linea).get("texto", "").strip()
                except json.JSONDecodeError:
                    continue
                if texto:
                    cuenta[texto] += 1
    except OSError:
        pass
    return cuenta


def guarda_alias(nuevos):
    """Añade los alias al catalogo propio, respetando lo que ya hubiera."""
    try:
        with open(PROPIO) as f:
            propio = json.load(f)
    except (OSError, json.JSONDecodeError):
        propio = {"version": 1, "acciones": []}

    porid = {a["id"]: a for a in propio.get("acciones", [])}
    catalogo = carga_catalogo()

    for ident, frases in nuevos.items():
        if ident in porid:
            entrada = porid[ident]
        else:
            # se copia la del paquete entera: el catalogo propio se carga antes
            # y sustituye a la original, asi que tiene que traer su comando
            entrada = dict(catalogo[ident])
            porid[ident] = entrada
            propio.setdefault("acciones", []).append(entrada)
        for f in frases:
            if f not in entrada["frases"]:
                entrada["frases"].append(f)

    with open(PROPIO, "w") as f:
        json.dump(propio, f, indent=2, ensure_ascii=False)
    return PROPIO


def main():
    fallos = carga_fallos()
    if not fallos:
        print("No hay nada sin entender todavia.")
        print(f"Se anota solo en {REGISTRO}")
        return 0

    catalogo = carga_catalogo()
    ids = sorted(catalogo)
    print(f"{len(fallos)} frases distintas no encajaron con ninguna accion.\n")

    nuevos = {}
    for texto, veces in fallos.most_common():
        print(f'  "{texto}"   ({veces} {"vez" if veces == 1 else "veces"})')
        print("     escribe el id de la accion, 'l' para listarlos, "
              "Enter para saltar, 'q' para terminar")
        try:
            resp = input("     > ").strip()
        except (EOFError, KeyboardInterrupt):
            print()
            break

        if resp == "q":
            break
        if resp == "l":
            for i, ident in enumerate(ids):
                print(f"       {ident:<24}", end="\n" if i % 3 == 2 else "")
            print()
            try:
                resp = input("     > ").strip()
            except (EOFError, KeyboardInterrupt):
                break
        if not resp:
            continue
        if resp not in catalogo:
            print(f"     '{resp}' no existe, saltada")
            continue

        nuevos.setdefault(resp, []).append(texto)
        print(f"     anotada como alias de {resp}")

    if not nuevos:
        print("\nNada que guardar.")
        return 0

    ruta = guarda_alias(nuevos)
    total = sum(len(v) for v in nuevos.values())
    print(f"\n{total} alias guardados en {ruta}")
    print("Se aplican al abrir la pestaña JARVIS otra vez.")

    if input("¿Vaciar el registro de frases no entendidas? [s/N] ").strip().lower() == "s":
        try:
            os.remove(REGISTRO)
            print("registro vaciado")
        except OSError as e:
            print(f"no se pudo vaciar: {e}")
    return 0


if __name__ == "__main__":
    sys.exit(main())