Update app.py

app.py

@@ -1,55 +1,50 @@
 # -*- coding: utf-8 -*-
-import re
-import unicodedata
+import re, unicodedata
 from pathlib import Path
 from typing import Tuple
-
 import gradio as gr
 import joblib
 import pandas as pd
 from scipy import sparse
 from sklearn.metrics.pairwise import cosine_similarity
+from sklearn.feature_extraction.text import TfidfVectorizer
+from sklearn.utils.validation import check_is_fitted
 
-# -----------------------------
-# Rutas (funciona en HF Spaces)
-# -----------------------------
+# =========================
+# Config
+# =========================
 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"
+TOP_N = 10  # ← Top global fijo (se eliminó el control del panel)
 
-# -----------------------------
-# Limpieza (sin NLTK)
-# -----------------------------
+# =========================
+# Utils de texto
+# =========================
 def strip_accents(s: str) -> str:
     return "".join(c for c in unicodedata.normalize("NFKD", s) if not unicodedata.combining(c))
 
-# 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"
+    "a","al","algo","algunas","algunos","ante","antes","aquel","aquella","aquellas","aquellos","aqui","así","aun","aunque",
+    "bajo","bien","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","eran","eres","es","esa","esas","ese","eso","esos","esta","estaba","estaban","estamos","estan","estar","estas",
+    "este","esto","estos","fue","fueron","ha","habia","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","quien","quienes","se","sea","sean","ser","si","sí","sido","sin","sobre","su","sus","tal","tambien","también",
+    "tampoco","tan","tanta","tantas","tanto","te","tenia","tenían","tendrá","tendrán","tenemos","tengo","ti","tiene",
+    "tienen","todo","todos","tu","tus","un","una","unas","uno","unos","usted","ustedes","y","ya"
 }
 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)
+    if not isinstance(s, str): s = "" if s is None else str(s)
     s = strip_accents(s.lower())
-    s = re.sub(r"[“”„‟‹›«»—–‐-‒–—―\-]", " ", s)  # comillas/guiones unicode
-    s = re.sub(r"[^\w\s]", " ", s)               # puntuación
+    s = re.sub(r"[“”„‟‹›«»—–‐-‒–—―\-]", " ", s)
+    s = re.sub(r"[^\w\s]", " ", s)
     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)
@@ -62,28 +57,17 @@ def catalog_tag(source_file: str) -> str:
     return "OTRO"
 
 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)
