怎么收钱

Telegram Stars 十几行代码就能收钱。真正的难点是支付回调会重复到达——我实测过，同一笔订单重推 5 次，用户会拿到 5 倍的东西。

约 20 分钟出错会直接损失钱aiogram 3.x · 最后验证 2026-08

两条路，先说该选哪条

Telegram 机器人收钱有两种方式，差别很大：

	Telegram Stars	第三方支付
卖什么	数字商品、会员、订阅 —— Telegram 规定虚拟商品只能用 Stars	实物、线下服务
用户怎么付	用他账号里的 Stars，苹果/谷歌内购充值	跳到支付服务商的页面
你要准备什么	什么都不用，代码里直接发账单	去 BotFather 绑一个支付服务商，还要资质
抽成	苹果谷歌抽 30%，提现有门槛	看服务商
上手难度	十几行代码	几天，卡在资质审核上

卖会员、卖数字内容 —— 用 Stars，而且只能用 Stars。这一章讲它，因为大部分 bot 的收费场景都属于这一类。

一笔支付要走三步

整个流程比想象中长，因为中间有个「你确认一下」的环节：

第 ② 步是很多人没料到的：不回复它，用户的付款会直接失败，而且只有 10 秒。

代码：三个 handler

Stars 支付有几个特殊约定，先记住：货币代码固定是 XTR，provider_token 传空字符串，金额单位就是星星个数（不像别的货币要乘 100）。

payment.pyaiogram 3.x · Telegram Stars

