Update app.py

app.py

@@ -1,9 +1,8 @@
-# app.py
 # -*- coding: utf-8 -*-
-import os
 import re
 import unicodedata
 from pathlib import Path
+from typing import Tuple
 
 import gradio as gr
 import joblib
@@ -11,67 +10,51 @@ import pandas as pd
 from scipy import sparse
 from sklearn.metrics.pairwise import cosine_similarity
 
-# ==========================
-# Ubicación de artefactos
-# ==========================
-ART = Path("artifacts")
+# -----------------------------
+# Rutas (funciona en HF Spaces)
+# -----------------------------
+ROOT = Path(__file__).parent
+ART = ROOT / "artifacts"
 VEC_PATH = ART / "tfidf_vectorizer.joblib"
 MAT_PATH = ART / "tfidf_matrix.npz"
 IDX_PATH = ART / "doc_index.csv"
 
-# ==========================
-# Utilidades de limpieza
-# ==========================
-import nltk
-from nltk.corpus import stopwords
-
-def _ensure_nltk():
-    try:
-        nltk.data.find("corpora/stopwords")
-    except LookupError:
-        nltk.download("stopwords")
-_ensure_nltk()
-
+# -----------------------------
+# Limpieza (sin NLTK)
+# -----------------------------
 def strip_accents(s: str) -> str:
     return "".join(c for c in unicodedata.normalize("NFKD", s) if not unicodedata.combining(c))
 
-STOPWORDS = {strip_accents(w.lower()) for w in stopwords.words("spanish")} | {"aun"}
-
-def limpiar_texto(s: str) -> str:
+# stopwords españolas normalizadas (compacta; puedes ampliar)
+STOPWORDS = {
+    "a","aca","ahi","ahí","al","algo","algunas","algunos","alla","allá","alli","allí","ante","antes",
+    "aquel","aquella","aquellas","aquellos","aqui","aquí","asi","así","aun","aunque","bajo","bien","cabe",
+    "cada","casi","cierta","ciertas","cierto","ciertos","como","con","contra","cual","cuales","cualquier",
+    "cualesquiera","cuyo","cuya","cuyas","cuyos","de","del","desde","donde","dos","el","ella","ellas",
+    "ellos","en","entre","era","erais","eramos","éramos","eran","eres","es","esa","esas","ese","eso",
+    "esos","esta","está","estaba","estaban","estamos","estan","están","estar","estas","este","esto",
+    "estos","etc","fue","fueron","ha","habia","había","habian","habían","haber","hay","hasta","la","las",
+    "le","les","lo","los","mas","más","me","mi","mis","mucha","muchas","mucho","muchos","muy","nada","ni",
+    "no","nos","nosotras","nosotros","nuestra","nuestras","nuestro","nuestros","o","otra","otras","otro",
+    "otros","para","pero","poco","por","porque","que","qué","quien","quién","quienes","quiénes","se","sea",
+    "sean","ser","si","sí","sido","sin","sobre","su","sus","tal","tambien","también","tampoco","tan",
+    "tanta","tantas","tanto","te","tenia","tenía","tenian","tenían","tendra","tendrá","tendran","tendrán",
+    "tenemos","tengo","ti","tiene","tienen","todo","todos","tu","tus","un","una","unas","uno","unos",
+    "usted","ustedes","y","ya"
+}
+# normalizar stopwords
+STOPWORDS = {strip_accents(w.lower()) for w in STOPWORDS} | {"aun"}
+
+def clean_text(s: str) -> str:
     if not isinstance(s, str):
         s = "" if s is None else str(s)
     s = strip_accents(s.lower())
-    s = re.sub(r"[“”„‟‹›«»—–‐-‒–—―\-]", " ", s)
-    s = re.sub(r"[^\w\s]", " ", s)
+    s = re.sub(r"[“”„‟‹›«»—–‐-‒–—―\-]", " ", s)  # comillas/guiones unicode
+    s = re.sub(r"[^\w\s]", " ", s)               # puntuación
     s = re.sub(r"\s+", " ", s).strip()
     toks = [t for t in s.split() if t not in STOPWORDS and not t.isdigit()]
     return " ".join(toks)
 