-    if not m:
-        m = re.search(r"CODIGO;NOMBRE:\s*([^;|]+)\s*;\s*([^|]+)", text_original, flags=re.I)
-    if m:
-        return m.group(1).strip(), m.group(2).strip()
-    code = None; name = None
-    m1 = re.search(r"CODIGO\s*:\s*([^|]+)", codes_raw, flags=re.I)
-    m2 = re.search(r"NOMBRE\s*:\s*([^|]+)", codes_raw, flags=re.I)
-    if m1: code = m1.group(1).strip()
-    if m2: name = m2.group(1).strip()
-    if code is None or name is None:
-        m1 = re.search(r"CODIGO\s*:\s*([^|]+)", text_original, flags=re.I)
-        m2 = re.search(r"NOMBRE\s*:\s*([^|]+)", text_original, flags=re.I)
-        if m1 and code is None: code = m1.group(1).strip()
-        if m2 and name is None: name = m2.group(1).strip()
-    return (code or "").strip(), (name or "").strip()
-
-# -----------------------------
-# Reglas duras (tu bloque)
-# -----------------------------
+    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) \
+        or re.search(r"CODIGO;NOMBRE:\s*([^;|]+)\s*;\s*([^|]+)", text_original, flags=re.I)
+    if m: return m.group(1).strip(), m.group(2).strip()
+    code = (re.search(r"CODIGO\s*:\s*([^|]+)", codes_raw, flags=re.I) or re.search(r"CODIGO\s*:\s*([^|]+)", text_original, flags=re.I))
+    name = (re.search(r"NOMBRE\s*:\s*([^|]+)", codes_raw, flags=re.I) or re.search(r"NOMBRE\s*:\s*([^|]+)", text_original, flags=re.I))
+    return (code.group(1).strip() if code else ""), (name.group(1).strip() if name else "")
+
+# =========================
+# Reglas
+# =========================
 REGLAS = [
     {
         "keywords": ["ops", "orden de prestacion de servicios", "contrato ops"],
@@ -106,8 +90,7 @@ REGLAS = [
     {
         'keywords': ["viatico", "viaticos"],
         'respuesta': {
-            'CICP': ("2.3.2.02.2.010".replace(".2.2."," .02.02.").replace(" ",""), "Servicios administrativos de apoyo") if False else
-                    ("2.3.2.02.02.010", "Servicios administrativos de apoyo"),
+            'CICP': ("2.3.2.02.02.010", "Servicios administrativos de apoyo"),
             'CPC': ("901", "Gastos directos de la administración pública"),
             'UNSPSC': ("20102301", "Gastos de viaje y manutención"),
         },
@@ -208,187 +191,142 @@ REGLAS = [
 def aplicar_reglas(query: str):
     q = clean_text(query)
     for r in REGLAS:
-        # Coincidencia OR por keyword (normalizado)
         for kw in r["keywords"]:
             pat = re.escape(kw).replace(r"\ ", r".*")
             if re.search(pat, q):
                 df = pd.DataFrame(
-                    [{"Catálogo": k, "Código": v[0], "Nombre": v[1], "Similaridad": 1.0} for k, v in r["respuesta"].items()]
+                    [{"Catálogo": k, "Código": v[0], "Nombre": v[1], "Similaridad": 1.0}
+                     for k, v in r["respuesta"].items()]
                 )
-                return df, f"⚙️ Regla activada: {r['motivo']}"
+                return df.sort_values("Catálogo"), f"⚙️ Regla activada: {r['motivo']}"
     return None, None
 
-# ----------------------------
-# Carga perezosa de artefactos
-# -----------------------------
+# =========================
+# Carga/entrenamiento TF-IDF
+# =========================
 VECTOR = None
 MATRIX = None
 INDEX = None
 
-from sklearn.feature_extraction.text import TfidfVectorizer
-from sklearn.utils.validation import check_is_fitted
-from sklearn.exceptions import NotFittedError
-
 def _is_fitted_vectorizer(vec) -> bool:
     try:
-        check_is_fitted(vec, attributes=["vocabulary_"])
-        check_is_fitted(vec._tfidf, attributes=["idf_"])
+        check_is_fitted(vec, attributes=["vocabulary_"]); check_is_fitted(vec._tfidf, attributes=["idf_"])
         return True
     except Exception:
         return False
 
 def _train_and_persist_from_index(index_df: pd.DataFrame):
-    # 1) Elegir corpus
-    if "tokens_lemmatized" in index_df.columns:
-        corpus = index_df["tokens_lemmatized"].fillna("").astype(str).tolist()
-    else:
-        corpus = index_df["text_original"].fillna("").astype(str).map(clean_text).tolist()
-
-    # 2) Entrenar vectorizador
-    vec = TfidfVectorizer(
-        analyzer="word",
-        token_pattern=r"(?u)\b\w+\b",
-        min_df=1,
-        max_df=0.9,
-        ngram_range=(1, 2),
-        sublinear_tf=True,
-        norm="l2",
-    )
-    X = vec.fit_transform(corpus)
-
-    # 3) Persistir
+    corpus = (index_df["tokens_lemmatized"] if "tokens_lemmatized" in index_df.columns else
+              index_df["text_original"].fillna("").astype(str).map(clean_text))
+    vec = TfidfVectorizer(analyzer="word", token_pattern=r"(?u)\b\w+\b",
+                          min_df=1, max_df=0.9, ngram_range=(1,2),
+                          sublinear_tf=True, norm="l2")
+    X = vec.fit_transform(list(corpus))
     ART.mkdir(exist_ok=True, parents=True)
-    joblib.dump(vec, VEC_PATH)
-    sparse.save_npz(MAT_PATH, X)
-
+    joblib.dump(vec, VEC_PATH); sparse.save_npz(MAT_PATH, X)
     return vec, X
 
 def ensure_loaded():
-    """Carga artefactos; si el vectorizador no está fit, reentrena desde el índice."""
     global VECTOR, MATRIX, INDEX
-
     if INDEX is None:
         INDEX = pd.read_csv(IDX_PATH)
-
-    vec = None
-    if VEC_PATH.exists():
-        try:
-            vec = joblib.load(VEC_PATH)
-        except Exception:
-            vec = None
-
-    X = None
-    if MAT_PATH.exists():
-        try:
-            X = sparse.load_npz(MAT_PATH)
-        except Exception:
-            X = None
-
+    vec = joblib.load(VEC_PATH) if VEC_PATH.exists() else None
+    X = sparse.load_npz(MAT_PATH) if MAT_PATH.exists() else None
     if vec is None or not _is_fitted_vectorizer(vec):
         vec, X = _train_and_persist_from_index(INDEX)
-    else:
-        if X is None:
-            if "tokens_lemmatized" in INDEX.columns:
-                corpus = INDEX["tokens_lemmatized"].fillna("").astype(str).tolist()
-            else:
-                corpus = INDEX["text_original"].fillna("").astype(str).map(clean_text).tolist()
-            X = vec.transform(corpus)
-            sparse.save_npz(MAT_PATH, X)
-
-    globals()["VECTOR"] = vec
-    globals()["MATRIX"] = X
-
-# -----------------------------
-# Motor TF-IDF (Top N global)
-# -----------------------------
-def recomendar(query: str, k: int):
+    elif X is None:
+        corpus = (INDEX["tokens_lemmatized"] if "tokens_lemmatized" in INDEX.columns else
+                  INDEX["text_original"].fillna("").astype(str).map(clean_text))
+        X = vec.transform(list(corpus))
+        sparse.save_npz(MAT_PATH, X)
+    VECTOR, MATRIX = vec, X
+
+# =========================
+# Búsqueda
+# =========================
+def recomendar(query: str):
+    # 1) Reglas
+    df_regla, motivo = aplicar_reglas(query)
+    if df_regla is not None:
+        return df_regla, 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)
+    parsed = df.apply(lambda r: parse_code_name(r.get("codes_raw",""), r.get("text_original","")), axis=1)
+    df["Código"] = [c for c,_ in parsed]; df["Nombre"] = [n for _,n in parsed]
+
+    out = (df[["Catálogo","Código","Nombre","Similaridad"]]
+           .sort_values("Similaridad", ascending=False)
+           .head(TOP_N)
+           .reset_index(drop=True))
+    if out.empty: return pd.DataFrame(), "Sin candidatos."
+    return out, "OK"
+
+# =========================
+# Exportar (xlsx con fallback a csv)
+# =========================
+def exportar(query: str) -> Tuple[str, str]:
+    df, _ = recomendar(query)
+    if df is None or df.empty:
+        df = pd.DataFrame(columns=["Catálogo","Código","Nombre","Similaridad"])
+
+    # Intento 1: xlsx con openpyxl
     try:
-        # 1) Reglas
-        df_regla, motivo = aplicar_reglas(query)
-        if df_regla is not None:
-            # Ya viene con 1.0 de Similaridad para cada catálogo de la regla
-            return df_regla.sort_values("Catálogo"), 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)
-        parsed = df.apply(lambda r: parse_code_name(r.get("codes_raw",""), r.get("text_original","")), axis=1)
-        df["Código"] = [c for c, _ in parsed]
-        df["Nombre"] = [n for _, n in parsed]
-
-        # Top N GLOBAL (ya no por catálogo)
-        out = (
-            df[["Catálogo","Código","Nombre","Similaridad"]]
-            .sort_values("Similaridad", ascending=False)
-            .head(int(k))
-            .reset_index(drop=True)
-        )
-        if out.empty:
-            return pd.DataFrame(), "Sin candidatos."
-        return out, "OK"
-    except Exception as e:
-        return pd.DataFrame(), f"Error: {type(e).__name__}: {e}"
-
-# -----------------------------
-# Exportar a Excel
-# -----------------------------
-def exportar_excel(query: str, k: int) -> str:
-    """Genera un Excel con los resultados actuales y retorna la ruta."""
-    df, _ = recomendar(query, k)
-    # Asegura algo descargable aunque no haya resultados
-    path = "/tmp/busqueda.xlsx"
-    if isinstance(df, pd.DataFrame) and not df.empty:
-        df.to_excel(path, index=False, sheet_name="Resultados")
-    else:
-        pd.DataFrame(columns=["Catálogo","Código","Nombre","Similaridad"]).to_excel(
-            path, index=False, sheet_name="Resultados"
-        )
-    return path
-
-# -----------------------------
-# Interfaz Gradio
-# -----------------------------
+        path = "/tmp/busqueda.xlsx"
+        with pd.ExcelWriter(path, engine="openpyxl") as w:
+            df.to_excel(w, index=False, sheet_name="Resultados")
+        return path, "Archivo Excel (.xlsx) generado."
+    except Exception:
+        # Intento 2: xlsx con xlsxwriter
+        try:
+            import xlsxwriter  # noqa: F401
+            path = "/tmp/busqueda.xlsx"
+            with pd.ExcelWriter(path, engine="xlsxwriter") as w:
+                df.to_excel(w, index=False, sheet_name="Resultados")
+            return path, "Archivo Excel (.xlsx) generado (xlsxwriter)."
+        except Exception:
+            # Fallback: CSV
+            path = "/tmp/busqueda.csv"
+            df.to_csv(path, index=False)
+            return path, "openpyxl/xlsxwriter no disponibles: se generó CSV."
+
+# =========================
+# UI
+# =========================
 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():
-        query = gr.Textbox(
-            label="Descripción técnica",
-            placeholder="reactivos de laboratorio para cromatografía hplc",
-            lines=3
-        )
-        k = gr.Slider(1, 30, value=10, step=1, label="Top N (global)")  # ← ya no 'Top por catálogo'
+    query = gr.Textbox(
+        label="Descripción técnica",
+        placeholder="reactivos de laboratorio para cromatografía hplc",
+        lines=3
+    )
     with gr.Row():
         btn = gr.Button("Buscar", variant="primary")
-        btn_xlsx = gr.Button("Descargar búsqueda en Excel")
+        btn_xlsx = gr.Button("Descargar búsqueda")
     out = gr.Dataframe(headers=["Catálogo","Código","Nombre","Similaridad"], label="Resultados", wrap=True)
     status = gr.Markdown()
     file_out = gr.File(label="Archivo generado", interactive=False)
 
-    def _on_click(q, topn):
-        df, msg = recomendar(q, topn)
+    def _on_search(q):
+        df, msg = recomendar(q)
         return df, (f"**Estado:** {msg}" if msg else "")
 
-    def _on_download(q, topn):
-        path = exportar_excel(q, topn)
-        return path
-
-    # Click en botón
-    btn.click(_on_click, inputs=[query, k], outputs=[out, status])
-
-    # Buscar al presionar ENTER en el Textbox
-    query.submit(_on_click, inputs=[query, k], outputs=[out, status])
+    def _on_download(q):
+        path, info = exportar(q)
+        # mostramos mensaje en status también
+        return path, f"**Descarga:** {info}"
 
-    # Descargar Excel con los resultados actuales
-    btn_xlsx.click(_on_download, inputs=[query, k], outputs=[file_out])
+    btn.click(_on_search, inputs=[query], outputs=[out, status])
+    query.submit(_on_search, inputs=[query], outputs=[out, status])  # ← buscar con ENTER
+    btn_xlsx.click(_on_download, inputs=[query], outputs=[file_out, status])
 
 if __name__ == "__main__":
     demo.launch()