import loggingfrom aiogram import F, Routerfrom aiogram.filters import Commandfrom aiogram.types import (    LabeledPrice, Message, PreCheckoutQuery,)import storagelog = logging.getLogger(__name__)router = Router()PLANS = {    "month": ("会员 30 天", 50, 30),    # (名称, 多少星, 给几天)    "year":  ("会员 365 天", 500, 365),}# ---------- ① 发账单 ----------@router.message(Command("buy"))async def cmd_buy(message: Message) -> None:    title, stars, days = PLANS["month"]    await message.answer_invoice(        title=title,        description="开通后立即生效，到期自动提醒",        # payload 是你自己的数据，支付成功时会原样带回来        payload=f"plan:month:{message.from_user.id}",        provider_token="",          # Stars 支付必须是空字符串        currency="XTR",             # Stars 的货币代码        prices=[LabeledPrice(label=title, amount=stars)],    )# ---------- ② 预校验：10 秒内必须回 ----------@router.pre_checkout_query()async def on_pre_checkout(query: PreCheckoutQuery) -> None:    # 这里做最后的检查：商品还有吗、用户还有资格吗    # 检查要快，超时不回等于让用户付款失败    await query.answer(ok=True)    # 要拒绝就这样，message 会显示给用户看：    # await query.answer(ok=False, error_message="这个套餐刚下架了")# ---------- ③ 付款成功：真正发货 ----------@router.message(F.successful_payment)async def on_paid(message: Message) -> None:    sp = message.successful_payment    plan_key = sp.invoice_payload.split(":")[1]    _, _, days = PLANS[plan_key]    # charge_id 是这笔交易的唯一标识 —— 幂等键就用它    granted = storage.grant_membership(        charge_id=sp.telegram_payment_charge_id,        user_id=message.from_user.id,        days=days,        stars=sp.total_amount,    )    if not granted:        log.info("重复回调，已忽略: %s", sp.telegram_payment_charge_id)        return              # 已经发过货了，什么都不做    await message.answer(f"到账了，会员延长 {days} 天。")    log.info("支付成功 user=%s stars=%s", message.from_user.id, sp.total_amount)

幂等：这一章最要紧的一件事

支付回调会重复到达。网络抖动、Telegram 内部重试，同一笔付款你可能收到好几次通知。

我实测了一下，同一笔 30 天的订单重推 5 次：

写法	实际发放	会员天数
直接加天数	5 次	150 天	用户付一笔钱，拿到 5 倍的东西
用 charge_id 做幂等	1 次	30 天	正确

并发场景（5 个回调同时到）结果一样。这不是概率问题，是必然会发生的事。

storage.py

CREATE TABLE IF NOT EXISTS payments (    charge_id TEXT PRIMARY KEY,     -- 幂等键：Telegram 给的交易号    user_id   INTEGER NOT NULL,    stars     INTEGER NOT NULL,    days      INTEGER NOT NULL,    ts        INTEGER NOT NULL)def grant_membership(charge_id: str, user_id: int,                     days: int, stars: int) -> bool:    """发货。这笔已经处理过就返回 False。"""    conn = sqlite3.connect(DB_PATH, timeout=10)    try:        conn.execute("BEGIN IMMEDIATE")        try:            conn.execute(                "INSERT INTO payments (charge_id, user_id, stars, days, ts) "                "VALUES (?, ?, ?, ?, strftime('%s','now'))",                (charge_id, user_id, stars, days),            )        except sqlite3.IntegrityError:            conn.execute("ROLLBACK")            return False          # 主键撞了 = 这笔处理过了        # 记账和发货在同一个事务里（第 6 章）        conn.execute(            "INSERT INTO members (user_id, expire_at) "            "VALUES (?, strftime('%s','now') + ?) "            "ON CONFLICT(user_id) DO UPDATE SET "            -- 已到期就从现在算，没到期就在原有基础上顺延            "  expire_at = MAX(expire_at, strftime('%s','now')) + ?",            (user_id, days * 86400, days * 86400),        )        conn.execute("COMMIT")        return True    except Exception:        conn.execute("ROLLBACK")        raise    finally:        conn.close()

✓ 那个 MAX 是有讲究的

用户没到期就续费时，新的到期时间应该从原到期日往后加，不是从今天加 —— 否则他提前续费反而亏了。
而如果已经过期了，就得从今天算起。MAX(expire_at, 现在) 一行同时处理这两种情况。

预校验那 10 秒

第 ② 步很容易被忽略，但不回复它，用户的付款直接失败，而且他看到的是一句莫名其妙的错误。

必须在 10 秒内回	所以别在这里查外部 API、别做慢操作
这是最后的拒绝机会	库存没了、活动结束了 —— 在这里说 `ok=False`，钱不会扣
过了这一步就得认账	第 ③ 步收到通知时钱已经扣了，这时候再说「没货了」就得走退款

所以真要检查库存，就在这一步：

payment.py

@router.pre_checkout_query()async def on_pre_checkout(query: PreCheckoutQuery) -> None:    try:        plan_key = query.invoice_payload.split(":")[1]        if plan_key not in PLANS:            await query.answer(ok=False, error_message="这个套餐已下架")            return        await query.answer(ok=True)    except Exception:        # 出错也要回，不然用户干等到超时        await query.answer(ok=False, error_message="系统繁忙，稍后再试")        raise

退款

Stars 支付可以原路退：

退款

await bot.refund_star_payment(    user_id=user_id,    telegram_payment_charge_id=charge_id,)

这也是为什么 payments 表要把 charge_id 存下来 —— 不存的话你想退都退不了。顺便，退款之后记得把发出去的会员天数扣回来，而且这个动作同样要幂等（同一笔别退两次）。

怎么测

好消息：Stars 支付有测试模式，不花真钱。

用测试环境	Telegram 客户端登录到 Test Server，那边的 Stars 是免费的
幂等逻辑单独测	不用真付款 —— 直接调 `grant_membership()` 传同一个 charge_id 五次，看天数是不是只加了一次
预校验超时	在 `on_pre_checkout` 里 `await asyncio.sleep(15)`，看看用户端报什么错

◆ 上线前一定要真付一笔小额的

测试环境跟正式环境的行为不完全一样。用最小面额（比如 1 星）在正式环境走一遍完整流程，确认钱到账、货发出、记录写进库了。
这笔钱值得花——支付是唯一一类出错会直接损失金钱和信誉的功能。

钱的事，几条硬规矩

规矩	为什么
发货必须幂等	回调一定会重复。这是本章唯一一条「不做就一定出事」的
charge_id 必须存下来	退款、对账、查纠纷，全靠它
金额和商品别信客户端	`payload` 是你自己塞的，但要在服务端重新按 plan 查价格，别直接用回调里带的数字
记账和发货在一个事务里	否则会出现「钱记了货没发」或者反过来
支付相关的日志都留着	用户说「我付了钱没到账」时，你要能查出到底发生了什么
先记录，后发消息	发消息可能失败（第 7 章）。让用户少收一条通知，也别让他没拿到东西

这一章的要点

数字商品只能用 Telegram Stars：currency="XTR"、provider_token=""、金额就是星星数
三步流程，中间的 pre_checkout 必须 10 秒内回，不回用户付款就失败
回调一定会重复到达 —— 实测重推 5 次会发出 5 倍的货
用 telegram_payment_charge_id 做幂等键，主键撞了就说明处理过了
续费用 MAX(原到期时间, 现在) + 时长，别让提前续费的人吃亏


上线前用最小面额在正式环境真跑一笔




✓ 试着改一个
把这一章和第 8 章接起来：付款成功后写 members.expire_at，定时任务扫到期的人 —— 然后验证第 8 章那个竞态：在扫描进行中付一笔款，看人会不会被踢。
这两章合起来就是一个完整的付费会员群，也是最常见的 Telegram 变现方式。





  支付这块有什么疑问？
  钱的事出错代价最大。哪一步不放心、或者你的场景这里没覆盖到，说一声。留不留联系方式都行。
  
    
    
    
      
      
    
  
  收到了，谢谢。留了联系方式的话我会回你。
  



  下一章
部署上线
  继续 →