Delete enhanced_search_v2.py

enhanced_search_v2.py

@@ -1,375 +0,0 @@
-# enhanced_search_v2.py (Versão Final, Corrigida e Comentada)
-###################################################################################################
-#
-# Este arquivo contém o motor de busca principal.
-#
-# O fluxo de dados da consulta foi corrigido para garantir que a correção ortográfica
-# seja propagada para todas as camadas da busca (lexical e semântica), restaurando a
-# funcionalidade original e mantendo as melhorias de ranking.
-#
-# A ordenação final usa a hierarquia:
-#   1. Feedback do Usuário
-#   2. "Golden Match" (Score Semântico de 100%)
-#   3. Score Híbrido (IA + Texto)
-#   4. Cobertura do Rol (desempate)
-#
-###################################################################################################
-
-import pandas as pd
-import re
-from thefuzz import process, fuzz
-from unidecode import unidecode
-import time
-from sentence_transformers import util
-import torch
-import math
-from collections import defaultdict
-from rank_bm25 import BM25Okapi
-
-# --- FUNÇÕES AUXILIARES DE NORMALIZAÇÃO --- #
-
-def literal_normalize_text(text):
-    """Normalização "agressiva" para a Camada 0 (Busca Literal)."""
-    if pd.isna(text): return ""
-    normalized = unidecode(str(text).lower())
-    normalized = re.sub(r'[^\w\s]', ' ', normalized)
-    return re.sub(r'\s+', ' ', normalized).strip()
-
-def normalize_text(text):
-    """Normalização padrão (minúsculas, sem acentos)."""
-    if pd.isna(text): return ""
-    return unidecode(str(text).lower().strip())
-
-def get_longest_word(query_text):
-    """Extrai a palavra mais longa de uma consulta como último recurso."""
-    words = re.findall(r'\b\w{4,}\b', query_text)
-    if not words: return ""
-    return max(words, key=len)
-
-
-# --- FUNÇÕES DE FORMATAÇÃO E DESTAQUE --- #
-
-def format_result(row_data, row_index, match_type="", score=0):
-    """Converte uma linha do DataFrame em um dicionário de resultado padronizado."""
-    data = row_data.copy()
-    is_rol = data.get('Correlacao_Rol', '').strip().lower() == 'sim'
-
-    if not is_rol:
-        data['Grupo'], data['Subgrupo'], data['Vigencia'], data['Resolucao_Normativa'] = '', '', '', ''
-        data['PAC'], data['DUT'] = '---', '---'
-    else:
-        data['PAC'] = 'Sim' if data.get('PAC', '').strip().lower() == 'pac' else 'Não'
-        original_dut_value = data.get('DUT', '').strip()
-        if original_dut_value and original_dut_value.replace('.', '', 1).isdigit():
-            data['DUT'] = f'Sim, DUT nº {original_dut_value}'
-        else: data['DUT'] = 'Não'
-
-    standard_columns = [
-        'Codigo_TUSS', 'Descricao_TUSS', 'Correlacao_Rol', 'Procedimento_Rol',
-        'Resolucao_Normativa', 'Vigencia', 'OD', 'AMB', 'HCO', 'HSO', 'PAC',
-        'DUT', 'SUBGRUPO', 'GRUPO', 'CAPITULO', 'Sinonimo_1', 'Sinonimo_2',
-        'Sinonimo_3', 'Sinonimo_4', 'Semantico'
-    ]
-    formatted_data = {col: data.get(col, '') for col in standard_columns}
-    
-    result = {
-        "row_index": row_index,
-        "score": round(score),
-        "text_score": round(score),
-        "semantic_score": 0,
-        "match_type": match_type,
-        "is_rol_procedure": is_rol
-    }
-    result.update(formatted_data)
-    return result
-
-def _highlight_matches(results, query):
-    """Adiciona tags <b> para destacar os termos da busca nos resultados."""
-    if not query or not results: return results
-    stopwords = {'de', 'do', 'da', 'dos', 'das', 'a', 'o', 'e', 'em', 'um', 'uma', 'para', 'com'}
-    query_words = {word for word in normalize_text(query).split() if len(word) > 2 and word not in stopwords}
-    cols_to_highlight = ['Descricao_TUSS', 'Procedimento_Rol', 'Sinonimo_1', 'Sinonimo_2', 'Sinonimo_3', 'Sinonimo_4', 'Semantico']
-    for result in results:
-        for col in cols_to_highlight:
-            original_text = result.get(col, '')
-            if original_text and query_words:
-                highlighted_text = original_text
-                for word in sorted(list(query_words), key=len, reverse=True):
-                    pattern = r'\b(' + re.escape(word) + r')\b'
-                    highlighted_text = re.sub(pattern, r'<b>\1</b>', highlighted_text, flags=re.IGNORECASE)
-                result[f"{col}_highlighted"] = highlighted_text
-            else:
-                result[f"{col}_highlighted"] = original_text
-    return results
-
-
-# --- FUNÇÕES DE CARREGAMENTO DE DADOS --- #
-
-def load_and_prepare_database(db_path):
-    """Carrega e pré-processa a base de dados principal para otimizar a busca."""
-    try:
-        print(f"Carregando e preparando a base de dados de: {db_path}...")
-        df_original = pd.read_csv(db_path, dtype=str).fillna('')
-        search_cols = ['Descricao_TUSS', 'Procedimento_Rol', 'Sinonimo_1', 'Sinonimo_2', 'Sinonimo_3', 'Sinonimo_4', 'Semantico', 'SUBGRUPO', 'GRUPO', 'CAPITULO']
-        df_normalized = df_original.copy()
-        for col in search_cols + ['Codigo_TUSS']:
-            if col in df_normalized.columns:
-                df_normalized[f'{col}_literal'] = df_normalized[col].apply(literal_normalize_text)
-                df_normalized[f'{col}_norm'] = df_normalized[col].apply(normalize_text)
-        
-        df_normalized['full_text_norm'] = df_normalized[[f'{col}_norm' for col in search_cols if f'{col}_norm' in df_normalized.columns]].agg(' '.join, axis=1)
-
-        print("Criando dicionário da base, modelo BM25...")
-        tokenized_corpus = [text.split() for text in df_normalized['full_text_norm']]
-        db_word_set = {word for doc in tokenized_corpus for word in doc if word}
-        bm25_model = BM25Okapi(tokenized_corpus)
-        
-        print("Criando corpus para busca fuzzy...")
-        fuzzy_search_corpus = []
-        for index, row in df_normalized.iterrows():
-            for col in ['Descricao_TUSS', 'Procedimento_Rol', 'Sinonimo_1', 'Sinonimo_2', 'Sinonimo_3', 'Sinonimo_4']:
-                if f'{col}_norm' in row and pd.notna(row[f'{col}_norm']):
-                    val = row[f'{col}_norm']
-                    if val: fuzzy_search_corpus.append((val, index, f'{col}_norm'))
-        
-        print(f"Base de dados pronta com {len(df_original)} procedimentos.")
-        return df_original, df_normalized, fuzzy_search_corpus, bm25_model, db_word_set, None, None
-    except Exception as e: print(f"Erro crítico ao carregar/preparar a base de dados: {e}"); raise
-
-def load_general_dictionary(path):
-    """Carrega um dicionário geral de palavras em português."""
-    try:
-        with open(path, 'r', encoding='utf-8') as f: words = {normalize_text(line.strip()) for line in f if line.strip()}
-        return words
-    except (FileNotFoundError, Exception): return set()
-
-def load_correction_corpus(dict_path, column_name='Termo_Correto'):
-    """Carrega um corpus de correções ortográficas de um CSV."""
-    try:
-        df_dict = pd.read_csv(dict_path, dtype=str).fillna('')
-        if column_name not in df_dict.columns: return [], []
-        original_corpus = df_dict[column_name].dropna().astype(str).tolist()
-        normalized_corpus = [normalize_text(term) for term in original_corpus]
-        return original_corpus, normalized_corpus
-    except (FileNotFoundError, Exception): return [], []
-
-
-# --- FUNÇÕES DE RECLASSIFICAÇÃO SEMÂNTICA (IA) --- #
-
-def create_unified_document_text(result_dict):
-    """Cria uma string de texto única para a análise da IA."""
-    text_parts = {
-        result_dict.get('Descricao_TUSS', ''), result_dict.get('Procedimento_Rol', ''),
-        result_dict.get('Semantico', ''), result_dict.get('SUBGRUPO', ''),
-        result_dict.get('GRUPO', ''), result_dict.get('CAPITULO', '')
-    }
-    for i in range(1, 5): text_parts.add(result_dict.get(f'Sinonimo_{i}', ''))
-    return ". ".join(sorted([part for part in text_parts if part and str(part).strip()]))
-
-def rerank_with_cross_encoder(query, results_list, model):
-    """Reclassifica resultados usando um Cross-Encoder e a lógica de Score Híbrido."""
-    if not model or not results_list or not query: return results_list, "Cross-Encoder não fornecido ou lista de candidatos vazia."
-    sentence_pairs = [[query, create_unified_document_text(result)] for result in results_list]
-    if not sentence_pairs: return results_list, "Não foram encontrados pares para reordenar."
-
-    try:
-        raw_scores = model.predict(sentence_pairs, show_progress_bar=False)
-        semantic_scores_normalized = torch.sigmoid(torch.tensor(raw_scores)).numpy() * 100
-        
-        for i, result in enumerate(results_list):
-            semantic_score = round(semantic_scores_normalized[i])
-            text_score = result.get('text_score', 0)
-            
-            result['semantic_score'] = semantic_score
-            result['match_type'] = "Relevância Híbrida (IA+Texto)"
-
-            if semantic_score >= 99.5:
-                result['is_golden_match'] = True
-                result['hybrid_score'] = semantic_score 
-            else:
-                result['is_golden_match'] = False
-                weights = (0.8, 0.2) if semantic_score >= 90 else (0.6, 0.4) if semantic_score >= 70 else (0.4, 0.6)
-                result['hybrid_score'] = (semantic_score * weights[0]) + (text_score * weights[1])
-
-        key_function = lambda x: (x.get('is_golden_match', False), x.get('hybrid_score', 0), x.get('is_rol_procedure', False))
-        reranked_results = sorted(results_list, key=key_function, reverse=True)
-        
-        log_message = "Reordenação final por: 1º Golden Match, 2º Score Híbrido, 3º Cobertura do Rol."
-        return reranked_results, log_message
-    
-    except Exception as e:
-        log_message = f"Erro no Cross-Encoder: {e}"; print(log_message)
-        return results_list, log_message
-
-
-# --- FUNÇÃO INTERNA DE BUSCA COM CAMADAS --- #
-
-def _run_search_layers(literal_query, normalized_query, response, df_original, df_normalized, fuzzy_search_corpus, bm25_model, limit_per_layer):
-    """Executa a busca em múltiplas camadas para encontrar candidatos."""
-    matched_indices = set()
-    stopwords = {'de', 'do', 'da', 'dos', 'das', 'a', 'o', 'e', 'em', 'um', 'uma', 'para', 'com'}
-    query_words = [word for word in normalized_query.split() if word not in stopwords and len(word) > 1]
-    
-    def sort_key(x):
-        return (x.get('score', 0), x.get('is_rol_procedure', False))
-
-    for layer in ["literal_matches", "exact_matches", "logical_matches", "almost_exact_matches", "term_matches", "keyword_matches"]:
-        response["results_by_layer"][layer] = []
-
-    # Camadas 0 e 1 (Exatas)
-    if literal_query:
-        for col in ['Codigo_TUSS_literal', 'Descricao_TUSS_literal', 'Procedimento_Rol_literal']:
-            matches = df_normalized[df_normalized[col] == literal_query]
-            for index, _ in matches.iterrows():
-                if index not in matched_indices:
-                    response["results_by_layer"]["literal_matches"].append(format_result(df_original.loc[index], index, "Texto Exato", 100))
-                    matched_indices.add(index)
-
-    if normalized_query:
-        matches = df_normalized[df_normalized['Codigo_TUSS_norm'] == normalized_query]
-        for index, _ in matches.iterrows():
-            if index not in matched_indices:
-                response["results_by_layer"]["exact_matches"].append(format_result(df_original.loc[index], index, "Código Exato", 100))
-                matched_indices.add(index)
-        for col in ['Descricao_TUSS_norm', 'Procedimento_Rol_norm']:
-            matches = df_normalized[df_normalized[col] == normalized_query]
-            for index, _ in matches.iterrows():
-                if index not in matched_indices:
-                    response["results_by_layer"]["exact_matches"].append(format_result(df_original.loc[index], index, "Exato (Normalizado)", 100))
-                    matched_indices.add(index)
-
-    # Camada 2 (Lógica)
-    if query_words:
-        mask = pd.Series(True, index=df_normalized.index)
-        for word in query_words: mask &= df_normalized['full_text_norm'].str.contains(r'\b' + re.escape(word) + r'\b', na=False)
-        for index, row in df_normalized[mask & ~df_normalized.index.isin(matched_indices)].iterrows():
-            score = fuzz.WRatio(normalized_query, row.get('full_text_norm', ''))
-            response["results_by_layer"]["logical_matches"].append(format_result(df_original.loc[index], index, "Busca Lógica", score))
-            matched_indices.add(index)
-
-    # Camada 3 (Fuzzy)
-    if fuzzy_search_corpus:
-        processed_indices = set()
-        for match_text, score in process.extractBests(normalized_query, [item[0] for item in fuzzy_search_corpus], scorer=fuzz.token_set_ratio, limit=limit_per_layer * 3, score_cutoff=90):
-            if score == 100 and match_text == normalized_query: continue
-            for _, original_index, _ in [item for item in fuzzy_search_corpus if item[0] == match_text]:
-                if original_index not in matched_indices and original_index not in processed_indices:
-                    response["results_by_layer"]["almost_exact_matches"].append(format_result(df_original.loc[original_index], original_index, "Busca por Aproximação", 98))
-                    matched_indices.add(original_index); processed_indices.add(original_index)
-
-    # Camada 4 (BM25)
-    if query_words and bm25_model:
-        doc_scores = bm25_model.get_scores(query_words)
-        max_score = max(doc_scores) if any(doc_scores) else 1.0
-        top_indices = sorted(range(len(doc_scores)), key=lambda i: doc_scores[i], reverse=True)[:limit_per_layer * 5]
-        for i in top_indices:
-            if doc_scores[i] > 0 and (original_index := df_normalized.index[i]) not in matched_indices:
-                normalized_score = (doc_scores[i] / max_score) * 90
-                response["results_by_layer"]["term_matches"].append(format_result(df_original.loc[original_index], original_index, "Relevância de Termos", normalized_score))
-                matched_indices.add(original_index)
-
-    # Ordena todas as camadas
-    for layer in response["results_by_layer"]:
-        response["results_by_layer"][layer] = sorted(response["results_by_layer"][layer], key=sort_key, reverse=True)[:limit_per_layer * 4]
-
-    return None
-
-
-# --- FUNÇÃO PRINCIPAL QUE ORQUESTRA A BUSCA --- #
-
-def search_procedure_with_log(query, df_original, df_normalized, fuzzy_search_corpus, correction_corpus,
-                              portuguese_word_set, bm25_model, db_word_set, 
-                              doc_freq, tuss_to_full_text_map,
-                              limit_per_layer=10,
-                              semantic_model=None,
-                              cross_encoder_model=None,
-                              user_best_matches_counts=None, user_feedback_threshold=10):
-    """Orquestra todo o processo de busca, da correção à reordenação final."""
-    RERANK_LIMIT = 50; start_time = time.time(); original_query = str(query).strip()
-    response = {"search_log": [], "results_by_layer": {}, "final_semantic_results": [], "was_corrected": False, "original_query": original_query, "corrected_query": ""}
-    if not original_query: response["search_log"].append("Query vazia."); return response
-    response["search_log"].append(f"Buscando por: '{original_query}'")
-    
-    # ETAPA 1: CORREÇÃO ORTOGRÁFICA
-    stopwords = {'de', 'do', 'da', 'dos', 'das', 'a', 'o', 'e', 'em', 'um', 'uma', 'para', 'com'}; query_after_correction = original_query
-    original_correction_corpus, normalized_correction_corpus = correction_corpus; valid_words = portuguese_word_set.union(db_word_set)
-    if valid_words and original_correction_corpus:
-        words, corrected_words, made_correction = query_after_correction.split(), [], False
-        for word in words:
-            norm_word = normalize_text(word); clean_norm_word = re.sub(r'[^\w]', '', norm_word)
-            if len(norm_word) < 4 or norm_word in stopwords or clean_norm_word in valid_words: corrected_words.append(word); continue
-            match_norm, score = process.extractOne(clean_norm_word, normalized_correction_corpus, scorer=fuzz.token_set_ratio)
-            if score >= 85:
-                corrected_word = original_correction_corpus[normalized_correction_corpus.index(match_norm)]
-                if word.istitle(): corrected_word = corrected_word.title()
-                elif word.isupper(): corrected_word = corrected_word.upper()
-                corrected_words.append(corrected_word); made_correction = True
-            else: corrected_words.append(word)
-        if made_correction: query_after_correction = " ".join(corrected_words); response.update({"was_corrected": True, "corrected_query": query_after_correction}); response["search_log"].append(f"Query corrigida para: '{query_after_correction}'.")
-
-    # <<< CORREÇÃO DO FLUXO DE DADOS >>>
-    # ETAPA 2: PREPARAÇÃO DAS QUERIES - Restaurado para a lógica original e correta.
-    # Cria uma versão da query para busca lexical (sem stopwords) e mantém a completa para a IA.
-    cleaned_query = " ".join([word for word in query_after_correction.split() if normalize_text(word) not in stopwords])
-    normalized_query = normalize_text(cleaned_query) # Para buscas por palavras-chave (BM25, Lógica, etc.)
-    if not cleaned_query.strip(): response["search_log"].append("Query resultante vazia."); return response
-    if cleaned_query != query_after_correction: response["search_log"].append(f"Query limpa (sem stop words): '{cleaned_query}'")
-    
-    # ETAPA 3: EXECUÇÃO DA BUSCA
-    _run_search_layers(literal_normalize_text(query_after_correction), normalized_query, response, df_original, df_normalized, fuzzy_search_corpus, bm25_model, limit_per_layer)
-    
-    # ETAPA 3.5: LÓGICA DE EARLY EXIT
-    high_confidence_results = response["results_by_layer"].get("literal_matches", []) + response["results_by_layer"].get("exact_matches", [])
-    if high_confidence_results:
-        response["search_log"].append("\n--- [MODO DE ALTA CONFIANÇA - SAÍDA ANTECIPADA] ---")
-        final_list = sorted(high_confidence_results, key=lambda x: (x.get('text_score', 0), x.get('is_rol_procedure', False)), reverse=True)
-        query_for_highlight = query_after_correction
-        response["final_semantic_results"] = _highlight_matches(final_list[:15], query_for_highlight)
-        end_time = time.time(); response["search_duration_seconds"] = round(end_time - start_time, 4)
-        response["search_log"].append(f"Busca completa (saída antecipada) em {response['search_duration_seconds']} segundos.")
-        print(f"\n\n==================== LOG DE DEPURAÇÃO (QUERY: '{original_query}') ====================")
-        for log_item in response["search_log"]: print(log_item)
-        return response
-
-    # ETAPA 4: AGREGAÇÃO HIERÁRQUICA E REORDENAÇÃO
-    all_candidates = []
-    # Camadas são agregadas em ordem de prioridade
-    for layer in ["logical_matches", "almost_exact_matches", "term_matches", "keyword_matches"]:
-        all_candidates.extend(response["results_by_layer"].get(layer, []))
-    
-    unique_candidates = list({r['row_index']: r for r in all_candidates}.values())
-    response["search_log"].append(f"\n--- [MODO DE BUSCA AMPLA] ---")
-    response["search_log"].append(f"Total de candidatos únicos (após desduplicação): {len(unique_candidates)}.")
-    
-    if user_best_matches_counts:
-        query_norm_fb = normalize_text(response.get("corrected_query") or original_query)
-        for r in unique_candidates:
-            votes = user_best_matches_counts.get(query_norm_fb, {}).get(r['Codigo_TUSS'], 0)
-            if votes >= user_feedback_threshold: r.update({'is_user_best_match': True, 'feedback_votes': votes})
-
-    response["search_log"].append(f"\n--- Análise e Reordenação ---")
-    
-    final_list = []
-    if unique_candidates:
-        # A IA recebe a consulta COMPLETA (com stopwords) para melhor contexto.
-        query_for_semantic = query_after_correction
-        
-        prioritized_by_feedback = sorted([r for r in unique_candidates if r.get('is_user_best_match')], key=lambda x: (x.get('feedback_votes', 0), x.get('semantic_score', 0), x.get('text_score', 0)), reverse=True)
-        to_rerank = [r for r in unique_candidates if not r.get('is_user_best_match')]
-        
-        final_list.extend(prioritized_by_feedback)
-        if prioritized_by_feedback: response["search_log"].append(f"{len(prioritized_by_feedback)} resultado(s) priorizado(s) por feedback.")
-        
-        if to_rerank:
-            to_rerank_sorted = sorted(to_rerank, key=lambda x: x.get('text_score', 0), reverse=True)
-            reranked_by_ia, log_msg = rerank_with_cross_encoder(query_for_semantic, to_rerank_sorted[:RERANK_LIMIT], cross_encoder_model)
-            response["search_log"].append(log_msg)
-            final_list.extend(reranked_by_ia)
-
-    query_for_highlight = query_after_correction
-    response["final_semantic_results"] = _highlight_matches(final_list[:15], query_for_highlight)
-    end_time = time.time(); response["search_duration_seconds"] = round(end_time - start_time, 4)
-    response["search_log"].append(f"Busca completa em {response['search_duration_seconds']} segundos.")
-    print(f"\n\n==================== LOG DE DEPURAÇÃO (QUERY: '{original_query}') ====================")
-    for log_item in response["search_log"]: print(log_item)
-    return response