简介：这是基于Python开发的开源电报发卡机器人源码包，面向需要在Telegram中实现卡密自动发货、订单管理与易支付收款对接的开发者或小商家，可快速搭建一套轻量发卡系统。程序在Python 3.6.7下测试通过，数据层采用sqlite3存储，备份与迁移方便。资源共9个文件，压缩包约3.08MB，核心为4个Python脚本，分别承担入口、配置、易支付接口和业务逻辑；另附依赖清单、说明文档、两个文本示例和一张演示动图，覆盖从环境安装、参数配置到支付回调的完整链路。目前已有1374人学习下载。该版本虽标注为开源最终发布版，但源码结构与易支付对接思路足够清晰，既适合Python新手学习机器人开发，也可作为生产工具按需二次开发，替换不合用的易支付返回逻辑后即可上线使用。
1. 电报发卡机器人到底在解决什么问题
把“基于Python的电报发卡机器人”这句话拆开看，核心不是“Python”，也不是“机器人”，而是“发卡”这两个字背后的账目逻辑。很多第一次做这个方向的人以为难点是让Bot回复消息，等真正上线才发现，最贵、最容易翻车的环节是这三个：用户付了钱能不能秒拿到货、同一张卡密会不会被卖两次、支付回调重发时会不会多发一张。所谓电报发卡机器人，本质是一个带商品库存、订单状态机和支付对账逻辑的无人值守销售系统，只是把店面开在了会话窗口里。
适合做这件事的人有两类：一类是手里有稳定货源、想省掉人工发货成本的小型经营者；另一类是已经写过几个Bot、想挑战真实交易闭环的开发者。前者关心机器人能不能管住库存和订单，后者关心事务、幂等和对账怎么做才不出乱子。所以我这篇不讲那些花哨的界面和框架，只讲怎么把一个发卡Bot从一张空表做到能收钱、能发货、能对账，以及中间那些你一定会踩的坑。
2. 动手前的四个关键认知：业务模型、消息回路、库选型和“无状态”设计
2.1 发卡业务的四层模型：商品、库存、订单、发货
我见过很多半路写发卡Bot的人，第一版代码里直接在商品表加了一个“卡密内容”字段，结果上架第二个商品时就懵了：价格一样但卡密分两批进货，库存怎么拆？有效期不同怎么标？所以第一件事是先把业务拆成四层，别混在一起。
- 商品层：只管“卖什么”，比如“某视频平台月卡”，定价、名称、分类都在这一层。
- 库存层：管“有多少可卖的卡密”，每一条卡密是一行，字段至少包含归属商品、卡密内容、状态。
- 订单层：管“谁买了什么、付了多少钱、当前到哪一步”，状态机从待支付到已发货到已完成。
- 发货层：管“怎么把卡密送到买家手里”，可能是私聊直接发送，也可能是发文件。
这个分层最直接的好处是：同一个商品可以追加库存，不影响历史订单；卡密可以提前批量导入，不用动商品表；你还能给卡密加“过期时间”“已使用标记”这类字段，而不会污染订单记录。一个发卡Bot最值钱的资产就是这套表结构设计，而不是Bot本身怎么回消息。
2.2 Bot消息回路：长轮询与Webhook的取舍
Telegram Bot本身没有常驻连接，一切交互都是“你的程序向Telegram服务器发起HTTP请求”或“Telegram服务器向你的程序发起HTTP请求”。前者叫长轮询（getUpdates），后者叫Webhook。发卡机器人这种场景里，绝大多数团队会从长轮询起步，原因很简单：开发环境没有公网HTTPS也能跑，断线后Bot会自动重连，逻辑直观好调试。
Webhook的优势是实时性更好、省去轮询开销，适合消息量很大的场景，但要求你有一台公网服务器、一个固定域名和合法的HTTPS证书。有一点必须记牢：同一个Token同一时间只能启用一种接收方式。你本地在跑长轮询，线上又挂了Webhook，Bot就会像人格分裂一样时而回复时而不回复。为了保险，生产环境启动脚本里我会先无条件调用一次删除Webhook的接口，再开始轮询：
curl -s "https://api.telegram.org/bot<你的Token>/deleteWebhook?drop_pending_updates=true"这段命令把之前可能残留的Webhook配置清掉，同时丢弃堆积的旧消息，避免Bot一上线就被一堆过期消息淹没。参数里drop_pending_updates设为true是让我在新环境启动时不处理历史积压更新，否则用户几小时前点的按钮可能在启动瞬间全部触发。Token就是Bot创建后拿到的那串数字:字符串格式的密钥，相当于这个机器人的登录凭证，凡是出现了"bot"前缀的HTTP调用都要带上它。
2.3 Python库选型：python-telegram-bot 还是 telebot
这个方向最常被问到的就是“用哪个库”。我的标准很简单：你更适应同步还是异步。python-telegram-bot从20.x开始全面转向异步，写起来像这样：
from telegram import Update from telegram.ext import Application, CommandHandler, ContextTypes async def start_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE): await update.message.reply_text("欢迎使用发卡机器人") app = Application.builder().token("123456:ABC-DEF").build() app.add_handler(CommandHandler("start", start_cmd)) app.run_polling()run_polling()是异步框架的入口，整个事件循环会一直跑下去。好处是并发处理能力强，同一个进程内同时服务大量用户时不阻塞；坏处是如果你的业务代码里混了同步的sqlite3操作，写不好会把事件循环卡住，反而比同步库更慢。
telebot（PyTelegramBotAPI）是同步风格，新手心智负担小得多：
import telebot bot = telebot.TeleBot("123456:ABC-DEF") @bot.message_handler(commands=["start"]) def start_cmd(message): bot.reply_to(message, "欢迎使用发卡机器人") bot.infinity_polling(timeout=60, long_polling_timeout=30)timeout控制的是连接超时秒数，long_polling_timeout表示长轮询最多挂起多少秒等新消息，Telegram服务器会在有新消息时提前返回。发卡Bot这种中等体量业务，同步库完全够用，代码也更容易维护。我的建议是：如果你一个人维护、业务量不大，直接上telebot；如果已经规划了异步HTTP回调、高并发秒杀场景，选python-telegram-bot并且把数据库访问也换成异步驱动，别混用。
2.4 Bot是“无状态”的：会话状态别指望框架帮你记
Telegram Bot API 设计上不维持会话状态，每次用户发消息都是一次独立的HTTP请求。这带来一个很实际的坑：用户点了一个“购买”按钮，你的程序开始等他转账，转账回来后用户随便发了条“你好”，你的Bot根本不知道这个人之前的下单流程走到哪了。
常见的解决方案是把业务状态编码进按钮回调数据里，或者在后端建一张临时的“会话状态表”来记录每个用户进行到哪一步。发卡场景里我强烈推荐前者：商品选择、数量确认、订单号这些信息全部拼进callback_data。这个字段有64字节长度限制，所以只放关键ID，比如buy:product_id:order_no。等用户后续确认时，你从数据库里把订单取出来，而不是靠内存里的变量。这样即使Bot进程重启，用户的下单流程也不会断。
3. 把核心购买链路跑通：建表、导卡密、下单、事务发货
3.1 建表：用SQLite起步时这三张表不能省
SQLite在做演示和小规模运营时完全够用，等日订单量过千再换MySQL不迟。建表脚本是整套代码的地基，我一般这样设计：
CREATE TABLE products ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, level TEXT NOT NULL DEFAULT '', price_cents INTEGER NOT NULL DEFAULT 0, status INTEGER NOT NULL DEFAULT 1 ); CREATE TABLE cards ( id INTEGER PRIMARY KEY AUTOINCREMENT, product_id INTEGER NOT NULL, content TEXT NOT NULL, status INTEGER NOT NULL DEFAULT 0, order_id INTEGER, sold_at INTEGER, UNIQUE(order_id) ); CREATE TABLE orders ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_no TEXT NOT NULL UNIQUE, product_id INTEGER NOT NULL, user_id INTEGER NOT NULL, amount_cents INTEGER NOT NULL, status TEXT NOT NULL DEFAULT 'pending', card_id INTEGER, paid_at INTEGER, created_at INTEGER ); CREATE INDEX idx_cards_product_status ON cards(product_id, status);价格用“分”为单位存储，是为了避免浮点数误差，这是线上收款系统的基本习惯，订单金额永远用整数。cards表里的UNIQUE(order_id)是防超卖的第二道保险：一张卡密只能绑定一个订单，数据库层面直接约束住。status在商品表里是上下架开关，在卡密表里是0 未售 / 1 已售，别混用。订单状态我建议用字符串而不是数字，pending / paid / shipped / cancelled这种可读性强，排错时一眼能看懂。
3.2 给Bot加一个入库命令：一行一个卡密地导入
卡密入库最朴素的交互是管理员直接给Bot发一条消息：第一行是/add_stock 商品ID，后面每行一条卡密。用telebot实现是这样的：
import sqlite3 import telebot bot = telebot.TeleBot("123456:ABC-DEF") ADMIN_IDS = {123456789} # 只允许这些ID操作 @bot.message_handler(commands=["add_stock"]) def add_stock(message): if message.from_user.id not in ADMIN_IDS: bot.reply_to(message, "无权限") return lines = message.text.split("\n") first = lines[0].split() if len(first) < 2: bot.reply_to(message, "用法：/add_stock 商品ID\n换行后每行一个卡密") return product_id = int(first[1]) secrets = [ln.strip() for ln in lines[1:] if ln.strip()] db = sqlite3.connect("faka.db") db.execute("BEGIN IMMEDIATE") try: db.executemany( "INSERT INTO cards (product_id, content) VALUES (?, ?)", [(product_id, s) for s in secrets], ) db.commit() except Exception as e: db.rollback() bot.reply_to(message, f"入库失败：{e}") return bot.reply_to(message, f"商品 {product_id} 入库 {len(secrets)} 条") db.close()这段代码里有三个细节值得说。第一，BEGIN IMMEDIATE是SQLite在写事务时提前拿写锁，避免多条插入之间被别的线程打断造成database is locked。第二，executemany用参数化方式批量插值，既快又安全，卡密内容里哪怕有单引号也不会破坏SQL。第三，ADMIN_IDS里面的ID要填你自己的Telegram数字ID，获取方式是在私聊里给userinfobot发一条消息，它会直接返回你的用户ID。
3.3 下单与确认发货：关键一步是“预占卡密”
用户侧流程我设计成三步：发/shop看到商品列表，点“购买”按钮生成订单，完成付款后由管理员执行/confirm_order 订单号。真正发货的逻辑集中在确认命令里：
@bot.message_handler(commands=["confirm_order"]) def confirm_order(message): if message.from_user.id not in ADMIN_IDS: return parts = message.text.split() if len(parts) < 2: bot.reply_to(message, "用法：/confirm_order 订单号") return order_no = parts[1] db = sqlite3.connect("faka.db") db.execute("BEGIN IMMEDIATE") try: order = db.execute( "SELECT id, product_id, status FROM orders WHERE order_no = ?", (order_no,), ).fetchone() if not order or order["status"] != "pending": db.rollback() bot.reply_to(message, "订单不存在或已处理") return cursor = db.execute( """ UPDATE cards SET status = 1, order_id = ?, sold_at = ? WHERE id = ( SELECT id FROM cards WHERE product_id = ? AND status = 0 LIMIT 1 ) """, (order["id"], int(time.time()), order["product_id"]), ) if cursor.rowcount == 0: db.rollback() bot.reply_to(message, "库存不足，订单保留，待补货后重试") return card_content = db.execute( "SELECT content FROM cards WHERE order_id = ?", (order["id"],) ).fetchone() db.execute( "UPDATE orders SET status = 'shipped', card_id = ? WHERE id = ?", (order["id"], order["id"]), ) db.commit() bot.send_message(order["user_id"], f"购买成功，卡密：\n{card_content['content']}") bot.reply_to(message, f"订单 {order_no} 已发货") except Exception as e: db.rollback() bot.reply_to(message, f"发货失败：{e}") finally: db.close()这里最重要的不是UPDATE本身，而是“先预占后发货”的顺序。UPDATE cards ... WHERE id = (SELECT id ... LIMIT 1)在一个事务里执行，意思是：从库存里取一张未售卡密，立刻标记为已售并绑定订单，然后才把卡密内容取出来发给用户。全程没有“先查询再删除”这种两步操作，因此两个管理员同时确认不同订单时，数据库锁和LIMIT 1会保证同一张卡密不会被两个人同时拿到。
3.4 为什么“读出来再从列表里删掉”是必翻车的写法
很多第一版代码会写成：先SELECT content FROM cards WHERE product_id=? AND status=0 LIMIT 5，把结果放到列表，删掉一张，再更新数据库。这在单用户测试时没问题，一旦两边同时下单，两个进程可能读到同一张卡密，然后各自发给不同买家。SQLite是单文件数据库，读操作默认不阻塞，两步查询之间完全可能插入另一个事务的更新。所以处理库存扣减，永远只有一个安全姿势：在同一个事务里做“条件更新”，并且用受影响行数判断是否抢到了库存。这个习惯放之四海皆准，无论你用的是SQLite还是PostgreSQL。
4. 支付、对账与防超卖：让出货这件事变得可信
4.1 支付接入：TRC20地址对账的关键参数
发卡Bot最常见的收款方式是USDT-TRC20，因为它是纯链上转账，没有第三方风控，到账快、手续费低。接入逻辑不是“让用户打钱到你的固定地址”，而是要能自动知道“谁给我打了钱”。常见做法是：每个订单生成一个独立的收款地址，或者所有订单共用一个地址、靠金额和时间来匹配。
独立地址方案更稳，但需要调用钱包服务商或交易所的API来批量生成地址。轮询链上交易的伪代码长这样：
def poll_payments(watch_addresses): txs = fetch_trc20_transactions(watch_addresses) for tx in txs: if tx["confirmations"] < int(os.getenv("MIN_CONF", "1")): continue if tx["amount"] < expected_amount(tx["to_address"]): continue if tx["contract_address"].lower() != USDT_CONTRACT.lower(): continue mark_order_paid(tx["to_address"], tx["tx_hash"])三个参数必须盯死。第一是MIN_CONF，TRC20一般建议等1次确认，大额交易等3次以上更稳；第二是合约地址白名单，因为TRC20链上有很多假币，必须校验转账合约地址确实是USDT的那一个；第三是金额匹配，Users可能少转、多转，匹配范围要留一个容差窗口，比如精确到分。法币支付网关的逻辑类似，区别只在于它是HTTP回调：收到支付结果通知时先验签再改订单，验签失败一律当伪造处理。
4.2 订单幂等：同一笔支付绝不能发两次货
支付回调不是可信的，它可能超时重发、网络重试、甚至被恶意伪造。所以每一笔支付记录都必须有唯一标识。我的做法是加一张支付流水表：
CREATE TABLE pay_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, tx_hash TEXT NOT NULL UNIQUE, order_no TEXT NOT NULL, amount_cents INTEGER NOT NULL, created_at INTEGER );tx_hash是链上交易哈希，天然唯一。支付回调进来后，先往pay_records插一条，如果IntegrityError说明这笔支付之前已经处理过，直接忽略。插入成功后才允许把订单状态改成paid。这两个动作放同一个事务里，代码结构是：
db.execute("BEGIN IMMEDIATE") try: db.execute( "INSERT INTO pay_records (tx_hash, order_no, amount_cents, created_at) VALUES (?, ?, ?, ?)", (tx_hash, order_no, amount_cents, now), ) db.execute( "UPDATE orders SET status = 'paid' WHERE order_no = ? AND status = 'pending'", (order_no,), ) db.commit() except sqlite3.IntegrityError: db.rollback() log("重复支付回调: " + tx_hash)这段代码解决的是“同一笔款被通知两次”的问题。订单更新条件里带上AND status = 'pending'是最后一道保险：即使支付流水插入成功，订单状态不是待支付，也不会覆盖已经被发货的状态。
4.3 防超卖的完整解法：事务预占卡密
第三章的预占写法已经解决了单库存的并发扣减，但实际运营里还有一个更隐蔽的场景：同一个买家连点两次“购买”，生成了两个订单，分别都抢了卡密，最后只付一笔钱，另一单变成坏账。我的解决办法是：生成订单前先检查该用户对同一商品是否有一个pending或paid状态的未完成订单，有就直接返回提示，不创建新单：
existing = db.execute( "SELECT order_no FROM orders WHERE user_id = ? AND product_id = ? AND status IN ('pending', 'paid')", (user_id, product_id), ).fetchone() if existing: bot.answer_callback_query(call.id, f"你已有未完成订单 {existing['order_no']}", show_alert=True) return这条查询要配合orders表上的联合索引(user_id, product_id, status)才会快。另外，订单表里order_no是唯一索引，生成订单号时习惯用时间戳 + 用户ID + 随机数，避免高并发下时间戳相同导致撞单。
4.4 对账任务：把所有“已付款未发货”的订单捞出来重查
支付回调偶尔会丢，可能是Bot进程宕机、网络抖动、回调服务重启。所以生产环境必须挂一个定时对账任务，每10分钟扫一次所有超过30分钟还没有发货的paid状态订单，去链上重新确认这笔款是否真的到账：
def reconcile(): cutoff = int(time.time()) - 30 * 60 rows = db.execute( "SELECT order_no, amount_cents FROM orders WHERE status = 'paid' AND updated_at < ?", (cutoff,), ).fetchall() for row in rows: paid = check_payment_on_chain(row["order_no"]) if paid: confirm_order_ship(row["order_no"]) else: log("对账未通过: " + row["order_no"])定时任务用系统cron即可，不必在Bot进程里再写一套调度器，进程重启不至于丢定时任务：
*/10 * * * * cd /opt/tg_faka_bot && /usr/bin/python3 reconcile.py >> /var/log/tg_faka_bot_reconcile.log 2>&1对账任务的意义不在于“处理成功的单子”，而在于捕获那些支付成功但回调丢失的漏网之鱼。发卡业务的信誉就靠这套兜底逻辑撑住：用户那边已经在链上看到转账成功，Bot这边因为回调丢了没发货，如果没有对账，这笔钱就成了糊涂账。
5. 电报发卡机器人的5个高频坑，每个都是真实翻车现场
5.1 回调没回应：用户点了按钮，Bot却毫无反应
现象是用户点“购买”按钮后，按钮上的转圈一直不消失，Bot也没任何回复。刚开始排查时容易怀疑是网络问题，实际上90%的情况是你在callback_query_handler里做完了所有逻辑，但忘了调用answer_callback_query。Telegram要求每个回调必须被“应答”，否则Telegram会把这条更新留在队列里反复重推。
解决的办法是每个回调处理函数的开头或结尾必须有一句bot.answer_callback_query(call.id)。它不需要传提示文本，空调用本身就代表“我已经读过这条回叫了”。如果消息量大，建议在进入业务逻辑前就应答，然后再慢慢处理数据，避免用户等太久。
5.2 卡密文本里带特殊字符，消息发送直接报400错误
现象是卡密明明是正常的，但发给买家时Bot报错“Bad Request: can't parse entities”。原因是你发送消息时开了parse_mode="Markdown"，而卡密内容里有下划线、星号、方括号这类Markdown保留字符。卡密这种东西是用户直接复制粘贴的，内容不可控，最常见的就是账号密码里带下划线。
解决方法是：发卡密内容时永远不要开启任何parse_mode。商品介绍、欢迎语这类你亲手写的文本可以用富文本格式，但涉及卡密原样的消息一律用默认纯文本发送。如果确实想让部分内容加粗，先对卡密内容做特殊字符转义再做拼接，但更稳妥的是干脆分开两条消息发。
5.3 Webhook和长轮询同时生效，Bot行为时好时坏
现象是本地调试一切正常，部署上线后用户反映“有时候回复，有时候不回”，而且日志里出现409 Conflict错误。原因是同一个Token被两个地方同时消费，要么本地还开着轮询进程，要么环境变量里设了Webhook又被手动轮询。
解决方法是固定“一套环境只用一种接收方式”。生产环境用Webhook就彻底停掉本地轮询进程；用轮询的话，在启动脚本的最前面加那句deleteWebhook调用。不要相信“我应该没配Webhook”这种记忆，启动时强制清理一次才是血泪经验换来的稳妥做法。
5.4 SQLite报“database is locked”导致发货失败
现象是流量起来后，后台看到大量sqlite3.OperationalError: database is locked，订单确认偶尔失败。原因是SQLite只允许一个写事务持有写锁，你开了多个线程或进程同时写库，其中一个事务长时间没提交，别的写操作只能一直等。
解决的办法分三层：第一，SQLite连接设置timeout=30，让写操作等待更久而不是立刻报错；第二，打开连接后执行一行PRAGMA journal_mode=WAL;，WAL模式显著降低读写互斥；第三，所有写操作必须短事务化，不要在事务里发网络请求、调外部API。发货逻辑里“先更新数据库提交事务，再调用send_message发卡密”，顺序不能反，这条规则能避开90%的锁问题。
5.5 对账差8小时：凌晨的订单怎么找都对不上
现象是对账脚本白天跑正常，每天凌晨的订单处理完后总是少几条，时间戳正好差8小时。原因是开发环境用本地时间，生产服务器用的是UTC，Python的datetime.now()在不同环境返回不同时区的时间，导致created_at和paid_at一个存的是东八区、一个存的是UTC。
解决方法是数据库里所有时间字段统一存整数时间戳，也就是int(time.time())。展示给用户看的时候再转换成当地时区字符串。这样无论服务器在哪个时区、对账脚本跑在哪台机器上，时间比较永远基于同一基准，不会有任何歧义。已经是字符串时间的库，尽早迁移成时间戳字段，否则越往后对账越痛苦。
6. 部署、守护进程与一个救命级的进阶习惯
6.1 用systemd把Bot守起来并做好日志轮转
开发时用终端跑python bot.py没问题，生产环境必须交给systemd守护，进程崩了自动拉起才是合格的部署方式。systemd service文件这样写：
[Unit] Description=tg_faka_bot After=network.target [Service] WorkingDirectory=/opt/tg_faka_bot ExecStart=/opt/tg_faka_bot/.venv/bin/python bot.py Restart=always RestartSec=5 EnvironmentFile=/opt/tg_faka_bot/.env StandardOutput=append:/var/log/tg_faka_bot.log StandardError=append:/var/log/tg_faka_bot.log [Install] WantedBy=multi-user.targetsaya_EnvironmentFile是很推荐的配置，把Token、管理员ID、数据库路径都放进.env文件，代码里用os.getenv读取，这样代码仓库就不会泄露密钥。日志文件会一直增长，配合一个logrotate配置就能自动切割：
/var/log/tg_faka_bot.log { daily rotate 14 compress copytruncate }copytruncate很重要，它复制日志内容后清空原文件，而不是通过重启进程来切换日志文件，这样Bot进程不用中断。数据库备份也别忘了，SQLite热备份不能用简单的cp，要用自带的备份命令：
sqlite3 /opt/tg_faka_bot/faka.db ".backup '/opt/tg_faka_bot/backup/faka_$(date +%F).db'"这个命令在数据库被占用时也能生成一致的快照，主进程不用停机。
6.2 高并发库存的一个进阶小技巧：预分配卡密池
如果某个商品突然爆单，所有订单同时走UPDATE cards ... WHERE status=0的预占逻辑，SQLite单库写入会成为瓶颈。我常用的做法是给热门商品做一个“预分配卡密池”：补货时一次性取出50张卡密放进内存队列，模块内部从这个池子里发卡，池子空了再回数据库补一次。这样对数据库的写压力从“每单一次”降到“每五十单一次”，代价是进程崩溃可能丢失未写入订单的卡密绑定关系，所以一定要配合对账任务兜底。
6.3 我至今保留的一个测试习惯
发卡系统最怕改坏逻辑，所以每次改完库存或订单相关代码，我会故意连点十次购买按钮，再加一个用户同时下单的模拟脚本，确认系统不会多发出一张卡密。这个测试习惯已经救过我很多次，尤其是那些看似“只改了一行”的提交。人工测试可能偶尔跳过，但“先测试再上线”这条底线不能丢，毕竟用户付了钱却拿不到货，口碑一崩就再也救不回来。希望这些经验能帮你在做电报发卡机器人的路上少踩几个坑。
本文还有配套的精品资源，点击获取
