手写一个最小可运行的转卡码系统 — 从下单到自动发卡的完整实现

一、转卡码系统到底在做什么

转卡码（自动发卡）系统的业务链路其实很短：用户下单支付 → 系统收到支付成功回调 → 自动把卡密发给用户。但就是这条看似简单的链路，牵扯出库存管理、支付回调验签、幂等发卡、卡密核销等一系列工程问题。网上卖转卡码源码的不少，但大多封装成了黑盒，出了问题只能干瞪眼。

本文不聊大架构，直接带你手写一个最小可运行的转卡码系统：Python + Flask + Redis + SQLite，核心代码只有两百行，覆盖下单、模拟支付回调、自动发卡、查询核销的完整闭环，本地几分钟就能跑起来。看完你就能彻底搞懂转卡码系统的核心原理——生产级的代理分润、防封入口池、多通道路由等模块，都是在这套骨架上长出来的。

二、先设计两张表

最小系统只需要两张表：orders（订单）和 cards（卡密库存）。生产环境用 MySQL，这里为了"零配置可运行"用 SQLite，SQL 基本通用：

-- cards：卡密库存表，status 驱动状态流转
CREATE TABLE cards (
  id         INTEGER PRIMARY KEY AUTOINCREMENT,
  product_id TEXT    NOT NULL,                    -- 商品ID，如 vip_month
  code       TEXT    NOT NULL UNIQUE,             -- 卡密明文
  status     TEXT    NOT NULL DEFAULT 'UNSOLD',   -- UNSOLD/SOLD/USED
  order_no   TEXT,                                -- 售出时绑定的订单号
  sold_at    TEXT
);

-- orders：订单表，PENDING → PAID → DELIVERED
CREATE TABLE orders (
  id         INTEGER PRIMARY KEY AUTOINCREMENT,
  order_no   TEXT    NOT NULL UNIQUE,
  product_id TEXT    NOT NULL,
  amount     INTEGER NOT NULL,                    -- 金额，单位：分
  status     TEXT    NOT NULL DEFAULT 'PENDING',
  card_code  TEXT,
  created_at TEXT    DEFAULT (datetime('now','localtime'))
);

两个关键设计：卡密状态机（未售 → 已售 → 已核销）和订单状态机（待支付 → 已支付 → 已发货）。后续所有并发问题，都靠这两组状态加上原子 UPDATE 解决，这是整篇文章最值得记住的点。

三、卡密批量生成：随机源 + 校验位

卡密是平台的"钱"，生成必须用密码学安全的随机源，绝不能用 random。这里用 secrets.token_hex，再拼一位校验位，防止用户手工乱填也能蒙对格式：

import secrets

def check_digit(body: str) -> str:
    # 简单校验位：所有字符 ASCII 码加权求和后取个位
    return str(sum(ord(c) for c in body) % 10)

def gen_card_code(batch: str, idx: int) -> str:
    # 2位批次 + 4位序号 + 10位随机 + 1位校验 = 17位卡密
    body = f"{batch}{idx:04d}{secrets.token_hex(5).upper()}"
    return body + check_digit(body)

# 批量生成 1000 张并入库（INSERT OR IGNORE 防重复导入）
def import_cards(product_id, batch, count=1000):
    rows = [(product_id, gen_card_code(batch, i)) for i in range(count)]
    db.executemany(
        "INSERT OR IGNORE INTO cards (product_id, code) VALUES (?, ?)", rows)
    print(f"导入 {len(rows)} 张卡密，库存同步到 Redis")
    r.set(f"stock:{product_id}", count)

INSERT OR IGNORE 配合 code UNIQUE 约束，即使脚本重复执行也不会产生重复卡密——这是数据层的最后一道保险。最后一行把库存同步进 Redis，为下单环节的原子扣减做准备。

四、下单接口：Redis 防重 + 原子扣库存

下单是最容易被刷的环节，做两层防护：SETNX 防重复提交（同一 IP 3 秒内只能下一次单），DECR 原子扣减库存（Redis 单线程执行，天然不会超卖）：

@app.post("/api/order")
def create_order():
    data = request.get_json()
    pid, ip = data["product_id"], request.remote_addr

    # 1. 防重：3 秒内同一 IP 重复下单直接拒绝
    if not r.set(f"dup:{pid}:{ip}", 1, nx=True, ex=3):
        return {"code": 429, "msg": "操作太频繁"}

    # 2. 原子扣减 Redis 库存，扣成负数则回补并拒绝
    remain = r.decr(f"stock:{pid}")
    if remain < 0:
        r.incr(f"stock:{pid}")
        return {"code": 400, "msg": "库存不足"}

    # 3. 生成订单号 + 支付宝当面付预下单，返回收款二维码
    order_no = f"{int(time.time())}{secrets.randbelow(10000):04d}"
    qr_url = alipay_precreate(order_no, pid, PRICES[pid])
    db.execute("INSERT INTO orders (order_no, product_id, amount) VALUES (?,?,?)",
               (order_no, pid, PRICES[pid]))
    return {"code": 0, "order_no": order_no, "qr_url": qr_url}

