"""Cliente REST mínimo para Cegid Jasmin. Validado contra Jasmin 3.02: - OAuth Client Credentials em identity.primaverabss.com/connect/token; - endpoints em https://my.jasminsoftware.com/api/{account}/{subscription}; - OData com $top máximo 100; - invoice from quotation exige body JSON vazio ({}) para evitar HTTP 411. """ from __future__ import annotations import time from dataclasses import dataclass from typing import Any, Dict, List, Optional import httpx from app.config import settings class JasminError(RuntimeError): pass @dataclass(frozen=True) class JasminConfig: base_url: str token_url: str account: str subscription: str client_id: str client_secret: str scope: str = "application" _TOKEN_CACHE: Dict[str, Any] = {"access_token": "", "expires_at": 0.0} def get_jasmin_config() -> JasminConfig: cfg = JasminConfig( base_url=(settings.jasmin_base_url or "https://my.jasminsoftware.com").rstrip("/"), token_url=(settings.jasmin_token_url or "https://identity.primaverabss.com/connect/token").strip(), account=(settings.jasmin_account or "").strip(), subscription=(settings.jasmin_subscription or "").strip(), client_id=(settings.jasmin_client_id or "").strip(), client_secret=(settings.jasmin_client_secret or "").strip(), scope=(settings.jasmin_scope or "application").strip() or "application", ) missing = [name for name, value in { "JASMIN_ACCOUNT": cfg.account, "JASMIN_SUBSCRIPTION": cfg.subscription, "JASMIN_CLIENT_ID": cfg.client_id, "JASMIN_CLIENT_SECRET": cfg.client_secret, }.items() if not value] if missing: raise JasminError("Configuração Jasmin incompleta: " + ", ".join(missing)) return cfg class JasminClient: def __init__(self, config: Optional[JasminConfig] = None, timeout: float = 30.0): self.config = config or get_jasmin_config() self.timeout = timeout @property def api_root(self) -> str: return f"{self.config.base_url}/api/{self.config.account}/{self.config.subscription}" async def get_token(self, *, force_refresh: bool = False) -> str: now = time.time() if not force_refresh and _TOKEN_CACHE.get("access_token") and float(_TOKEN_CACHE.get("expires_at") or 0) > now + 60: return str(_TOKEN_CACHE["access_token"]) data = {"grant_type": "client_credentials", "scope": self.config.scope} async with httpx.AsyncClient(timeout=self.timeout) as client: response = await client.post( self.config.token_url, data=data, auth=(self.config.client_id, self.config.client_secret), headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"}, ) if response.status_code >= 400: raise JasminError(f"OAuth Jasmin falhou {response.status_code}: {response.text}") payload = response.json() token = payload.get("access_token") if not token: raise JasminError(f"OAuth Jasmin não devolveu access_token: {payload}") expires_in = int(payload.get("expires_in") or 3600) _TOKEN_CACHE.update({"access_token": token, "expires_at": now + expires_in}) return str(token) async def _request( self, method: str, path: str, *, params: Optional[Dict[str, Any]] = None, json_body: Any = None, headers: Optional[Dict[str, str]] = None, ) -> Any: token = await self.get_token() url = f"{self.api_root}/{path.lstrip('/')}" request_headers = { "Authorization": f"Bearer {token}", "Accept": "application/json", } if json_body is not None: request_headers["Content-Type"] = "application/json" if headers: request_headers.update(headers) async with httpx.AsyncClient(timeout=self.timeout, follow_redirects=True) as client: response = await client.request( method.upper(), url, params=params, json=json_body, headers=request_headers, ) if response.status_code >= 400: detail: Any = response.text try: detail = response.json() except Exception: pass raise JasminError(f"Jasmin error {response.status_code} em {method.upper()} {path}: {detail}") if not response.content: return None try: return response.json() except Exception: return response.text async def _raw_request( self, method: str, path: str, *, accept: str = "application/pdf", json_body: Any = None, ) -> tuple[bytes, str]: """Pedido autenticado para respostas binárias, usado por PDFs Jasmin.""" token = await self.get_token() url = f"{self.api_root}/{path.lstrip('/')}" headers = {"Authorization": f"Bearer {token}", "Accept": accept} if json_body is not None: headers["Content-Type"] = "application/json" async with httpx.AsyncClient(timeout=self.timeout, follow_redirects=True) as client: response = await client.request(method.upper(), url, json=json_body, headers=headers) if response.status_code >= 400: detail: Any = response.text try: detail = response.json() except Exception: pass raise JasminError(f"Jasmin error {response.status_code} em {method.upper()} {path}: {detail}") return response.content, response.headers.get("content-type", accept) async def print_quotation_pdf(self, quotation_id: str) -> tuple[bytes, str]: return await self._raw_request("GET", f"/sales/quotations/{quotation_id}/print", accept="application/pdf") async def print_invoice_pdf(self, invoice_id: str) -> tuple[bytes, str]: return await self._raw_request("GET", f"/billing/invoices/{invoice_id}/print", accept="application/pdf") async def get_versions(self) -> Dict[str, Any]: return await self._request("GET", "/businessCore/productInfos/getVersions") async def list_customers_odata(self, *, top: int = 100, skip: int = 0) -> Dict[str, Any]: top = min(max(int(top), 1), 100) return await self._request("GET", "/salesCore/customerParties/extension/odata", params={"$top": top, "$skip": int(skip)}) async def get_customer_by_tax_id(self, tax_id: str) -> Any: return await self._request("GET", f"/salesCore/customerParties/getCustomerByCompanyTaxId/{tax_id}") async def get_customer_by_party_key(self, party_key: str) -> Dict[str, Any]: return await self._request("GET", f"/salesCore/customerParties/{party_key}") async def create_customer(self, payload: Dict[str, Any]) -> str: result = await self._request("POST", "/salesCore/customerParties", json_body=payload) return str(result).strip('"') async def list_sales_items(self, *, top: int = 100, skip: int = 0) -> Dict[str, Any]: top = min(max(int(top), 1), 100) return await self._request("GET", "/salesCore/salesItems/extension/odata", params={"$top": top, "$skip": int(skip)}) async def get_sales_item(self, item_key: str) -> Dict[str, Any]: return await self._request("GET", f"/salesCore/salesItems/{item_key}") async def create_quotation(self, payload: Dict[str, Any]) -> str: result = await self._request("POST", "/sales/quotations", json_body=payload) return str(result).strip('"') async def get_quotation(self, quotation_id: str) -> Dict[str, Any]: return await self._request("GET", f"/sales/quotations/{quotation_id}") async def list_quotations( self, *, top: int = 100, skip: int = 0, filter: Optional[str] = None, orderby: Optional[str] = None, ) -> Dict[str, Any]: top = min(max(int(top), 1), 100) params: Dict[str, Any] = {"$top": top, "$skip": int(skip)} if filter: params["$filter"] = filter if orderby: params["$orderby"] = orderby return await self._request("GET", "/sales/quotations/odata", params=params) async def create_invoice_from_quotation(self, quotation_id: str) -> str: # Body vazio é obrigatório no tenant testado; sem body devolve HTTP 411. result = await self._request("POST", f"/billing/invoices/fromQuotation/{quotation_id}", json_body={}) return str(result).strip('"') async def get_invoice(self, invoice_id: str) -> Dict[str, Any]: return await self._request("GET", f"/billing/invoices/{invoice_id}") async def list_invoices( self, *, top: int = 100, skip: int = 0, filter: Optional[str] = None, orderby: Optional[str] = None, ) -> Dict[str, Any]: top = min(max(int(top), 1), 100) params: Dict[str, Any] = {"$top": top, "$skip": int(skip)} if filter: params["$filter"] = filter if orderby: params["$orderby"] = orderby return await self._request("GET", "/billing/invoices/odata", params=params)