306 lines
13 KiB
Python
306 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from .client import TimApiClient
|
|
|
|
|
|
class ContasDomainService:
|
|
"""Business services for Contas.
|
|
|
|
Important: this is *not* an agent runtime or workflow engine. Conversation,
|
|
confirmation, clarification, checkpoints and branching remain in LangGraph/
|
|
agent_framework_oci. These methods implement only TIM business operations.
|
|
"""
|
|
|
|
def __init__(self, client: TimApiClient | None = None) -> None:
|
|
self.client = client or TimApiClient()
|
|
|
|
@staticmethod
|
|
def _products(payload: Any) -> list[dict[str, Any]]:
|
|
if not isinstance(payload, dict):
|
|
return []
|
|
value = payload.get("products") or payload.get("services") or []
|
|
return [x for x in value if isinstance(x, dict)]
|
|
|
|
@staticmethod
|
|
def _match_service(products: list[dict[str, Any]], subject: str) -> dict[str, Any] | None:
|
|
needle = (subject or "").strip().lower()
|
|
if not needle:
|
|
return None
|
|
exact = [p for p in products if str(p.get("name") or p.get("description") or "").strip().lower() == needle]
|
|
if exact:
|
|
return exact[0]
|
|
partial = [p for p in products if needle in str(p.get("name") or p.get("description") or "").lower() or str(p.get("name") or p.get("description") or "").lower() in needle]
|
|
return partial[0] if partial else None
|
|
|
|
def consultar_faturas(self, *, msisdn: str, **_: Any) -> Any:
|
|
return self.client.consultar_faturas(msisdn)
|
|
|
|
def consultar_plano(self, *, msisdn: str, **args: Any) -> dict[str, Any]:
|
|
"""Retorna somente os planos presentes na evidência de billing analysis.
|
|
|
|
A consulta é deliberadamente determinística: não usa VAS, RAG nem o
|
|
workflow de invoice_explanation para responder qual é o plano do cliente.
|
|
"""
|
|
billing = args.get("billing_analysis")
|
|
if not isinstance(billing, dict):
|
|
billing = self.client.billing_analysis(
|
|
msisdn,
|
|
invoice_id=args.get("invoice_id"),
|
|
customer_id=args.get("customer_id"),
|
|
)
|
|
|
|
plans: list[dict[str, Any]] = []
|
|
for section in billing.get("currentInvoice") or [] if isinstance(billing, dict) else []:
|
|
if not isinstance(section, dict):
|
|
continue
|
|
section_type = str(section.get("type") or "").strip().casefold()
|
|
section_desc = str(section.get("desc") or "").strip().casefold()
|
|
if section_type not in {"plano", "planos"} and section_desc not in {"plano", "planos"}:
|
|
continue
|
|
for item in section.get("items") or []:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
name = str(item.get("desc") or item.get("name") or "").strip()
|
|
if not name:
|
|
continue
|
|
plans.append({
|
|
"name": name,
|
|
"value": item.get("value"),
|
|
"type": str(item.get("type") or section.get("type") or "plano"),
|
|
"contestable": item.get("contestable"),
|
|
})
|
|
|
|
return {
|
|
"plans": plans,
|
|
"count": len(plans),
|
|
"source": "billing_analysis.currentInvoice",
|
|
}
|
|
|
|
def invoice_explanation(self, *, msisdn: str, **args: Any) -> dict[str, Any]:
|
|
# Reuse framework-prefetched evidence when available; the domain never owns
|
|
# a second cache/session implementation. Explanation itself is produced by
|
|
# the framework LLM.
|
|
complete = args.get("complete_invoices_payload")
|
|
if not isinstance(complete, dict):
|
|
complete = self.client.consultar_faturas(msisdn)
|
|
billing = args.get("billing_analysis")
|
|
if not isinstance(billing, dict):
|
|
billing = self.client.billing_analysis(msisdn, invoice_id=args.get("invoice_id"), customer_id=args.get("customer_id"))
|
|
return {
|
|
"complete_invoices": complete,
|
|
"billing_analysis": billing,
|
|
"instruction": "Use estes dados como evidência para explicar composição/variação da fatura. Não invente cobranças ausentes.",
|
|
}
|
|
|
|
def consultar_vas(self, *, msisdn: str, **_: Any) -> Any:
|
|
return self.client.consultar_vas(msisdn)
|
|
|
|
def consultar_historico_vas(self, *, msisdn: str, **_: Any) -> Any:
|
|
return self.client.historico_vas(msisdn)
|
|
|
|
@staticmethod
|
|
def _history_products(payload: Any) -> list[dict[str, Any]]:
|
|
if not isinstance(payload, dict):
|
|
return []
|
|
services = payload.get("services") or payload.get("products") or []
|
|
out: list[dict[str, Any]] = []
|
|
for item in services if isinstance(services, list) else []:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
row = dict(item)
|
|
row.setdefault("name", row.get("description") or row.get("billDescription") or "")
|
|
row.setdefault("cspId", row.get("csp_id"))
|
|
row.setdefault("appId", row.get("app_id"))
|
|
if "can" not in row:
|
|
row["can"] = {"cancel": bool(row.get("canCancel", False))}
|
|
out.append(row)
|
|
return out
|
|
|
|
@staticmethod
|
|
def _service_can_cancel(service: dict[str, Any]) -> bool:
|
|
can = service.get("can") if isinstance(service.get("can"), dict) else {}
|
|
details = service.get("details") if isinstance(service.get("details"), dict) else {}
|
|
details_can = details.get("can") if isinstance(details.get("can"), dict) else {}
|
|
values = [
|
|
can.get("cancel"),
|
|
details_can.get("cancel"),
|
|
service.get("canCancel"),
|
|
service.get("can_cancel"),
|
|
]
|
|
explicit = [value for value in values if value is not None]
|
|
return bool(explicit[0]) if explicit else True
|
|
|
|
def cancelar_vas_avulso(self, *, msisdn: str, subject: str, protocol: str = "", **_: Any) -> dict[str, Any]:
|
|
current = self.client.consultar_vas(msisdn)
|
|
active_products = self._products(current)
|
|
service = self._match_service(active_products, subject)
|
|
source = "active"
|
|
history = None
|
|
if not service:
|
|
history = self.client.historico_vas(msisdn)
|
|
history_products = self._history_products(history)
|
|
service = self._match_service(history_products, subject)
|
|
source = "history" if service else "none"
|
|
if not service:
|
|
return {
|
|
"success": False,
|
|
"error": f"Serviço '{subject}' não encontrado na consulta VAS nem no histórico",
|
|
"reason": "service_not_found",
|
|
"services": active_products,
|
|
"history": history,
|
|
"evidence_source": source,
|
|
}
|
|
if not self._service_can_cancel(service):
|
|
return {
|
|
"success": False,
|
|
"error": "Serviço já se encontra cancelado, não sendo possível realizar recancelamento.",
|
|
"reason": "already_inactive_or_blocked",
|
|
"service": service,
|
|
"evidence_source": source,
|
|
}
|
|
try:
|
|
block = self.client.bloquear_vas(msisdn, service)
|
|
except Exception as exc:
|
|
return {
|
|
"success": False,
|
|
"service": service,
|
|
"error": str(exc),
|
|
"reason": "block_vas_failed",
|
|
"evidence_source": source,
|
|
"eligible_for_contestation": True,
|
|
}
|
|
try:
|
|
cancel = self.client.cancelar_vas(msisdn, service, protocol=protocol)
|
|
except Exception as exc:
|
|
return {
|
|
"success": False,
|
|
"service": service,
|
|
"block": block,
|
|
"error": str(exc),
|
|
"reason": "cancel_vas_failed",
|
|
"evidence_source": source,
|
|
"eligible_for_contestation": True,
|
|
}
|
|
return {
|
|
"success": True,
|
|
"service": service,
|
|
"block": block,
|
|
"cancellation": cancel,
|
|
"evidence_source": source,
|
|
"eligible_for_contestation": True,
|
|
}
|
|
|
|
def tratar_vas_estrategico(self, *, msisdn: str, subject: str, accepted_explanation: bool | None = None, **_: Any) -> dict[str, Any]:
|
|
current = self.client.consultar_vas(msisdn)
|
|
service = self._match_service(self._products(current), subject)
|
|
return {
|
|
"success": bool(service),
|
|
"service": service,
|
|
"accepted_explanation": accepted_explanation,
|
|
"guidance": "VAS estratégico/bundle usa orientação conversacional do agente; não executar cancelamento automático sem tool transacional específica e confirmação do framework.",
|
|
}
|
|
|
|
def contestar_cobranca(self, *, msisdn: str, subject: str, valor: Any, motivo: str = "", **args: Any) -> dict[str, Any]:
|
|
# Domain orchestration only: all conversational confirmation has already happened in framework runtime.
|
|
invoices = self.client.consultar_faturas(msisdn)
|
|
contract = self.client.contrato(msisdn)
|
|
profile = self.client.profile_full(msisdn)
|
|
protocol_payload = {
|
|
"msisdn": msisdn,
|
|
"source": os_source(args),
|
|
"reason1": "CONTESTACAO",
|
|
"reason2": subject,
|
|
"reason3": motivo,
|
|
"status": "OPENED",
|
|
"requestStatus": "Aberto",
|
|
}
|
|
protocol = self.client.abrir_protocolo(protocol_payload)
|
|
protocol_number = str(protocol.get("interactionProtocol") or protocol.get("protocolNumber") or args.get("protocol") or "") if isinstance(protocol, dict) else ""
|
|
customer = (((invoices or {}).get("billingProfile") or {}).get("customer") or {}) if isinstance(invoices, dict) else {}
|
|
contest_payload = {
|
|
"msisdn": msisdn,
|
|
"sr": protocol_number,
|
|
"socialSecNo": customer.get("document") or args.get("social_sec_no") or "",
|
|
"customerId": customer.get("customerId") or customer.get("id") or args.get("customer_id") or "",
|
|
"invoiceNumber": args.get("invoice_id") or "",
|
|
"userId": "AIAGENTCR",
|
|
"items": [{"itemName": subject, "itemType": "VAS_AVULSO", "claimedAmount": str(valor), "validatedAmount": str(valor)}],
|
|
"description": motivo,
|
|
}
|
|
contestation = self.client.contestar(contest_payload)
|
|
tracking = self.client.tracking({"msisdn": msisdn, "protocolNumber": protocol_number, "activityType": "Contestação", "activityStatus": "Aberto"})
|
|
return {"success": True, "protocol": protocol, "contestation": contestation, "tracking": tracking, "invoices": invoices, "contract": contract, "profile": profile}
|
|
|
|
def consultar_status_solicitacao(self, *, msisdn: str = "", protocol: str = "", **_: Any) -> Any:
|
|
return self.client.status_sr({"msisdn": msisdn, "protocolNumber": protocol, "status": "CONSULTA", "channel": "AIAGENTCR"})
|
|
|
|
def enviar_sms(self, *, msisdn: str, message: str, **_: Any) -> Any:
|
|
return self.client.sms(msisdn, message)
|
|
|
|
def buscar_fatura_detalhada(
|
|
self,
|
|
*,
|
|
msisdn: str,
|
|
invoice_id: str,
|
|
customer_id: str = "",
|
|
include_danfe: bool = False,
|
|
output: str = "",
|
|
**_: Any,
|
|
) -> Any:
|
|
return self.client.bill_pdf(
|
|
msisdn, invoice_id, customer_id, include_danfe=include_danfe, output=output
|
|
)
|
|
|
|
def perfil_fatura(self, *, msisdn: str, **_: Any) -> Any:
|
|
return self.client.profile_bill(msisdn)
|
|
|
|
def info_linha(self, *, msisdn: str, **_: Any) -> Any:
|
|
return self.client.line_info(msisdn)
|
|
|
|
def recuperar_fatura_pdf(self, *, msisdn: str, invoice_id: str, customer_id: str = "", **_: Any) -> Any:
|
|
return self.client.secure_pdf(msisdn, invoice_id, customer_id)
|
|
|
|
@staticmethod
|
|
def _normalize_final_status(status: str) -> str:
|
|
valid = {
|
|
"resolvido", "nao_resolvido", "resolvido_outros_assuntos",
|
|
"outros_assuntos", "erro_falha_sistema", "erro_no_match", "erro_no_input",
|
|
}
|
|
aliases = {
|
|
"0": "resolvido", "1": "nao_resolvido",
|
|
"2": "resolvido_outros_assuntos", "3": "outros_assuntos",
|
|
"final": "resolvido",
|
|
}
|
|
value = str(status or "").strip().lower()
|
|
value = aliases.get(value, value)
|
|
return value if value in valid else "erro_falha_sistema"
|
|
|
|
@staticmethod
|
|
def _normalize_final_summary(summary: str) -> str:
|
|
value = str(summary or "").strip()
|
|
return "Encerramento realizado pelo agente." + (f" {value}" if value else "")
|
|
|
|
def finalizar_atendimento(self, *, status: str = "resolvido", summary: str = "", **args: Any) -> dict[str, Any]:
|
|
normalized_status = self._normalize_final_status(status)
|
|
normalized_summary = self._normalize_final_summary(summary)
|
|
protocol = args.get("protocol") or args.get("ura_call_id") or ""
|
|
response: dict[str, Any] = {
|
|
"success": True,
|
|
"status": normalized_status,
|
|
"summary": normalized_summary,
|
|
}
|
|
if protocol:
|
|
response["service_request_status"] = self.client.status_sr({
|
|
"protocolNumber": protocol,
|
|
"status": "Fechado",
|
|
"channel": "AIAGENTCR",
|
|
"notes": normalized_summary,
|
|
})
|
|
return response
|
|
|
|
|
|
def os_source(args: dict[str, Any]) -> str:
|
|
return str(args.get("channel") or "AIAGENTCR")
|