注意"先扣 Redis 再落订单"的顺序：如果先查库存再扣，高并发下两个请求会同时读到"还剩 1 张"，双双放行就超卖了；DECR 是原子的，永远不会出现这种竞态。订单号用时间戳加随机数拼出来，保证全局唯一，也方便后面做对账。

五、支付回调：验签 + 幂等 + 自动发卡

回调是整条链路的心脏，也是最容易出错的地方。三个原则缺一不可：必须验签（否则任何人伪造一个 TRADE_SUCCESS 就能白嫖卡密）、必须幂等（支付宝回调会重试多次，重复发卡就是事故）、发卡必须原子（并发回调不能发出两张卡）：

@app.post("/api/alipay/notify")
def alipay_notify():
    params = request.form.to_dict()
    # 1. 验签：用支付宝公钥验证签名，验不过直接拒绝
    if not verify_alipay_sign(params):
        return "failure"

    order_no = params["out_trade_no"]
    # 2. 幂等锁：SETNX 抢到锁才处理，回调重试直接返回 success
    if not r.set(f"paylock:{order_no}", 1, nx=True, ex=120):
        return "success"

    # 3. 状态机推进 + 原子取卡，双保险防重复发卡
    if params["trade_status"] == "TRADE_SUCCESS":
        row = db.get_order(order_no)
        if row and row["status"] == "PENDING":
            card = db.take_one_card(row["product_id"], order_no)
            db.mark_delivered(order_no, card)
            push_card_to_user(order_no, card)   # 短信/订阅消息推送
    return "success"

take_one_card 的核心是一条原子 SQL：UPDATE cards SET status='SOLD', order_no=? WHERE product_id=? AND status='UNSOLD' LIMIT 1，执行后再把这张卡查回来。两个并发回调同时进来，只有一条 UPDATE 能命中"未售"状态，另一条影响行数为 0——配合上面的 SETNX 锁，重复发卡的可能性被彻底堵死。

六、查询与核销接口

用户付完钱要能查到卡密，买家或下游渠道要能核销。核销同样要带状态条件，防止同一张卡被核销两次：

# 查询订单：返回卡密（生产环境建议先脱敏展示，点"查看"再明文）
@app.get("/api/order/<order_no>")
def get_order(order_no):
    row = db.get_order(order_no)
    if not row:
        return {"code": 404, "msg": "订单不存在"}
    return {"code": 0, "order_no": order_no, "status": row["status"],
            "card_code": row["card_code"]}

# 核销：只有 SOLD 状态能变成 USED，重复核销返回失败
@app.post("/api/card/use")
def use_card():
    code = request.get_json()["code"]
    n = db.execute(
        "UPDATE cards SET status='USED' WHERE code=? AND status='SOLD'",
        (code,)).rowcount
    return {"code": 0 if n else 400,
            "msg": "核销成功" if n else "卡密不存在或已使用"}

七、跑起来：三分钟验证全流程

环境只需要一个 Redis 和 Python 3.10+，依赖就三个包：

# 1. 启动 Redis（Docker 一行搞定）
docker run -d --name redis -p 6379:6379 redis:7-alpine

# 2. 安装依赖并启动服务
pip install flask redis requests
python app.py     # 监听 8000 端口

# 3. 模拟下单（拿到订单号）
curl -s -X POST http://127.0.0.1:8000/api/order \
  -H 'Content-Type: application/json' -d '{"product_id":"vip_month"}'

# 4. 模拟支付宝支付成功回调（生产环境由支付宝服务器发起）
curl -s -X POST http://127.0.0.1:8000/api/alipay/notify \
  -d 'out_trade_no=17550000001234&trade_status=TRADE_SUCCESS'

# 5. 查询订单：卡密已自动发放
curl -s http://127.0.0.1:8000/api/order/17550000001234

本地就能完整走通"下单 → 回调 → 发卡 → 查卡"全流程。把这套骨架部署到服务器，换上真实的支付宝应用密钥（当面付或电脑网站支付均可），就是一个能收钱的转卡码系统。对账、退款、超时关单这些进阶能力，可以在此基础上按订单状态机逐步补上。

八、小结：从最小实现到生产级

这个最小实现只有两百行，却把转卡码系统的三个核心机制讲透了：状态机驱动订单与卡密流转、Redis 原子操作扛住并发、验签加幂等守住资金安全。生产环境还需要补上防封入口池、代理分润、多通道路由、自动对账告警等模块——这些正是商城在售的转卡码系统 V3 内置的能力，全面重构的架构、开箱即用，比从零造轮子划算得多；转卡码系统 V2 则是久经运营验证的稳定版。写这套代码时我全程用 Codex Desktop 辅助调试，遇到验签、SQL 竞态这类问题直接丢给它看报错改代码，效率提升非常明显。建议你也动手敲一遍，再对照商城源码看生产版是如何补全的，收获会大得多。