-# ==========================
-# Reglas heurísticas (ejemplo OPS)
-# ==========================
-REGLAS = [
-    {
-        "keywords": ["ops", "orden de prestacion de servicios", "contrato ops"],
-        "respuesta": {
-            "CICP":  ("2.1.2.02.02.008", "Servicios prestados a las empresas y servicios de producción"),
-            "CPC":   ("8",               "Servicios prestados a las empresas y servicios de producción"),
-            "UNSPSC":("80111600",        "Servicios de personal temporal"),
-        },
-        "motivo": "Coincidencia con palabra clave OPS",
-    },
-]
-
-def aplicar_reglas(consulta: str):
-    texto = limpiar_texto(consulta)
-    for regla in REGLAS:
-        if any(k in texto for k in regla["keywords"]):
-            rows = []
-            for cat, (cod, nom) in regla["respuesta"].items():
-                rows.append({"Catálogo": cat, "Código": cod, "Nombre": nom, "Similaridad": 1.0, "Origen": "Regla"})
-            return pd.DataFrame(rows)
-    return None
-
 def catalog_tag(source_file: str) -> str:
     s = (source_file or "").lower()
     if "cicp" in s: return "CICP"
@@ -79,7 +62,7 @@ def catalog_tag(source_file: str) -> str:
     if "unspsc" in s: return "UNSPSC"
     return "OTRO"
 
-def parse_code_name(codes_raw: str, text_original: str):
+def parse_code_name(codes_raw: str, text_original: str) -> Tuple[str, str]:
     codes_raw = str(codes_raw or "")
     text_original = str(text_original or "")
     m = re.search(r"CODIGO;NOMBRE:\s*([^;|]+)\s*;\s*([^|]+)", codes_raw, flags=re.I)
@@ -99,75 +82,112 @@ def parse_code_name(codes_raw: str, text_original: str):
         if m2 and name is None: name = m2.group(1).strip()
     return (code or "").strip(), (name or "").strip()
 
