"""Commercial safety validation for generated BLIF replies. Core approved rule: A BLIF não presta serviço direto de instalação. """ from __future__ import annotations import re from typing import Any, Dict, List, Optional from app.business_knowledge_service import KnowledgeMatch, retrieve_business_knowledge _PRICE_RE = re.compile(r"(? str: text = str(text or "").lower() for src, dst in [("á", "a"), ("à", "a"), ("ã", "a"), ("â", "a"), ("é", "e"), ("ê", "e"), ("í", "i"), ("ó", "o"), ("õ", "o"), ("ô", "o"), ("ú", "u"), ("ç", "c")]: text = text.replace(src, dst) return text def _known_prices() -> set[str]: from app.business_knowledge_service import load_business_knowledge data = load_business_knowledge() prices = set() for item in list(data.get("products") or []) + list(data.get("accessories") or []): value = item.get("price_without_vat") if value is None: continue prices.add(str(value)) prices.add(f"{value},00") prices.add(f"{value}.00") return prices def _is_approved_installation_explanation(low: str) -> bool: approved = [ "sem instalacao incluida", "valor indicado e referente ao equipamento", "valor indicado e referente ao equipamento, sem instalacao", "blif nao presta servico direto de instalacao", "blif nao presta instalacao", "qualquer eletricista qualificado", "suporte remoto ao eletricista", ] return any(phrase in low for phrase in approved) def _states_installation_included(low: str) -> bool: allowed_negations = [ "sem instalacao incluida", "sem a instalacao incluida", "nao inclui instalacao", "nao esta incluida instalacao", "nao esta incluida a instalacao", ] if any(phrase in low for phrase in allowed_negations): return False blocked_patterns = [ r"\bcom instalacao incluida\b", r"\binclui(?: a)? instalacao\b", r"\binstalacao (?:esta |fica )?incluida\b", r"\binstalacao gratuita\b", ] return any(re.search(pattern, low) for pattern in blocked_patterns) def validate_generated_reply( message_body: str, *, knowledge: Optional[KnowledgeMatch] = None, task: Optional[Dict[str, Any]] = None, selected_documents: Optional[List[Dict[str, Any]]] = None, ) -> Dict[str, List[str]]: """Return blockers/warnings for a generated or edited customer reply. The validator catches the main risks for BLIF: wrong installation promise, invented stock/delivery guarantees, fiscal/IVA overclaims and unexpected prices. It is deliberately conservative for LLM-generated drafts, while still allowing operators to edit and retry. """ text = str(message_body or "").strip() low = _n(text) blockers: List[str] = [] warnings: List[str] = [] docs = selected_documents or [] # BLIF does not provide direct installation. Mentioning support or that an # electrician can install is OK; saying BLIF installs or installation is # included should be blocked unless explicitly from a document, which this # validator cannot prove. if re.search(r"\bblif\b.{0,60}\b(instala|monta|faz a instalacao|trata da instalacao)", low) and not _is_approved_installation_explanation(low): blockers.append("A resposta sugere que a BLIF presta instalação direta; contradiz a regra comercial aprovada.") if _states_installation_included(low): blockers.append("A resposta indica instalação incluída sem prova documental.") # Avoid stock promises from free text unless a stock system/doc explicitly exists. if any(phrase in low for phrase in ["temos stock garantido", "stock garantido", "disponibilidade garantida", "envio imediato garantido"]): blockers.append("A resposta promete stock/envio garantido sem validação externa.") # IVA: catalog prices are without VAT. Allow general explanation; warn when # text says a price includes VAT without selected document context. if "com iva" in low and "sem iva" not in low and not docs: warnings.append("A resposta menciona valores com IVA sem documento selecionado; confirma se está correto antes de enviar.") # Delivery promise must remain tied to payment confirmation. if ("2-3 dias" in low or "2 a 3 dias" in low) and "pagamento" not in low: warnings.append("Prazo de entrega deve ficar associado a confirmação de pagamento.") # Objective/document safety. When the operator selected an invoice, the # draft must be an invoice-delivery message, not a payment/proforma request. task_action = str((task or {}).get("action_code") or "").upper() selected_kinds = {str(doc.get("document_kind") or "").lower() for doc in docs} has_invoice_doc = "invoice" in selected_kinds if task_action == "SEND_INVOICE" and not has_invoice_doc: blockers.append("SEND_INVOICE requer uma fatura selecionada como anexo/contexto.") if task_action == "SEND_INVOICE" or has_invoice_doc: forbidden_invoice_delivery = [ "fatura sera emitida", "fatura sera enviada apos", "apos confirmacao do pagamento", "após confirmação do pagamento", "envie-nos o comprovativo", "envie o comprovativo", "para que possamos emitir a fatura", "para emitirmos a fatura", ] if any(_n(phrase) in low for phrase in forbidden_invoice_delivery): blockers.append("A resposta trata a fatura como ainda não emitida ou pede comprovativo indevido; para SEND_INVOICE deve dizer que a fatura segue em anexo.") if has_invoice_doc and "fatura" in low and "anexo" not in low and "em anexo" not in low: warnings.append("Há fatura selecionada; é recomendável mencionar que segue em anexo.") # In ClientFlow, SEND_PROFORMA means sending the Jasmin ORC.* as the # operational proforma. If such a document is selected, the draft must not # say that a formal proposal will be sent later; it is already selected. has_quotation_doc = "quotation" in selected_kinds if task_action == "SEND_PROFORMA": if not has_quotation_doc: blockers.append("SEND_PROFORMA requer orçamento Jasmin ORC.* selecionado como pró-forma.") if has_quotation_doc: forbidden_proforma_phrases = [ "fico ao dispor para enviar proposta formal", "posso enviar proposta formal", "enviaremos a proposta formal", "posteriormente enviaremos a proposta", ] if any(_n(phrase) in low for phrase in forbidden_proforma_phrases): blockers.append("A resposta diz que a proposta formal ainda será enviada, mas o ORC/pró-forma já está selecionado como anexo.") if "anexo" not in low and "em anexo" not in low: warnings.append("Há ORC/proforma selecionado; é recomendável mencionar que segue em anexo.") # Catch unexpected prices. This does not block known catalog values or values # that likely came from selected documents. known = _known_prices() doc_values = set() for doc in docs: for key in ["total_amount", "amount"]: val = doc.get(key) if val is not None: raw = str(val).replace(".", ",") doc_values.add(raw) doc_values.add(raw.split(",")[0]) for price in _PRICE_RE.findall(text): normalized = price.replace(".", ",") integer = normalized.split(",")[0] if normalized not in known and integer not in known and normalized not in doc_values and integer not in doc_values: warnings.append(f"Preço '{price} €' não foi reconhecido no catálogo/documentos selecionados.") # Topic-specific forbidden rules become warnings unless they are covered above. if knowledge is None: knowledge = retrieve_business_knowledge(text) for topic in knowledge.topics: for forbidden in topic.forbidden: f = _n(forbidden) if "nao prometer" in f or "nao afirmar" in f or "nao dizer" in f: # Already represented by the generic checks; show as context once. continue if topic.id == "installation_policy" and "instalacao" in low and "eletricista" not in low and "suporte remoto" not in low: warnings.append("Ao responder sobre instalação, é recomendável referir eletricista qualificado e suporte remoto BLIF.") # Deduplicate while preserving order. blockers = list(dict.fromkeys(blockers)) warnings = list(dict.fromkeys(warnings)) return {"blockers": blockers, "warnings": warnings}