JARVIS/config/indexar.py

#!/usr/bin/env python3
"""Prepara el corpus de COFRE para que JARVIS lo consulte.

    python3 ~/COFRE/CODERS/JARVIS/config/indexar.py            # ver que entraria
    python3 ~/COFRE/CODERS/JARVIS/config/indexar.py --hacerlo  # copiarlo
    python3 ~/COFRE/CODERS/JARVIS/config/indexar.py --hacerlo --solo-notas

Que hace: recorre las carpetas de COFRE, se queda SOLO con lo que es
conocimiento y lo copia a la carpeta de documentos de Newelle, que es de donde
lee el RAG. Despues hay que pulsar "Update Index" en los ajustes de la app.

Por que hace falta filtrar. Medido en esta maquina:

    COFRE entero (las nueve carpetas)   ~120 GB
    de eso, ISOs, maquinas virtuales,
    APKs, audio y un diccionario de
    contraseñas de 9,1 GB              ~100 GB
    texto de menos de 100 KB             76,6 MB en 35 098 ficheros
    quitando exploitdb                   20,8 MB en  4 548 ficheros

exploitdb se queda fuera a proposito: son 32 000 ficheros que aplastarian el
indice y enterrarian tus notas, y ademas NO hacen falta ahi. Tienes
`searchsploit` instalado, que busca en ellos de forma exacta e instantanea. Un
indice de vectores es peor que una busqueda exacta cuando lo que buscas es el
nombre literal de un exploit.

Los nombres se aplanan conservando la ruta —PENTESTERS__OSCP__nmap.md— para que
al responder se pueda ver de donde salio cada cosa.
"""
import argparse
import hashlib
import os
import shutil
import sys

COFRE = os.path.expanduser("~/COFRE")
DESTINO = os.path.expanduser(
    "~/.var/app/io.github.qwersyk.Newelle/config/models/documents")

CARPETAS = ["CODERS", "PENTESTERS", "AUTOMATAS", "DOSER", "HARD-WARERS",
            "PHISHERS", "PROTECTORS", "REVERSERS", "SNEAKERS"]

EXTENSIONES = {".md", ".txt", ".rst", ".adoc"}

# Por tamaño: por encima de esto ya no es prosa, son datos. El diccionario
# kaonashi.txt son 9,1 GB de contraseñas.
TAMANO_MAXIMO = 100 * 1024
TAMANO_MINIMO = 120          # menos de esto no dice nada

# Carpetas que no aportan conocimiento y si mucho ruido
FUERA = {
    "exploitdb",        # 32 000 ficheros; para eso esta searchsploit
    "node_modules", ".git", "venv", ".venv", "site-packages", "__pycache__",
    "isos", "VirtualBox VMs", "build", "dist", ".cache", "target",
    "flatpak-app", ".flatpak-builder", "rag_cache", "entrenamiento",
}

# Ficheros que son plantilla o licencia y solo ensucian
NOMBRES_FUERA = {"LICENSE.txt", "LICENSE.md", "CHANGELOG.md", "requirements.txt"}


def interesa(ruta: str, nombre: str) -> bool:
    ext = os.path.splitext(nombre)[1].lower()
    if ext not in EXTENSIONES or nombre in NOMBRES_FUERA:
        return False
    try:
        tam = os.path.getsize(ruta)
    except OSError:
        return False
    return TAMANO_MINIMO <= tam <= TAMANO_MAXIMO


def recorre(solo_notas: bool):
    """Devuelve [(origen, nombre_plano)] de todo lo que merece indexarse."""
    salida = []
    for carpeta in CARPETAS:
        raiz = os.path.join(COFRE, carpeta)
        if not os.path.isdir(raiz):
            continue
        for dirpath, dirnames, ficheros in os.walk(raiz):
            dirnames[:] = [d for d in dirnames
                           if d not in FUERA and not d.startswith(".")]
            for nombre in ficheros:
                if solo_notas and not nombre.lower().endswith(".md"):
                    continue
                origen = os.path.join(dirpath, nombre)
                if not interesa(origen, nombre):
                    continue
                relativa = os.path.relpath(origen, COFRE)
                plano = relativa.replace(os.sep, "__")
                if len(plano) > 180:      # algunos nombres son kilometricos
                    corto = hashlib.sha256(plano.encode()).hexdigest()[:8]
                    plano = plano[:150] + "_" + corto + os.path.splitext(plano)[1]
                salida.append((origen, plano))
    return salida


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--hacerlo", action="store_true", help="copiar de verdad")
    ap.add_argument("--solo-notas", action="store_true",
                    help="solo .md, sin los .txt sueltos")
    ap.add_argument("--limpiar", action="store_true",
                    help="vaciar el destino antes de copiar")
    args = ap.parse_args()

    print("recorriendo COFRE...", flush=True)
    lista = recorre(args.solo_notas)

    porcarpeta, total = {}, 0
    for origen, _ in lista:
        carpeta = os.path.relpath(origen, COFRE).split(os.sep)[0]
        tam = os.path.getsize(origen)
        n, b = porcarpeta.get(carpeta, (0, 0))
        porcarpeta[carpeta] = (n + 1, b + tam)
        total += tam

    print(f"\n{'carpeta':<14}{'ficheros':>10}{'texto':>12}")
    for carpeta in CARPETAS:
        if carpeta in porcarpeta:
            n, b = porcarpeta[carpeta]
            print(f"{carpeta:<14}{n:>10}{b/1048576:>10.1f} MB")
    print(f"{'TOTAL':<14}{len(lista):>10}{total/1048576:>10.1f} MB")
    print(f"\nfragmentos de 256 tokens: ~{total // 1024}")

    if not args.hacerlo:
        print("\nesto es solo el recuento; para copiarlo: --hacerlo")
        return 0

    if args.limpiar and os.path.isdir(DESTINO):
        shutil.rmtree(DESTINO)
    os.makedirs(DESTINO, exist_ok=True)

    copiados = saltados = 0
    for origen, plano in lista:
        destino = os.path.join(DESTINO, plano)
        try:
            # solo si cambio: reindexar 4500 ficheros iguales no tiene sentido
            if (os.path.exists(destino)
                    and os.path.getmtime(destino) >= os.path.getmtime(origen)):
                saltados += 1
                continue
            shutil.copy2(origen, destino)
            copiados += 1
        except OSError as e:
            print(f"  no se pudo copiar {plano}: {e}")

    print(f"\n{copiados} copiados, {saltados} ya estaban al dia")
    print(f"en {DESTINO}")
    print("\nAhora, en Newelle: Ajustes -> RAG -> Update Index")
    print("La primera vez tarda; son unos 21 000 fragmentos.")
    return 0


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