136 lines
4.0 KiB
Python
136 lines
4.0 KiB
Python
"""Telegram Stars: курс, пакеты и создание инвойсов."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
from aiogram.types import LabeledPrice
|
|
|
|
from services.db import Database
|
|
|
|
# Суммы пополнения в рублях (как у крипты)
|
|
STAR_TOPUP_AMOUNTS_RUB: tuple[int, ...] = (100, 200, 500, 1000, 2000)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class StarsQuote:
|
|
amount_rub: float
|
|
stars: int
|
|
rub_per_star: float
|
|
|
|
@property
|
|
def stars_label(self) -> str:
|
|
return f"{self.stars} ⭐"
|
|
|
|
@property
|
|
def rub_label(self) -> str:
|
|
if self.amount_rub == int(self.amount_rub):
|
|
return f"{int(self.amount_rub)} ₽"
|
|
return f"{self.amount_rub:.2f} ₽"
|
|
|
|
|
|
def rub_to_stars(amount_rub: float, *, rub_per_star: float) -> int:
|
|
rate = float(rub_per_star)
|
|
if rate <= 0:
|
|
raise ValueError("rub_per_star must be > 0")
|
|
stars = int(math.ceil(float(amount_rub) / rate))
|
|
return max(1, stars)
|
|
|
|
|
|
def quote_topup(amount_rub: float, *, rub_per_star: float) -> StarsQuote:
|
|
amount = float(amount_rub)
|
|
if amount <= 0:
|
|
raise ValueError("amount_rub must be > 0")
|
|
stars = rub_to_stars(amount, rub_per_star=rub_per_star)
|
|
return StarsQuote(amount_rub=amount, stars=stars, rub_per_star=float(rub_per_star))
|
|
|
|
|
|
def parse_stars_payload(payload: str) -> tuple[int, int, float] | None:
|
|
"""
|
|
payload: stars:{invoice_id}:{telegram_id}:{amount_rub}
|
|
→ (invoice_id, telegram_id, amount_rub)
|
|
"""
|
|
parts = (payload or "").split(":")
|
|
if len(parts) != 4 or parts[0] != "stars":
|
|
return None
|
|
try:
|
|
return int(parts[1]), int(parts[2]), float(parts[3])
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def build_stars_payload(*, invoice_id: int, telegram_id: int, amount_rub: float) -> str:
|
|
return f"stars:{invoice_id}:{telegram_id}:{int(amount_rub)}"
|
|
|
|
|
|
def packs_payload(*, rub_per_star: float) -> list[dict[str, Any]]:
|
|
out: list[dict[str, Any]] = []
|
|
for amount in STAR_TOPUP_AMOUNTS_RUB:
|
|
q = quote_topup(amount, rub_per_star=rub_per_star)
|
|
out.append(
|
|
{
|
|
"amount_rub": q.amount_rub,
|
|
"stars": q.stars,
|
|
"rub_label": q.rub_label,
|
|
"stars_label": q.stars_label,
|
|
"title": f"{q.rub_label} · {q.stars_label}",
|
|
}
|
|
)
|
|
return out
|
|
|
|
|
|
async def create_stars_topup(
|
|
*,
|
|
bot: Any,
|
|
db: Database,
|
|
telegram_id: int,
|
|
amount_rub: float,
|
|
rub_per_star: float,
|
|
) -> dict[str, Any]:
|
|
"""Создать локальный инвойс + invoice link для Mini App / Telegram Web."""
|
|
if telegram_id <= 0:
|
|
raise ValueError("Stars доступны только для Telegram-аккаунта")
|
|
quote = quote_topup(amount_rub, rub_per_star=rub_per_star)
|
|
local_id = db.next_local_invoice_id()
|
|
payload = build_stars_payload(
|
|
invoice_id=local_id,
|
|
telegram_id=telegram_id,
|
|
amount_rub=quote.amount_rub,
|
|
)
|
|
invoice_link = await bot.create_invoice_link(
|
|
title="Пополнение баланса",
|
|
description=(
|
|
f"Зачисление {quote.rub_label} на баланс VPN Service "
|
|
f"({quote.stars_label})"
|
|
),
|
|
payload=payload,
|
|
currency="XTR",
|
|
prices=[
|
|
LabeledPrice(
|
|
label=f"Баланс {quote.rub_label}",
|
|
amount=quote.stars,
|
|
)
|
|
],
|
|
provider_token="",
|
|
)
|
|
db.create_invoice_record(
|
|
telegram_id=telegram_id,
|
|
invoice_id=local_id,
|
|
kind="topup",
|
|
amount_rub=quote.amount_rub,
|
|
pay_url=invoice_link,
|
|
payload=payload,
|
|
provider="stars",
|
|
)
|
|
return {
|
|
"invoice_id": local_id,
|
|
"invoice_link": invoice_link,
|
|
"amount_rub": quote.amount_rub,
|
|
"stars": quote.stars,
|
|
"rub_label": quote.rub_label,
|
|
"stars_label": quote.stars_label,
|
|
"payload": payload,
|
|
}
|