自动发卡系统源码部署教程：环境配置与运行流程

自动发卡系统通常用于销售可在线交付的数字商品：买家付款后，系统按订单发放卡密、兑换码、授权凭证或下载信息。它可以减少人工核对和发送的工作量，但并不意味着接入程序后就能安全地自动确认收款。商品授权、支付回调验证、库存保护和凭证保密都需要单独设计。

本文用 Flask 和 SQLite 搭建一个可本地运行的最小示例，演示添加卡密、创建订单、核验收款后发货的基本流程。示例不连接真实支付渠道，也不提供公开的“买家自行确认已付款”接口。它适合开发学习、内部验证和小规模原型，不应未经加固直接用于公开售卖。

一、适用场景与不适用场景

这类系统适合交付权属清晰、允许转售或分发的数字商品，例如自有产品的兑换码、经授权销售的软件许可证，以及由商家合法发行的数字服务凭证。具体能否销售，应以商品授权协议、平台规则和适用法律为准。

自动发卡不适用于绕过账号安全机制、转售未经授权的密钥、分发盗版内容、出售个人信息或其他违法商品。系统能自动交付，不代表商品本身获得了销售授权。部署者应先核实商品来源、使用范围、退款规则和售后责任。

还要区分“订单创建”和“付款完成”。买家访问网站或提交订单，只能说明产生了一个订单记录，不能证明款项已到账。自动交付必须依赖可信的支付渠道通知或经过授权人员的核验；不能仅凭浏览器跳转、前端参数或买家提交的截图发货。

二、运行环境与部署边界

示例使用 Python 3.10 或更新版本、Flask 和 Python 自带的 SQLite。Python 3.10 是本文代码的建议运行环境，不是 Flask 对所有版本的强制要求。安装 Flask 时，pip 会根据当前 Python 版本选择可兼容的包；若安装失败，应检查 Python、pip 和 Flask 的兼容性，而不要直接套用未经核实的固定版本号。

SQLite 适合本地学习和低并发原型。它是单文件数据库，不需要单独安装数据库服务，但在多个进程同时写入、需要高可用或需要精细数据库运维时，应评估 PostgreSQL、MySQL 等数据库。切换数据库时，事务、锁和字段类型的行为可能不同，需要重新测试库存与发货逻辑。

以下示例只绑定本机地址，不提供生产环境 Web 服务器配置。生产部署还需要 HTTPS、反向代理、进程管理、访问控制、日志轮转、备份和恢复方案。不得把调试服务器直接暴露到公网。

三、创建项目并安装依赖

新建项目目录，并在其中创建虚拟环境。虚拟环境可以避免项目依赖与系统 Python 包相互影响。

mkdir card-store
cd card-store
python -m venv .venv

在 macOS 或 Linux 上激活环境：

source .venv/bin/activate

在 Windows PowerShell 上激活环境：

.venv\Scripts\Activate.ps1

安装 Flask：

python -m pip install Flask

如需记录依赖，可在当前环境中导出实际安装的版本：

python -m pip freeze > requirements.txt

这会将当前环境中安装的包及版本写入文件。若环境中还安装了其他无关包，文件也会包含它们。正式项目应在干净的虚拟环境中安装和锁定依赖，并在目标部署环境验证。

四、实现一个最小可运行服务

创建 app.py。代码提供三个本地演示功能：查看商品库存数量、创建待核验订单，以及由本机管理员通过命令行确认指定订单的款项并发货。它不接收银行卡信息，不模拟支付渠道，也没有公开的确认付款接口。

示例中的卡密存储在 SQLite 明文列中，因此只适合本地演示。生产环境应限制数据库文件权限、限制数据库访问，并根据威胁模型评估凭证加密、密钥管理和审计要求。加密实现不能只把密钥硬编码在源码里。

import os
import sqlite3
import sys
import uuid
from pathlib import Path

from flask import Flask, jsonify, request

app = Flask(__name__)

BASE_DIR = Path(__file__).resolve().parent
DB_PATH = Path(os.environ.get("CARD_DB_PATH", BASE_DIR / "store.db"))


def connect_db():
    connection = sqlite3.connect(DB_PATH, timeout=10)
    connection.row_factory = sqlite3.Row
    connection.execute("PRAGMA foreign_keys = ON")
    return connection


