import uuid from dataclasses import dataclass from django.conf import settings from chat.knowledge.asf_faq_ptbr import ASF_FAQ_LITE, try_asf_canned_answer from chat.knowledge.fianca_faq_ptbr import ( FIANCA_FAQ_COMPACT, FIANCA_FAQ_LITE, try_fianca_canned_answer, ) from chat.knowledge.modulos_guia_ptbr import try_modulos_guia_canned_answer from chat.knowledge.operacional_faq_ptbr import try_operacional_canned_answer from chat.services.openai_service import OpenAIChatService @dataclass(frozen=True) class ChatResponse: request_id: str answer_text: str conversation_id: str model: str status: str error: dict | None usage: dict | None class ChatService: def __init__(self, openai_service=None): self.openai_service = openai_service def send_message( self, *, message, request_id=None, source=None, tenant=None, actor=None, context=None, conversation=None, case_data=None, options=None, conversation_id=None, ): request_id = request_id or str(uuid.uuid4()) normalized_conversation_id = conversation_id or str(uuid.uuid4()) case_data = case_data or {} conversation = conversation or {} # Resposta local (sem tokens OpenAI): FAQ frequente, sem histórico e sem dados injetados pelo servidor. if not self._case_data_has_server_lookup(case_data): history = conversation.get("history") or [] if not history: canned = try_asf_canned_answer(message or "") if not canned: canned = try_fianca_canned_answer(message or "") if not canned: canned = try_operacional_canned_answer(message or "") if not canned: canned = try_modulos_guia_canned_answer(message or "") if canned: return ChatResponse( request_id=request_id, answer_text=canned, conversation_id=normalized_conversation_id, model="canned-faq", status="ok", error=None, usage={"input_tokens": 0, "output_tokens": 0}, ) openai_service = self.openai_service or OpenAIChatService() messages = self._build_messages( source=source, tenant=tenant, actor=actor, context=context, conversation=conversation, case_data=case_data, message=message, ) result = openai_service.generate_reply(messages=messages, options=options) answer_text = result.content.strip() if not answer_text: answer_text = ( "Não consegui gerar uma resposta agora. " "Tente novamente em instantes." ) return ChatResponse( request_id=request_id, answer_text=answer_text, conversation_id=normalized_conversation_id, model=result.model, status="ok", error=None, usage={ "input_tokens": result.input_tokens, "output_tokens": result.output_tokens, }, ) def _build_messages( self, *, source, tenant, actor, context, conversation, case_data, message, ): conversation = conversation or {} cd = case_data or {} messages = [{"role": "system", "content": self._build_system_prompt(cd)}] for item in conversation.get("history", []): messages.append({"role": item["role"], "content": item["text"]}) payload = { "source": source or "unknown", "tenant": tenant or {}, "actor": actor or {}, "context": context or {}, "case_data": cd, "user_message": message, } messages.append({"role": "user", "content": self._build_user_prompt(payload)}) return messages def _case_data_has_server_lookup(self, case_data): """Resumo ou lookup injetado pelo PHP (não confiar em case_data só do cliente).""" if not isinstance(case_data, dict): return False if case_data.get("resumo_consulta_seguros_vigentes") or case_data.get( "resumo_consulta_fianca" ): return True if case_data.get("resumo_consulta_operacional"): return True if case_data.get("resumo_consulta_codigo_referencia"): return True if case_data.get("lookup_seguros_vigentes") or case_data.get("fianca_lookup"): return True if case_data.get("lookup_operacional"): return True if case_data.get("lookup_codigo_referencia"): return True if case_data.get("resumo_consulta_contagens") or case_data.get( "lookup_contagens" ): return True return False def _build_system_prompt(self, case_data=None): case_data = case_data or {} faq_block = ( f"{ASF_FAQ_LITE}\n\n{FIANCA_FAQ_COMPACT}" if self._case_data_has_server_lookup(case_data) else f"{ASF_FAQ_LITE}\n\n{FIANCA_FAQ_LITE}" ) return ( f"{settings.CHAT_SYSTEM_PROMPT}\n\n" "Voce atende usuarios leigos (imobiliarias, locatarios, equipe operacional). " "Use portugues do Brasil, frases curtas, tom cordial. " "Para duvidas sobre a ASF, produtos (fiança, titulo de capitalizacao, incendio), processo e atendimento, " "use o FAQ abaixo como base; seja breve (paragrafo ou dois), sem listas longas, a menos que o usuario peca detalhes. " "Para perguntas sobre seguro fiança (produto, documentos, status, pagamento, cancelamento, etc.), " "use o FAQ abaixo como base; seja breve (paragrafo ou dois), sem listas longas, a menos que o usuario peca detalhes. " "Nunca mencione: JSON, payloads, case_data, fianca_lookup, lookup_seguros_vigentes, itens_json, " "nomes de campos, APIs, banco de dados, Django ou detalhes de implementacao. " "Prioridade: se existir case_data.resumo_consulta_operacional (faturamento/sinistro/administrativo), use esse. " "Senao, se existir case_data.resumo_consulta_codigo_referencia (codigo AN/IN/CAP/FAT/SIN), use esse. " "Senao, se existir case_data.resumo_consulta_contagens (totais por status no escopo do usuario), use esse. " "Senao, se existir case_data.resumo_consulta_seguros_vigentes, use esse texto " "(consulta unificada: fiança, incendio, titulo de capitalizacao). Senao, se existir " "case_data.resumo_consulta_fianca, use esse. Responda com base nesses resumos em linguagem humana. " "Se nao houver resumo nem dados uteis no contexto, diga de forma simples que " "nao encontrou informacao e o que a pessoa pode fazer (sem pedir 'dados tecnicos'). " "Nunca invente numeros de apolice, CPF completo ou vigencias.\n\n" f"{faq_block}" ) def _build_user_prompt(self, payload): case_data = (payload or {}).get("case_data") or {} extra = "" if isinstance(case_data, dict) and case_data.get("resumo_consulta_operacional"): extra = ( "\n\nPrioridade maxima: resumo_consulta_operacional (faturamento/sinistro/administrativo). " "Responda com base nele, em linguagem natural; nao peca dados tecnicos. " "Evite promessas vagas e finalize com proximo passo pratico.\n" ) elif isinstance(case_data, dict) and case_data.get( "resumo_consulta_codigo_referencia" ): extra = ( "\n\nPrioridade maxima: resumo_consulta_codigo_referencia (status por codigo AN/IN/CAP/FAT/SIN). " "Responda com base nele, em linguagem natural; nao peca dados tecnicos. " "Nao use frases como 'posso verificar novamente' ou 'consultar de novo' — os dados ja sao desta consulta. " "Finalize com orientacao pratica (proximo passo no sistema ou contato imobiliaria/seguradora), sem promessas vagas.\n" ) elif isinstance(case_data, dict) and case_data.get( "resumo_consulta_contagens" ): extra = ( "\n\nPrioridade maxima: resumo_consulta_contagens (totais agregados no escopo do usuario). " "Use apenas os numeros e rotulos do resumo; nao invente contagens. " "Se o usuario perguntou por um dia especifico, respeite a nota do resumo sobre filtro de data nos vigentes.\n" ) elif isinstance(case_data, dict) and case_data.get( "resumo_consulta_seguros_vigentes" ): extra = ( "\n\nPrioridade maxima: resumo_consulta_seguros_vigentes (consulta unificada no sistema). " "Responda com base nele, em linguagem natural; nao peca dados tecnicos.\n" ) elif isinstance(case_data, dict) and case_data.get("resumo_consulta_fianca"): extra = ( "\n\nPrioridade: o texto em resumo_consulta_fianca e o resultado oficial da consulta. " "Responda com base nele, em linguagem natural; nao peca dados tecnicos.\n" ) return ( "Use o contexto abaixo para responder a pergunta do usuario.\n" "Priorize precisao e linguagem acessivel.\n" f"{extra}\n" f"{self._to_compact_json(payload)}" ) def _to_compact_json(self, payload): import json return json.dumps(payload, ensure_ascii=False, separators=(",", ":"), sort_keys=True)