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")