Update app.py

app.py

@@ -18,7 +18,6 @@ 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)
 
 # =========================
 # Utils de texto
@@ -27,15 +26,15 @@ def strip_accents(s: str) -> str:
     return "".join(c for c in unicodedata.normalize("NFKD", s) if not unicodedata.combining(c))
 
 STOPWORDS = {
-    "a","al","algo","algunas","algunos","ante","antes","aquel","aquella","aquellas","aquellos","aqui","así","aun","aunque",
+    "a","al","algo","algunas","algunos","ante","antes","aquel","aquella","aquellas","aquellos","aqui","asi","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",
+    "este","esto","estos","fue","fueron","ha","habia","habian","haber","hay","hasta","la","las","le","les","lo","los",
+    "mas","mas","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",
+    "que","quien","quienes","se","sea","sean","ser","si","si","sido","sin","sobre","su","sus","tal","tambien","tambien",
+    "tampoco","tan","tanta","tantas","tanto","te","tenia","tenian","tendra","tendran","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"}
@@ -43,64 +42,24 @@ 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"[“”„‟‹›«»—–‐‒–—―\-]", " ", 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)
 
+def _kw_pattern(kw_norm: str) -> str:
+    # "medidor ph" -> r"\bmedidor\b.*\bph\b"
+    parts = [re.escape(p) for p in kw_norm.split()]
+    if not parts: return ""
+    return r"\b" + r".*".join(parts) + r"\b"
+
 def catalog_tag(source_file: str) -> str:
     s = (source_file or "").lower()
     if "cicp" in s: return "CICP"
-    if "cpc" in s:  return "CPC"
+    if "cpc"  in s: return "CPC"
     if "unspsc" in s: return "UNSPSC"
     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) \
-        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 "")
-
-# --- añade esto cerca de tus utilidades, debajo de parse_code_name ---
-ORDER_CATS = ["CICP", "CPC", "UNSPSC"]
-
-def normalize_unspsc_if_cpc_901(rows):
-    """rows: lista de dicts [{'Catálogo','Código','Nombre','Similaridad'}]"""
-    out = []
-    for r in rows:
-        if r["Catálogo"] == "CPC" and str(r["Código"]).strip() == "901":
-            out.append({"Catálogo":"UNSPSC","Código":"N/A","Nombre":"N/A","Similaridad":1.0})
-        else:
-            out.append(r)
-    return out
-
-def order_and_fill_one_per_catalog(df):
-    """Garantiza 1 por catálogo (CICP,CPC,UNSPSC), con orden fijo y normalización 901->N/A."""
-    # Tomar el mejor por catálogo
-    best = (
-        df.sort_values("Similaridad", ascending=False)
-          .groupby("Catálogo", as_index=False)
-          .head(1)
-    )
-    # Pasar a lista para poder normalizar UNSPSC si CPC=901
-    rows = [{"Catálogo":r["Catálogo"], "Código":r["Código"], "Nombre":r["Nombre"], "Similaridad":r["Similaridad"]}
-            for _, r in best.iterrows()]
-    rows = normalize_unspsc_if_cpc_901(rows)
-
-    # Asegurar orden y devolver sólo los catálogos esperados
-    ordered = [r for r in rows if r["Catálogo"] in ORDER_CATS]
-    ordered.sort(key=lambda x: ORDER_CATS.index(x["Catálogo"]))
-    # Si alguno faltó, crear placeholders vacíos (opcional)
-    seen = {r["Catálogo"] for r in ordered}
-    for cat in ORDER_CATS:
-        if cat not in seen:
-            ordered.append({"Catálogo":cat, "Código":"", "Nombre":"", "Similaridad":0.0})
-    ordered.sort(key=lambda x: ORDER_CATS.index(x["Catálogo"]))
-    return pd.DataFrame(ordered, columns=["Catálogo","Código","Nombre","Similaridad"])
     
 # =========================
 # Reglas
@@ -674,22 +633,95 @@ REGLAS = [
     },
 ]
 
