160 lines
5.0 KiB
Python
160 lines
5.0 KiB
Python
"""Клиент 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")),
|
|
)
|