from __future__ import annotations import json import os import time import uuid from pathlib import Path from typing import Any from urllib.parse import quote import requests 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 []) 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") def __init__(self) -> None: self.mock = os.getenv("TIM_USE_MOCK_GATEWAY", "true").lower() in {"1", "true", "yes", "on"} or os.getenv("TIM_GATEWAY_MODE", "mock").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")) 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 = os.getenv("TIM_COMPLETE_INVOICES_URL", "") return self.request("POST", url, payload={"msisdn": msisdn}, headers=self._headers(client_id=None, auth=os.getenv("TIM_COMPLETE_INVOICES_AUTH", ""), extra={"ClientID": os.getenv("TIM_COMPLETE_INVOICES_CLIENT_ID", "AIAGENTCR")})) def billing_analysis(self, msisdn: str, **context: Any) -> Any: if self.mock: return self.fixture("divergencia") base = os.getenv("TIM_DIVERGENCIA_URL", "").rstrip("/") url = f"{base}/{quote(msisdn)}" auth = os.getenv("TIM_DIVERGENCIA_AUTH", "") or os.getenv("TIM_SECURE_PDF_AUTH", "") headers = self._headers(client_id=None, auth=auth, extra={ "clientID": os.getenv("TIM_DIVERGENCIA_CLIENT_ID", "AIAGENTCR"), }) params = {"channel": context.get("channel") or "AIAGENTCR"} attempts: list[dict[str, Any]] = [] try: result = self.request("GET", url, params=params, headers=headers, timeout=int(os.getenv("TIM_DIVERGENCIA_TIMEOUT", "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 = os.getenv("TIM_URL_CONSULTA_VAS", "") url = base.replace("{msisdn}", quote(msisdn)) return self.request("GET", url, headers=self._headers(client_id=None, auth=os.getenv("TIM_CONSULTA_AUTH", ""), extra={"clientId": os.getenv("TIM_CONSULTA_CLIENT_ID", "AIAAGENTCR")})) 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": os.getenv("TIM_VAS_HISTORY_CLIENT_ID", os.getenv("TIM_DEFAULT_CLIENT_ID", "CHAT")), "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 = os.getenv("TIM_URL_BLOQUEIO_VAS", "") 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 os.getenv("TIM_DEFAULT_CSP_ID", "740")) op = os.getenv("TIM_BLOQUEIO_OPERATION_TYPE", "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 = os.getenv("TIM_BLOQUEIO_PAYLOAD_MODE", "auto").strip().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=os.getenv("TIM_BLOQUEIO_AUTH", ""), extra={ "Accept-Encoding": os.getenv("TIM_BLOQUEIO_ACCEPT_ENCODING", "gzip,deflate"), "clientId": os.getenv("TIM_BLOQUEIO_CLIENT_ID", "AIAGENTCR"), "messageId": str(uuid.uuid4()), }, ) last = None for index, payload in enumerate(payloads): try: return self.request("POST", url, payload=payload, headers=headers, timeout=int(os.getenv("TIM_BLOQUEIO_TIMEOUT", "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 = os.getenv("TIM_CANCELAMENTO_URL", "") payload = { "channel": "AIAGENTCR", "msisdn": msisdn, "appId": str(service.get("appId") or service.get("app_id") or ""), "cspId": str(service.get("cspId") or service.get("csp_id") or os.getenv("TIM_DEFAULT_CSP_ID", "740")), "interactionProtocol": protocol, } headers = self._headers( client_id=None, auth=os.getenv("TIM_CANCELAMENTO_AUTH", ""), extra={ "clientId": os.getenv("TIM_CANCELAMENTO_CLIENT_ID", os.getenv("TIM_DEFAULT_CLIENT_ID", "AIAGENTCR")), "messageId": str(uuid.uuid4()), "AuthorizationOAM": os.getenv("TIM_CANCELAMENTO_AUTHORIZATION_OAM", ""), "Cn_field": os.getenv("TIM_CANCELAMENTO_CN_FIELD", ""), "Type_field": os.getenv("TIM_CANCELAMENTO_TYPE_FIELD", ""), }, ) attempts: list[dict[str, Any]] = [] result = self.request("DELETE", url, payload=payload, headers=headers, timeout=int(os.getenv("TIM_CANCELAMENTO_TIMEOUT", "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": os.getenv("TIM_CONTRATO_CLIENT_ID", os.getenv("TIM_DEFAULT_CLIENT_ID", "CHAT"))}, ) 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 = os.getenv("TIM_PROFILE_FULL_URL", "").replace("{msisdn}", quote(msisdn)) headers = self._headers( client_id=None, auth=os.getenv("TIM_PROFILE_FULL_AUTH", ""), extra={"ClientID": os.getenv("TIM_PROFILE_FULL_CLIENT_ID", "AIAGENTCR")}, ) return self.request("GET", url, headers=headers) 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": "AIAGENTCR", "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": "AIAGENTCR", "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 os.getenv("TIM_PROTOCOL_CLIENT_ID", "BFFDIGITAL")), "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", "AIAGENTCR") 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 os.getenv("TIM_CUSTOMER_CONTESTATION_CLIENT_ID", "AIAGENTCR")) message_id = str(data.pop("messageId", "") or uuid.uuid4()) user_id = str(data.get("userId") or "AIAGENTCR").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 "AIAGENTCR", "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 os.getenv("TIM_SERVICE_REQUEST_STATUS_CLIENT_ID", "AIAGENTCR")), "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 os.getenv("TIM_TRACKING_ACTIVITIES_CHANNEL", "AIAGENTCR"), "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 os.getenv("TIM_TRACKING_ACTIVITIES_USER_LOGIN", "SIEBELPOS_INBOUND")}, } headers = self._headers( client_id=None, auth=os.getenv("TIM_TRACKING_ACTIVITIES_AUTH", ""), extra={"clientId": data.get("clientId") or os.getenv("TIM_TRACKING_ACTIVITIES_CLIENT_ID", "AIAGENTCR")}, ) 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 os.getenv("TIM_SMS_SENDER_NAME", "TIM Brasil"), "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": os.getenv("TIM_SMS_CLIENT_ID", "AIAGENTCR")}, ) 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 = os.getenv("TIM_URL_PERFIL_FATURA", "") or os.getenv("TIM_COMPLETE_INVOICES_URL", "") return self.request( "POST", url, payload={"msisdn": msisdn}, headers=self._headers( client_id=None, auth=os.getenv("TIM_PROFILE_BILL_AUTH", "") or os.getenv("TIM_COMPLETE_INVOICES_AUTH", ""), extra={"ClientID": os.getenv("TIM_PROFILE_BILL_CLIENT_ID", "AIAGENTCR")}, ), timeout=int(os.getenv("TIM_PROFILE_BILL_TIMEOUT", "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 = os.getenv("TIM_PROFILE_FULL_URL", "").replace("{msisdn}", quote(msisdn)) payload = self.request( "GET", url, headers=self._headers( client_id=None, auth=os.getenv("TIM_PROFILE_FULL_AUTH", ""), extra={"ClientID": os.getenv("TIM_PROFILE_FULL_CLIENT_ID", "AIAGENTCR")}, ), ) 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=os.getenv("TIM_BILL_PDF_AUTH", "") or os.getenv("TIM_SECURE_PDF_AUTH", ""), extra={ "clientId": os.getenv("TIM_BILL_PDF_CLIENT_ID", "AIAGENTCR"), "Accept": "application/pdf", }, ) result = self.request( "POST", url, payload=payload, headers=headers, timeout=int(os.getenv("TIM_BILL_PDF_TIMEOUT", os.getenv("TIM_INVOICE_RECOVER_TIMEOUT", "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=os.getenv("TIM_SECURE_PDF_AUTH", ""), extra={"clientId": os.getenv("TIM_INVOICE_RECOVER_CLIENT_ID", "AIAGENTCR")}, ), timeout=int(os.getenv("TIM_INVOICE_RECOVER_TIMEOUT", "30")), )