-# ==========================
-# Carga en startup
-# ==========================
-VEC = joblib.load(VEC_PATH)
-MAT = sparse.load_npz(MAT_PATH)
-IDX = pd.read_csv(IDX_PATH)
-IDX["catalogo"] = IDX["source_file"].apply(catalog_tag)
-
-# ==========================
-# Endpoint de predicción
-# ==========================
-def predecir(consulta: str, top_por_catalogo: int = 1):
-    if not consulta or not consulta.strip():
-        return pd.DataFrame([{"Catálogo": "", "Código": "", "Nombre": "", "Similaridad": 0.0, "Origen": "—"}])
-
-    # 1) Reglas
-    out_regla = aplicar_reglas(consulta)
-    if out_regla is not None:
-        return out_regla.sort_values("Catálogo")
-
-    # 2) Modelo TF-IDF
-    q = limpiar_texto(consulta)
-    vec_q = VEC.transform([q])
-    sims = cosine_similarity(vec_q, MAT)[0]
-
-    df = IDX.copy()
-    df["Similaridad"] = sims
-
-    frames = []
-    for cat in ["CICP", "CPC", "UNSPSC"]:
-        sub = (
-            df[df["catalogo"] == cat]
-            .sort_values("Similaridad", ascending=False)
-            .head(top_por_catalogo)
-            .copy()
-        )
-        parsed = sub.apply(lambda r: parse_code_name(r.get("codes_raw",""), r.get("text_original","")), axis=1)
-        sub["Código"] = [c for c, _ in parsed]
-        sub["Nombre"] = [n for _, n in parsed]
-        sub["Catálogo"] = cat
-        sub["Origen"] = "TF-IDF"
-        frames.append(sub[["Catálogo","Código","Nombre","Similaridad","Origen"]])
-
-    res = pd.concat(frames, ignore_index=True)
-    res["Similaridad"] = res["Similaridad"].round(4)
-    return res.sort_values("Catálogo")
-
-# ==========================
-# Gradio UI
-# ==========================
-with gr.Blocks(title="Recomendador CICP / CPC / UNSPSC") as demo:
-    gr.Markdown("## Recomendador por texto (CICP / CPC / UNSPSC)\n*TF-IDF + reglas*")
+# -----------------------------
+# Reglas duras (ejemplo OPS)
+# -----------------------------
+REGLAS = [
+    {
+        "keywords": ["ops", "orden de prestacion de servicios", "contrato ops"],
+        "respuesta": {
+            "CICP":  ("2.1.2.02.02.008", "Servicios prestados a las empresas y servicios de producción"),
+            "CPC":   ("8",               "Servicios prestados a las empresas y servicios de producción"),
+            "UNSPSC":("80111600",        "Servicios de personal temporal"),
+        },
+        "motivo": "Coincidencia con palabra clave OPS",
+    },
+]
+
+def aplicar_reglas(query: str):
+    q = clean_text(query)
+    for r in REGLAS:
+        if any(k in q for k in r["keywords"]):
+            df = pd.DataFrame(
+                [{"Catálogo": k, "Código": v[0], "Nombre": v[1], "Similaridad": 1.0} for k, v in r["respuesta"].items()]
+            )
+            return df, r["motivo"]
+    return None, None
+
+# -----------------------------
+# Carga perezosa de artefactos
+# -----------------------------
+VECTOR = None
+MATRIX = None
+INDEX = None
+
+def ensure_loaded():
+    global VECTOR, MATRIX, INDEX
+    if VECTOR is None:
+        VECTOR = joblib.load(VEC_PATH)
+    if MATRIX is None:
+        MATRIX = sparse.load_npz(MAT_PATH)
+    if INDEX is None:
+        INDEX = pd.read_csv(IDX_PATH)
+
+# -----------------------------
+# Motor TF-IDF agrupado
+# -----------------------------
+def recomendar(query: str, k: int):
+    try:
+        # 1) Reglas
+        df_regla, motivo = aplicar_reglas(query)
+        if df_regla is not None:
+            return df_regla, f"⚙️ Regla activada: {motivo}"
+
+        # 2) Modelo
+        ensure_loaded()
+        q = clean_text(query)
+        if not q:
+            return pd.DataFrame(), "La consulta quedó vacía tras limpieza."
+
+        xq = VECTOR.transform([q])
+        sims = cosine_similarity(xq, MATRIX).flatten()
+
+        df = INDEX.copy()
+        df["Similaridad"] = sims
+        df["Catálogo"] = df["source_file"].apply(catalog_tag)
+
+        # top-k por catálogo
+        out = []
+        for cat in ["CICP", "CPC", "UNSPSC"]:
+            sub = (
+                df[df["Catálogo"] == cat]
+                .sort_values("Similaridad", ascending=False)
+                .head(int(k))
+                .copy()
+            )
+            if sub.empty:
+                continue
+            parsed = sub.apply(lambda r: parse_code_name(r.get("codes_raw",""), r.get("text_original","")), axis=1)
+            sub["Código"] = [c for c, _ in parsed]
+            sub["Nombre"] = [n for _, n in parsed]
+            out.append(sub[["Catálogo", "Código", "Nombre", "Similaridad"]])
+
+        if not out:
+            return pd.DataFrame(), "Sin candidatos."
+        res = pd.concat(out, ignore_index=True)
+        return res, "OK"
+    except Exception as e:
+        # mostrará el error en la interfaz
+        return pd.DataFrame(), f"Error: {type(e).__name__}: {e}"
+
+# -----------------------------
+# Interfaz Gradio
+# -----------------------------
+with gr.Blocks(title="Recomendador por texto (CICP / CPC / UNSPSC)") as demo:
+    gr.Markdown("# Recomendador por texto (CICP / CPC / UNSPSC)\n\n_TF-IDF + reglas_")
     with gr.Row():
-        consulta = gr.Textbox(label="Descripción técnica", lines=3, placeholder="Ej: Vinculación joven investigadora OPS ...")
-        topk = gr.Slider(1, 5, value=1, step=1, label="Top por catálogo")
+        query = gr.Textbox(label="Descripción técnica", placeholder="reactivos de laboratorio para cromatografía hplc", lines=3)
+        k = gr.Slider(1, 5, value=1, step=1, label="Top por catálogo")
     btn = gr.Button("Buscar")
-    salida = gr.Dataframe(headers=["Catálogo","Código","Nombre","Similaridad","Origen"], interactive=False)
+    out = gr.Dataframe(headers=["Catálogo","Código","Nombre","Similaridad"], label="Resultados", wrap=True)
+    status = gr.Markdown()
 
-    ejemplos = gr.Examples(
-        examples=[
-            ["Vinculación joven investigadora, OPS gastos de operación y servicios técnicos", 1],
-            ["contrato de personal temporal", 1],
-            ["reactivos de laboratorio para cromatografía hplc", 1],
-        ],
-        inputs=[consulta, topk],
-        label="Ejemplos",
-    )
+    def _on_click(q, topk):
+        df, msg = recomendar(q, topk)
+        return df, (f"**Estado:** {msg}" if msg else "")
 
-    btn.click(predecir, inputs=[consulta, topk], outputs=[salida])
+    btn.click(_on_click, inputs=[query, k], outputs=[out, status])
 
 if __name__ == "__main__":
+    # Para pruebas locales
     demo.launch()