235 lines
9.2 KiB
Python
235 lines
9.2 KiB
Python
"""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)
|