80 lines
3.0 KiB
Python
80 lines
3.0 KiB
Python
import frappe
|
|
import requests
|
|
from kapital_bank.auth import refresh_token, get_default_login, BROWSER_HEADERS
|
|
|
|
BASE_URL = "https://my.birbank.business/api/b2b"
|
|
|
|
|
|
class BIRBankClient:
|
|
"""HTTP client for BIRBank B2B API with automatic token refresh on 401."""
|
|
|
|
def __init__(self, login_name=None):
|
|
if not login_name:
|
|
login_name = get_default_login()
|
|
if not login_name:
|
|
frappe.throw("No authenticated Kapital Bank Login found")
|
|
self.login_name = login_name
|
|
self._load_token()
|
|
|
|
def _load_token(self):
|
|
doc = frappe.get_doc("Kapital Bank Login", self.login_name)
|
|
self.token = doc.get_password("jwt_token") if doc.jwt_token else ""
|
|
|
|
def _headers(self):
|
|
return {**BROWSER_HEADERS, "Authorization": f"Bearer {self.token}"}
|
|
|
|
def _request(self, method, path, **kwargs):
|
|
url = f"{BASE_URL}{path}"
|
|
kwargs.setdefault("timeout", 30)
|
|
|
|
resp = requests.request(method, url, headers=self._headers(), **kwargs)
|
|
|
|
if resp.status_code == 401:
|
|
# Try refreshing once
|
|
frappe.logger().info(f"[KB] 401 on {path}, refreshing token for {self.login_name}")
|
|
self.token = refresh_token(self.login_name)
|
|
resp = requests.request(method, url, headers=self._headers(), **kwargs)
|
|
|
|
if not resp.ok:
|
|
frappe.log_error(
|
|
f"[KB] {method} {url} → HTTP {resp.status_code}\nResponse body: {resp.text[:2000]}",
|
|
"KB API Response Error"
|
|
)
|
|
|
|
return resp
|
|
|
|
def get(self, path, params=None):
|
|
return self._request("GET", path, params=params)
|
|
|
|
def post(self, path, json=None, data=None, files=None, headers_override=None):
|
|
if files:
|
|
# For multipart requests, don't set Content-Type (requests sets it automatically)
|
|
hdrs = {k: v for k, v in BROWSER_HEADERS.items() if k != "Content-Type"}
|
|
hdrs["Authorization"] = f"Bearer {self.token}"
|
|
url = f"{BASE_URL}{path}"
|
|
resp = requests.post(url, headers=hdrs, data=data, files=files, timeout=30)
|
|
if resp.status_code == 401:
|
|
self.token = refresh_token(self.login_name)
|
|
hdrs["Authorization"] = f"Bearer {self.token}"
|
|
resp = requests.post(url, headers=hdrs, data=data, files=files, timeout=30)
|
|
if not resp.ok:
|
|
frappe.log_error(
|
|
f"[KB] POST {url} (multipart) → HTTP {resp.status_code}\nResponse body: {resp.text[:2000]}",
|
|
"KB API Response Error"
|
|
)
|
|
return resp
|
|
return self._request("POST", path, json=json)
|
|
|
|
def delete(self, path, params=None):
|
|
return self._request("DELETE", path, params=params)
|
|
|
|
def get_json(self, path, params=None):
|
|
resp = self.get(path, params=params)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
def post_json(self, path, json=None):
|
|
resp = self.post(path, json=json)
|
|
resp.raise_for_status()
|
|
return resp.json()
|