from __future__ import annotations import re from typing import Any _STRATEGIC = ( "paramount", "netflix", "disney", "hbo", "youtube premium", "aya ", "deezer", "globoplay", "truecaller", "food balance", ) def _norm(value: Any) -> str: text = str(value or "").casefold() text = re.sub(r"[^a-z0-9áàâãéèêíìîóòôõúùûç+ ]+", " ", text) return " ".join(text.split()) def _iter_invoice_rows(invoice_detail: Any): if not isinstance(invoice_detail, dict): return for _, sections in invoice_detail.items(): if not isinstance(sections, dict): continue for section, rows in sections.items(): if not isinstance(rows, list): continue for row in rows: if isinstance(row, dict): yield str(section or ""), row def infer_informational_context(query: str, invoice_detail: Any = None) -> dict[str, list[str]]: """Infere a marca informacional usada pela finalização após uma consulta RAG. Não executa RAG e não guarda sessão. Apenas transforma a pergunta + evidência de fatura em metadata de domínio que o grafo/framework poderá persistir. """ q = _norm(query) if not q: return {} names: list[str] = [] types: list[str] = [] for section, row in _iter_invoice_rows(invoice_detail): name = str(row.get("desc") or row.get("name") or row.get("description") or "").strip() if not name: continue n = _norm(name) if n and (n in q or q in n): names.append(name) sec = _norm(section) if "sva" in sec or "servico" in sec or "serviço" in sec: types.append("avulso") # Serviços estratégicos podem ser reconhecidos mesmo sem detalhe da fatura. if not names: for alias in _STRATEGIC: if _norm(alias) in q: names.append(alias.strip().title()) types.append("estrategico") break if not names: return {} dedup_names = list(dict.fromkeys(names)) dedup_types = list(dict.fromkeys(types)) return { "informational_service_names": dedup_names, "informational_vas_types": dedup_types, "informational_rag_vas_types": dedup_types, }