"""Aplica `data/rag_alcance.json` a un índice RAG ya construido, sin reingerir. Dos pasadas, las mismas que hace la ingesta: 1. **Descartar** índices alfabéticos, sumarios y preliminares. 2. **Reetiquetar** `especie` en las secciones que no son de perro ni gato. Reingerir cuesta OCR sobre cientos de MB de PDF; esto sólo toca metadatos y filas, deja los vectores calculados intactos y corre en segundos. Ver `app/rag/alcance_corpus.py` para por qué hace falta (el filtro de especie estaba inerte: todos los chunks con `especie` vacía). uv run --group rag python ../scripts/curar_indice.py # simulacro, no escribe uv run --group rag python ../scripts/curar_indice.py --aplicar # escribe Tras aplicarlo hay que republicar el índice (`make publish-index`) para que las builds y las demás máquinas se lo lleven; si no, sólo queda arreglado en local. """ from __future__ import annotations import argparse import json import sys from collections import Counter from pathlib import Path RAIZ = Path(__file__).resolve().parent.parent sys.path.insert(0, str(RAIZ / "backend")) from app.config import obtener_config # noqa: E402 from app.rag.alcance_corpus import ( # noqa: E402 cargar_descartes, cargar_rangos, debe_descartarse, especie_de, ) def actualizar_manifiesto(ruta: Path, n_fragmentos: int) -> None: """Deja `manifest.json` de acuerdo con lo que hay realmente en la tabla.""" destino = ruta / "manifest.json" if not destino.exists(): print("⚠ Sin manifest.json: no se puede dejar constancia del recuento.") return datos = json.loads(destino.read_text(encoding="utf-8")) antes = datos.get("n_fragmentos") datos["n_fragmentos"] = n_fragmentos datos["curado"] = "data/rag_alcance.json" destino.write_text(json.dumps(datos, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print(f"manifest.json actualizado: n_fragmentos {antes} → {n_fragmentos}") def main() -> int: parser = argparse.ArgumentParser() parser.add_argument( "--aplicar", action="store_true", help="escribe los cambios; sin esta bandera sólo informa de qué haría", ) parser.add_argument("--indice", type=Path, help="ruta del índice (por defecto, la de config)") args = parser.parse_args() import lancedb cfg = obtener_config() ruta = args.indice or cfg.rag_index_dir if not ruta.exists(): print(f"❌ No hay índice en {ruta}. Usa `make fetch-index`.") return 1 descartes, rangos = cargar_descartes(), cargar_rangos() if not descartes and not rangos: print("❌ data/rag_alcance.json no declara nada: no habría qué hacer.") return 1 print(f"Índice: {ruta}") print(f"\nDescartes ({len(descartes)}):") for d in descartes: print(f" · {d.libro_empieza_por[:42]}… pp. {d.desde}–{d.hasta} ({d.motivo[:58]})") print(f"Rangos de especie ({len(rangos)}):") for r in rangos: print(f" · {r.libro_empieza_por[:42]}… pp. {r.desde}–{r.hasta} → {r.especie}") db = lancedb.connect(str(ruta)) tabla = db.open_table("literatura") df = tabla.to_pandas() total = len(df) fuera = df.apply(lambda f: debe_descartarse(f["libro"], f["pagina"], f["texto"]), axis=1) df_quedan = df[~fuera].copy() nueva = [especie_de(f.libro, f.pagina, "") for f in df_quedan.itertuples()] reetiquetados = sum( 1 for antes, ahora in zip(df_quedan["especie"], nueva, strict=True) if antes != ahora ) print(f"\nchunks: {total}") print(f" a descartar (índices/sumarios/prelim.): {int(fuera.sum())}") print(f" quedan: {len(df_quedan)}") print(f" cambian de especie: {reetiquetados}") print(f" especies tras curar: {dict(Counter(nueva))}") manifiesto = ruta / "manifest.json" if manifiesto.exists(): declarado = json.loads(manifiesto.read_text(encoding="utf-8")).get("n_fragmentos") if declarado != len(df_quedan): print(f"\n⚠ manifest.json declara {declarado} fragmentos y quedarían {len(df_quedan)}.") if not fuera.any() and not reetiquetados: if manifiesto.exists() and declarado != len(df_quedan): if not args.aplicar: print("Repite con --aplicar para corregir sólo el manifiesto.") return 0 actualizar_manifiesto(ruta, len(df_quedan)) return 0 print("\nNada que hacer.") return 0 if not args.aplicar: print("\nSimulacro. Repite con --aplicar para escribir.") return 0 # Se reescribe la tabla entera desde el DataFrame: `mode='overwrite'` conserva el esquema y # los vectores ya calculados, que es justo lo que no queremos recomputar. El índice FTS sí # se reconstruye después, porque la sobrescritura lo invalida. df_quedan["especie"] = nueva db.create_table("literatura", data=df_quedan, mode="overwrite") tabla = db.open_table("literatura") try: tabla.create_fts_index("texto", replace=True) print("Índice FTS reconstruido.") except Exception as exc: # noqa: BLE001 - informativo; el denso sigue sirviendo print(f"⚠ No se pudo reconstruir el FTS ({exc}). Reconstrúyelo antes de confiar en el híbrido.") # Compactar no es cosmético: LanceDB versiona, así que sobrescribir DEJA los datos viejos en # disco y el índice CRECE al quitarle filas (medido: 73 MB → 140 MB). Como este artefacto se # sube al Hub y se hornea en la imagen, publicarlo sin compactar multiplicaría su tamaño. import datetime tabla.optimize(cleanup_older_than=datetime.timedelta(0), delete_unverified=True) tamano = sum(f.stat().st_size for f in ruta.rglob("*") if f.is_file()) / 1e6 print(f"Versiones antiguas purgadas. Tamaño en disco: {tamano:.0f} MB") # El manifiesto lo escribe la ingesta y aquí acaba de cambiar el número de fragmentos. Si no # se actualiza, miente: `fetch-index` lo imprime y `publish-index` lo usa como mensaje de # commit, así que el Hub anunciaría un recuento que no es el del índice que sirve. actualizar_manifiesto(ruta, len(df_quedan)) print(f"\n✅ {total} → {len(df_quedan)} fragmentos. Recuerda `make publish-index`.") return 0 if __name__ == "__main__": raise SystemExit(main())