# motor_pesquisa.py # Motor de pesquisa local sobre OTs + comentários import os import json import re import unicodedata from collections import Counter, defaultdict from rapidfuzz import fuzz DESKTOP = os.path.expanduser("~/Desktop") WORKORDERS_FILE = os.path.join(DESKTOP, "workorders.json") COMENTARIOS_FILE = os.path.join(DESKTOP, "comentarios.jsonl") # ---------------------------------------------------------------- # Normalização de texto (remove acentos, minúsculas, pontuação) # ---------------------------------------------------------------- def _normalizar(texto): if not texto: return "" # Remover acentos texto = unicodedata.normalize("NFKD", str(texto)) texto = "".join(c for c in texto if not unicodedata.combining(c)) # Minúsculas texto = texto.lower() # Trocar pontuação por espaço texto = re.sub(r"[^\w\s]", " ", texto) # Colapsar espaços texto = re.sub(r"\s+", " ", texto).strip() return texto # ---------------------------------------------------------------- # Expansão de sinónimos em português (técnico) # ---------------------------------------------------------------- SINONIMOS = { "nao arranca": ["nao liga", "nao parte", "nao funciona", "sem arranque", "nao inicia", "parada", "parou", "desligada"], "erro": ["alarme", "falha", "avaria", "defeito", "problema", "erro"], "carregador": ["alimentador", "iemca", "bar feeder", "charger"], "emergencia": ["emerg", "emergency", "botao emergencia", "seta"], "carregamento": ["carrega", "alimentacao", "introducao de barra"], "vibracao": ["vibra", "trepida", "treme"], "fuga": ["vazamento", "escape", "perda"], "oleo": ["lubrificante", "fluido", "óleo hidraulico", "oleo hidraulico"], "ar comprimido": ["pneumatico", "ar", "pressao ar"], "barulho": ["ruido", "rúido", "som estranho"], "sobreaquecimento": ["aquecimento", "quente", "temperatura alta"], } def _expandir_sinonimos(texto_norm): """Adiciona sinónimos ao texto normalizado para melhorar o match.""" extras = [] for chave, alts in SINONIMOS.items(): if chave in texto_norm: extras.extend(alts) if extras: return texto_norm + " " + " ".join(extras) return texto_norm # ---------------------------------------------------------------- # Carregamento de dados # ---------------------------------------------------------------- def carregar_workorders(): with open(WORKORDERS_FILE, encoding="utf-8") as f: return json.load(f) def carregar_comentarios(): """Devolve dict {ot_id: {...}} do comentarios.jsonl.""" if not os.path.exists(COMENTARIOS_FILE): return {} resultado = {} with open(COMENTARIOS_FILE, encoding="utf-8") as f: for linha in f: try: obj = json.loads(linha) resultado[obj["id"]] = obj except Exception: continue return resultado # ---------------------------------------------------------------- # Construção do índice # ---------------------------------------------------------------- class MotorPesquisa: def __init__(self): self.workorders = carregar_workorders() self.comentarios = carregar_comentarios() self.docs = [] # lista de dicts com texto indexado self._construir_indice() def _construir_indice(self): print(f"📚 A indexar {len(self.workorders)} OTs...") for wo in self.workorders: ot_id = wo.get("id") com = self.comentarios.get(ot_id, {}) # Combinar todos os campos textuais relevantes partes = [ wo.get("code", ""), wo.get("name", ""), wo.get("assetCode", ""), wo.get("assetName", ""), wo.get("brand", ""), wo.get("brandModel", ""), wo.get("serialNumber", ""), wo.get("remedy") or "", ] # Adicionar comentários (texto livre) se existirem for c in com.get("comments", []): partes.append(c.get("text", "")) texto_bruto = " ".join(str(p) for p in partes if p) texto_norm = _normalizar(texto_bruto) texto_norm = _expandir_sinonimos(texto_norm) self.docs.append({ "id": ot_id, "code": wo.get("code"), "name": wo.get("name"), "assetCode": wo.get("assetCode"), "assetFamily": _familia_asset(wo.get("assetCode", "")), "assetName": wo.get("assetName"), "brand": wo.get("brand"), "brandModel": wo.get("brandModel"), "technician": wo.get("technician"), "urgency": wo.get("urgency"), "criticality": wo.get("criticality"), "beginDateTime": wo.get("beginDateTime"), "stateId": wo.get("stateId"), "remedy": wo.get("remedy"), "texto_norm": texto_norm, "comentarios_obj": com, }) print(f"✅ Índice pronto: {len(self.docs)} documentos") # ------------------------------------------------------------ # Pesquisa # ------------------------------------------------------------ def pesquisar(self, pergunta, top_n=10, min_score=30): """ Devolve top_n casos mais similares + agregação de remedies. Args: pergunta (str): texto do utilizador top_n (int): número de casos a devolver min_score (int): score mínimo (0-100) para incluir """ # 1. Detetar códigos de máquina na pergunta (754, 754A, EC9, ...) maquinas_mencionadas = _extrair_codigos_maquina(pergunta) pergunta_norm = _normalizar(pergunta) pergunta_norm = _expandir_sinonimos(pergunta_norm) resultados = [] for doc in self.docs: score = self._calcular_score(pergunta_norm, doc, maquinas_mencionadas) if score >= min_score: resultados.append((score, doc)) resultados.sort(key=lambda x: -x[0]) top = resultados[:top_n] # 2. Agregação de soluções (só das OTs com remedy preenchido) return self._formatar_resposta(pergunta, maquinas_mencionadas, top, resultados) def _calcular_score(self, pergunta_norm, doc, maquinas): """ Score composto 0-100: - 40% match de máquina exata (ou família) - 40% similaridade textual (rapidfuzz) - 10% bónus se tem remedy preenchido - 10% bónus se estado Encerrada (foi resolvida) """ score_maq = 0 asset_code = (doc.get("assetCode") or "").upper() familia = doc.get("assetFamily", "") if maquinas: if asset_code in maquinas: score_maq = 40 elif familia and familia in maquinas: score_maq = 25 elif any(m in asset_code or asset_code in m for m in maquinas): score_maq = 30 # Similaridade textual (parcial, tolerante a ordem) texto_doc = doc["texto_norm"] if pergunta_norm and texto_doc: s1 = fuzz.partial_ratio(pergunta_norm, texto_doc) s2 = fuzz.token_set_ratio(pergunta_norm, texto_doc) score_txt = (s1 * 0.4 + s2 * 0.6) * 0.40 else: score_txt = 0 bonus_remedy = 10 if doc.get("remedy") else 0 bonus_encerrada = 10 if doc.get("stateId") == "ada41e02-5e6a-4dc7-9f5c-e6b14340f1a6" else 0 return score_maq + score_txt + bonus_remedy + bonus_encerrada def _formatar_resposta(self, pergunta, maquinas, top, todos_resultados): # Contagem de remedies nos top resultados contagem = Counter() for score, doc in top: r = doc.get("remedy") if r: contagem[r] += 1 total_com_remedy = sum(contagem.values()) probabilidades = [] if total_com_remedy > 0: for remedy, n in contagem.most_common(): probabilidades.append({ "solucao": remedy, "casos": n, "probabilidade": round(n / total_com_remedy * 100, 1), }) return { "pergunta": pergunta, "maquinas_detetadas": sorted(maquinas), "total_resultados": len(todos_resultados), "top_casos": [ { "score": round(score, 1), "code": doc["code"], "name": doc["name"], "assetCode": doc["assetCode"], "assetName": doc["assetName"], "brand": doc["brand"], "brandModel": doc["brandModel"], "technician": doc["technician"], "beginDateTime": doc["beginDateTime"], "remedy": doc["remedy"], "comentarios": doc["comentarios_obj"].get("comments", []), } for score, doc in top ], "solucoes_probaveis": probabilidades, } # ---------------------------------------------------------------- # Utilitários # ---------------------------------------------------------------- def _familia_asset(asset_code): """'754A' -> '754'; 'EC9B' -> 'EC9'; '170' -> '170'""" if not asset_code: return "" return re.sub(r"[A-Z]+$", "", asset_code.upper()) def _extrair_codigos_maquina(texto): """Extrai códigos tipo 754, 754A, EC9, EC9B, X41, P035, SF6...""" if not texto: return set() padroes = [ r"\b\d{3,4}[A-Z]?\b", # 754, 754A, 770, 807 r"\b[A-Z]{2,3}\d{1,3}[A-Z]?\b", # EC9, EC9A, SF6, SBF r"\b[A-Z]\d{2,3}[A-Z]?\b", # X41, Z69, Z7C r"\bP\d{3}\b", # P035, P125 ] encontrados = set() texto_upper = texto.upper() for p in padroes: for m in re.findall(p, texto_upper): encontrados.add(m) return encontrados # ---------------------------------------------------------------- # Apresentação (CLI bonito) # ---------------------------------------------------------------- def imprimir_resposta(resp): print() print("=" * 78) print(f"🔎 PERGUNTA: {resp['pergunta']}") if resp["maquinas_detetadas"]: print(f"🖥️ Máquinas detetadas: {', '.join(resp['maquinas_detetadas'])}") print(f"📊 Total de casos encontrados: {resp['total_resultados']}") print("=" * 78) print("\n🏆 CASOS MAIS SIMILARES\n") for i, caso in enumerate(resp["top_casos"], 1): print(f"{i}. [{caso['score']}%] {caso['code']} — {caso['name'][:60]}") print(f" 🖥️ {caso['assetCode']} - {caso['assetName']}") print(f" 🏷️ {caso['brand']} {caso['brandModel']}") print(f" 👤 {caso['technician']} 📅 {caso['beginDateTime'][:10] if caso['beginDateTime'] else '-'}") print(f" 💡 Solução: {caso['remedy'] or '(sem solução registada)'}") if caso['comentarios']: print(f" 💬 {len(caso['comentarios'])} comentário(s)") for c in caso['comentarios'][:2]: texto = c.get('text', '')[:120] print(f" - {c.get('author')}: {texto}") print() if resp["solucoes_probaveis"]: print("📊 SOLUÇÕES POR PROBABILIDADE\n") for s in resp["solucoes_probaveis"]: barra = "█" * int(s["probabilidade"] / 3) print(f" {s['probabilidade']:5.1f}% {barra:<34} {s['solucao']} ({s['casos']} caso(s))") print() else: print("⚠️ Nenhuma solução registada nos casos encontrados.\n") # ---------------------------------------------------------------- # CLI interativo # ---------------------------------------------------------------- def main(): motor = MotorPesquisa() print("\n🎯 Motor de pesquisa pronto. Escreve a tua pergunta (ou 'sair').") print(" Exemplo: '754 erro emergencia carregador nao arranca'\n") while True: try: pergunta = input("❓ > ").strip() except (EOFError, KeyboardInterrupt): print("\n👋 Até logo!") break if not pergunta: continue if pergunta.lower() in ("sair", "exit", "quit", "q"): print("👋 Até logo!") break resp = motor.pesquisar(pergunta, top_n=10, min_score=30) imprimir_resposta(resp) if __name__ == "__main__": main()