"""Пополнение баланса с сайта (Lava / AnyPay / Heleket / Crypto Pay).""" from __future__ import annotations import time from dataclasses import dataclass from typing import Any from services.anypay import Anypay, AnypayError from services.cryptopay import CryptoPay, CryptoPayError from services.db import Database from services.heleket import Heleket, HeleketError from services.lava import Lava, LavaError TOPUP_MIN_RUB = 10.0 TOPUP_MAX_RUB = 3000.0 TOPUP_PRESETS_RUB: tuple[int, ...] = (50, 100, 200, 500, 1000, 2000, 3000) PROVIDERS = ("lava", "anypay", "heleket", "cryptobot") class TopupError(ValueError): pass @dataclass(frozen=True, slots=True) class TopupInvoice: provider: str amount_rub: float pay_url: str invoice_id: int order_id: str | None = None title: str = "" def parse_topup_amount(raw: Any) -> float: try: amount = float(raw) except (TypeError, ValueError) as exc: raise TopupError("Укажите сумму числом") from exc amount = round(amount, 2) if amount < TOPUP_MIN_RUB: raise TopupError(f"Минимум {int(TOPUP_MIN_RUB)} ₽ за одну оплату") if amount > TOPUP_MAX_RUB: raise TopupError(f"Максимум {int(TOPUP_MAX_RUB)} ₽ за одну оплату") return amount def normalize_provider(raw: Any) -> str: provider = str(raw or "").strip().lower() if provider in {"crypto", "cryptopay", "crypto_bot"}: provider = "cryptobot" if provider not in PROVIDERS: raise TopupError("Неизвестный способ оплаты") return provider def _safe_order_id(raw: str) -> str: cleaned = "".join(ch if ch.isalnum() or ch in "_-" else "_" for ch in raw) return cleaned[:128] def available_providers( *, cryptopay: CryptoPay, heleket: Heleket, lava: Lava, anypay: Anypay | None = None, ) -> list[dict[str, Any]]: items: list[dict[str, Any]] = [] if lava.enabled: items.append( { "id": "lava", "title": "Карта / СБП · Lava", "hint": "Lava.ru · зачисление автоматически", } ) if anypay and anypay.enabled: items.append( { "id": "anypay", "title": "AnyPay", "hint": "Карта, СБП, крипта и др. · anypay.io", } ) if heleket.enabled: items.append( { "id": "heleket", "title": "Крипта · Heleket", "hint": "USDT и др. · зачисление по webhook", } ) if cryptopay.enabled: items.append( { "id": "cryptobot", "title": "Крипта · Crypto Bot", "hint": "Оплата в @CryptoBot", } ) return items def topup_methods_payload( *, cryptopay: CryptoPay, heleket: Heleket, lava: Lava, anypay: Anypay | None = None, ) -> dict[str, Any]: return { "min_amount": int(TOPUP_MIN_RUB), "max_amount": int(TOPUP_MAX_RUB), "presets": list(TOPUP_PRESETS_RUB), "providers": available_providers( cryptopay=cryptopay, heleket=heleket, lava=lava, anypay=anypay ), } async def create_web_topup_invoice( *, db: Database, telegram_id: int, amount_rub: float, provider: str, cryptopay: CryptoPay, heleket: Heleket, lava: Lava, anypay: Anypay | None = None, ) -> TopupInvoice: amount = parse_topup_amount(amount_rub) provider = normalize_provider(provider) stamp = int(time.time()) order_id = _safe_order_id(f"web_topup_{telegram_id}_{int(amount)}_{stamp}") if provider == "lava": if not lava.enabled: raise TopupError("Lava не настроена") try: inv = await lava.create_invoice( amount_rub=amount, order_id=order_id, comment=f"Web topup {int(amount)} RUB uid={telegram_id}", custom_fields=str(telegram_id), ) except LavaError as exc: raise TopupError(str(exc)) from exc local_id = db.next_local_invoice_id() db.create_invoice_record( telegram_id=telegram_id, invoice_id=local_id, kind="topup", amount_rub=amount, pay_url=inv.pay_url, payload=None, provider="lava", order_id=order_id, external_uuid=inv.invoice_id, ) return TopupInvoice( provider="lava", amount_rub=amount, pay_url=inv.pay_url, invoice_id=local_id, order_id=order_id, title="Карта / СБП · Lava", ) if provider == "anypay": if not anypay or not anypay.enabled: raise TopupError("AnyPay не настроен") local_id = db.next_local_invoice_id() pay_id = str(local_id) try: inv = await anypay.create_invoice( pay_id=pay_id, amount_rub=amount, desc=f"Web topup {int(amount)} RUB uid={telegram_id}", ) except AnypayError as exc: raise TopupError(str(exc)) from exc db.create_invoice_record( telegram_id=telegram_id, invoice_id=local_id, kind="topup", amount_rub=amount, pay_url=inv.pay_url, payload=pay_id, provider="anypay", order_id=pay_id, external_uuid=inv.transaction_id, ) return TopupInvoice( provider="anypay", amount_rub=amount, pay_url=inv.pay_url, invoice_id=local_id, order_id=pay_id, title="AnyPay", ) if provider == "heleket": if not heleket.enabled: raise TopupError("Heleket не настроен") try: inv = await heleket.create_invoice( amount_rub=amount, order_id=order_id, description=f"Web topup {int(amount)} RUB uid={telegram_id}", ) except HeleketError as exc: raise TopupError(str(exc)) from exc local_id = db.next_local_invoice_id() db.create_invoice_record( telegram_id=telegram_id, invoice_id=local_id, kind="topup", amount_rub=amount, pay_url=inv.pay_url, payload=order_id, provider="heleket", order_id=order_id, external_uuid=inv.uuid, ) return TopupInvoice( provider="heleket", amount_rub=amount, pay_url=inv.pay_url, invoice_id=local_id, order_id=order_id, title="Крипта · Heleket", ) if not cryptopay.enabled: raise TopupError("Crypto Bot не настроен") payload = f"topup:{telegram_id}:{int(amount)}:web" try: invoice = await cryptopay.create_fiat_invoice( amount_rub=amount, description=f"Пополнение баланса на {int(amount)} ₽", payload=payload, ) except CryptoPayError as exc: raise TopupError(str(exc)) from exc db.create_invoice_record( telegram_id=telegram_id, invoice_id=invoice.invoice_id, kind="topup", amount_rub=amount, pay_url=invoice.pay_url, payload=payload, provider="cryptobot", ) return TopupInvoice( provider="cryptobot", amount_rub=amount, pay_url=invoice.pay_url or invoice.bot_invoice_url, invoice_id=invoice.invoice_id, order_id=None, title="Крипта · Crypto Bot", )