""" TConnect Payment SDK (single-file) — Python Wraps the hard parts of the TConnect Payment Open API: - JWT login + automatic token refresh - AES-256-CBC encryption / decryption for every request and IPN - Header normalization; call any endpoint in one line Install dependencies: pip install requests cryptography 7-line example — accept your first QR payment: from tconnect import TConnect tc = TConnect(partner_code="81234567", username="demo@yourdomain.vn", password="Password#123", client_id="...", client_secret="...", secret_key="", service_code="vccb-qr", va="VA100023312", bincode="970454") qr = tc.create_qr(order_id="ORDER123", amount=150000) print(qr["image_png_base64"]) # drop straight into

Full documentation: https://docs.tconnect.vn """ import os import json import time import uuid import binascii import requests from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from cryptography.hazmat.primitives import padding from cryptography.hazmat.backends import default_backend DEFAULT_BASE_URL = "https://sme-open-api-sandbox.tconnect.vn" class TConnectError(Exception): """Error returned by the TConnect API (carries HTTP status + body when available).""" def __init__(self, message, status=None, body=None): super().__init__(message) self.status = status self.body = body class TConnect: def __init__( self, partner_code, secret_key, username=None, password=None, client_id=None, client_secret=None, service_code=None, va=None, bincode=None, base_url=DEFAULT_BASE_URL, timeout=30, ): """ partner_code Merchant identifier (provided by TConnect). secret_key AES-256 key as a hex string (provided by TConnect). username/password/client_id/client_secret JWT login credentials. service_code/va/bincode Defaults for create_qr (can be overridden per call). """ self.partner_code = str(partner_code) self.key = bytes.fromhex(secret_key) self.username = username self.password = password self.client_id = client_id self.client_secret = client_secret self.service_code = service_code self.va = va self.bincode = bincode self.base_url = base_url.rstrip("/") self.timeout = timeout self._access_token = None self._refresh_token = None self._access_expire_at = 0 self._refresh_expire_at = 0 # ---------- AES-256-CBC ---------- def encrypt(self, plain): """Encrypt a string/JSON -> hex (IV prepended to the ciphertext).""" if not isinstance(plain, str): plain = json.dumps(plain, ensure_ascii=False, separators=(",", ":")) iv = os.urandom(16) padder = padding.PKCS7(128).padder() data = padder.update(plain.encode()) + padder.finalize() cipher = Cipher(algorithms.AES(self.key), modes.CBC(iv), backend=default_backend()) enc = cipher.encryptor() ct = enc.update(data) + enc.finalize() return binascii.hexlify(iv + ct).decode() def decrypt(self, encrypted): """Decrypt hex -> original string.""" raw = binascii.unhexlify(encrypted) if len(raw) < 16: raise TConnectError("ciphertext too short") iv, ct = raw[:16], raw[16:] cipher = Cipher(algorithms.AES(self.key), modes.CBC(iv), backend=default_backend()) dec = cipher.decryptor() padded = dec.update(ct) + dec.finalize() unpadder = padding.PKCS7(128).unpadder() return (unpadder.update(padded) + unpadder.finalize()).decode() # ---------- Auth ---------- def login(self): payload = { "username": self.username, "password": self.password, "client_id": self.client_id, "client_secret": self.client_secret, } res = self._post("/openapi/v1/auth/login", payload, auth=False, encrypted=True) self._store_token(res) return res def refresh(self): if not self._refresh_token or time.time() >= self._refresh_expire_at: return self.login() res = self._post( "/openapi/v1/auth/refresh", {"refresh_token": self._refresh_token}, auth=False, encrypted=True, ) self._store_token(res) return res def _store_token(self, res): now = time.time() self._access_token = res.get("access_token") self._refresh_token = res.get("refresh_token") # subtract 60s of headroom so we never use a token right at expiry self._access_expire_at = now + max(res.get("expires_in", 0) - 60, 0) self._refresh_expire_at = now + max(res.get("refresh_expires_in", 0) - 60, 0) def _token(self): if not self._access_token or time.time() >= self._access_expire_at: if self._refresh_token and time.time() < self._refresh_expire_at: self.refresh() else: self.login() return self._access_token # ---------- Services (not encrypted) ---------- def get_services(self, service_type=None, code=None, payment_method=None, limit=10, page=1): params = {"limit": limit, "page": page} if service_type: params["service_type"] = service_type if code: params["code"] = code if payment_method: params["payment_method"] = payment_method headers = {"Partner-Code": self.partner_code, "Authorization": f"Bearer {self._token()}"} r = requests.get( f"{self.base_url}/openapi/v1/services", params=params, headers=headers, timeout=self.timeout, ) return self._parse(r, encrypted=False) # ---------- Virtual Account ---------- def create_va(self, full_name, bank_account_no, internal_code, request_id=None): payload = { "request_id": request_id or str(uuid.uuid4()), "full_name": full_name, "bank_account_no": bank_account_no, "internal_code": internal_code, } return self._post("/openapi/v1/va/va-account/create", payload) # ---------- QR ---------- def create_qr(self, order_id, amount=0, va=None, bincode=None, service_code=None, req_id=None): """Create a payment QR (amount=0 -> static QR, amount>0 -> dynamic QR).""" va = va or self.va bincode = bincode or self.bincode service_code = service_code or self.service_code if not (va and bincode and service_code): raise TConnectError("create_qr requires va, bincode and service_code (set at init or pass in)") payload = { "req_id": req_id or str(uuid.uuid4()), "order_id": order_id, "va": va, "bincode": bincode, "amount": amount, } return self._post( "/openapi/v1/transaction/qr/generate", payload, extra_headers={"x-service-code": service_code}, ) # ---------- Push to devices (POS card / Soundbox QR) ---------- def push_to_device(self, serial_no, amount=None, type="pos", order_id=None, qr_string=None, action=None, action_data=None, trace_no=None, txn_id=None, request_id=None, extra_data=None, customer_code=None): """Push a payment command to a physical device at the counter. type="pos" POS terminal — CARD payment. Pick the operation via `action`: SALE (default) / MOTO / PREAUTH / PREAUTH_COMPLETE / VOID / SETTLEMENT. VOID and PREAUTH_COMPLETE require `action_data` (retrievalRefNo); also send `trace_no` (and `txn_id` for PREAUTH_COMPLETE). SETTLEMENT (end-of-day batch) needs no order_id/amount. type="soundbox" Soundbox speaker — requires `qr_string` (from create_qr). customer_code Sets the X-Customer-Code header (required when the token is a Service Provider token). """ payload = {"serial_no": serial_no, "type": type} if amount is not None: payload["amount"] = amount if order_id: payload["order_id"] = order_id if type == "soundbox": if qr_string: payload["qr_string"] = qr_string else: # pos if action: payload["action"] = action if action_data: payload["action_data"] = action_data if trace_no: payload["trace_no"] = trace_no if txn_id: payload["txn_id"] = txn_id payload["request_id"] = request_id or str(uuid.uuid4()) if extra_data is not None: payload["extra_data"] = extra_data extra_headers = {"X-Customer-Code": customer_code} if customer_code else None return self._post("/openapi/v1/devices/payments/push", payload, extra_headers=extra_headers) # ---------- Ecom (online payment link via Digistore) ---------- def create_payment_link(self, order_id, amount, currency="VND", description=None, return_url=None, ipn_url=None, expires_in=900, customer=None): """Create an ECOM payment link for an order -> returns a payment_url the customer opens in a browser. customer: dict {"name": ..., "email": ..., "phone": ...} (optional). Returns a dict with payment_id, payment_url, status, expires_at, ... """ payload = {"order_id": order_id, "amount": amount, "currency": currency, "expires_in": expires_in} if description: payload["description"] = description if return_url: payload["return_url"] = return_url if ipn_url: payload["ipn_url"] = ipn_url if customer: payload["customer"] = customer return self._post("/openapi/v1/ecom/payments/link", payload) def cancel_payment_link(self, order_id, description=None): """Cancel a previously created ECOM payment link by order_id (the link becomes disabled).""" payload = {"order_id": order_id} if description: payload["description"] = description return self._post("/openapi/v1/ecom/payments/cancel", payload, extra_headers={"x-service-code": "ecom"}) # ---------- Transaction queries ---------- def get_qr_transactions(self, limit=10, page=1, from_date=None, to_date=None, order_id=None, acc_no=None): payload = {"limit": limit, "page": page} if from_date and to_date: payload["from_date"], payload["to_date"] = from_date, to_date if order_id: payload["order_id"] = order_id if acc_no: payload["acc_no"] = acc_no return self._post("/openapi/v1/transaction/qr", payload) def check_qr_order(self, order_id, acc_no=None): payload = {"order_id": order_id} if acc_no: payload["acc_no"] = acc_no return self._post("/openapi/v1/transaction/qr/order/status", payload) def get_card_transactions(self, limit=10, page=1, from_date=None, to_date=None, serial_no=None): """List SUCCESSFUL CARD transactions (POS). Filter by POS serial and/or a time range. Can be used to confirm a card payment: if the order_id appears in the result, it was paid.""" payload = {"limit": limit, "page": page} if from_date and to_date: payload["from_date"], payload["to_date"] = from_date, to_date if serial_no: payload["serial_no"] = serial_no return self._post("/openapi/v1/transaction/card", payload) def get_cash_transactions(self, limit=10, page=1, from_date=None, to_date=None): payload = {"limit": limit, "page": page} if from_date and to_date: payload["from_date"], payload["to_date"] = from_date, to_date return self._post("/openapi/v1/transaction/cash", payload) # ---------- IPN / Webhook ---------- def parse_ipn(self, body): """Decrypt an IPN (dict or JSON string with a 'data' field) -> transaction dict.""" if isinstance(body, (bytes, bytearray)): body = body.decode() if isinstance(body, str): body = json.loads(body) return json.loads(self.decrypt(body["data"])) # ---------- HTTP core ---------- def _post(self, path, payload, auth=True, encrypted=True, extra_headers=None): headers = {"Partner-Code": self.partner_code, "Content-Type": "application/json"} if auth: headers["Authorization"] = f"Bearer {self._token()}" if extra_headers: headers.update(extra_headers) body = {"data": self.encrypt(payload)} if encrypted else payload r = requests.post(f"{self.base_url}{path}", json=body, headers=headers, timeout=self.timeout) return self._parse(r, encrypted=False) def _parse(self, r, encrypted=False): if r.status_code >= 400: raise TConnectError(f"HTTP {r.status_code}", status=r.status_code, body=r.text) try: data = r.json() except ValueError: return r.text if encrypted and isinstance(data, dict) and "data" in data: return json.loads(self.decrypt(data["data"])) return data