108 lines
3.1 KiB
Python
108 lines
3.1 KiB
Python
"""Клиент Crypto Pay API (@CryptoBot)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
|
|
class CryptoPayError(RuntimeError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class Invoice:
|
|
invoice_id: int
|
|
status: str
|
|
amount: str
|
|
pay_url: str
|
|
bot_invoice_url: str
|
|
payload: str | None = None
|
|
fiat: str | None = None
|
|
|
|
|
|
class CryptoPay:
|
|
def __init__(self, token: str, *, testnet: bool = False) -> None:
|
|
self.token = token.strip()
|
|
self.base_url = (
|
|
"https://testnet-pay.crypt.bot/api"
|
|
if testnet
|
|
else "https://pay.crypt.bot/api"
|
|
)
|
|
|
|
@property
|
|
def enabled(self) -> bool:
|
|
return bool(self.token)
|
|
|
|
async def _call(self, method: str, **params: Any) -> Any:
|
|
if not self.enabled:
|
|
raise CryptoPayError("CRYPTO_BOT_TOKEN не задан")
|
|
headers = {"Crypto-Pay-API-Token": self.token}
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
resp = await client.post(
|
|
f"{self.base_url}/{method}",
|
|
headers=headers,
|
|
json={k: v for k, v in params.items() if v is not None},
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
if not data.get("ok"):
|
|
raise CryptoPayError(str(data.get("error") or data))
|
|
return data["result"]
|
|
|
|
async def get_me(self) -> dict[str, Any]:
|
|
return await self._call("getMe")
|
|
|
|
async def create_fiat_invoice(
|
|
self,
|
|
*,
|
|
amount_rub: float,
|
|
description: str,
|
|
payload: str,
|
|
expires_in: int = 1800,
|
|
accepted_assets: str = "USDT,TON,BTC,ETH,LTC",
|
|
) -> Invoice:
|
|
result = await self._call(
|
|
"createInvoice",
|
|
currency_type="fiat",
|
|
fiat="RUB",
|
|
amount=f"{amount_rub:.2f}",
|
|
description=description[:1024],
|
|
payload=payload[:4096],
|
|
accepted_assets=accepted_assets,
|
|
expires_in=expires_in,
|
|
allow_comments=False,
|
|
allow_anonymous=True,
|
|
)
|
|
return self._parse_invoice(result)
|
|
|
|
async def get_invoice(self, invoice_id: int) -> Invoice | None:
|
|
result = await self._call(
|
|
"getInvoices",
|
|
invoice_ids=str(invoice_id),
|
|
)
|
|
items = result if isinstance(result, list) else result.get("items") or []
|
|
if not items:
|
|
return None
|
|
return self._parse_invoice(items[0])
|
|
|
|
@staticmethod
|
|
def _parse_invoice(raw: dict[str, Any]) -> Invoice:
|
|
pay_url = (
|
|
raw.get("bot_invoice_url")
|
|
or raw.get("pay_url")
|
|
or raw.get("mini_app_invoice_url")
|
|
or ""
|
|
)
|
|
return Invoice(
|
|
invoice_id=int(raw["invoice_id"]),
|
|
status=str(raw.get("status") or "active"),
|
|
amount=str(raw.get("amount") or ""),
|
|
pay_url=pay_url,
|
|
bot_invoice_url=str(raw.get("bot_invoice_url") or pay_url),
|
|
payload=raw.get("payload"),
|
|
fiat=raw.get("fiat"),
|
|
)
|