def init_db():
    with connect_db() as db:
        db.executescript("""
        CREATE TABLE IF NOT EXISTS products (
            id INTEGER PRIMARY KEY,
            name TEXT NOT NULL,
            price_cents INTEGER NOT NULL CHECK (price_cents >= 0)
        );

        CREATE TABLE IF NOT EXISTS cards (
            id INTEGER PRIMARY KEY,
            product_id INTEGER NOT NULL REFERENCES products(id),
            secret TEXT NOT NULL,
            order_id TEXT UNIQUE,
            FOREIGN KEY (order_id) REFERENCES orders(id)
        );

        CREATE TABLE IF NOT EXISTS orders (
            id TEXT PRIMARY KEY,
            product_id INTEGER NOT NULL REFERENCES products(id),
            amount_cents INTEGER NOT NULL CHECK (amount_cents >= 0),
            status TEXT NOT NULL CHECK (status IN ('pending', 'paid', 'fulfilled')),
            created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
        );
        """)
        db.execute(
            "INSERT OR IGNORE INTO products(id, name, price_cents) VALUES (?, ?, ?)",
            (1, "演示兑换码", 500)
        )


def add_card(product_id, secret):
    if not secret or not secret.strip():
        raise ValueError("卡密不能为空")

    with connect_db() as db:
        db.execute(
            "INSERT INTO cards(product_id, secret) VALUES (?, ?)",
            (product_id, secret.strip())
        )


def create_order(product_id):
    order_id = uuid.uuid4().hex

    with connect_db() as db:
        product = db.execute(
            "SELECT id, price_cents FROM products WHERE id = ?",
            (product_id,)
        ).fetchone()

        if product is None:
            return None, "商品不存在"

        stock = db.execute(
            "SELECT COUNT(*) AS n FROM cards "
            "WHERE product_id = ? AND order_id IS NULL",
            (product_id,)
        ).fetchone()["n"]

        if stock == 0:
            return None, "库存不足"

        db.execute(
            "INSERT INTO orders(id, product_id, amount_cents, status) "
            "VALUES (?, ?, ?, 'pending')",
            (order_id, product_id, product["price_cents"])
        )

    return order_id, None


def fulfill_order(order_id):
    # BEGIN IMMEDIATE 获取写锁，避免两个本地写入同时领取同一张卡。
    with connect_db() as db:
        db.execute("BEGIN IMMEDIATE")

        order = db.execute(
            "SELECT id, product_id, status FROM orders WHERE id = ?",
            (order_id,)
        ).fetchone()

        if order is None:
            return None, "订单不存在"

        if order["status"] == "fulfilled":
            card = db.execute(
                "SELECT secret FROM cards WHERE order_id = ?",
                (order_id,)
            ).fetchone()
            return card["secret"], None

        if order["status"] != "paid":
            return None, "订单尚未核验为已付款"

        card = db.execute(
            "SELECT id, secret FROM cards "
            "WHERE product_id = ? AND order_id IS NULL "
            "ORDER BY id LIMIT 1",
            (order["product_id"],)
        ).fetchone()

        if card is None:
            return None, "库存不足，请人工处理"

        db.execute(
            "UPDATE cards SET order_id = ? WHERE id = ?",
            (order_id, card["id"])
        )
        db.execute(
            "UPDATE orders SET status = 'fulfilled' WHERE id = ?",
            (order_id,)
        )

        return card["secret"], None


@app.get("/health")
def health():
    return jsonify({"status": "ok"})


@app.get("/products")
def products():
    with connect_db() as db:
        rows = db.execute("""
            SELECT p.id, p.name, p.price_cents,
                   (SELECT COUNT(*) FROM cards c
                    WHERE c.product_id = p.id AND c.order_id IS NULL) AS stock
            FROM products p
            ORDER BY p.id
        """).fetchall()

    return jsonify([
        {
            "id": row["id"],
            "name": row["name"],
            "price_cents": row["price_cents"],
            "stock": row["stock"],
        }
        for row in rows
    ])


@app.post("/orders")
def orders():
    if not request.is_json:
        return jsonify({"error": "请求必须使用 JSON"}), 415

    data = request.get_json(silent=True)
    if not isinstance(data, dict):
        return jsonify({"error": "无效的 JSON 请求"}), 400

    product_id = data.get("product_id")
    if isinstance(product_id, bool) or not isinstance(product_id, int):
        return jsonify({"error": "product_id 必须是整数"}), 400

    order_id, error = create_order(product_id)
    if error:
        return jsonify({"error": error}), 400

    return jsonify({
        "order_id": order_id,
        "status": "pending",
        "message": "订单已创建，尚未确认付款"
    }), 201


