from __future__ import annotations import base64 import json import os import time import uuid from pathlib import Path from typing import Any from urllib.parse import quote import requests import yaml from .integrations.secure_pdf_crypto import encrypt_secure_pdf_value from .parsers import parse_tim_bill_pdf class TimApiError(RuntimeError): def __init__(self, message: str, *, status_code: int | None = None, body: Any = None, attempts: list[dict[str, Any]] | None = None): super().__init__(message) self.status_code = status_code self.body = body self.attempts = list(attempts or []) _INTEGRATION_DEFAULTS_CACHE: dict[str, str] | None = None def get_tim_integration_value(*names: str, default: str = "") -> str: """Env > config/tim_integration_defaults.yaml > default explícito.""" global _INTEGRATION_DEFAULTS_CACHE for name in names: value = os.getenv(name) if value is not None and str(value).strip(): return str(value).strip() if _INTEGRATION_DEFAULTS_CACHE is None: path = Path(os.getenv("TIM_INTEGRATION_DEFAULTS_PATH", "./config/tim_integration_defaults.yaml")) try: raw = yaml.safe_load(path.read_text(encoding="utf-8")) if path.exists() else {} except Exception as exc: raise TimApiError(f"Falha ao ler defaults de integração TIM: {path}: {exc}") from exc _INTEGRATION_DEFAULTS_CACHE = {str(k): str(v) for k, v in (raw or {}).items() if v not in (None, "")} for name in names: value = _INTEGRATION_DEFAULTS_CACHE.get(name) if value is not None and str(value).strip(): return str(value).strip() return default class TimApiClient: """Thin TIM integration adapter. This class deliberately contains no agent, LangGraph, memory, routing or LLM logic. Those responsibilities belong to agent_framework_oci. """ FIXTURES = Path(__file__).with_name("fixtures") INTEGRATION_DEFAULTS_PATH = Path(os.getenv("TIM_INTEGRATION_DEFAULTS_PATH", "./config/tim_integration_defaults.yaml")) _integration_defaults_cache: dict[str, str] | None = None def __init__(self) -> None: self.mock = os.getenv("TIM_USE_MOCK_GATEWAY", "false").lower() in {"1", "true", "yes", "on"} or os.getenv("TIM_GATEWAY_MODE", "real").lower() == "mock" self.timeout = int(os.getenv("TIM_GATEWAY_DEFAULT_TIMEOUT", "30")) self.max_retries = int(os.getenv("TIM_GATEWAY_RETRY_MAX_RETRIES", "3")) self.backoff = float(os.getenv("TIM_GATEWAY_RETRY_BACKOFF_FACTOR", "0.5")) def fixture(self, name: str) -> Any: path = self.FIXTURES / f"{name}.json" return json.loads(path.read_text(encoding="utf-8")) @classmethod def _integration_defaults(cls) -> dict[str, str]: get_tim_integration_value("__warmup__") return dict(_INTEGRATION_DEFAULTS_CACHE or {}) @classmethod def _env_first(cls, *names: str, default: str = "") -> str: return get_tim_integration_value(*names, default=default) @classmethod def _auth_value( cls, *auth_names: str, user_names: tuple[str, ...] = (), password_names: tuple[str, ...] = (), ) -> str: user = cls._env_first(*user_names) if user_names else "" password = cls._env_first(*password_names) if password_names else "" if user and password: token = base64.b64encode(f"{user}:{password}".encode("utf-8")).decode("ascii") return f"Basic {token}" raw = cls._env_first(*auth_names) if not raw: return "" low = raw.lower() if low.startswith("basic ") or low.startswith("bearer "): return raw if ":" in raw and " " not in raw: return "Basic " + base64.b64encode(raw.encode("utf-8")).decode("ascii") return raw @staticmethod def _url_with_msisdn(base: str, msisdn: str, *, normalize_country: bool = False) -> str: text = str(base or "").strip() digits = "".join(ch for ch in str(msisdn or "") if ch.isdigit()) if normalize_country and digits and not digits.startswith("55"): digits = f"55{digits}" value = digits or str(msisdn or "").strip() if "{msisdn}" in text: return text.replace("{msisdn}", quote(value)) return f"{text.rstrip('/')}/{quote(value)}" if text else value def _headers(self, *, client_id: str | None = None, auth: str | None = None, extra: dict[str, str] | None = None) -> dict[str, str]: h = {"Content-Type": "application/json", "Accept": "application/json"} if client_id: h["client_id"] = client_id if auth: h["Authorization"] = auth if extra: h.update({k: v for k, v in extra.items() if v not in (None, "")}) return h def request(self, method: str, url: str, *, payload: dict[str, Any] | None = None, params: dict[str, Any] | None = None, headers: dict[str, str] | None = None, timeout: int | None = None, attempt_log: list[dict[str, Any]] | None = None) -> Any: if not url: raise TimApiError("Endpoint TIM não configurado") last: Exception | None = None attempts = max(1, self.max_retries) for attempt in range(attempts): started = time.monotonic() try: resp = requests.request(method, url, json=payload, params=params, headers=headers or {}, timeout=timeout or self.timeout) if resp.status_code in {204, 205}: if attempt_log is not None: attempt_log.append({"attempt": attempt + 1, "success": True, "status_code": resp.status_code, "latency_ms": int((time.monotonic()-started)*1000), "api_url": url}) return {"status": resp.status_code, "message": "OK"} resp.raise_for_status() if attempt_log is not None: attempt_log.append({"attempt": attempt + 1, "success": True, "status_code": resp.status_code, "latency_ms": int((time.monotonic()-started)*1000), "api_url": url}) if not resp.content: return {"status": resp.status_code, "message": "OK"} ctype = resp.headers.get("content-type", "").lower() if "json" in ctype: return resp.json() return {"status": resp.status_code, "raw_content": resp.content, "content_type": ctype} except requests.RequestException as exc: last = exc if attempt_log is not None: response = getattr(exc, "response", None) attempt_log.append({"attempt": attempt + 1, "success": False, "status_code": getattr(response, "status_code", None), "latency_ms": int((time.monotonic()-started)*1000), "error": str(exc), "api_url": url}) if attempt >= attempts - 1: break time.sleep(self.backoff * (2 ** attempt)) response = getattr(last, "response", None) status_code = getattr(response, "status_code", None) body = None if response is not None: try: body = response.json() except Exception: body = getattr(response, "text", None) raise TimApiError(str(last), status_code=status_code, body=body, attempts=list(attempt_log or [])) @staticmethod def _attach_transport(result: Any, *, operation: str, attempts: list[dict[str, Any]]) -> Any: metadata = { "rct_operation": operation, "attempts": list(attempts), "api_response_payload": result if isinstance(result, (dict, list, str, int, float, bool)) else str(result), } if isinstance(result, dict): out = dict(result) out["_transport"] = metadata return out return {"data": result, "_transport": metadata} def consultar_faturas(self, msisdn: str) -> Any: if self.mock: return self.fixture("complete_invoices") url = self._env_first("TIM_COMPLETE_INVOICES_URL") return self.request( "POST", url, payload={"msisdn": msisdn}, headers=self._headers( client_id=None, auth=self._env_first("TIM_COMPLETE_INVOICES_AUTH"), extra={"ClientID": self._env_first("TIM_COMPLETE_INVOICES_CLIENT_ID", )}, ), timeout=int(self._env_first("TIM_COMPLETE_INVOICES_TIMEOUT", default="30")), ) def billing_analysis(self, msisdn: str, **context: Any) -> Any: if self.mock: return self.fixture("divergencia") base = self._env_first("TIM_DIVERGENCIA_URL").rstrip("/") url = f"{base}/{quote(msisdn)}" auth = self._auth_value( "TIM_DIVERGENCIA_AUTH", "TIM_DIVERGENCE_AUTH", user_names=("TIM_DIVERGENCIA_USER", "TIM_DIVERGENCIA_USERNAME", "TIM_DIVERGENCE_USER", "TIM_DIVERGENCE_USERNAME"), password_names=("TIM_DIVERGENCIA_PASSWORD", "TIM_DIVERGENCIA_PASS", "TIM_DIVERGENCE_PASSWORD", "TIM_DIVERGENCE_PASS"), ) headers = self._headers(client_id=None, auth=auth, extra={ "clientID": self._env_first("TIM_DIVERGENCIA_CLIENT_ID", ), }) params = {"channel": context.get("channel") or self._env_first("TIM_DIVERGENCIA_CHANNEL")} attempts: list[dict[str, Any]] = [] try: result = self.request("GET", url, params=params, headers=headers, timeout=int(self._env_first("TIM_DIVERGENCIA_TIMEOUT", "TIM_DIVERGENCE_TIMEOUT", default="120")), attempt_log=attempts) return self._attach_transport(result, operation="base_conhecimento", attempts=attempts) except TimApiError as exc: if not exc.attempts: exc.attempts = attempts raise def consultar_vas(self, msisdn: str) -> Any: if self.mock: return self.fixture("query_vas") base = self._env_first("TIM_URL_CONSULTA_VAS", "TIM_CONSULTA_URL") url = self._url_with_msisdn(base, msisdn, normalize_country=True) return self.request( "GET", url, headers=self._headers( client_id=None, auth=self._env_first("TIM_QUERY_AUTH", "TIM_CONSULTA_AUTH"), extra={"clientId": self._env_first("TIM_CONSULTA_CLIENT_ID", )}, ), timeout=int(self._env_first("TIM_QUERY_TIMEOUT", "TIM_CONSULTA_TIMEOUT", default="30")), ) def historico_vas(self, msisdn: str) -> Any: if self.mock: return self.fixture("vas_history") base = os.getenv("TIM_VAS_HISTORY_URL", "") sep = "&" if "?" in base else "?" url = f"{base}{sep}msisdn={quote(msisdn)}" headers = self._headers( client_id=None, auth=os.getenv("TIM_VAS_HISTORY_AUTH", ""), extra={ "clientId": self._env_first("TIM_VAS_HISTORY_CLIENT_ID", ), "messageId": os.getenv("TIM_VAS_HISTORY_MESSAGE_ID", "") or str(uuid.uuid4()), }, ) return self.request("GET", url, headers=headers, timeout=int(os.getenv("TIM_VAS_HISTORY_TIMEOUT", "30"))) def bloquear_vas(self, msisdn: str, service: dict[str, Any]) -> Any: if self.mock: return self.fixture("block_vas") url = self._env_first("TIM_URL_BLOQUEIO_VAS", "TIM_BLOQUEIO_URL") digits = "".join(ch for ch in str(msisdn) if ch.isdigit()) normalized = digits[2:] if len(digits) == 13 else digits app_id = str(service.get("appId") or service.get("app_id") or "") csp_id = str(service.get("cspId") or service.get("csp_id") or self._env_first("TIM_DEFAULT_CSP_ID")) op = self._env_first("TIM_BLOCK_OPERATION_TYPE", "TIM_BLOQUEIO_OPERATION_TYPE", default="block") base = {"Customer": {"Msisdn": normalized}, "AppId": app_id, "CspId": csp_id, "TypeOperation": op} payloads = [ {"customer": {"msisdn": normalized}, "appId": app_id, "cspId": csp_id, "typeOperation": op}, {"input": base, "Input": base}, {"vasBlock": {"msisdn": normalized, "appId": app_id, "cspId": csp_id, "type": op}}, ] mode = self._env_first("TIM_BLOCK_PAYLOAD_MODE", "TIM_BLOQUEIO_PAYLOAD_MODE", default="auto").lower() if mode == "vasblock": payloads = [payloads[2], payloads[0]] elif mode == "input": payloads = [payloads[1], payloads[0]] elif mode == "pmid": payloads = [payloads[0], payloads[1]] headers = self._headers( client_id=None, auth=self._env_first("TIM_BLOCK_AUTH", "TIM_BLOQUEIO_AUTH"), extra={ "Accept-Encoding": self._env_first("TIM_BLOCK_ACCEPT_ENCODING", "TIM_BLOQUEIO_ACCEPT_ENCODING", default="gzip,deflate"), "clientId": self._env_first("TIM_BLOCK_CLIENT_ID", "TIM_BLOQUEIO_CLIENT_ID", ), "messageId": str(uuid.uuid4()), }, ) last = None for index, payload in enumerate(payloads): try: return self.request("POST", url, payload=payload, headers=headers, timeout=int(self._env_first("TIM_BLOCK_TIMEOUT", "TIM_BLOQUEIO_TIMEOUT", default="30"))) except TimApiError as exc: last = exc if exc.status_code != 400 or index >= len(payloads) - 1: raise raise last or TimApiError("Bloqueio VAS falhou") def cancelar_vas(self, msisdn: str, service: dict[str, Any], *, protocol: str = "") -> Any: if self.mock: return self.fixture("cancel_vas") url = self._env_first("TIM_CANCELLATION_URL", "TIM_CANCELAMENTO_URL") payload = { "channel": self._env_first("TIM_CANCELAMENTO_CHANNEL"), "msisdn": msisdn, "appId": str(service.get("appId") or service.get("app_id") or ""), "cspId": str(service.get("cspId") or service.get("csp_id") or self._env_first("TIM_DEFAULT_CSP_ID")), "interactionProtocol": protocol, } headers = self._headers( client_id=None, auth=self._env_first("TIM_CANCELLATION_AUTH", "TIM_CANCELAMENTO_AUTH"), extra={ "clientId": os.getenv("TIM_CANCELAMENTO_CLIENT_ID", self._env_first("TIM_DEFAULT_CLIENT_ID")), "messageId": str(uuid.uuid4()), "AuthorizationOAM": self._env_first("TIM_CANCELLATION_AUTH_OAM", "TIM_CANCELAMENTO_AUTH_OAM", "TIM_CANCELAMENTO_AUTHORIZATION_OAM"), "Cn_field": self._env_first("TIM_CANCELLATION_CN_FIELD", "TIM_CANCELAMENTO_CN_FIELD"), "Type_field": self._env_first("TIM_CANCELLATION_TYPE_FIELD", "TIM_CANCELAMENTO_TYPE_FIELD"), }, ) attempts: list[dict[str, Any]] = [] result = self.request("DELETE", url, payload=payload, headers=headers, timeout=int(self._env_first("TIM_CANCELLATION_TIMEOUT", "TIM_CANCELAMENTO_TIMEOUT", default="30")), attempt_log=attempts) return self._attach_transport(result, operation="cancela_vas", attempts=attempts) def contrato(self, msisdn: str) -> Any: if self.mock: return self.fixture("contrato") base = os.getenv("TIM_CONTRATO_URL", "").rstrip("/") headers = self._headers( client_id=None, auth=os.getenv("TIM_CONTRATO_AUTH", ""), extra={"clientId": self._env_first("TIM_CONTRATO_CLIENT_ID", )}, ) return self.request("GET", f"{base}/{quote(msisdn)}", headers=headers, timeout=int(os.getenv("TIM_CONTRATO_TIMEOUT", "30"))) def profile_full(self, msisdn: str) -> Any: if self.mock: # contract fixture carries representative customer identity in local mode. return self.fixture("contrato") url = self._url_with_msisdn(self._env_first("TIM_PROFILE_FULL_URL"), msisdn) headers = self._headers( client_id=None, auth=os.getenv("TIM_PROFILE_FULL_AUTH", ""), extra={"ClientID": self._env_first("TIM_PROFILE_FULL_CLIENT_ID")}, ) return self.request("GET", url, headers=headers, timeout=int(self._env_first("TIM_PROFILE_FULL_TIMEOUT", default="30"))) def abrir_protocolo(self, payload: dict[str, Any]) -> Any: """Registra protocolo V2 preservando o contrato externo do Contas original. As workflow actions trabalham com um payload lógico/achatado. O adapter é responsável por converter esse modelo para o contrato PMid/Siebel, sem contaminar o domínio ou o WorkflowRuntime com detalhes HTTP. """ if self.mock: return self.fixture("protocol") data = dict(payload or {}) msisdn = str(data.get("msisdn") or data.get("assetId") or "") service_status = str(data.get("requestStatus") or data.get("status") or "") message_id = str( data.get("messageId") or data.get("interactionCallId") or data.get("ura_call_id") or data.get("session_id") or uuid.uuid4() ) body = { "socialSecNo": str(data.get("socialSecNo") or data.get("social_sec_no") or ""), "channel": self._env_first("TIM_PROTOCOL_CHANNEL"), "customerId": str(data.get("customerId") or data.get("customer_id") or ""), "assetId": str(data.get("assetId") or data.get("asset_id") or msisdn), "customerName": str(data.get("customerName") or data.get("customer_name") or ""), "customerEmail": str(data.get("customerEmail") or data.get("customer_email") or ""), "customerPhone1": str(data.get("customerPhone1") or data.get("customer_phone1") or ""), "accessType": str(data.get("accessType") or data.get("access_type") or ""), "serviceRequest": { "userId": str(data.get("serviceRequestUserId") or data.get("service_request_user_id") or ""), "reason1": str(data.get("reason1") or ""), "reason2": str(data.get("reason2") or ""), "reason3": str(data.get("reason3") or ""), "status": service_status, "notes": str(data.get("serviceRequestNotes") or data.get("service_request_notes") or data.get("notes") or ""), "type": str(data.get("type") or "CLIENTE"), }, "interaction": { "protocol": str(data.get("interactionProtocol") or data.get("interaction_protocol") or ""), "flagSms": True, "source": self._env_first("TIM_DEFAULT_SOURCE"), "callId": str(data.get("interactionCallId") or data.get("interaction_call_id") or ""), "crmSource": "Siebel Pós", "reasonId": str(data.get("interactionReasonId") or data.get("interaction_reason_id") or ""), "amount": str(data.get("interactionAmount") or data.get("interaction_amount") or ""), "directionContact": str(data.get("directionContact") or data.get("direction_contact") or "FROM-CLIENT"), "requestFlag": False, "requestSla": str(data.get("interactionRequestSla") or data.get("interaction_request_sla") or ""), "status": str(data.get("status") or service_status), }, } headers = self._headers( client_id=None, auth=str(data.get("Authorization") or os.getenv("TIM_PROTOCOL_AUTH", "")), extra={ "clientId": str(data.get("clientId") or data.get("client_id") or self._env_first("TIM_PROTOCOL_CLIENT_ID")), "messageId": message_id, "Authorizationoam": str(data.get("authorization_oam") or os.getenv("TIM_PROTOCOL_AUTHORIZATION_OAM", "")), "Cn_field": str(data.get("cn_field") or os.getenv("TIM_PROTOCOL_CN_FIELD", "")), "Type_field": str(data.get("type_field") or os.getenv("TIM_PROTOCOL_TYPE_FIELD", "")), }, ) attempts: list[dict[str, Any]] = [] result = self.request( "POST", os.getenv("TIM_PROTOCOL_URL", ""), payload=body, headers=headers, timeout=int(os.getenv("TIM_PROTOCOL_TIMEOUT", "30")), attempt_log=attempts, ) operation = str(data.get("rct_operation") or "") return self._attach_transport(result, operation=operation, attempts=attempts) if operation else result def contestar(self, payload: dict[str, Any]) -> Any: if self.mock: return self.fixture("contestacao_tool") data = dict(payload) data.setdefault("userId", self._env_first("TIM_CUSTOMER_CONTESTATION_USER_ID", )) data.setdefault("customerIdCurrent", data.get("customerId") or "") data.setdefault("customerType", "2") data.setdefault("customerStatus", "1") status_map = {"ABERTA": "0", "ABERTO": "0", "OPEN": "0", "FECHADA": "1", "FECHADO": "1", "CLOSED": "1"} raw_status = str(data.get("invoiceStatus") or "").strip() data["invoiceStatus"] = status_map.get(raw_status.upper(), raw_status or "0") data.setdefault("invoiceAmountOpen", "0") data.setdefault("invoiceAmount", "0") due = "".join(ch for ch in str(data.get("invoiceDueDate") or "") if ch.isdigit()) if len(due) == 8 and str(data.get("invoiceDueDate") or "").startswith(tuple(str(y) for y in range(19, 22))): # YYYYMMDD already normalized pass elif len(due) == 8: # common ddMMyyyy -> yyyyMMdd due = due[4:]+due[2:4]+due[:2] data["invoiceDueDate"] = due data.setdefault("contestationType", "0") data.setdefault("adjustReason", "SERVICO_NAO_SOLICITADO") data.setdefault("observation", data.pop("description", "")) data.setdefault("refundOption", "0") data.setdefault("doubleRefund", False) data.setdefault("manualContaCertaIndicator", False) client_id = str(data.pop("clientId", "") or self._env_first("TIM_CUSTOMER_CONTESTATION_CLIENT_ID")) message_id = str(data.pop("messageId", "") or uuid.uuid4()) user_id = str(data.get("userId") or self._env_first("TIM_DEFAULT_CLIENT_ID")).upper() headers = self._headers( client_id=None, auth=os.getenv("TIM_CUSTOMER_CONTESTATION_AUTH", ""), extra={"clientId": client_id, "messageId": message_id, "X-Agent-Id": user_id}, ) attempts: list[dict[str, Any]] = [] result = self.request("POST", os.getenv("TIM_CUSTOMER_CONTESTATION_URL", ""), payload=data, headers=headers, timeout=int(os.getenv("TIM_CUSTOMER_CONTESTATION_TIMEOUT", "30")), attempt_log=attempts) return self._attach_transport(result, operation="contestacao", attempts=attempts) def status_sr(self, payload: dict[str, Any]) -> Any: if self.mock: return self.fixture("service_request_status") data = dict(payload) service_request = data.get("serviceRequest") if isinstance(data.get("serviceRequest"), dict) else { "status": data.get("status") or "", "notes": data.get("notes") or "", "protocolNumber": data.get("protocolNumber") or data.get("protocol") or "", "auditResult": data.get("auditResult") or "", "auditReasonResult": data.get("auditReasonResult") or "", "auditConvenienceTime": data.get("auditConvenienceTime") or "", "auditSubStatus": data.get("auditSubStatus") or "", } for key in ("reason1", "reason2", "reason3", "auditChecks", "attachments"): if data.get(key): service_request[key] = data[key] body = {"channel": data.get("channel") or self._env_first("TIM_DEFAULT_CLIENT_ID"), "serviceRequest": service_request} for key in ("msisdn", "customers", "date"): if data.get(key): body[key] = data[key] message_id = str(data.get("messageId") or data.get("ura_call_id") or data.get("session_id") or uuid.uuid4()) headers = self._headers( client_id=None, auth=os.getenv("TIM_SERVICE_REQUEST_STATUS_AUTH", ""), extra={ "clientId": str(data.get("clientId") or self._env_first("TIM_SERVICE_REQUEST_STATUS_CLIENT_ID")), "messageId": message_id, "Authorizationoam": data.get("authorization_oam") or os.getenv("TIM_SERVICE_REQUEST_STATUS_AUTHORIZATION_OAM", ""), "Cn_field": data.get("cn_field") or os.getenv("TIM_SERVICE_REQUEST_STATUS_CN_FIELD", ""), "Type_field": data.get("type_field") or os.getenv("TIM_SERVICE_REQUEST_STATUS_TYPE_FIELD", ""), }, ) attempts: list[dict[str, Any]] = [] result = self.request( "POST", os.getenv("TIM_SERVICE_REQUEST_STATUS_URL", ""), payload=body, headers=headers, timeout=int(os.getenv("TIM_SERVICE_REQUEST_STATUS_TIMEOUT", "30")), attempt_log=attempts ) return self._attach_transport(result, operation="", attempts=attempts) def tracking(self, payload: dict[str, Any]) -> Any: if self.mock: return self.fixture("tracking_activities") data = dict(payload) invoice = data.get("invoice") if isinstance(data.get("invoice"), dict) else { "emissionDate": data.get("invoiceEmissionDate") or "", "expirationDate": data.get("invoiceExpirationDate") or "", "number": data.get("invoiceNumber") or "", "status": data.get("invoiceStatus") or "", "openAmount": data.get("invoiceOpenAmount") or "", "totalAmount": data.get("invoiceTotalAmount") or "", } body = { "channel": data.get("channel") or self._env_first("TIM_TRACKING_ACTIVITIES_CHANNEL"), "customer": {"socialSecNo": data.get("socialSecNo") or "", "msisdn": data.get("msisdn") or ""}, "protocolNumber": data.get("protocolNumber") or "", "invoice": invoice, "activity": { "type": data.get("activityType") or "", "status": data.get("activityStatus") or "", "id": data.get("activityId") or "", }, "user": {"login": data.get("userLogin") or self._env_first("TIM_TRACKING_ACTIVITIES_USER_LOGIN")}, } headers = self._headers( client_id=None, auth=os.getenv("TIM_TRACKING_ACTIVITIES_AUTH", ""), extra={"clientId": data.get("clientId") or self._env_first("TIM_TRACKING_ACTIVITIES_CLIENT_ID")}, ) return self.request("POST", os.getenv("TIM_TRACKING_ACTIVITIES_URL", ""), payload=body, headers=headers, timeout=int(os.getenv("TIM_TRACKING_ACTIVITIES_TIMEOUT", "30"))) def sms(self, msisdn: str, message: str, **context: Any) -> Any: if self.mock: return self.fixture("sms") payload: dict[str, Any] = { "msisdn": msisdn, "senderAddress": context.get("sender_address") or os.getenv("TIM_SMS_SENDER_ADDRESS", "324"), "senderName": context.get("sender_name") or self._env_first("TIM_SMS_SENDER_NAME"), "message": message, "longURL": context.get("long_url") or message, } if context.get("notify_url"): payload["receiptRequest"] = {"notifyURL": context["notify_url"]} headers = self._headers( client_id=None, auth=os.getenv("TIM_SMS_AUTH", ""), extra={"clientId": self._env_first("TIM_SMS_CLIENT_ID")}, ) attempts: list[dict[str, Any]] = [] result = self.request("POST", os.getenv("TIM_SMS_URL", ""), payload=payload, headers=headers, timeout=int(os.getenv("TIM_SMS_TIMEOUT", "30")), attempt_log=attempts) return self._attach_transport(result, operation="sgr_codbar", attempts=attempts) def profile_bill(self, msisdn: str) -> Any: """Perfil de faturamento usando o mesmo contrato CompleteInvoices do original.""" if self.mock: return self.fixture("profile_bill") url = self._env_first("TIM_COMPLETE_INVOICES_URL", "TIM_URL_PERFIL_FATURA") return self.request( "POST", url, payload={"msisdn": msisdn}, headers=self._headers( client_id=None, auth=self._env_first("TIM_COMPLETE_INVOICES_AUTH", "TIM_PROFILE_BILL_AUTH"), extra={"ClientID": self._env_first("TIM_COMPLETE_INVOICES_CLIENT_ID", "TIM_PROFILE_BILL_CLIENT_ID", )}, ), timeout=int(self._env_first("TIM_COMPLETE_INVOICES_TIMEOUT", "TIM_PROFILE_BILL_TIMEOUT", default="30")), ) def line_info(self, msisdn: str) -> Any: """Consulta dados cadastrais da linha/dependente e preserva o CPF correto.""" if self.mock: payload = self.fixture("contrato") else: url = self._url_with_msisdn(self._env_first("TIM_PROFILE_FULL_URL"), msisdn) payload = self.request( "GET", url, headers=self._headers( client_id=None, auth=os.getenv("TIM_PROFILE_FULL_AUTH", ""), extra={"ClientID": self._env_first("TIM_PROFILE_FULL_CLIENT_ID")}, ), timeout=int(self._env_first("TIM_PROFILE_FULL_TIMEOUT", default="30")), ) def extract(value: Any) -> str: if not isinstance(value, dict): return "" direct = value.get("socialSecNo") if direct not in (None, ""): return str(direct).strip() for key in ("customer", "contract", "billingProfile", "subscriber"): found = extract(value.get(key)) if found: return found return "" return {"social_sec_no": extract(payload), "raw": payload if isinstance(payload, dict) else {}} def bill_pdf( self, msisdn: str, invoice_id: str, customer_id: str, *, include_danfe: bool = False, output: str = "", ) -> Any: """Recupera a fatura detalhada via POST e normaliza seu PDF. Esta operação é distinta de ``secure_pdf`` e corresponde ao BillPdfCommand original, usado por pró-rata, matching e explicação detalhada de fatura. """ if self.mock: parsed = self.fixture("invoice_pdf_include_danfe_true" if include_danfe else "invoice_pdf_include_danfe_false") return { "status": "SUCCESS", "message": "Fatura recuperada com sucesso", "file_content": None, "file_name": output or "antiga.pdf", "parsed_content": parsed, } url = os.getenv("TIM_BILL_PDF_URL", "") or os.getenv("TIM_SECURE_PDF_URL", "") or os.getenv("TIM_URL_INVOICE_RECOVER", "") payload = { "invoiceId": encrypt_secure_pdf_value(invoice_id), "customerId": encrypt_secure_pdf_value(customer_id), "invoiceType": "DETALHADA", } headers = self._headers( client_id=None, auth=self._env_first("TIM_BILL_PDF_AUTH", "TIM_SECURE_PDF_AUTH", "TIM_INVOICE_RECOVER_AUTH"), extra={ "clientId": self._env_first("TIM_BILL_PDF_CLIENT_ID", "TIM_SECURE_PDF_CLIENT_ID", ), "Accept": "application/pdf", }, ) result = self.request( "POST", url, payload=payload, headers=headers, timeout=int(self._env_first("TIM_BILL_PDF_TIMEOUT", "TIM_SECURE_PDF_TIMEOUT", "TIM_INVOICE_RECOVER_TIMEOUT", default="30")), ) raw_content = result.get("raw_content") if isinstance(result, dict) else None parsed = result.get("parsed_content") if isinstance(result, dict) else None if raw_content and parsed is None: parsed = parse_tim_bill_pdf(raw_content, include_danfe=include_danfe) return { "status": "SUCCESS" if raw_content or parsed is not None else str((result or {}).get("status") if isinstance(result, dict) else ""), "message": "Fatura recuperada com sucesso" if raw_content or parsed is not None else "Resposta de fatura sem conteúdo", "file_content": raw_content, "file_name": output or "antiga.pdf", "parsed_content": parsed, "raw": result, } def secure_pdf(self, msisdn: str, invoice_id: str, customer_id: str = "") -> Any: if self.mock: return self.fixture("invoice_pdf_include_danfe_false") url = os.getenv("TIM_URL_INVOICE_RECOVER", "") params = { "invoiceId": encrypt_secure_pdf_value(invoice_id), "msisdn": encrypt_secure_pdf_value(msisdn), "customerId": encrypt_secure_pdf_value(customer_id), } return self.request( "GET", url, params=params, headers=self._headers( client_id=None, auth=self._env_first("TIM_SECURE_PDF_AUTH", "TIM_BILL_PDF_AUTH", "TIM_INVOICE_RECOVER_AUTH"), extra={"clientid": self._env_first("TIM_INVOICE_RECOVER_CLIENT_ID", "TIM_SECURE_PDF_CLIENT_ID", )}, ), timeout=int(os.getenv("TIM_INVOICE_RECOVER_TIMEOUT", "30")), )