Python实战：构建好友裂变福利自动化分发系统

一、技术选型与环境依赖搭建

本方案采用Python作为核心开发语言，结合Flask轻量级框架提供API接口，MySQL存储用户关系数据，Redis处理高并发下的缓存与锁机制。请确保你的服务器已安装Python 3.8及以上版本。

在项目根目录下执行以下命令安装核心依赖库：

pip install flask pymysql redis cryptography

这里引入cryptography库是为了生成更安全的邀请码，避免被轻易遍历。

二、MySQL数据库表结构设计

我们需要设计两张核心表：用户表（users）和邀请记录表（invite_logs）。用户表存储基础信息，邀请记录表用于审计和防止重复领奖。

请在MySQL客户端执行以下SQL语句完成建表：

```sql CREATE DATABASE fission_system DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; USE fission_system; CREATE TABLE users ( id INT AUTO_INCREMENT PRIMARY KEY COMMENT '用户ID', openid VARCHAR(64) NOT NULL UNIQUE COMMENT '用户唯一标识，如微信openid', invite_code VARCHAR(16) NOT NULL UNIQUE COMMENT '用户的专属邀请码', inviter_id INT DEFAULT 0 COMMENT '邀请人ID，0表示无邀请人', reward_status TINYINT DEFAULT 0 COMMENT '福利领取状态：0未领取，1已领取', invite_count INT DEFAULT 0 COMMENT '累计邀请人数', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, INDEX idx_openid (openid), INDEX idx_invite_code (invite_code) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='用户信息表'; CREATE TABLE invite_logs ( id INT AUTO_INCREMENT PRIMARY KEY, inviter_id INT NOT NULL COMMENT '邀请人ID', invitee_id INT NOT NULL COMMENT '被邀请人ID', reward_given TINYINT DEFAULT 0 COMMENT '是否已发放奖励', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, INDEX idx_inviter (inviter_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='邀请日志表'; ```

三、Redis缓存键值规范设计

为了减轻数据库压力并实现防抖，我们定义以下Redis Key规范：

• 用户信息缓存： user:info:{user_id}，存储用户基础JSON数据，过期时间1小时。
• 邀请码映射： map:code:user:{invite_code}，存储邀请码对应的User ID，永不过期。
• 分布式锁： lock:invite:{openid}，防止同一用户并发请求重复绑定，过期时间5秒。

四、后端核心逻辑代码实现

创建名为app.py的文件，将以下完整代码直接复制进去。代码包含了数据库连接池、邀请码生成算法、防并发锁以及事务处理逻辑。

