from __future__ import annotations import json import math import os import re import socket from collections.abc import Callable, Mapping from email.message import Message from typing import Any, cast from urllib.error import HTTPError, URLError from urllib.parse import quote, urlsplit, urlunsplit from urllib.request import HTTPRedirectHandler, Request, build_opener from .types import ( LeadsListRequest, LeadsListResult, MarketingFunnelResult, MessagesListRequest, MessagesListResult, PipelinesGetRequest, PipelinesGetResult, PipelinesListRequest, PipelinesListResult, ProductsBrandsResult, ProductsCategoriesResult, ProductsGetRequest, ProductsGetResult, ProductsListResult, ProductsQueryRequest, PublicMethod, ShopBannersListRequest, ShopBannersListResult, ShopDescontoValidarRequest, ShopDescontoValidarResult, ShopFreteCalcularRequest, ShopFreteCalcularResult, ShopFreteConfigResult, ShopLojaConfigResult, StagesListRequest, StagesListResult, TasksGetRequest, TasksGetResult, TasksListRequest, TasksListResult, ) PUBLIC_METHODS: frozenset[str] = frozenset({ "messages.list", "marketing.funnel", "leads.list", "pipelines.list", "pipelines.get", "stages.list", "tasks.list", "tasks.get", "products.list", "products.search", "products.get", "products.brands", "products.categories", "shop.banners.list", "shop.frete.config", "shop.frete.calcular", "shop.desconto.validar", "shop.loja.config", }) _TOKEN_RE = re.compile(r"^[a-f0-9]{64}$", re.IGNORECASE) _TOKEN_IN_PATH_RE = re.compile(r"/api/v1/[a-f0-9]{64}(?=/)", re.IGNORECASE) Transport = Callable[[Request, float], Any] class SysledAPIError(RuntimeError): def __init__( self, message: str, *, status: int | None = None, code: str | None = None, method: str | None = None, request_id: str | None = None, retry_after_seconds: float | None = None, ) -> None: super().__init__(message) self.status = status self.code = code self.method = method self.request_id = request_id self.retry_after_seconds = retry_after_seconds def redact_secrets(value: object, *explicit_secrets: str) -> str: text = str(value) text = _TOKEN_IN_PATH_RE.sub("/api/v1/[REDACTED]", text) text = re.sub(r"\b(?:bearer\s+)?[a-f0-9]{64}\b", "[REDACTED]", text, flags=re.IGNORECASE) for secret in explicit_secrets: if secret: text = text.replace(secret, "[REDACTED]") return text class _NoRedirect(HTTPRedirectHandler): def redirect_request(self, req: Any, fp: Any, code: int, msg: str, headers: Any, newurl: str) -> None: # A redirect can forward the token-bearing path to another origin. return None def _default_transport(request: Request, timeout: float) -> Any: return build_opener(_NoRedirect()).open(request, timeout=timeout) def _normalize_base_url(raw: str) -> str: parsed = None try: parsed = urlsplit(raw) parsed.port # Validate malformed ports without retaining the input in an error. except (ValueError, TypeError): parsed = None if parsed is None or not isinstance(raw, str) or any(c.isspace() for c in raw): raise SysledAPIError("SYSLED_URL deve ser uma origem absoluta válida.", code="INVALID_CONFIGURATION") local = parsed.hostname in {"localhost", "127.0.0.1", "::1"} if not parsed.hostname or (parsed.scheme != "https" and not (local and parsed.scheme == "http")): raise SysledAPIError("SYSLED_URL deve usar HTTPS; HTTP só é aceito em loopback.", code="INVALID_CONFIGURATION") if parsed.username or parsed.password or parsed.query or parsed.fragment: raise SysledAPIError("SYSLED_URL não pode conter credenciais, query string ou fragmento.", code="INVALID_CONFIGURATION") if parsed.path not in {"", "/"}: raise SysledAPIError("SYSLED_URL deve conter somente a origem, sem /api/v1 ou outro caminho.", code="INVALID_CONFIGURATION") return urlunsplit((parsed.scheme, parsed.netloc, "", "", "")) def _header(headers: Message | Mapping[str, str] | Any, name: str) -> str | None: value = headers.get(name) if headers is not None and hasattr(headers, "get") else None if value is None and hasattr(headers, "items"): value = next((v for k, v in headers.items() if k.lower() == name.lower()), None) return str(value) if value is not None else None def _retry_after(headers: Any) -> float | None: raw = _header(headers, "Retry-After") if raw is None: return None try: value = float(raw) return value if math.isfinite(value) and value >= 0 else None except ValueError: return None class SysledClient: def __init__( self, *, base_url: str | None = None, token: str | None = None, timeout: float = 10.0, transport: Transport | None = None, max_response_bytes: int = 5_000_000, ) -> None: base_url = base_url or os.environ.get("SYSLED_URL") token = token or os.environ.get("SYSLED_TOKEN") if not base_url: raise SysledAPIError("Defina SYSLED_URL no ambiente do servidor.", code="INVALID_CONFIGURATION") if not isinstance(token, str) or not _TOKEN_RE.fullmatch(token): raise SysledAPIError("Defina SYSLED_TOKEN como segredo hexadecimal de 64 caracteres.", code="INVALID_CONFIGURATION") if not isinstance(timeout, (int, float)) or isinstance(timeout, bool) or not math.isfinite(timeout) or timeout <= 0 or not isinstance(max_response_bytes, int) or isinstance(max_response_bytes, bool) or max_response_bytes <= 0: raise SysledAPIError("timeout e max_response_bytes devem ser positivos.", code="INVALID_CONFIGURATION") self._base_url = _normalize_base_url(base_url) self._token = token self._timeout = float(timeout) self._transport = transport or _default_transport self._max_response_bytes = max_response_bytes def call(self, method: PublicMethod, params: Mapping[str, object] | None = None) -> dict[str, Any]: if method not in PUBLIC_METHODS: raise SysledAPIError("Método fora do perfil público public-consultations-v1.", code="METHOD_NOT_PUBLIC") url = f"{self._base_url}/api/v1/{quote(self._token, safe='')}/{quote(method, safe='.') }" body = None try: body = json.dumps(dict(params or {}), ensure_ascii=False, separators=(",", ":"), allow_nan=False).encode("utf-8") except (TypeError, ValueError): pass if body is None: raise SysledAPIError(f"Parâmetros não serializáveis em {method}.", code="INVALID_REQUEST", method=method) request = Request(url, data=body, method="POST", headers={"Content-Type": "application/json", "Accept": "application/json"}) response: Any failure = None try: response = self._transport(request, self._timeout) except HTTPError as error: response = error except (TimeoutError, socket.timeout): failure = SysledAPIError(f"A chamada {method} excedeu {self._timeout:g} s.", code="TIMEOUT", method=method) except (URLError, OSError) as error: failure = self._transport_error(error, method) except Exception as error: failure = self._transport_error(error, method) # Raise outside except: __context__/traceback must not retain the raw exception/URL. if failure is not None: raise failure status = 0 headers = None raw = b"" try: status = int(response.status if hasattr(response, "status") else response.getcode()) headers = getattr(response, "headers", None) raw = response.read(self._max_response_bytes + 1) except Exception as error: failure = self._transport_error(error, method) finally: close = getattr(response, "close", None) if callable(close): try: close() except Exception as error: failure = failure or self._transport_error(error, method) if failure is not None: raise failure if 300 <= status < 400: raise SysledAPIError("Redirecionamento recusado; confirme o host com o administrador.", status=status, code="REDIRECT_REFUSED", method=method) if len(raw) > self._max_response_bytes: raise SysledAPIError(f"Resposta excedeu o limite em {method}.", status=status, code="INVALID_RESPONSE", method=method) payload = None try: payload = json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError): pass if payload is None: raise SysledAPIError(f"Resposta JSON inválida em {method}.", status=status, code="INVALID_RESPONSE", method=method) request_id_raw = _header(headers, "X-Request-Id") request_id = redact_secrets(request_id_raw, self._token) if request_id_raw else None if status < 200 or status >= 300: code = payload.get("code") if isinstance(payload, dict) and isinstance(payload.get("code"), str) else None candidate = None if isinstance(payload, dict): candidate = payload.get("message") if isinstance(payload.get("message"), str) else payload.get("error") message = redact_secrets(candidate or f"A API SYSLED respondeu HTTP {status}.", self._token) raise SysledAPIError(message, status=status, code=redact_secrets(code, self._token) if code else None, method=method, request_id=request_id, retry_after_seconds=_retry_after(headers)) if not isinstance(payload, dict) or payload.get("ok") is not True or payload.get("method") != method or "result" not in payload: raise SysledAPIError(f"Envelope incoerente em {method}.", status=status, code="INVALID_RESPONSE", method=method, request_id=request_id) if not isinstance(payload["result"], dict): raise SysledAPIError(f"result não é um objeto em {method}.", status=status, code="INVALID_RESPONSE", method=method, request_id=request_id) return cast(dict[str, Any], payload["result"]) def _transport_error(self, error: Exception, method: str) -> SysledAPIError: reason = error.reason if isinstance(error, URLError) else error if isinstance(reason, (TimeoutError, socket.timeout)): return SysledAPIError(f"A chamada {method} excedeu {self._timeout:g} s.", code="TIMEOUT", method=method) safe = redact_secrets(error, self._token) return SysledAPIError(f"Falha de transporte em {method}: {safe}", code="NETWORK_ERROR", method=method) def messages_list(self, params: MessagesListRequest) -> MessagesListResult: return cast(MessagesListResult, self.call("messages.list", params)) def marketing_funnel(self, *, from_date: str, to: str, pipeline_id: int) -> MarketingFunnelResult: return cast(MarketingFunnelResult, self.call("marketing.funnel", {"from": from_date, "to": to, "pipeline_id": pipeline_id})) def leads_list(self, params: LeadsListRequest | None = None) -> LeadsListResult: return cast(LeadsListResult, self.call("leads.list", params)) def pipelines_list(self, params: PipelinesListRequest | None = None) -> PipelinesListResult: return cast(PipelinesListResult, self.call("pipelines.list", params)) def pipelines_get(self, params: PipelinesGetRequest) -> PipelinesGetResult: return cast(PipelinesGetResult, self.call("pipelines.get", params)) def stages_list(self, params: StagesListRequest) -> StagesListResult: return cast(StagesListResult, self.call("stages.list", params)) def tasks_list(self, params: TasksListRequest | None = None) -> TasksListResult: return cast(TasksListResult, self.call("tasks.list", params)) def tasks_get(self, params: TasksGetRequest) -> TasksGetResult: return cast(TasksGetResult, self.call("tasks.get", params)) def products_list(self, params: ProductsQueryRequest | None = None) -> ProductsListResult: return cast(ProductsListResult, self.call("products.list", params)) def products_search(self, params: ProductsQueryRequest | None = None) -> ProductsListResult: return cast(ProductsListResult, self.call("products.search", params)) def products_get(self, params: ProductsGetRequest) -> ProductsGetResult: return cast(ProductsGetResult, self.call("products.get", params)) def products_brands(self) -> ProductsBrandsResult: return cast(ProductsBrandsResult, self.call("products.brands", {})) def products_categories(self) -> ProductsCategoriesResult: return cast(ProductsCategoriesResult, self.call("products.categories", {})) def shop_banners_list(self, params: ShopBannersListRequest | None = None) -> ShopBannersListResult: return cast(ShopBannersListResult, self.call("shop.banners.list", params)) def shop_frete_config(self) -> ShopFreteConfigResult: return cast(ShopFreteConfigResult, self.call("shop.frete.config", {})) def shop_frete_calcular(self, params: ShopFreteCalcularRequest) -> ShopFreteCalcularResult: return cast(ShopFreteCalcularResult, self.call("shop.frete.calcular", params)) def shop_desconto_validar(self, params: ShopDescontoValidarRequest) -> ShopDescontoValidarResult: return cast(ShopDescontoValidarResult, self.call("shop.desconto.validar", params)) def shop_loja_config(self) -> ShopLojaConfigResult: return cast(ShopLojaConfigResult, self.call("shop.loja.config", {}))