def cli():
    if len(sys.argv) < 2:
        print("用法：python app.py init | add-card PRODUCT_ID CARD_CODE | mark-paid ORDER_ID")
        raise SystemExit(2)

    command = sys.argv[1]

    if command == "init":
        init_db()
        print("数据库已初始化")
    elif command == "add-card" and len(sys.argv) == 4:
        add_card(int(sys.argv[2]), sys.argv[3])
        print("卡密已加入库存")
    elif command == "mark-paid" and len(sys.argv) == 3:
        order_id = sys.argv[2]
        with connect_db() as db:
            result = db.execute(
                "UPDATE orders SET status = 'paid' "
                "WHERE id = ? AND status = 'pending'",
                (order_id,)
            )
            if result.rowcount != 1:
                print("订单不存在或状态不是 pending")
                raise SystemExit(1)

        secret, error = fulfill_order(order_id)
        if error:
            print(error)
            raise SystemExit(1)
        print("发货凭证：", secret)
    else:
        print("参数不正确")
        raise SystemExit(2)


if __name__ == "__main__":
    if len(sys.argv) > 1:
        cli()
    else:
        init_db()
        app.run(host="127.0.0.1", port=5000, debug=False)

这段代码使用 SQL 参数占位符传入外部值，避免将用户输入拼接进 SQL 语句。订单金额由服务端商品记录读取，而不是接受客户端提交的价格。订单 ID 使用随机 UUID 生成；这有助于降低被顺序猜测的风险，但不等于访问控制，生产环境不能把“订单 ID 难猜”当作保护凭证的唯一措施。

库存查询和订单创建不是一项完整的支付事务：用户创建待核验订单后，库存仍可能被其他订单占用。因此，真正确认付款时必须再次原子地检查并分配库存。示例的 fulfill_order 使用 SQLite 写事务，并在分配时只选择尚未绑定订单的卡密；若付款后库存已耗尽，程序会报错，运营方需要退款或按规则处理，不能静默丢单。

五、初始化、录入库存与本机运行

先初始化数据库：

python app.py init

加入一条演示卡密。命令行参数会出现在终端历史或进程信息中，因此不要用这种方式录入真实敏感凭证，也不要把真实卡密写入 shell 历史、工单或版本库。

python app.py add-card 1 DEMO-ONLY-0001

启动本地服务：

python app.py

默认监听 127.0.0.1:5000，仅接受本机连接。可用浏览器或命令行检查健康状态和商品信息：

curl http://127.0.0.1:5000/health
curl http://127.0.0.1:5000/products

创建待付款订单：

curl -X POST http://127.0.0.1:5000/orders \
  -H "Content-Type: application/json" \
  -d '{"product_id": 1}'

响应会包含 order_id。此时状态是 pending，订单没有被付款，也不会收到卡密。示例没有向买家提供领取卡密的接口，所以订单 ID 本身不会直接泄漏凭证。

六、演示核验与发货流程

为了测试完整的本地流程，运营者先通过支付渠道后台或其他可信方式核验订单确实到账，再在终端执行：

python app.py mark-paid 在创建订单时返回的订单ID

该命令会将状态为 pending 的订单改为 paid，随后从对应商品库存中分配一条未使用卡密，并将订单设为 fulfilled。重复处理已发货订单时，代码返回同一条已绑定卡密，避免再次分配另一条库存。实际业务是否允许重复查看、需要怎样的审计记录，应根据售后和安全要求设计。

命令行中的 mark-paid 是本地演示工具，不是支付接口。把它改造成无需身份验证的网页路由，或让买家可以自行调用，都会使任何人能够伪造付款并领取库存。生产系统应由可信支付渠道的服务端通知触发核验，或者由受权限控制的后台操作人员确认；每次操作都应留下可追踪的记录。

七、接入支付时必须处理的边界

不同支付渠道的回调格式、签名算法、字段名称、重试行为和金额单位并不相同，不能假定一个通用示例适用于所有渠道。接入前应以所选渠道当前的官方文档为准，至少明确以下事项：

• 验签：按渠道要求验证签名，并使用服务端保存的密钥或证书。不能仅检查请求中出现了“支付成功”字样。
• 订单对应：确认回调中的商户订单号与本系统订单一致，并核对商户身份、商品或业务类型。
• 金额与币种：与服务端订单金额、币种逐项核对。若渠道以最小货币单位传金额，应遵循该渠道文档，不能自行按浮点数换算。
• 状态流转：只接受符合业务规则的状态变化。重复通知应能够安全处理，不能重复扣库存或重复发货。
• 响应与重试：按渠道文档返回正确应答，并为超时、重复通知和处理失败设计重试及对账机制。
• 密钥管理：把支付密钥放在权限受控的环境变量或密钥管理服务中，不提交到源码仓库，不输出到日志。