-
 def aplicar_reglas(query: str):
-    q = clean_text(query)
+    texto = clean_text(query)
     for r in REGLAS:
         for kw in r["keywords"]:
-            pat = re.escape(kw).replace(r"\ ", r".*")
-            if re.search(pat, q):
-                df = pd.DataFrame(
+            kw_norm = clean_text(kw)
+            if not kw_norm: continue
+            pat = _kw_pattern(kw_norm)
+            if re.search(pat, texto):
+                tmp = pd.DataFrame(
                     [{"Catálogo": k, "Código": v[0], "Nombre": v[1], "Similaridad": 1.0}
                      for k, v in r["respuesta"].items()]
                 )
-                return df.sort_values("Catálogo"), f"⚙️ Regla activada: {r['motivo']}"
+                return tmp, f"⚙️ Regla activada: {r['motivo']}"
     return None, None
 
 # =========================
-# Carga/entrenamiento TF-IDF
+# Parsing de códigos (robusto, mismo que search_tfidf.py)
+# =========================
+ORDER_CATS = ["CICP", "CPC", "UNSPSC"]
+
+def _s(x) -> str:
+    """string seguro ('' si None/NaN)"""
+    try:
+        if x is None: return ""
+        if isinstance(x, float) and x != x:  # NaN
+            return ""
+        return str(x)
+    except Exception:
+        return "" if x is None else str(x)
+
+def parse_code_name(catalogo: str, codes_raw, text_original) -> Tuple[str,str]:
+    cat = _s(catalogo).strip().upper()
+    cr  = _s(codes_raw)
+    to  = _s(text_original)
+
+    if cat == "UNSPSC":
+        m = re.search(r"UNSPSC:\s*([^;]+)\s*;\s*(.+)", cr, flags=re.I)
+        if m: return m.group(1).strip(), m.group(2).strip()
+
+    if cat == "CPC":
+        m = re.search(r"CPC:\s*([^;]+)\s*;\s*(.+)", cr, flags=re.I)
+        if m: return m.group(1).strip(), m.group(2).strip()
+
+    if cat == "CICP":
+        code = None
+        m1 = re.search(r"CODIGO:\s*([^\s\|;]+)", cr, flags=re.I)
+        if m1: code = m1.group(1).strip()
+        name = None
+        m2 = re.search(r"CICP:\s*([^|]+)$", to, flags=re.I)
+        if m2: name = m2.group(1).strip()
+        if code or name:
+            return _s(code).strip(), _s(name).strip()
+
+    # Fallback genérico
+    if ";" in cr:
+        parts = [p.strip() for p in cr.split(";", 2)]
+        if len(parts) >= 2:
+            return parts[-2], parts[-1]
+    return cr.strip(), (to if to else cr).strip()
+
+def normalize_unspsc_if_cpc_901(rows):
+    """Si el CPC seleccionado es 901, fuerza UNSPSC=N/A."""
+    out = []
+    cpc_is_901 = any(r["Catálogo"]=="CPC" and str(r["Código"]).strip()=="901" for r in rows)
+    for r in rows:
+        if r["Catálogo"]=="UNSPSC" and cpc_is_901:
+            out.append({"Catálogo":"UNSPSC","Código":"N/A","Nombre":"N/A","Similaridad":1.0})
+        else:
+            out.append(r)
+    return out
+
+def order_and_one_per_catalog(df_like):
+    """Top-1 por catálogo + orden CICP→CPC→UNSPSC + normalización 901."""
+    df = pd.DataFrame(df_like)
+    best = (df.sort_values("Similaridad", ascending=False)
+              .groupby("Catálogo", as_index=False)
+              .head(1))
+    rows = [{"Catálogo": r["Catálogo"], "Código": r["Código"], "Nombre": r["Nombre"],
+             "Similaridad": r["Similaridad"]} for _, r in best.iterrows()]
+    rows = normalize_unspsc_if_cpc_901(rows)
+    have = {r["Catálogo"] for r in rows}
+    for cat in ORDER_CATS:
+        if cat not in have:
+            rows.append({"Catálogo":cat,"Código":"", "Nombre":"", "Similaridad":0.0})
+    rows.sort(key=lambda r: ORDER_CATS.index(r["Catálogo"]))
+    return pd.DataFrame(rows, columns=["Catálogo","Código","Nombre","Similaridad"])
+
+# =========================
+# Carga/entrenamiento TF-IDF (como app (2).py)
 # =========================
 VECTOR = None
 MATRIX = None
@@ -697,14 +729,16 @@ INDEX = None
 
 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):
