简介一套基于某发卡系统二次开发的Telegram个人发卡机器人源码支持中英双语面向需对接Telegram机器人自动发卡的个人开发者或中小团队适用于电商、游戏等虚拟卡管理与交易处理场景。运行要求Linux或Windows服务器PHP 8.1与MySQL 5.7安装教程已随包提供后台默认admin请及时修改。资源共2000个文件压缩包约27.48MB以JavaScript和TypeScript源码为主1409个js、124个ts辅以Vue组件、JSON配置、Markdown文档、CSS样式及SQL脚本前后端逻辑分离便于二次扩展。已有105人学习下载。阅读源码可掌握事务处理、多语言切换与机器人接口对接方式进而定制卡密生成规则、优化交易流程、增强安全性。1. 一个 TG 发卡机器人到底解决了什么事先说结论再说值不值得做我见过不少卖软件授权码、虚拟卡密、会员兑换码的小店白天在电脑前手动发卡晚上客户下单没人理第二天起来一堆“在吗”“怎么还没发”的消息。也有人把店铺挂进发卡系统但客户还是要打开网页、付款、等页面跳转流程长了流失就多。这个标题的「TG 个人发卡机器人」就是给现成的发卡系统补一个 Telegram 前端客户在对话里输入 /buy 1机器人返回收款二维码付款后卡密自动私聊发过去中英文文案随用户偏好切换。它解决的是「自动发货」和「跨语言沟通」两件事适合手里已经有发卡系统、想多开一个 TG 渠道的个人卖家和两三人的小团队。值不值得做我的结论是如果单量还在几百单一天用现成发卡系统加一个机器人外壳是最省钱的方案不用重写后台也不用碰支付回调。2. 基于发卡系统二次开发职责边界、数据表与对接方式怎么定2.1 发卡系统管什么机器人管什么职责边界先分清很多人一上来就把下单、扣库存、发卡密的逻辑写死在机器人里这是二开最常见的翻车点。发卡系统之所以叫发卡系统是因为它已经帮你把几件麻烦事做了支付通道配置、订单状态流转、卡密库存管理、掉单检测、后台统计。你要做的机器人只是把 Telegram 对话翻译成对发卡系统的操作而不是再造一个发卡系统。我一般会把职责切成这样模块归属方负责的事商品、卡密、库存发卡系统商品上下架、卡密入库、库存扣减支付收款、回调发卡系统收款二维码、支付平台回调、订单改为已支付用户对话、下单入口TG 机器人接收 /buy、展示商品、回答“怎么还没收到”发货通知TG 机器人查订单状态把卡密私聊发给买家语言切换TG 机器人读用户偏好渲染对应语言的文案分界的核心逻辑是每笔订单的「事实状态」只在发卡系统里维护机器人只是它的一个客户端。这样做的好处是你换支付通道、改卡密批次都不用来回改机器人的代码哪天不想用 TG 了关了机器人网页端还在正常卖。机器人侧哪怕内存里的缓存全丢了重新启动后扫一遍订单表就能恢复。2.2 双语言只在机器人层做不用动发卡系统后台标题里的「支持双语言」容易让人想多是不是要把发卡系统的后台管理界面也翻成英文对个人发卡来说完全没必要。后台是店主自己看的商品名、货架、批量导入卡密这些操作固定用一种语言反而少出错。真正需要双语的只有买家接触到的部分——欢迎语、商品列表、订单提示、发货通知、错误提示这些对话文案。这就是二开的优势发卡系统保持原样机器人侧做一个语言包模块就够了。我在后面第五章会专门讲语言包怎么设计这里先给结论用 JSON 键值对存文案代码里通过一个 _t(lang, key) 函数取文案不要在业务代码里到处写 if lang zh。这么做的第二个理由和排查有关。如果双语文案散落在各个函数里哪天买家反馈“英文模式下下单按钮不见了”你得翻遍所有消息发送点才能找到是哪条分支漏了。用语言包的话打开 JSON 查一下缺失的 key 就定位了。2.3 对接方式三选一直连数据库、HTTP API、中间表这是二开第一道选择题也是决定后面所有坑走向的选择。直连数据库是最常见的做法个人二开我通常也推荐先从这里开始。常见发卡系统的数据模型其实很固定一张商品表字段大致是 id、name、price、stock一张卡密表字段是 id、product_id、card_content、status0 未售 1 已售一张订单表字段是 order_no、product_id、user_id、amount、statuspending/paid/shipped、created_at。机器人拿到用户输入的商品 ID直接操作这三张表。优点是逻辑直白、事务可控缺点是得先看懂原系统表结构而且业务写死了表名发卡系统升级表结构时你得跟着改。HTTP API 是更保守的路线。如果发卡系统已经提供了下单、订单查询的接口机器人只做 HTTP 调用。优点是解耦缺点是很多个人发卡系统的 API 没有完整文档二次开发时你得一边抓包一边猜参数调试成本高出一截。中间表适合订单量已经大到直连成为瓶颈的场景机器人把下单请求写进一张待处理队列表发卡系统后台的守护进程消费这张表。这个方案我放到第六章配合 Redis 一起讲日常几百单量级别用不上。我自己做的时候选的是直连数据库加事务因为发货链路里有一个硬需求从「扣库存」到「生成订单」到「标记卡密已售」必须是一个原子操作用 API 做反而要多处理网络超时和重试。下面第三章就按这个路线展开。3. 把机器人和发卡系统接起来最小可运行代码与下单链路3.1 从 BotFather 拿到 token两步准备动手写代码之前先在 Telegram 里找到 BotFather发 /newbot 按提示起名拿到一串 token格式类似123456:ABC-DEF...。这串 token 是机器人唯一的身份凭证泄露了别人就能控制你的机器人我习惯把它放到环境变量里而不是写死在代码文件。第二步是给机器人设置命令菜单。给 BotFather 发 /setcommands然后提交buy - 下单购买, 用法 /buy 商品ID list - 查看商品列表 order - 查询订单状态 lang - 切换语言命令菜单的作用是让用户点输入框旁边的菜单按钮就能看到可用指令而不是靠猜。注意菜单里的描述会直接展示给用户中英混合会影响体验建议写英文短句中文含义在机器人回复的欢迎语里说明。3.2 最小机器人骨架长轮询跑起来这一节的目标是先让机器人在本地能回复消息。我常用 python-telegram-bot 的异步版本写骨架代码量少回调机制清晰。# bot.py — TG 发卡机器人最小骨架 import os from telegram import Update from telegram.ext import Application, CommandHandler, ContextTypes TOKEN os.environ[TG_BOT_TOKEN] async def start(update: Update, context: ContextTypes.DEFAULT_TYPE): lang get_user_lang(update.effective_user.id) # 读用户偏好, 默认 en text _t(lang, start_welcome) await update.message.reply_text(text) async def buy(update: Update, context: ContextTypes.DEFAULT_TYPE): product_id context.args[0] if context.args else None if product_id is None: await update.message.reply_text(_t(get_user_lang(update.effective_user.id), buy_usage)) return # 真正的下单逻辑在 3.3 节, 这里先占位 await update.message.reply_text(f收到商品ID: {product_id}) def main(): app Application.builder().token(TOKEN).build() app.add_handler(CommandHandler(start, start)) app.add_handler(CommandHandler(buy, buy)) # 长轮询; drop_pending_updates 表示启动时丢弃离线期积压的旧消息 app.run_polling(drop_pending_updatesTrue) if __name__ __main__: main()逻辑说明这里的 get_user_lang 和 _t 是双语言模块的接口本章先留了函数名第五章会给出实现。buy 命令从 context.args 里取用户输入的第一个参数作为商品 ID这是 TG 命令最基础的参数获取方式。参数说明run_polling 有两个常用参数值得关注。drop_pending_updatesTrue 的意思是机器人停机期间收到的消息不补发否则你一上线离线时所有的 /buy 会瞬间涌进来触发下单。timeout 参数默认为 20 秒如果你的服务器到 Telegram 的网络不稳定可以显式调成 50减少连接被服务端断开的概率。3.3 下单链路从 /buy 2 到卡密自动发放下单链路是整个二开的核心我把它拆成三步扣库存生成订单、返回收款信息、检测到支付成功后发货。注意一个关键设计TG 机器人接收不到支付平台的回调回调只会打到发卡系统自己的地址上。所以个人二开最省事的方案是让机器人定期去扫订单表把「已支付但未发货」的订单捞出来处理。# order.py — 下单、付款轮询、自动发货 import time, uuid import pymysql DB_CONFIG {host: 127.0.0.1, user: root, password: yourpass, database: shop, charset: utf8mb4} def create_order(user_id: int, product_id: int) - dict: conn pymysql.connect(**DB_CONFIG) try: with conn.cursor() as cur: # 扣库存, 条件里带 stock 0 防止超卖 sql UPDATE products SET stock stock - 1 WHERE id %s AND stock 0 n cur.execute(sql, (product_id,)) if n 0: return {ok: False, msg: sold_out} # 读商品价格, 生成订单 cur.execute(SELECT name, price FROM products WHERE id %s, (product_id,)) name, price cur.fetchone() order_no TG time.strftime(%Y%m%d%H%M%S) uuid.uuid4().hex[:8] cur.execute( INSERT INTO orders (order_no, product_id, user_id, amount, status) VALUES (%s, %s, %s, %s, pending), (order_no, product_id, user_id, price), ) conn.commit() return {ok: True, order_no: order_no, amount: price, product_name: name} finally: conn.close() def ship_paid_orders(): 轮询扫描已支付未发货的订单, 执行发货 conn pymysql.connect(**DB_CONFIG) try: with conn.cursor() as cur: # 找出所有待发货订单; shipped 标记字段要提前在订单表里加好 cur.execute(SELECT order_no, product_id, user_id FROM orders WHERE status paid AND shipped 0) for order_no, product_id, user_id in cur.fetchall(): # 从卡密表取一张未售卡密, 标记已售 cur.execute( SELECT id, card_content FROM cards WHERE product_id %s AND status 0 LIMIT 1 FOR UPDATE, (product_id,), ) row cur.fetchone() if row is None: # 库存不多了: 标记缺货, 发通知让管理员补卡密 mark_shortage(conn, order_no) continue card_id, card_content row cur.execute(UPDATE cards SET status 1 WHERE id %s, (card_id,)) cur.execute(UPDATE orders SET shipped 1 WHERE order_no %s, (order_no,)) send_to_telegram(user_id, card_delivery, card_contentcard_content) # 发送函数见下 conn.commit() finally: conn.close()逻辑说明create_order 里把扣库存的 UPDATE 语句条件写成了stock 0这行至关重要——它保证了即便两个请求同时进来数据库也只会让一条语句真正扣减成功。ship_paid_orders 里的FOR UPDATE是行级锁目的是防止两个并发轮询线程取到同一张卡密。参数说明轮询频率我一般设 10 到 15 秒一次。太频繁会给发卡系统数据库造成无谓压力太慢则买家付款后等得着急。如果你用的是云服务器上跑的 cron 或 systemd timer 来触发 ship_paid_orders注意任务必须加锁或保证单实例运行否则重复执行会重复发货——这个坑在第四章会专门展开。3.4 把卡密安全地发出去避开 Markdown 解析陷阱发货消息里最容易翻车的是 Telegram 的 Markdown 解析。卡密内容里出现_、*、[、#这些字符时如果启用了解析模式TG 会尝试把它们当格式标记处理轻则显示错乱重则整条消息发送失败。我的处理方式很粗暴卡密内容一律用 HTML 模式包在 pre 标签里并且去掉对卡密内容的格式解析。# sender.py — 用 HTML 模式发送卡密 from telegram import Bot async def send_card(bot: Bot, chat_id: int, card_content: str): text fpre{card_content}/pre # parse_mode 只对模板生效, 卡密原文不会参与解析 await bot.send_message(chat_idchat_id, texttext, parse_modeHTML)这样做的好处是万无一失卡密里的任何字符都按原样展示。代价是买家长按复制时要多点一下但比起消息发送失败这点体验损失完全可以接受。4. 二开避坑指南我在 TG 发卡机器人上踩过的 5 个坑4.1 并发下单导致库存超卖扣减必须带条件现象两个买家几乎同时点下单后台出现库存为 -1或者两张订单对应同一张卡密。原因代码写成了「先 SELECT stock再在应用层判断大于 0最后 UPDATE」。两个请求都读到了 stock1都通过判断于是都执行了 UPDATE。解决扣库存的 SQL 永远带条件像第三章那样写成UPDATE products SET stock stock - 1 WHERE id %s AND stock 0然后通过受影响行数判断是否成功。这个写法把并发控制交给了数据库行锁比自己加应用层锁可靠得多。上线前记得写个并发测试脚本开 50 个线程同时下单同一个商品断言库存不为负数。4.2 重复发货轮询任务被 cron 叠加执行现象买家反馈收到了两条一模一样的卡密后台看同一订单的发货记录有两条。原因我一度把发货任务同时挂在了 cron 每 2 分钟执行和 asyncio 循环里。两个定时器触发时间重叠两个进程都扫描到了同一笔 paid 订单都执行了发货。解决发货前必须把订单状态更新和卡密标记做成一个带条件的事务。在 orders 表加 shipped 字段发货语句统一写成UPDATE orders SET shipped 1 WHERE order_no %s AND shipped 0判断行数为 1 才继续取卡密行数为 0 说明别的进程已经在处理直接跳过。这是幂等性设计比任何「定时器错开」的方案都可靠。4.3 机器人突然没反应先查是不是起了两个进程现象日志里没有报错但机器人对消息不再响应重启后正常跑一段时间又哑火。原因最常见的是run_polling抛出了 409 Conflict——同一个 bot token 被两个进程占用了Telegram 只允许一个长轮询连接。常常是开发调试时终端里跑着一个旧进程没关又用 systemd 起了一个新的。解决遇到机器人失联第一步不要重启先执行ps aux | grep python检查有没有重复进程第二步看日志里有没有Conflict: terminated by other getUpdates request。之后养成习惯进程守护统一交给 systemd开发调试时明确关掉终端进程再起服务。4.4 订单时间差 8 小时时区问题让「今日订单」统计对不上现象后台统计今日订单只有凌晨 3 点到现在的数据但打开数据库看下午的订单创建时间却是早上。原因发卡系统服务器的 PHP 时区设置是 Asia/Shanghai而订单表用的字段类型是 timestamp数据库连接会话时区是 UTC。同一个时间一边存成 UTC一边展示成 UTC8所有日期分组都错位了。解决统一在应用层处理。订单表创建时间我用 datetime 类型存储插入时由代码显式写入本地时间字符串统计时用 date(order_date) 直接分组其中 order_date 是插入时冗余写入的Y-m-d字段。这样数据库会话时区怎么变都不影响统计。4.5 双语文案缺失用户切到英文后看到中文原文现象买家把语言切成 English订单状态消息里「支付成功」还是中文只有欢迎语是英文。原因语言包 JSON 里漏了 key代码里的 _t 函数没有 fallback 机制直接返回了空或抛异常导致走了默认分支硬编码了中文。解决_t 函数的实现强制带两层 fallback优先用户所选语言其次回退到默认语言比如英语再其次返回 key 本身。上线前写个校验脚本扫描代码里所有 _t 调用检查每个 key 在两种语言包里都存在缺了就报错。5. 双语言不是翻译两遍语言包、用户偏好与文案渲染的正确做法5.1 语言包是 JSON 键值对够用就好别上重型框架个人发卡机器人的文案满打满算几十条不需要引入 gettext 之类的完整国际化框架。我用最朴素的 JSON 文件组织放在项目 lang 目录下{ start_welcome: { zh: 欢迎使用小店发卡机器人。输入 /buy 商品ID 下单输入 /list 查看商品。, en: Welcome to our shop bot. Send /buy PRODUCT_ID to order, or /list to browse products. }, buy_usage: { zh: 用法/buy 商品ID例如 /buy 2, en: Usage: /buy PRODUCT_ID, e.g. /buy 2 }, card_delivery: { zh: 您的卡密\n{card_content}\n感谢购买。, en: Your card:\n{card_content}\nThank you for your purchase. } }加载逻辑也简单程序启动时把两个 JSON 读进内存查 key 时先查用户语言没有就查默认语言# i18n.py — 最小双语言实现 import json, os _packs {} for lang in (zh, en): with open(flang/{lang}.json, r, encodingutf-8) as f: _packs[lang] json.load(f) DEFAULT_LANG en def _t(lang: str, key: str, **kwargs) - str: template _packs.get(lang, {}).get(key) or _packs[DEFAULT_LANG].get(key, key) for k, v in kwargs.items(): template template.replace({ k }, str(v)) return template参数说明kwargs 是模板参数用于渲染订单号、金额这类动态内容。注意生成环境要确保磁盘上的 JSON 不包含注释——JSON 标准不支持注释编辑器里的绿色注释行会导致 json.load 直接报错。5.2 用户偏好存哪单独一张偏好表别动发卡系统用户的语言偏好只对机器人有意义发卡系统不需要知道买家说的是中文还是英文。所以我在机器人自己的 SQLite 里建了一张表和发卡系统的业务库隔离CREATE TABLE user_lang ( user_id INTEGER PRIMARY KEY, lang TEXT NOT NULL DEFAULT en, updated_at DATETIME );机器人在每条命令处理前查一次。查不到就走默认语言。切换命令 /lang 的实现就是 upsert# lang.py — 用户切换语言 async def set_lang(update: Update, context: ContextTypes.DEFAULT_TYPE): user_id update.effective_user.id lang context.args[0].lower() if context.args else None if lang not in (zh, en): await update.message.reply_text(Usage: /lang zh or /lang en) return sqlite_exec(INSERT INTO user_lang (user_id, lang, updated_at) VALUES (?, ?, ?) ON CONFLICT(user_id) DO UPDATE SET lang excluded.lang, (user_id, lang, time.strftime(%Y-%m-%d %H:%M:%S))) await update.message.reply_text(_t(lang, lang_switched))逻辑说明ON CONFLICT 子句是 SQLite 的 upsert 语法避免「先 SELECT 再 UPDATE」的竞态。这里有个值得注意的细节机器人回复的文案应该用新语言而不是用户之前的语言所以最后一行用的是新传入的 lang。5.3 文案渲染用占位符不要字符串拼接跨语言最隐蔽的坑是词序。中文说「订单号xxx金额100 元」英文说「Order xxx, amount 100 CNY」把动态值硬拼出来会让翻译完全没法做。所以语言包里一律写{order_no}、{amount}这种占位符渲染时按参数替换整句的顺序交给翻译人员。我踩过一次亏早期代码里写f您的订单号是 {order_no}后来加英文包时发现没法直接复用了只好把这一句拆掉重写。从那以后所有用户可见文案都进语言包代码里不允许出现裸的 f-string 拼接文案。这算一条纪律不是风格偏好。6. 上线后的进阶配置队列削峰、结构化日志与一张订单看板6.1 用 Redis 队列给爆单场景削峰先返回「正在处理」再慢慢消化平时几百单一天直接操作发卡系统没问题。可一旦你在 TG 群里发了个限时优惠几十人同时点 /buy发卡系统的下单接口可能几秒内被打爆订单没生成买家却收到了「下单失败」。这时候我给下单入口加一层 Redis 队列消息进队后立刻回复「正在处理」worker 进程按固定速率消费队列逐个创建订单。# worker.py — Redis 队列消费下单请求 import redis, json r redis.Redis(host127.0.0.1, port6379, db0) def enqueue_order(user_id: int, product_id: int): r.lpush(order_queue, json.dumps({user_id: user_id, product_id: product_id})) def order_worker(): while True: _, payload r.brpop(order_queue, timeout0) data json.loads(payload) try: create_order(user_iddata[user_id], product_iddata[product_id]) except Exception as e: # 失败先回队, 最多重试 3 次, 之后写死信日志人工介入 r.lpush(order_queue_dead, payload)逻辑说明brpop 的 timeout0 表示阻塞等待直到队列里有新任务才返回。这是抢占式消费多个 worker 进程同时跑也不会把同一单取走。要不要上队列的判断标准很简单你的发卡系统在下单高峰期有没有出现过连接超时或数据库死锁。没有就别加这是典型的「没有痛点就不要造轮子」的组件。6.2 结构化日志一条订单的生命周期可以一键追踪排障效率一半取决于日志格式。不要把日志写成一行行自由文本而是每条日志输出一个 JSON 字典字段固定让 grep 能按订单号把整个链路拉出来。import logging, json def log_order_event(order_no: str, event: str, result: str, user_id: int, cost_ms: int): logging.info(json.dumps({ order_no: order_no, user_id: user_id, event: event, # order_created / payment_confirmed / card_shipped result: result, # ok / failed / retry cost_ms: cost_ms, time: time.strftime(%Y-%m-%d %H:%M:%S) }))排查流程就变成grep TGB20250101... app.log订单从创建到发货的每一步都串起来了。没有 trace_id 的经验我吃过亏——买家说没收到货我却要在几千行日志里人肉翻这个习惯建议从一开始就养。6.3 一张订单看板几条 SQL 顶一个后台不需要额外的监控平台发卡系统数据库里跑几条 SQL 就能得到今天最重要的经营指标-- 今日成单量与销售额 SELECT DATE(create_time) AS day, COUNT(*) AS orders, SUM(amount) AS revenue FROM orders WHERE status paid GROUP BY DATE(create_time); -- 热销商品 Top 5 SELECT product_id, COUNT(*) AS sold FROM orders WHERE status paid GROUP BY product_id ORDER BY sold DESC LIMIT 5; -- 卡密库存预警: 列出剩余不足 10 张的商品 SELECT product_id, COUNT(*) AS remaining FROM cards WHERE status 0 GROUP BY product_id HAVING remaining 10;这些查询建议做成只读账号执行避免误操作。我个人的习惯是每周一看一眼库存预警低于阈值就提前补卡密宁可在没人买的时候补也不要等爆单了才手忙脚乱。这个方向做到这一步已经能覆盖一个个人发卡业务从接单到发货到统计的全部环节。回想我自己第一次把机器人跑通线上遇到的第一批问题几乎都集中在并发、幂等和时区这三类希望上面这些经验能让你少走一轮弯路。祝顺利希望帮到你。本文还有配套的精品资源点击获取