验签成功也不等于所有业务检查都完成。回调可能重复到达，也可能在本系统暂时无法发货时到达。推荐使用数据库事务和唯一约束控制幂等；记录支付事件的渠道交易号，并让“核验付款”和“分配库存”在可恢复的状态机中运行。若支付已确认但发货失败，应保留待处理记录并告警，不能把失败伪装成成功。

八、配置与安全检查清单

• 不要暴露调试服务：示例关闭了 Flask 调试模式并监听本机地址，但这不构成生产部署。使用正式 WSGI 服务和反向代理，并在代理层配置 HTTPS。
• 限制后台权限：录入卡密、查看凭证、人工确认和退款等操作应有身份验证、最小权限和审计日志。管理端不应只靠隐藏路径保护。
• 保护数据库及备份：限制数据库文件和备份文件的操作系统权限。备份也含有库存凭证，必须纳入访问控制、加密和保留策略。
• 防止凭证泄漏：避免把完整卡密写入访问日志、异常堆栈、监控标签、邮件主题或公开订单页面。按业务需要设置凭证查看和售后验证流程。
• 限制输入：示例只检查商品 ID 的类型，没有实现速率限制、请求大小上限、跨站请求防护、登录验证或完整的参数校验。公网服务必须补齐这些措施。
• 防止滥用：对创建订单、查询订单和后台操作实施合理的速率限制与访问控制。只使用随机订单号并不能防止刷单、资源耗尽或自动化攻击。
• 明确售后策略：提前定义缺货、支付成功但发货失败、重复支付、退款、卡密已使用和争议订单的处理方式，并保留必要的业务记录。

九、常见错误与排查方式

1. 提示 Flask 找不到

通常是虚拟环境没有激活，或 Flask 被装到了另一套 Python 环境。先确认 python 和 pip 指向同一环境，再运行 python -m pip show Flask。不要在不确定环境的情况下同时使用多个 Python 安装路径。

2. 数据库中没有商品或卡密

检查是否在项目目录执行了初始化命令，以及 CARD_DB_PATH 是否指向了另一个数据库文件。示例默认使用 app.py 同目录下的 store.db。更换数据库路径后，需要在新路径中重新初始化并录入库存。

3. 创建订单提示库存不足

确认卡密已加入正确的商品 ID。示例默认商品 ID 为 1；卡密只会分配给对应商品。库存被绑定到订单后不会再被视为可用库存。不要直接手工改数据库状态来“恢复”卡密，应先核实该凭证是否已交付或使用。

4. 订单一直处于 pending

这是预期行为：示例没有支付接入，创建订单不会自动变成已付款。只有通过可信核验后，演示者才使用命令行执行 mark-paid。不能因为页面跳转成功或用户声称已付款就改变订单状态。

5. 报告订单已付款，但没有卡密

可能是商品库存已耗尽，或订单状态不符合发货条件。示例会在库存不足时停止发货，不会生成虚假凭证。生产系统应将此情况记录为待处理故障，通知运营人员按退款、补货或联系买家等规则处理。

6. 数据库出现锁定或并发异常

SQLite 支持事务，但并不适合无限增加的并发写入。本文设置了有限的连接等待时间，并使用写事务串行化发货；它不能替代容量评估和压力测试。并发量上升时，应考虑迁移到适合业务的数据库，并重新验证事务隔离、幂等约束和库存竞争处理。

十、上线前的最小验收

至少使用测试商品检查以下场景：正常创建订单、无效商品 ID、无库存、未核验订单不能发货、同一订单重复执行不重复占用库存、不同订单并发领取时不拿到同一张卡，以及付款成功但库存不足时能否留下可追踪的待处理状态。对真实支付渠道还要覆盖验签失败、金额不匹配、重复通知、通知超时和渠道重试。

上线前还应确认：商品授权和销售规则明确；密钥不在代码仓库；管理端有访问控制；数据库和备份有权限保护；服务通过 HTTPS 对外提供；日志不会记录明文卡密；监控能够发现支付成功但交付失败；有经过演练的备份恢复和退款流程。本文代码没有实现这些生产能力，应将其视为学习用骨架，而不是完整的商业系统。

总结

自动发卡的核心并非把卡密放进数据库后自动显示，而是可靠地区分待付款、已付款和已交付，确保每笔真实付款最多触发一次正确发货，并在库存或网络故障时留下可恢复的状态。本文示例展示了本地环境、商品和库存、订单创建以及人工核验后的事务化发货流程；真实部署还必须按照支付渠道官方规范实现验签、金额核对、幂等和安全控制。