mirror of
https://github.com/hoshikawa2/compass_backoffice.git
synced 2026-07-09 13:54:20 +00:00
344 lines
18 KiB
Python
344 lines
18 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Dict, List
|
|
import asyncio
|
|
|
|
from .clients import BackofficeRESTClient
|
|
from .settings import get_settings
|
|
|
|
|
|
def _jsonable(value: Any) -> Any:
|
|
if value is None or isinstance(value, (str, int, float, bool)):
|
|
return value
|
|
if isinstance(value, dict):
|
|
return {str(k): _jsonable(v) for k, v in value.items()}
|
|
if isinstance(value, (list, tuple, set)):
|
|
return [_jsonable(v) for v in value]
|
|
if hasattr(value, "model_dump"):
|
|
return _jsonable(value.model_dump())
|
|
if hasattr(value, "dict"):
|
|
return _jsonable(value.dict())
|
|
return str(value)
|
|
|
|
|
|
def _run_async(coro):
|
|
return asyncio.run(coro)
|
|
|
|
|
|
def _tim_clients_enabled() -> bool:
|
|
settings = get_settings()
|
|
return (not settings.use_mock) and settings.backend_type == "tim_clients"
|
|
|
|
|
|
def _tim_unavailable(reason: str) -> Dict[str, Any]:
|
|
return {"ok": False, "backend_type": "tim_clients", "source": "tim_clients", "reason": reason}
|
|
|
|
|
|
_RECLAMACOES: Dict[str, Dict[str, Any]] = {
|
|
"12345": {
|
|
"protocol_id": "12345",
|
|
"status": "em_analise",
|
|
"canal_origem": "ANATEL",
|
|
"motivo": "Contestação de cobrança em fatura",
|
|
"prioridade": "alta",
|
|
"prazo_sla_horas": 48,
|
|
"customer_key": "11999999999",
|
|
"contract_key": "3000131180",
|
|
"resumo": "Cliente questiona valor de serviço adicional na última fatura.",
|
|
},
|
|
"98765": {
|
|
"protocol_id": "98765",
|
|
"status": "pendente_cliente",
|
|
"canal_origem": "Ouvidoria",
|
|
"motivo": "Falha recorrente de atendimento",
|
|
"prioridade": "media",
|
|
"prazo_sla_horas": 72,
|
|
"customer_key": "11888888888",
|
|
"contract_key": "3000999999",
|
|
"resumo": "Cliente solicita retorno formal sobre atendimento anterior.",
|
|
},
|
|
}
|
|
|
|
_CLIENTES: Dict[str, Dict[str, Any]] = {
|
|
"11999999999": {
|
|
"customer_key": "11999999999",
|
|
"nome": "Cliente Exemplo TIM",
|
|
"segmento": "consumer",
|
|
"plano": "TIM Controle Exemplo",
|
|
"contratos": ["3000131180"],
|
|
"flags": ["possui_reclamacao_aberta", "nao_ofertar_sem_contexto"],
|
|
"ultimas_interacoes": [
|
|
{"data": "2026-06-01", "canal": "URA", "resumo": "Consulta de fatura."},
|
|
{"data": "2026-06-03", "canal": "Backoffice", "resumo": "Abertura de análise."},
|
|
],
|
|
},
|
|
"11888888888": {
|
|
"customer_key": "11888888888",
|
|
"nome": "Cliente Exemplo Ouvidoria",
|
|
"segmento": "consumer",
|
|
"plano": "TIM Pós Exemplo",
|
|
"contratos": ["3000999999"],
|
|
"flags": ["ouvidoria"],
|
|
"ultimas_interacoes": [],
|
|
},
|
|
}
|
|
|
|
_ACOES: List[Dict[str, Any]] = []
|
|
|
|
|
|
def _backend_error(exc: Exception) -> Dict[str, Any]:
|
|
settings = get_settings()
|
|
if settings.fail_open_on_backend_error:
|
|
return {"ok": False, "backend_error": str(exc), "backend_type": settings.backend_type}
|
|
raise exc
|
|
|
|
|
|
def consultar_reclamacao(protocol_id: str | None, customer_key: str | None = None, interaction_key: str | None = None) -> Dict[str, Any]:
|
|
settings = get_settings()
|
|
if not settings.use_mock and settings.backend_type == "rest":
|
|
try:
|
|
return BackofficeRESTClient(settings).get_json("reclamacoes", {"protocol_id": protocol_id, "customer_key": customer_key, "interaction_key": interaction_key})
|
|
except Exception as exc:
|
|
return _backend_error(exc)
|
|
if not settings.use_mock and settings.backend_type == "oracle":
|
|
return _backend_error(RuntimeError("Backend Oracle ainda precisa receber a query/procedure real do backoffice original"))
|
|
|
|
key = protocol_id or interaction_key
|
|
if not key:
|
|
return {"found": False, "reason": "protocol_id ausente"}
|
|
data = _RECLAMACOES.get(str(key))
|
|
if not data:
|
|
return {"found": False, "protocol_id": key, "reason": "reclamacao_nao_encontrada"}
|
|
if customer_key and data.get("customer_key") != customer_key:
|
|
return {"found": False, "protocol_id": key, "reason": "protocolo_nao_pertence_ao_cliente"}
|
|
return {"found": True, **data}
|
|
|
|
|
|
def consultar_cliente_backoffice(customer_key: str | None = None, contract_key: str | None = None, session_key: str | None = None, cpf: str | None = None, cnpj: str | None = None, document: str | None = None, document_type: str | None = None) -> Dict[str, Any]:
|
|
customer_key = customer_key or cpf or cnpj or document
|
|
settings = get_settings()
|
|
if _tim_clients_enabled():
|
|
return consultar_imdb_cliente(customer_key=customer_key, contract_key=contract_key, session_key=session_key)
|
|
if not settings.use_mock and settings.backend_type == "rest":
|
|
try:
|
|
return BackofficeRESTClient(settings).get_json("clientes", {"customer_key": customer_key, "contract_key": contract_key, "session_key": session_key})
|
|
except Exception as exc:
|
|
return _backend_error(exc)
|
|
if not settings.use_mock and settings.backend_type == "oracle":
|
|
return _backend_error(RuntimeError("Backend Oracle ainda precisa receber a query/procedure real do backoffice original"))
|
|
|
|
if not customer_key:
|
|
return {"found": False, "reason": "customer_key ausente"}
|
|
data = _CLIENTES.get(str(customer_key))
|
|
if not data:
|
|
return {"found": False, "customer_key": customer_key, "reason": "cliente_nao_encontrado"}
|
|
if contract_key and contract_key not in data.get("contratos", []):
|
|
return {"found": True, "warning": "contrato_nao_localizado_para_cliente", **data}
|
|
return {"found": True, "session_key": session_key, **data}
|
|
|
|
|
|
def registrar_acao_backoffice(protocol_id: str | None, action_text: str | None, operator_session: str | None = None) -> Dict[str, Any]:
|
|
settings = get_settings()
|
|
if not settings.use_mock and settings.backend_type == "rest":
|
|
try:
|
|
return BackofficeRESTClient(settings).post_json("acoes", {"protocol_id": protocol_id, "action_text": action_text, "operator_session": operator_session})
|
|
except Exception as exc:
|
|
return _backend_error(exc)
|
|
if not settings.use_mock and settings.backend_type == "oracle":
|
|
return _backend_error(RuntimeError("Backend Oracle ainda precisa receber a query/procedure real do backoffice original"))
|
|
|
|
if not protocol_id:
|
|
return {"registered": False, "reason": "protocol_id ausente"}
|
|
if not action_text:
|
|
return {"registered": False, "reason": "action_text ausente"}
|
|
action = {
|
|
"id": len(_ACOES) + 1,
|
|
"protocol_id": str(protocol_id),
|
|
"action_text": action_text,
|
|
"operator_session": operator_session,
|
|
"created_at": datetime.now(timezone.utc).isoformat(),
|
|
"status": "registrado",
|
|
}
|
|
_ACOES.append(action)
|
|
return {"registered": True, "action": action}
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tools adicionais reintroduzidas da branch funcional do backoffice original.
|
|
# Elas mantêm o mesmo padrão mock/rest/oracle do servidor convertido. Em modo
|
|
# REST, os endpoints esperados são configurados no serviço corporativo:
|
|
# /siebel/cases, /imdb/customer, /speech/history, /tais-kb/search, etc.
|
|
# Em modo Oracle, os pontos ficam preparados para procedures/queries reais.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _rest_or_mock(endpoint: str, params: Dict[str, Any], mock_payload: Dict[str, Any]) -> Dict[str, Any]:
|
|
settings = get_settings()
|
|
if not settings.use_mock and settings.backend_type == "rest":
|
|
try:
|
|
return BackofficeRESTClient(settings).get_json(endpoint, params)
|
|
except Exception as exc:
|
|
return _backend_error(exc)
|
|
if not settings.use_mock and settings.backend_type == "oracle":
|
|
return _backend_error(RuntimeError(f"Backend Oracle pendente para endpoint lógico {endpoint}"))
|
|
if not settings.use_mock and settings.backend_type == "tim_clients":
|
|
return _tim_unavailable(f"Tool {endpoint} ainda não tem client TIM direto mapeado; use tool específica ou configure BACKOFFICE_MCP_BACKEND_TYPE=rest com BACKOFFICE_MCP_REST_BASE_URL")
|
|
return mock_payload
|
|
|
|
|
|
def consultar_siebel_caso(protocol_id: str | None = None, interaction_key: str | None = None, customer_key: str | None = None) -> Dict[str, Any]:
|
|
case_id = protocol_id or interaction_key or "12345"
|
|
base = consultar_reclamacao(case_id, customer_key=customer_key, interaction_key=interaction_key)
|
|
mock = {
|
|
"found": bool(base.get("found")),
|
|
"source": "mock_siebel",
|
|
"case_id": case_id,
|
|
"sr_status": "open" if base.get("found") else "not_found",
|
|
"payload": base,
|
|
}
|
|
return _rest_or_mock("siebel/cases", {"protocol_id": protocol_id, "interaction_key": interaction_key, "customer_key": customer_key}, mock)
|
|
|
|
|
|
def registrar_acao_siebel(protocol_id: str | None = None, action_text: str | None = None, operator_session: str | None = None) -> Dict[str, Any]:
|
|
settings = get_settings()
|
|
payload = {"protocol_id": protocol_id, "action_text": action_text, "operator_session": operator_session}
|
|
if not settings.use_mock and settings.backend_type == "rest":
|
|
try:
|
|
return BackofficeRESTClient(settings).post_json("siebel/actions", payload)
|
|
except Exception as exc:
|
|
return _backend_error(exc)
|
|
if not settings.use_mock and settings.backend_type == "oracle":
|
|
return _backend_error(RuntimeError("Backend Oracle pendente para registrar_acao_siebel"))
|
|
result = registrar_acao_backoffice(protocol_id, action_text, operator_session)
|
|
return {"source": "mock_siebel", "siebel_registered": result.get("registered", False), **result}
|
|
|
|
|
|
def consultar_imdb_cliente(customer_key: str | None = None, contract_key: str | None = None, session_key: str | None = None, cpf: str | None = None, cnpj: str | None = None, document: str | None = None, document_type: str | None = None) -> Dict[str, Any]:
|
|
customer_key = customer_key or cpf or cnpj or document
|
|
settings = get_settings()
|
|
if _tim_clients_enabled():
|
|
if not customer_key:
|
|
return {"found": False, "ok": False, "source": "tim_imdb", "reason": "customer_key/msisdn ausente"}
|
|
try:
|
|
from src.components.clients.imdb_client import ImdbClient
|
|
from src.core.config import settings as app_settings
|
|
data, http_meta = _run_async(ImdbClient().get_imdb_access_data_with_retry(str(customer_key), app_settings.PMID_API_CLIENT_ID))
|
|
return {
|
|
"ok": True,
|
|
"found": data is not None,
|
|
"source": "tim_imdb",
|
|
"customer_key": customer_key,
|
|
"contract_key": contract_key,
|
|
"session_key": session_key,
|
|
"data": _jsonable(data),
|
|
"http_meta": _jsonable(http_meta),
|
|
}
|
|
except Exception as exc:
|
|
return _backend_error(exc)
|
|
data = consultar_cliente_backoffice(customer_key, contract_key, session_key)
|
|
mock = {"source": "mock_imdb", "pmid": f"PMID-{customer_key or 'UNKNOWN'}", **data}
|
|
return _rest_or_mock("imdb/customer", {"customer_key": customer_key, "contract_key": contract_key, "session_key": session_key}, mock)
|
|
|
|
|
|
def consultar_speech_analytics(protocol_id: str | None = None, customer_key: str | None = None, interaction_key: str | None = None, cpf: str | None = None, cnpj: str | None = None, document: str | None = None, document_type: str | None = None) -> Dict[str, Any]:
|
|
customer_key = customer_key or cpf or cnpj or document
|
|
if _tim_clients_enabled():
|
|
if not customer_key:
|
|
return {"found": False, "ok": False, "source": "tim_speech_analytics", "reason": "customer_key/msisdn ausente"}
|
|
# A API de histórico precisa de CPF/CNPJ. Se o fluxo ainda não trouxe via IMDB, retornamos erro controlado.
|
|
social_sec_no = None
|
|
try:
|
|
imdb = consultar_imdb_cliente(customer_key=customer_key)
|
|
social_sec_no = (((imdb.get("data") or {}).get("socialSecNo")) if isinstance(imdb, dict) else None)
|
|
except Exception:
|
|
social_sec_no = None
|
|
if not social_sec_no:
|
|
return {"found": False, "ok": False, "source": "tim_speech_analytics", "reason": "socialSecNo ausente; consulte IMDB/PMID antes ou envie cpf_cnpj mascarado/canônico"}
|
|
try:
|
|
from src.components.clients.speech_analytics_client import SpeechAnalyticsClient
|
|
history, http_meta = _run_async(SpeechAnalyticsClient().get_history(str(social_sec_no), str(customer_key)))
|
|
return {"ok": True, "found": bool(history), "source": "tim_speech_analytics", "protocol_id": protocol_id or interaction_key, "customer_key": customer_key, "history": _jsonable(history), "http_meta": _jsonable(http_meta)}
|
|
except Exception as exc:
|
|
return _backend_error(exc)
|
|
mock = {
|
|
"found": True,
|
|
"source": "mock_speech_analytics",
|
|
"protocol_id": protocol_id or interaction_key,
|
|
"customer_key": customer_key,
|
|
"summary": "Histórico de fala simulado: cliente questionou cobrança e solicitou retorno formal.",
|
|
"sentiment": "neutral",
|
|
"events": [],
|
|
}
|
|
return _rest_or_mock("speech/history", {"protocol_id": protocol_id, "customer_key": customer_key, "interaction_key": interaction_key}, mock)
|
|
|
|
|
|
def consultar_tais_kb(query: str | None = None, protocol_id: str | None = None, customer_key: str | None = None) -> Dict[str, Any]:
|
|
if _tim_clients_enabled():
|
|
q = query or protocol_id or customer_key
|
|
if not q:
|
|
return {"found": False, "ok": False, "source": "tim_tais_kb", "reason": "query/protocol_id/customer_key ausente"}
|
|
try:
|
|
from src.components.clients.tais_kb_client import TaisKbClient, Product
|
|
from src.core.config import settings as app_settings
|
|
result = _run_async(TaisKbClient().search_documents(str(q), product=Product.MOVEL, top_k=int(app_settings.TAIS_TOP_K), preprocess=True, postprocess=True))
|
|
return {"ok": True, "found": True, "source": "tim_tais_kb", "query": q, "result": _jsonable(result)}
|
|
except Exception as exc:
|
|
return _backend_error(exc)
|
|
mock = {
|
|
"found": True,
|
|
"source": "mock_tais_kb",
|
|
"query": query or protocol_id or customer_key,
|
|
"documents": [
|
|
{"id": "KB-ANATEL-001", "title": "Tratativa de contestação de cobrança", "score": 0.91},
|
|
{"id": "TPL-RESP-001", "title": "Template de resposta formal ANATEL", "score": 0.87},
|
|
],
|
|
"templates": [
|
|
{"id": "TPL-RESP-001", "text": "Prezado cliente, analisamos sua solicitação e registramos a tratativa..."}
|
|
],
|
|
}
|
|
return _rest_or_mock("tais-kb/search", {"query": query, "protocol_id": protocol_id, "customer_key": customer_key}, mock)
|
|
|
|
|
|
def consultar_abrt(customer_key: str | None = None, protocol_id: str | None = None, social_sec_no: str | None = None) -> Dict[str, Any]:
|
|
if _tim_clients_enabled():
|
|
if not customer_key or not social_sec_no:
|
|
return {"found": False, "ok": False, "source": "tim_abrt", "reason": "customer_key/msisdn e social_sec_no são obrigatórios para ABRT"}
|
|
try:
|
|
from src.components.clients.abrt_client import AbrtClient
|
|
data, http_meta = _run_async(AbrtClient().get_abrt_data_with_retry(str(social_sec_no), str(customer_key)))
|
|
return {"ok": True, "found": data is not None, "source": "tim_abrt", "customer_key": customer_key, "protocol_id": protocol_id, "data": _jsonable(data), "http_meta": _jsonable(http_meta)}
|
|
except Exception as exc:
|
|
return _backend_error(exc)
|
|
mock = {"found": True, "source": "mock_abrt", "customer_key": customer_key, "protocol_id": protocol_id, "has_open_case": False}
|
|
return _rest_or_mock("abrt/status", {"customer_key": customer_key, "protocol_id": protocol_id}, mock)
|
|
|
|
|
|
def consultar_portabilidade(customer_key: str | None = None, contract_key: str | None = None, social_sec_no: str | None = None) -> Dict[str, Any]:
|
|
if _tim_clients_enabled():
|
|
if not customer_key or not social_sec_no:
|
|
return {"found": False, "ok": False, "source": "tim_portability", "reason": "customer_key/msisdn e social_sec_no são obrigatórios para Portabilidade"}
|
|
try:
|
|
from src.components.clients.portability_client import PortabilityClient
|
|
data, http_meta = _run_async(PortabilityClient().get_portability_history_with_retry(str(social_sec_no), str(customer_key)))
|
|
return {"ok": True, "found": bool(data), "source": "tim_portability", "customer_key": customer_key, "contract_key": contract_key, "data": _jsonable(data), "http_meta": _jsonable(http_meta)}
|
|
except Exception as exc:
|
|
return _backend_error(exc)
|
|
mock = {"found": True, "source": "mock_portability", "customer_key": customer_key, "contract_key": contract_key, "portability_status": "none"}
|
|
return _rest_or_mock("portability/status", {"customer_key": customer_key, "contract_key": contract_key}, mock)
|
|
|
|
|
|
def buscar_templates_emulador(protocol_id: str | None = None, query: str | None = None) -> Dict[str, Any]:
|
|
kb = consultar_tais_kb(query=query, protocol_id=protocol_id)
|
|
return {"source": "mock_emulator_rag", "found": True, "templates": kb.get("templates", []), "documents": kb.get("documents", [])}
|
|
|
|
|
|
def gerar_rascunho_emulador(protocol_id: str | None = None, selected_actions: list | None = None, operator_instructions: str | None = None) -> Dict[str, Any]:
|
|
actions = selected_actions or []
|
|
text = operator_instructions or "Rascunho gerado a partir das evidências disponíveis."
|
|
return {
|
|
"draft_created": True,
|
|
"protocol_id": protocol_id,
|
|
"selected_actions": actions,
|
|
"draft": f"Resposta formal ao cliente: {text}",
|
|
"status": "draft",
|
|
}
|