-    corpus = (index_df["tokens_lemmatized"] if "tokens_lemmatized" in index_df.columns else
-              index_df["text_original"].fillna("").astype(str).map(clean_text))
+    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")
@@ -722,8 +756,9 @@ def ensure_loaded():
     if vec is None or not _is_fitted_vectorizer(vec):
         vec, X = _train_and_persist_from_index(INDEX)
     elif X is None:
-        corpus = (INDEX["tokens_lemmatized"] if "tokens_lemmatized" in INDEX.columns else
-                  INDEX["text_original"].fillna("").astype(str).map(clean_text))
+        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
@@ -735,7 +770,8 @@ def recomendar(query: str):
     # 1) Reglas
     df_regla, motivo = aplicar_reglas(query)
     if df_regla is not None:
-        return df_regla, motivo
+        df_out = order_and_one_per_catalog(df_regla)
+        return df_out, motivo
 
     # 2) Modelo
     ensure_loaded()
@@ -743,35 +779,37 @@ def recomendar(query: str):
     if not q:
         return pd.DataFrame(), "La consulta quedó vacía tras limpieza."
 
-    xq = VECTOR.transform([q]); sims = cosine_similarity(xq, MATRIX).flatten()
+    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]
 
-    # ⬇️ NUEVO: 1 por catálogo, orden CICP→CPC→UNSPSC y normalización CPC=901
-    df = df[["Catálogo","Código","Nombre","Similaridad"]]
-    df_out = order_and_fill_one_per_catalog(df)
+    # Evitar NaN antes del parser
+    if "codes_raw" in df.columns: df["codes_raw"] = df["codes_raw"].fillna("")
+    if "text_original" in df.columns: df["text_original"] = df["text_original"].fillna("")
+
+    parsed = df.apply(lambda r: parse_code_name(r["Catálogo"], 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]
 
+    df = df[["Catálogo","Código","Nombre","Similaridad"]]
+    df_out = order_and_one_per_catalog(df)
     return df_out, "OK"
 
 # =========================
 # Exportar (xlsx con fallback a csv)
 # =========================
-def exportar(query: str) -> Tuple[str, str]:
+def exportar(query: 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:
         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"
@@ -779,16 +817,15 @@ def exportar(query: str) -> Tuple[str, str]:
                 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
+# UI (Gradio)
 # =========================
 with gr.Blocks(title="Recomendador de Códigos (CICP / CPC / UNSPSC)") as demo:
-    gr.Markdown("# Recomendador de Códigos (CICP / CPC / UNSPSC)\n")
+    gr.Markdown("# Recomendador de Códigos (CICP / CPC / UNSPSC)")
     query = gr.Textbox(
         label="Descripción técnica",
         placeholder="reactivos de laboratorio para cromatografía hplc",
@@ -807,11 +844,10 @@ with gr.Blocks(title="Recomendador de Códigos (CICP / CPC / UNSPSC)") as demo:
 
     def _on_download(q):
         path, info = exportar(q)
-        # mostramos mensaje en status también
         return path, f"**Descarga:** {info}"
 
     btn.click(_on_search, inputs=[query], outputs=[out, status])
-    query.submit(_on_search, inputs=[query], outputs=[out, status])  # ← buscar con ENTER
+    query.submit(_on_search, inputs=[query], outputs=[out, status])
     btn_xlsx.click(_on_download, inputs=[query], outputs=[file_out, status])
 
 if __name__ == "__main__":