如何在 aiogram 3 的机器人中开通 Telegram Stars 支付功能：附代码的分步指南

pip install -U aiogram

import asyncio
import logging
import os

from aiogram import Bot, Dispatcher, F, Router
from aiogram.exceptions import TelegramBadRequest
from aiogram.filters import Command, CommandObject, CommandStart
from aiogram.types import LabeledPrice, Message, PreCheckoutQuery

ADMIN_ID = int(os.environ["ADMIN_ID"])  # ваш Telegram ID — только ему доступен /refund

# Товар: название, описание, цена в звёздах (целое число) и метка заказа
PRODUCT_TITLE = "Доступ к курсу"
PRODUCT_DESCRIPTION = "Полный доступ ко всем урокам курса навсегда"
PRODUCT_PAYLOAD = "course_full"
PRICE_STARS = 50

router = Router()


@router.message(CommandStart())
async def start(message: Message) -> None:
    await message.answer(f"Привет! Курс стоит {PRICE_STARS} ⭐. Купить: /buy")


@router.message(Command("buy"))
async def buy(message: Message) -> None:
    await message.answer_invoice(
        title=PRODUCT_TITLE,
        description=PRODUCT_DESCRIPTION,
        payload=PRODUCT_PAYLOAD,  # вернётся в pre_checkout_query и successful_payment
        currency="XTR",           # XTR — Telegram Stars, provider_token не нужен
        prices=[LabeledPrice(label=PRODUCT_TITLE, amount=PRICE_STARS)],
    )


@router.pre_checkout_query()
async def pre_checkout(query: PreCheckoutQuery) -> None:
    # Последняя проверка перед списанием: товар существует и в наличии.
    # Ответить нужно за 10 секунд, иначе Telegram отменит оплату.
    if query.invoice_payload != PRODUCT_PAYLOAD:
        await query.answer(ok=False, error_message="Товар не найден. Создайте счёт заново: /buy")
        return
    await query.answer(ok=True)


@router.message(F.successful_payment)
async def paid(message: Message) -> None:
    payment = message.successful_payment
    if payment.invoice_payload != PRODUCT_PAYLOAD or payment.total_amount != PRICE_STARS:
        logging.warning("Неожиданный платёж: %s", payment.invoice_payload)
        return
    # charge_id сохраните в базе вместе с заказом — без него не сделать возврат
    logging.info("Оплата %s ⭐ от %s, charge_id=%s",
                 payment.total_amount, message.from_user.id, payment.telegram_payment_charge_id)
    # Здесь выдайте товар: откройте доступ, отправьте файл, начислите баланс
    await message.answer("Спасибо за покупку! Доступ к курсу открыт ✅")


@router.message(Command("refund"), F.from_user.id == ADMIN_ID)
async def refund(message: Message, command: CommandObject, bot: Bot) -> None:
    # /refund <user_id> <charge_id> — вернуть звёзды покупателю
    args = (command.args or "").split()
    if len(args) != 2 or not args[0].isdigit():
        await message.answer("Формат: /refund <user_id> <charge_id>")
        return
    try:
        await bot.refund_star_payment(user_id=int(args[0]), telegram_payment_charge_id=args[1])
    except TelegramBadRequest as e:
        await message.answer(f"Возврат не прошёл: {e.message}")
        return
    await message.answer("Звёзды возвращены покупателю")


@router.message(Command("paysupport"))
async def pay_support(message: Message) -> None:
    await message.answer("Вопросы по оплате: напишите @your_support — ответим в течение дня.")


async def main() -> None:
    logging.basicConfig(level=logging.INFO)
    bot = Bot(token=os.environ["BOT_TOKEN"])
    dp = Dispatcher()
    dp.include_router(router)
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())

export BOT_TOKEN="токен_от_BotFather"
export ADMIN_ID="ваш_telegram_id"
python bot.py

link = await bot.create_invoice_link(
    title="Доступ к курсу",
    description="Полный доступ ко всем урокам курса",
    payload="course_full",
    currency="XTR",
    prices=[LabeledPrice(label="Доступ к курсу", amount=50)],
)