"""Клиент Heleket Payment API (резервная крипто-оплата).""" from __future__ import annotations import base64 import hashlib import json import logging from dataclasses import dataclass from typing import Any import httpx logger = logging.getLogger(__name__) BASE_URL = "https://api.heleket.com/v1" PAID_STATUSES = frozenset({"paid", "paid_over"}) class HeleketError(RuntimeError): pass @dataclass(frozen=True, slots=True) class HeleketInvoice: uuid: str order_id: str amount: str status: str pay_url: str currency: str | None = None is_final: bool = False def _json_body(payload: dict[str, Any]) -> str: # Как в PHP json_encode: без пробелов, unicode as-is, слэши экранируются raw = json.dumps(payload, ensure_ascii=False, separators=(",", ":")) return raw.replace("/", "\\/") def sign_request(payload: dict[str, Any], api_key: str) -> str: body = _json_body(payload) encoded = base64.b64encode(body.encode("utf-8")).decode("ascii") return hashlib.md5((encoded + api_key).encode("utf-8")).hexdigest() def verify_webhook_sign(data: dict[str, Any], api_key: str) -> bool: payload = dict(data) received = str(payload.pop("sign", "") or "") if not received: return False return sign_request(payload, api_key) == received class Heleket: def __init__( self, merchant_id: str, api_key: str, *, callback_url: str = "", currency: str = "RUB", ) -> None: self.merchant_id = merchant_id.strip() self.api_key = api_key.strip() self.callback_url = (callback_url or "").strip().rstrip("/") self.currency = (currency or "RUB").strip().upper() or "RUB" @property def enabled(self) -> bool: return bool(self.merchant_id and self.api_key) async def _post(self, path: str, payload: dict[str, Any] | None = None) -> dict[str, Any]: if not self.enabled: raise HeleketError("Heleket не настроен (HELEKET_MERCHANT_ID / HELEKET_API_KEY)") body_obj = payload if payload is not None else {} body = _json_body(body_obj) headers = { "merchant": self.merchant_id, "sign": sign_request(body_obj, self.api_key), "Content-Type": "application/json", } async with httpx.AsyncClient(timeout=30.0) as client: resp = await client.post( f"{BASE_URL}{path}", headers=headers, content=body.encode("utf-8"), ) try: data = resp.json() except Exception as exc: # noqa: BLE001 raise HeleketError(f"HTTP {resp.status_code}: {resp.text[:300]}") from exc state = data.get("state") if resp.status_code >= 400 or (state not in (0, "0", None) and state != 0): errors = data.get("errors") or data.get("message") or data raise HeleketError(str(errors)) result = data.get("result") if not isinstance(result, dict): raise HeleketError(f"Неожиданный ответ: {data}") return result async def create_invoice( self, *, amount_rub: float, order_id: str, description: str = "", lifetime: int = 3600, ) -> HeleketInvoice: payload: dict[str, Any] = { "amount": f"{amount_rub:.2f}", "currency": self.currency, "order_id": order_id, "lifetime": max(300, min(lifetime, 43200)), "is_payment_multiple": False, "theme": "dark", } if self.callback_url: payload["url_callback"] = self.callback_url if description: payload["additional_data"] = description[:255] result = await self._post("/payment", payload) return self._parse(result) async def get_invoice( self, *, order_id: str | None = None, uuid: str | None = None, ) -> HeleketInvoice | None: payload: dict[str, Any] = {} if order_id: payload["order_id"] = order_id if uuid: payload["uuid"] = uuid if not payload: raise HeleketError("Нужен order_id или uuid") try: result = await self._post("/payment/info", payload) except HeleketError: return None return self._parse(result) @staticmethod def is_paid(status: str) -> bool: return status in PAID_STATUSES @staticmethod def _parse(raw: dict[str, Any]) -> HeleketInvoice: status = str(raw.get("payment_status") or raw.get("status") or "") return HeleketInvoice( uuid=str(raw.get("uuid") or ""), order_id=str(raw.get("order_id") or ""), amount=str(raw.get("amount") or ""), status=status, pay_url=str(raw.get("url") or ""), currency=raw.get("currency"), is_final=bool(raw.get("is_final")), )