"""TBank payment adapter.""" import asyncio from collections.abc import Awaitable, Callable import hashlib import hmac from typing import Any import httpx from app.adapters.tbank.base import ( TBankPaymentAdapterError, TBankPaymentNotificationTokenError, TBankPaymentRequestError, ) from app.config import TBankPaymentConfig from app.schemas.payment import TBankPaymentNotification class TBankAdapter: def __init__( self, http_client: httpx.AsyncClient, *, init_url: str, notification_url: str, success_url: str, terminal_key: str, password: str, timeout_seconds: float = 10.0, retry_attempts: int = 2, retry_backoff_seconds: float = 0.2, sleep: Callable[[float], Awaitable[None]] = asyncio.sleep, ) -> None: self._http_client = http_client self._init_url = init_url self._notification_url = notification_url self._success_url = success_url self._terminal_key = terminal_key self._password = password self._timeout_seconds = timeout_seconds self._retry_attempts = retry_attempts self._retry_backoff_seconds = retry_backoff_seconds self._sleep = sleep @classmethod def from_config( cls, *, http_client: httpx.AsyncClient, config: TBankPaymentConfig, ) -> "TBankAdapter": return cls( http_client=http_client, init_url=config.init_url, notification_url=config.notification_url, success_url=config.success_url, terminal_key=config.auth.terminal_key, password=config.auth.password, timeout_seconds=config.timeout_seconds, retry_attempts=config.retry_attempts, retry_backoff_seconds=config.retry_backoff_seconds, ) async def create_payment_link(self, order_uuid: str, amount_kopecks: int) -> str: payload = self._build_payload( order_uuid=order_uuid, amount_kopecks=amount_kopecks, ) for attempt in range(self._retry_attempts + 1): try: response = await self._http_client.post( self._init_url, json=payload, headers={ "Accept": "application/json", "Content-Type": "application/json", }, timeout=self._timeout_seconds, ) except (httpx.TimeoutException, httpx.TransportError) as exc: if attempt < self._retry_attempts: await self._sleep(self._retry_delay(attempt)) continue raise TBankPaymentAdapterError( "TBank payment init request failed after retry attempts." ) from exc if self._should_retry(response.status_code): if attempt < self._retry_attempts: await self._sleep(self._retry_delay(attempt)) continue raise TBankPaymentAdapterError( "TBank payment init request failed with retriable status " f"{response.status_code}." ) if 400 <= response.status_code < 500: raise _build_tbank_request_error( "TBank payment init request was rejected.", status_code=response.status_code, payload=_response_json_or_none(response), ) try: response.raise_for_status() raw_payload = response.json() except (httpx.HTTPError, TypeError, ValueError) as exc: raise TBankPaymentAdapterError( "TBank payment init returned invalid payload." ) from exc return self._map_payment_url(raw_payload) raise TBankPaymentAdapterError("TBank payment init request failed unexpectedly.") def verify_payment_notification( self, notification: TBankPaymentNotification, ) -> None: payload = notification.model_dump(mode="python") expected_token = _build_tbank_token(payload, password=self._password) if not hmac.compare_digest(expected_token, notification.Token): raise TBankPaymentNotificationTokenError( "TBank payment notification token is invalid." ) def _build_payload(self, *, order_uuid: str, amount_kopecks: int) -> dict[str, Any]: if not order_uuid.strip(): raise TBankPaymentRequestError("TBank payment order id must be non-empty.") if amount_kopecks <= 0: raise TBankPaymentRequestError("TBank payment amount must be positive.") payload: dict[str, Any] = { "TerminalKey": self._terminal_key, "Amount": amount_kopecks, "OrderId": order_uuid, "NotificationURL": self._notification_url, "SuccessURL": self._success_url, } payload["Token"] = _build_tbank_token(payload, password=self._password) return payload def _retry_delay(self, attempt: int) -> float: return self._retry_backoff_seconds * (attempt + 1) @staticmethod def _should_retry(status_code: int) -> bool: return status_code == 429 or status_code >= 500 @staticmethod def _map_payment_url(payload: object) -> str: if not isinstance(payload, dict): raise TBankPaymentAdapterError( "TBank payment init payload must be a JSON object." ) if payload.get("Success") is False: raise _build_tbank_request_error( "TBank payment init request was rejected.", payload=payload, ) payment_url = payload.get("PaymentURL") if not isinstance(payment_url, str) or not payment_url.strip(): raise TBankPaymentAdapterError( "TBank payment init response must include PaymentURL." ) return payment_url.strip() def _build_tbank_token(payload: dict[str, Any], *, password: str) -> str: token_payload = { key: value for key, value in payload.items() if key != "Token" and not isinstance(value, (dict, list)) } token_payload["Password"] = password token_source = "".join( _stringify_token_value(token_payload[key]) for key in sorted(token_payload) ) return hashlib.sha256(token_source.encode("utf-8")).hexdigest() def _stringify_token_value(value: Any) -> str: if isinstance(value, bool): return "true" if value else "false" return str(value) def _response_json_or_none(response: httpx.Response) -> object | None: try: return response.json() except (TypeError, ValueError): return None def _build_tbank_request_error( message: str, *, status_code: int | None = None, payload: object | None = None, ) -> TBankPaymentRequestError: error_code = _payload_text_value(payload, "ErrorCode") provider_message = _payload_text_value(payload, "Message") details = _payload_text_value(payload, "Details") return TBankPaymentRequestError( _format_tbank_request_error_message( message, status_code=status_code, error_code=error_code, provider_message=provider_message, details=details, ), status_code=status_code, error_code=error_code, provider_message=provider_message, details=details, ) def _payload_text_value(payload: object | None, key: str) -> str | None: if not isinstance(payload, dict): return None value = payload.get(key) if value is None: return None text = str(value).strip() if not text: return None return text def _format_tbank_request_error_message( message: str, *, status_code: int | None, error_code: str | None, provider_message: str | None, details: str | None, ) -> str: fields: list[str] = [] if status_code is not None: fields.append(f"status_code={status_code}") if error_code is not None: fields.append(f"error_code={error_code}") if provider_message is not None: fields.append(f"message={provider_message}") if details is not None: fields.append(f"details={details}") if not fields: return message return f"{message} {' '.join(fields)}"