新媒体复购自动化：搭建自动发放优惠券系统的完整教程

环境依赖安装与初始化

本文将基于Python语言和Flask框架，搭建一套完整的自动化复购系统。该系统会在用户支付成功后，自动判断其历史订单数量，针对老用户自动发放复购优惠券，并触发微信服务通知。需要在本地环境中安装必要的依赖库。

打开终端，直接执行以下命令安装核心依赖包：

```bash pip install flask pymysql requests redis xml.etree.ElementTree cryptography ```

安装完成后，创建项目目录结构。新建一个名为 `repurchase_system` 的文件夹，并在其中创建 `app.py`（主程序）、`config.py`（配置文件）和 `requirements.txt`。请确保你的开发环境中已安装 MySQL 5.7+ 和 Redis 5.0+。

数据库表结构设计

为了实现复购逻辑，我们需要设计三张核心表：用户表、订单表和优惠券表。请直接在 MySQL 数据库中执行以下 SQL 语句完成建表。这里假设数据库名为 `ecommerce_db`。

```sql CREATE DATABASE IF NOT EXISTS ecommerce_db; USE ecommerce_db; -- 用户表：存储基础信息 CREATE TABLE `users` ( `id` int(11) NOT NULL AUTO_INCREMENT, `openid` varchar(64) NOT NULL COMMENT '微信openid', `nickname` varchar(100) DEFAULT NULL, `created_at` datetime DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (`id`), UNIQUE KEY `uk_openid` (`openid`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; -- 订单表：记录交易流水 CREATE TABLE `orders` ( `id` int(11) NOT NULL AUTO_INCREMENT, `order_no` varchar(32) NOT NULL COMMENT '订单号', `user_id` int(11) NOT NULL, `total_fee` int(11) NOT NULL COMMENT '金额，单位：分', `status` tinyint(1) DEFAULT '0' COMMENT '0:待支付, 1:已支付', `transaction_id` varchar(64) DEFAULT NULL COMMENT '微信支付流水号', `paid_at` datetime DEFAULT NULL, PRIMARY KEY (`id`), UNIQUE KEY `uk_order_no` (`order_no`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; -- 优惠券表：记录发放详情 CREATE TABLE `coupons` ( `id` int(11) NOT NULL AUTO_INCREMENT, `user_id` int(11) NOT NULL, `coupon_code` varchar(20) NOT NULL, `amount` int(11) NOT NULL COMMENT '面额，单位：分', `status` tinyint(1) DEFAULT '0' COMMENT '0:未使用, 1:已使用', `source` varchar(50) DEFAULT 'REPURCHASE' COMMENT '来源：复购赠送', `created_at` datetime DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (`id`), KEY `idx_user_id` (`user_id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; ```

核心配置与工具类编写

在 `config.py` 文件中，填入你的数据库连接、Redis 连接以及微信支付相关的密钥信息。请务必替换为你自己在微信商户平台和公众号后台获取的真实密钥。

```python config.py import os class Config: 数据库配置 MYSQL_HOST = '127.0.0.1' MYSQL_PORT = 3306 MYSQL_USER = 'root' MYSQL_PASS = 'your_mysql_password' MYSQL_DB = 'ecommerce_db' Redis配置 REDIS_HOST = '127.0.0.1' REDIS_PORT = 6379 REDIS_DB = 0 微信支付配置 WECHAT_MCH_ID = '1234567890' 商户号 WECHAT_API_KEY = 'your_api_key_v3_key' API v3 密钥 WECHAT_APPID = 'wx1234567890abcdef' 小程序或公众号AppID WECHAT_APP_SECRET = 'your_app_secret' AppSecret 模板消息ID (需在公众号后台申请) TEMPLATE_ID_COUPON = 'abc123xyz456' ```

接下来在 `app.py` 中初始化数据库连接池和 Redis 实例。为了保证高并发下的数据一致性，我们将使用 Redis 来实现分布式锁，防止微信重复回调导致多次发放优惠券。

```python app.py 首部引入与初始化 from flask import Flask, request, render_template_string import pymysql from redis import Redis import hashlib import random import string import time import xml.etree.ElementTree as ET import requests import json from config import Config app = Flask(__name__) 数据库连接 def get_db_conn(): return pymysql.connect( host=Config.MYSQL_HOST, port=Config.MYSQL_PORT, user=Config.MYSQL_USER, password=Config.MYSQL_PASS, database=Config.MYSQL_DB, cursorclass=pymysql.cursors.DictCursor ) Redis 连接 redis_client = Redis(host=Config.REDIS_HOST, port=Config.REDIS_PORT, db=Config.REDIS_DB) ```

支付回调监听与复购判断

这是系统的核心部分。我们需要创建一个接口接收微信支付的成功通知。微信支付通知以 XML 格式 POST 过来。我们需要解析 XML，验证签名（此处省略复杂签名验证逻辑，生产环境必须加上），更新订单状态，并触发复购逻辑。

在 `app.py` 中添加以下路由和处理逻辑：

```python def generate_coupon_code(): """生成随机优惠券码""" return ''.join(random.choices(string.ascii_uppercase + string.digits, k=12)) def check_repurchase_and_issue_coupon(user_id, current_order_no): """ 检查是否为复购并发放优惠券 :param user_id: 用户ID :param current_order_no: 当前订单号（排除本次订单计算历史） """ conn = get_db_conn() try: with conn.cursor() as cursor: 1. 查询该用户在当前订单之前的已支付订单数量 sql = """ SELECT COUNT() as count FROM orders WHERE user_id = %s AND status = 1 AND order_no != %s """ cursor.execute(sql, (user_id, current_order_no)) result = cursor.fetchone() history_order_count = result['count'] 2. 定义复购策略：如果是第2次购买（即历史有1单），发放优惠券 也可以修改为 history_order_count > 0 即只要买过就送 if history_order_count == 1: print(f"用户 {user_id} 检测到复购，准备发放优惠券...") 3. 防重锁：确保同一用户在同一时间窗口内不会重复发放 lock_key = f"lock:coupon:{user_id}" if redis_client.set(lock_key, 1, nx=True, ex=60): try: 生成面额为500分（5元）的优惠券 code = generate_coupon_code() issue_sql = """ INSERT INTO coupons (user_id, coupon_code, amount, source) VALUES (%s, %s, %s, 'REPURCHASE') """ cursor.execute(issue_sql, (user_id, code, 500)) conn.commit() 获取用户OpenID用于发送通知 cursor.execute("SELECT openid FROM users WHERE id = %s", (user_id,)) user = cursor.fetchone() if user: send_coupon_notification(user['openid'], code, 500) finally: redis_client.delete(lock_key) finally: conn.close() @app.route('/api/pay/callback', methods=['POST']) def pay_callback(): """ 微信支付回调接口 """ xml_data = request.data try: root = ET.fromstring(xml_data) data = {child.tag: child.text for child in root} 核心参数提取 return_code = data.get('return_code') out_trade_no = data.get('out_trade_no') 商户订单号 transaction_id = data.get('transaction_id') 微信流水号 if return_code == 'SUCCESS': conn = get_db_conn() try: with conn.cursor() as cursor: 查询订单是否存在 cursor.execute("SELECT id, user_id, status FROM orders WHERE order_no = %s", (out_trade_no,)) order = cursor.fetchone() if not order: return '' 幂等性检查：如果订单已经是支付状态，直接返回成功，防止重复处理 if order['status'] == 1: return '' 更新订单状态 update_sql = """ UPDATE orders SET status = 1, transaction_id = %s, paid_at = NOW() WHERE id = %s """ cursor.execute(update_sql, (transaction_id, order['id'])) conn.commit() 触发复购检查逻辑（异步或同步执行，此处为同步演示） check_repurchase_and_issue_coupon(order['user_id'], out_trade_no) finally: conn.close() 响应微信 return '' else: return '' except Exception as e: print(f"Callback Error: {e}") return '' ```

优惠券发放与消息推送

为了提升用户体验，优惠券发放后需要立即触达用户。我们使用微信的“模板消息”接口发送通知。这需要获取 Access Token，这是一个基础且必须的步骤。

在 `app.py` 中补充以下函数：

```python def get_wechat_access_token(): """ 获取微信 Access Token，带有本地缓存机制 """ cache_key = "wechat:access_token" token = redis_client.get(cache_key) if token: return token.decode('utf-8') url = f"https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid={Config.WECHAT_APPID}&secret={Config.WECHAT_APP_SECRET}" response = requests.get(url) data = response.json() if 'access_token' in data: token = data['access_token'] expires_in = data['expires_in'] - 300 提前5分钟过期 redis_client.setex(cache_key, expires_in, token) return token else: raise Exception(f"Get Token Failed: {data}") def send_coupon_notification(openid, coupon_code, amount): """ 发送优惠券到账模板消息 """ try: token = get_wechat_access_token() url = f"https://api.weixin.qq.com/cgi-bin/message/template/send?access_token={token}" 构造模板消息数据 注意：keyword1, keyword2 等需根据你申请的模板具体变量名进行修改 data = { "touser": openid, "template_id": Config.TEMPLATE_ID_COUPON, "page": "pages/coupon/coupon", 用户点击跳转的小程序页面 "data": { "thing1": { 假设 thing1 是优惠券名称 "value": "复购回馈满减券" }, "amount2": { 假设 amount2 是金额 "value": f"{amount / 10:.1f}" 假设单位是元，展示格式 }, "time3": { 假设 time3 是到期时间 "value": "7天内有效" }, "character_string4": { 假设 character_string4 是券码 "value": coupon_code } } } resp = requests.post(url, json=data) print(f"Message sent to {openid}, result: {resp.json()}") except Exception as e: print(f"Send notification failed: {e}") ```

本地调试与线上部署

代码编写完成后，需要进行测试。由于微信支付回调必须要求公网可访问的 URL，我们在本地开发时可以使用 Ngrok 进行内网穿透。

1. 启动 Flask 应用：

```bash export FLASK_APP=app.py export FLASK_ENV=development flask run --port=5000 ```

2. 下载并解压 Ngrok（根据你的操作系统），在终端执行：

```bash ./ngrok http 5000 ```

3. Ngrok 会生成一个临时的公网地址（例如 `https://abc123.ngrok.io`）。将这个地址加上你的回调路由，填入微信商户平台的“支付结果通知URL”配置中，例如：`https://abc123.ngrok.io/api/pay/callback`。

4. 模拟测试：你可以使用微信支付沙箱环境，或者直接在数据库中手动插入一条“待支付”的订单，然后使用 Postman 模拟微信的 XML 回调报文发送到你的 Ngrok 地址。

以下是用于 Postman 测试的 XML 报文模板（请将 `out_trade_no` 替换为你数据库中存在的订单号）：

```xml ```

发送请求后，观察你的终端控制台日志。如果看到“检测到复购，准备发放优惠券...”字样，并且数据库 `coupons` 表中新增了记录，同时收到了微信模板消息，说明整个链路已经跑通。上线时，只需将代码部署至云服务器（如腾讯云、阿里云），并将 Nginx 配置为反向代理至 Flask 端口，并配置好 SSL 证书即可正式投入使用。