```python import pymysql from flask import Flask, request, jsonify from redis import Redis import secrets import time app = Flask(__name__) 配置信息 DB_CONFIG = { 'host': '127.0.0.1', 'user': 'root', 'password': 'your_password', 请修改为实际密码 'db': 'fission_system', 'charset': 'utf8mb4', 'cursorclass': pymysql.cursors.DictCursor } 初始化Redis连接 redis_client = Redis(host='127.0.0.1', port=6379, db=0) def get_db_connection(): return pymysql.connect(DB_CONFIG) def generate_invite_code(user_id): """生成基于用户ID和随机盐的邀请码""" salt = secrets.token_hex(2) return f"{user_id:04x}{salt}".upper() @app.route('/api/user/register', methods=['POST']) def register(): """ 用户注册并绑定邀请关系接口 参数：openid (必填), invite_code (选填) """ data = request.json openid = data.get('openid') invite_code = data.get('invite_code') if not openid: return jsonify({'code': 400, 'msg': '缺少必要参数openid'}), 400 1. 获取分布式锁，防止并发注册 lock_key = f"lock:invite:{openid}" is_locked = redis_client.set(lock_key, '1', nx=True, ex=5) if not is_locked: return jsonify({'code': 429, 'msg': '操作过于频繁，请稍后重试'}), 429 try: conn = get_db_connection() with conn.cursor() as cursor: 2. 检查用户是否已存在 cursor.execute("SELECT id, inviter_id, reward_status FROM users WHERE openid = %s", (openid,)) user = cursor.fetchone() if user: 用户已存在，不做处理或返回信息 return jsonify({'code': 200, 'msg': '注册成功', 'data': {'user_id': user['id'], 'is_new': False}}) 3. 新用户逻辑：验证邀请码 inviter_id = 0 if invite_code: 优先查Redis缓存 cache_key = f"map:code:user:{invite_code}" cached_inviter_id = redis_client.get(cache_key) if cached_inviter_id: inviter_id = int(cached_inviter_id) else: Redis未命中，查数据库 cursor.execute("SELECT id FROM users WHERE invite_code = %s", (invite_code,)) inviter = cursor.fetchone() if inviter: inviter_id = inviter['id'] 回写Redis redis_client.set(cache_key, str(inviter_id)) 4. 开启事务插入数据 try: 生成新用户ID（这里简化处理，实际依赖DB自增） cursor.execute("INSERT INTO users (openid, invite_code, inviter_id) VALUES (%s, %s, %s)", (openid, '', 0)) 先插入占位，获取ID new_user_id = cursor.lastrowid real_invite_code = generate_invite_code(new_user_id) 更新自己的邀请码 cursor.execute("UPDATE users SET invite_code = %s, inviter_id = %s WHERE id = %s", (real_invite_code, inviter_id, new_user_id)) 缓存邀请码映射 redis_client.set(f"map:code:user:{real_invite_code}", str(new_user_id)) 5. 处理裂变福利（如果存在邀请人） if inviter_id > 0: 插入邀请日志 cursor.execute("INSERT INTO invite_logs (inviter_id, invitee_id, reward_given) VALUES (%s, %s, 0)", (inviter_id, new_user_id)) 更新邀请人计数 cursor.execute("UPDATE users SET invite_count = invite_count + 1 WHERE id = %s", (inviter_id,)) 福利发放核心逻辑 假设规则：每邀请3人发放一次福利 cursor.execute("SELECT invite_count FROM users WHERE id = %s FOR UPDATE", (inviter_id,)) inviter_data = cursor.fetchone() 重新获取锁后的最新计数（虽然上面update了，但为了逻辑严谨演示查询） 实际业务中，这里可以调用优惠券服务 if inviter_data['invite_count'] % 3 == 0: 标记奖励状态（此处简化，实际应写入另一张奖励表） cursor.execute("UPDATE users SET reward_status = 1 WHERE id = %s", (inviter_id,)) TODO: 此处插入发券代码 conn.commit() return jsonify({ 'code': 200, 'msg': '注册并绑定成功', 'data': { 'user_id': new_user_id, 'my_invite_code': real_invite_code, 'inviter_id': inviter_id } }) except Exception as e: conn.rollback() app.logger.error(f"Database Error: {str(e)}") return jsonify({'code': 500, 'msg': '系统内部错误'}), 500 finally: 释放锁 redis_client.delete(lock_key) if 'conn' in locals(): conn.close() if __name__ == '__main__': app.run(debug=True, port=5000) ```

五、接口测试与全流程验证

代码启动后，我们需要模拟真实场景进行测试。假设我们有两个用户：User A（老用户）和 User B（新用户）。

步骤1：注册邀请人User A

使用Postman或curl发送请求，User A没有邀请码。

```bash curl -X POST http://127.0.0.1:5000/api/user/register \ -H "Content-Type: application/json" \ -d '{"openid": "user_a_wx_001"}' ```

返回结果中会包含 my_invite_code，例如 0001A3B2。请记录此码。

步骤2：新用户User B通过邀请码注册

User B使用User A的邀请码进行注册。

```bash curl -X POST http://127.0.0.1:5000/api/user/register \ -H "Content-Type: application/json" \ -d '{"openid": "user_b_wx_002", "invite_code": "0001A3B2"}' ```

此时观察数据库 users 表，User B 的 inviter_id 应为 User A 的 ID，且 User A 的 invite_count 变为 1。

步骤3：验证防刷机制

使用相同的 openid 再次发送请求，系统应直接返回“注册成功”且 is_new 为 false，不会重复插入数据。

```bash curl -X POST http://127.0.0.1:5000/api/user/register \ -H "Content-Type: application/json" \ -d '{"openid": "user_b_wx_002", "invite_code": "0001A3B2"}' ```

六、生产环境部署注意事项

上述代码已具备核心逻辑，但在部署到生产环境前，必须注意以下细节：

• 连接池配置：代码中每次请求都创建新连接，高并发下会导致连接数耗尽。建议使用 DBUtils 或 SQLAlchemy 的连接池。
• 异步任务：发放福利（如调用第三方优惠券接口）属于耗时操作，切勿在主线程同步执行。应引入 Celery 或 RQ 将发券逻辑放入异步队列中处理。
• 密码安全：切勿将数据库密码硬编码在代码中，应使用环境变量 os.environ.get 读取。
• 日志监控：建议接入 Sentry 或 ELK 监控异常日志，特别是奖励发放失败的情况，以便后续补发。