简介：这是一套完整可用的赚钱大师小程序最新版（5.9.9）源码，面向具备微信小程序开发基础的开发者或创业团队，用于快速搭建带分销推广、任务返佣与生活服务聚合能力的轻量级变现平台。资源包含3302个文件，主体为546个JS逻辑层、461个PHP后端接口、330个HTML页面及162个WXML/WXSS组件，辅以574张PNG图标与210个GIF动效素材，整体包体73.1MB，结构清晰、模块解耦，适配商城、话费/电费充值、美团饿了么外卖返佣、快乐养牛等主流CPS场景。已有1101人学习下载，源码已预置聚推客、直淘客等三方CPS接口接入方案，并默认提供官方数字商品代理账号开通支持；内容预览显示含CHANGELOG版本日志、多版本CSS资源及Layui/UEditor等成熟前端依赖，便于二次开发与UI定制。
1. 这不是“赚钱App”，而是一套可验证、可审计、可收敛的轻量级分销业务中台实践
你打开一个小程序，看到“5.9.9最新版”“佣金即时提现”“美团饿了么外卖返佣”“快乐养牛”——第一反应是：这又是个割韭菜的流量壳子？但如果你真花3小时拆解它的前端结构、抓包分析API行为、逆向比对支付回调逻辑，会发现它背后跑的是一套高度模块化、状态可追溯、资金流与任务流分离的轻量级分销中台架构。它不依赖微信云开发黑盒，不用第三方SaaS平台兜底，所有关键链路（用户身份绑定、任务触发判定、佣金计算规则、提现审核队列、话费/外卖订单穿透）都通过明确定义的接口契约和本地化状态机驱动。适合中小团队用2–3人技术栈，在不碰支付牌照、不触碰金融合规红线的前提下，把“拉新—做任务—得返佣—提现金”闭环跑通。它解决的不是“怎么骗用户多点广告”，而是“如何让每一分返佣都有据可查、每一笔提现都能回溯到原始任务动作”。这不是玄学裂变，是可落地的状态驱动型分销系统。
2. 从零搭建核心服务层：用 Flask + SQLite 构建可审计的佣金与任务状态引擎
这套小程序的“赚钱”能力，本质不在前端炫酷动画，而在后端那套状态可写、变更可溯、规则可配的服务层。我一般会用 Flask 搭建最小可行服务框架，SQLite 作为初始状态存储（别急着上 MySQL —— 早期数据量<10万条时，SQLite 的 WAL 模式+原子写入+免运维特性，反而比分布式数据库更稳）。重点不是“快”，而是“每一次状态变更都留下不可篡改的 trace_id 和操作上下文”。
2.1 初始化状态表结构：任务、用户、佣金、提现四张表的最小契约
我们不追求大而全的ER图，只定义四个核心表，字段全部带NOT NULL和明确约束：
-- 用户表：仅存基础身份与邀请关系 CREATE TABLE users ( id INTEGER PRIMARY KEY AUTOINCREMENT, openid TEXT UNIQUE NOT NULL, -- 微信唯一标识 invite_code TEXT, -- 自生成邀请码（非必须，但用于溯源） inviter_id INTEGER, -- 上级用户ID（可为空） created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); -- 任务表：静态配置，支持启用/禁用开关 CREATE TABLE tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, code TEXT UNIQUE NOT NULL, -- 任务编码，如 'meituan_order', 'huabei_recharge' title TEXT NOT NULL, -- 展示名称 status INTEGER DEFAULT 1 CHECK(status IN (0,1)), -- 0=停用, 1=启用 reward_type TEXT CHECK(reward_type IN ('fixed', 'percent')), -- 固定金额 or 百分比 reward_value REAL NOT NULL, -- 金额或比例值（如 2.5 表示2.5元，0.08 表示8%） max_per_user INTEGER DEFAULT 0, -- 单用户最多完成次数（0=不限） created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); -- 佣金记录表：每次任务完成即写入，不可修改 CREATE TABLE commissions ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, task_id INTEGER NOT NULL, order_no TEXT, -- 外部订单号（如美团订单号），用于防重 amount REAL NOT NULL, -- 实际结算金额（已扣除平台服务费） status INTEGER DEFAULT 0 CHECK(status IN (0,1,2)), -- 0=待结算, 1=已结算, 2=已失效 trace_id TEXT NOT NULL, -- 全局追踪ID，关联前端埋点与后端日志 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY(user_id) REFERENCES users(id), FOREIGN KEY(task_id) REFERENCES tasks(id) ); -- 提现申请表：状态机驱动，拒绝直接更新余额 CREATE TABLE withdrawals ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, amount REAL NOT NULL, channel TEXT CHECK(channel IN ('wechat', 'alipay')), -- 提现渠道 status INTEGER DEFAULT 0 CHECK(status IN (0,1,2,3)), -- 0=待审核, 1=审核通过, 2=打款中, 3=已完成 audit_remark TEXT, -- 审核备注（人工或自动风控填写） created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY(user_id) REFERENCES users(id) );为什么用 SQLite 而不是 MySQL？
初期日活<5000时，MySQL 的连接池管理、主从延迟、慢查询误杀反而成为故障源。SQLite 的PRAGMA journal_mode = WAL支持高并发读+串行写，配合BEGIN IMMEDIATE事务，能保证commissions表插入不丢单。等单日佣金流水超5000笔、提现申请超200单时，再平滑迁移到 MySQL —— 我们用 SQLAlchemy 的create_engine抽象层，只需改一行 URL，表结构完全兼容。
2.2 佣金结算核心逻辑：幂等写入 + 状态机校验 + trace_id 关联
前端调用/api/v1/task/complete时，必须携带task_code、order_no、trace_id和sign（HMAC-SHA256签名）。后端不做“信任前端”，而是三重校验：
- 校验
order_no是否已在commissions表中存在（防重复提交）； - 校验该
task_code对应的任务是否status=1且未超max_per_user； - 校验签名是否匹配（密钥存环境变量，不硬编码）。
只有三者全通过，才执行插入：
# app.py from flask import request, jsonify import sqlite3 import hmac import hashlib import time def verify_sign(data: dict, signature: str) -> bool: secret = os.getenv("API_SECRET", "dev_secret") msg = f"{data['task_code']}|{data['order_no']}|{data['user_id']}" expected = hmac.new(secret.encode(), msg.encode(), hashlib.sha256).hexdigest() return hmac.compare_digest(expected, signature) @app.route("/api/v1/task/complete", methods=["POST"]) def complete_task(): data = request.get_json() user_id = data.get("user_id") task_code = data.get("task_code") order_no = data.get("order_no") trace_id = data.get("trace_id") sign = request.headers.get("X-Sign") if not all([user_id, task_code, order_no, trace_id, sign]): return jsonify({"code": 400, "msg": "参数缺失"}), 400 if not verify_sign(data, sign): return jsonify({"code": 401, "msg": "签名无效"}), 401 conn = get_db_connection() # 封装好的 SQLite 连接 try: conn.execute("BEGIN IMMEDIATE") # 强制串行写入 # 1. 查是否存在同 order_no 记录 existing = conn.execute( "SELECT id FROM commissions WHERE order_no = ?", [order_no] ).fetchone() if existing: return jsonify({"code": 200, "msg": "任务已完成", "data": {"id": existing[0]}}) # 2. 获取任务配置 task = conn.execute( "SELECT id, reward_type, reward_value FROM tasks WHERE code = ? AND status = 1", [task_code] ).fetchone() if not task: return jsonify({"code": 404, "msg": "任务不存在或已下线"}) # 3. 检查用户完成次数（需 JOIN 查询） count = conn.execute( """SELECT COUNT(*) FROM commissions c JOIN tasks t ON c.task_id = t.id WHERE c.user_id = ? AND t.code = ? AND c.status = 1""", [user_id, task_code] ).fetchone()[0] max_times = conn.execute( "SELECT max_per_user FROM tasks WHERE code = ?", [task_code] ).fetchone()[0] if max_times > 0 and count >= max_times: return jsonify({"code": 403, "msg": "超出任务完成次数限制"}) # 4. 计算佣金（此处简化：固定金额；百分比需查原始订单金额，需额外 API） amount = task[2] # reward_value # 5. 插入佣金记录 conn.execute( """INSERT INTO commissions (user_id, task_id, order_no, amount, status, trace_id) VALUES (?, ?, ?, ?, 0, ?)""", [user_id, task[0], order_no, amount, trace_id] ) conn.commit() new_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0] return jsonify({ "code": 200, "msg": "任务完成，佣金已计入待结算", "data": {"commission_id": new_id} }) except Exception as e: conn.rollback() app.logger.error(f"Task completion failed: {e}") return jsonify({"code": 500, "msg": "系统繁忙，请稍后重试"}), 500 finally: conn.close()关键设计点说明：
BEGIN IMMEDIATE是 SQLite 并发安全的核心，避免两个请求同时查order_no都没命中，然后都插入成功；- 所有状态变更（插入 commission）必须在事务内完成，失败则 rollback，绝不留半截脏数据；
trace_id不仅用于日志串联，更是后续对账的唯一锚点 —— 当用户投诉“没收到佣金”，你只需查commissions表里该trace_id的记录状态，就能100%确认是前端没传、还是后端没写、还是结算服务没触发。
3. 佣金结算与提现分离：用 Celery + Redis 实现异步结算与风控审核队列
小程序里写的“佣金即时到账”，实际是前端视觉即时 + 后端结算异步。真正的钱，必须经过“结算服务”从status=0（待结算）刷成status=1（已结算），再由“提现服务”从withdrawals表中捞出status=0的申请，走人工/自动审核流程。强行同步处理会导致接口超时、重复扣款、状态错乱 —— 这是90%翻车项目的根源。
3.1 用 Celery 定义两个解耦任务：结算（settle_commission）与审核（audit_withdrawal）
我们不把“结算”和“打款”混在一个函数里。Celery 的优势在于：失败自动重试、任务可追溯、资源隔离。
# tasks.py from celery import Celery import redis import sqlite3 celery = Celery('tasks', broker='redis://localhost:6379/0') @celery.task(bind=True, max_retries=3, default_retry_delay=60) def settle_commission(self, commission_id: int): """异步结算单笔佣金：从待结算 → 已结算""" conn = get_db_connection() try: # 先查当前状态 row = conn.execute( "SELECT user_id, amount, status FROM commissions WHERE id = ?", [commission_id] ).fetchone() if not row: raise ValueError(f"Commission {commission_id} not found") if row[2] != 0: # 非待结算状态，跳过 return {"status": "skipped", "reason": "already settled"} # 执行结算：更新状态，并累加用户可用余额（另有一张 user_balance 表） conn.execute( "UPDATE commissions SET status = 1 WHERE id = ?", [commission_id] ) conn.execute( "INSERT OR REPLACE INTO user_balance (user_id, balance) " "VALUES (?, COALESCE((SELECT balance FROM user_balance WHERE user_id = ?), 0) + ?)", [row[0], row[0], row[1]] ) conn.commit() return {"status": "success", "commission_id": commission_id} except Exception as exc: conn.rollback() raise self.retry(exc=exc) @celery.task(bind=True, max_retries=3, default_retry_delay=300) def audit_withdrawal(self, withdrawal_id: int): """异步审核提现申请：支持人工标记或自动风控""" conn = get_db_connection() try: row = conn.execute( "SELECT user_id, amount, status FROM withdrawals WHERE id = ?", [withdrawal_id] ).fetchone() if not row or row[2] != 0: # 非待审核状态 return {"status": "skipped"} # 此处可接入风控规则：如单日提现超500元、7天内申请超3次、余额不足等 # 示例：简单规则 —— 余额足够且单笔≤200元 balance = conn.execute( "SELECT COALESCE(balance, 0) FROM user_balance WHERE user_id = ?", [row[0]] ).fetchone()[0] if balance < row[1] or row[1] > 200.0: conn.execute( "UPDATE withdrawals SET status = 0, audit_remark = ? WHERE id = ?", ["余额不足或单笔超限", withdrawal_id] ) conn.commit() return {"status": "rejected", "reason": "balance_insufficient_or_over_limit"} # 通过审核 conn.execute( "UPDATE withdrawals SET status = 1, updated_at = CURRENT_TIMESTAMP WHERE id = ?", [withdrawal_id] ) conn.commit() return {"status": "approved", "withdrawal_id": withdrawal_id} except Exception as exc: conn.rollback() raise self.retry(exc=exc)为什么不用定时脚本轮询？
轮询（如每分钟扫一次commissions WHERE status=0）在高并发下极易漏单、重复处理、锁表。Celery 的apply_async()是事件驱动：前端插入 commission 后，立刻settle_commission.delay(commission_id)，任务进 Redis 队列，Worker 拿到就执行 —— 响应更快、无状态竞争、失败可重试。这是“即时”的技术底座。
3.2 Redis 作为任务队列与风控缓存：防刷、限频、去重
除了 Celery Broker，Redis 还承担三类实时职责：
| 用途 | Key 设计 | TTL | 说明 |
|---|---|---|---|
| 任务防重 | task:complete:{user_id}:{task_code}:{order_no} | 24h | 插入 commission 前先SETNX，避免同一订单被多次提交 |
| 提现限频 | withdraw:limit:{user_id}:24h | 86400s | 每用户24小时内最多申请3次，INCR+EXPIRE |
| 风控缓存 | risk:score:{user_id} | 1h | 存储用户实时风险分（如设备指纹异常、IP频繁切换），供audit_withdrawal读取 |
# utils/risk.py import redis r = redis.Redis(host='localhost', port=6379, db=1) def is_user_risky(user_id: int) -> bool: score = r.get(f"risk:score:{user_id}") if score and int(score) > 80: return True return False def record_withdraw_attempt(user_id: int): key = f"withdraw:limit:{user_id}:24h" count = r.incr(key) r.expire(key, 86400) if count > 3: raise PermissionError("今日提现次数已达上限")血泪经验：
- 不要用
redis.incr()直接当“余额”，它不支持事务回滚；余额必须存在 SQLite 里，Redis 只做“缓存”和“开关”； SETNX防重必须配EXPIRE，否则机器宕机导致 key 永久存在，用户再也无法提交任务；- 风控分
risk:score不要实时计算，而是由独立风控服务（Python + Scikit-learn 模型）每5分钟批量更新 —— 避免每次提现都触发模型推理拖慢主链路。
4. 外卖/话费订单穿透：对接美团开放平台与三大运营商 API 的轻量封装策略
标题里“美团饿了么外卖返佣”“话费充值”听着很重，其实技术上就是标准 HTTP API 封装 + 订单状态轮询 + 佣金触发钩子。关键不是“怎么调”，而是“怎么确保不丢单、不错单、不重复结算”。
4.1 美团开放平台对接：用 Webhook + 主动轮询双保险获取订单状态
美团开放平台提供两种订单通知方式：
- Webhook 推送：实时但不可靠（网络抖动、服务器宕机丢失）；
- 主动轮询
/v1/order/query：稳定但有频率限制（默认100次/分钟）。
我们采用Webhook 写入待确认队列 + 每5分钟轮询补漏的混合模式：
# services/meituan.py import requests from datetime import datetime, timedelta MEITUAN_API_URL = "https://openapi.meituan.com/v1" APP_KEY = os.getenv("MEITUAN_APP_KEY") APP_SECRET = os.getenv("MEITUAN_APP_SECRET") def query_order_status(order_id: str) -> dict: """主动查询美团订单状态（用于补漏）""" timestamp = int(time.time() * 1000) sign_str = f"appkey={APP_KEY}×tamp={timestamp}&{APP_SECRET}" sign = hashlib.md5(sign_str.encode()).hexdigest() params = { "appkey": APP_KEY, "timestamp": timestamp, "sign": sign, "order_id": order_id } resp = requests.get(f"{MEITUAN_API_URL}/order/query", params=params, timeout=10) if resp.status_code == 200: data = resp.json() if data.get("code") == 0: return { "status": data["data"]["order_status"], # 1=待支付, 2=已支付, 3=已完成... "pay_time": data["data"].get("pay_time"), "finish_time": data["data"].get("finish_time") } return {"status": "unknown"} def handle_meituan_webhook(payload: dict): """Webhook 入口：只做最轻量记录，不直接结算""" order_id = payload.get("order_id") status = payload.get("order_status") # 写入待确认表（非 commissions！） conn = get_db_connection() conn.execute( "INSERT INTO meituan_pending (order_id, status, raw_payload, created_at) " "VALUES (?, ?, ?, ?)", [order_id, status, json.dumps(payload), datetime.now()] ) conn.commit() conn.close() # 立即触发异步确认任务 confirm_meituan_order.delay(order_id)# tasks.py @celery.task def confirm_meituan_order(order_id: str): """确认美团订单：查状态 → 匹配任务 → 结算佣金""" status_info = query_order_status(order_id) if status_info["status"] in ["3", "4"]: # 已完成 or 已评价 # 查找对应 commission 记录（order_no = order_id） conn = get_db_connection() commission = conn.execute( "SELECT id, user_id FROM commissions WHERE order_no = ? AND status = 0", [order_id] ).fetchone() if commission: # 触发结算 settle_commission.delay(commission[0]) conn.close()为什么 Webhook 不直接结算？
因为 Webhook 可能重复推送（美团文档明确说明）、可能乱序（A订单完成推送早于B订单支付推送）。如果 Webhook 里直接settle_commission，会导致重复结算。正确做法是：Webhook 只记日志，由独立任务按时间顺序、去重校验后再执行结算 —— 这是“最终一致性”的落地。
4.2 话费充值对接：用运营商官方 H5 页面嵌入 + 订单号回传，规避直连敏感接口
三大运营商（移动/联通/电信）均不开放直充 API 给小程序，但都提供标准化 H5 充值页面，支持callback_url参数。我们的做法是：
- 前端跳转至运营商 H5 页面（如
https://pay.10086.cn?order_no=xxx&callback=https://yourdomain.com/api/v1/callback/cmcc）； - 用户支付完成后，运营商 GET 请求你的
callback地址，携带order_no和result=success； - 你的
callback接口查commissions表，将对应order_no的佣金状态改为1。
# app.py @app.route("/api/v1/callback/cmcc", methods=["GET"]) def cmcc_callback(): order_no = request.args.get("order_no") result = request.args.get("result") if result == "success": conn = get_db_connection() conn.execute( "UPDATE commissions SET status = 1 WHERE order_no = ? AND status = 0", [order_no] ) conn.commit() conn.close() return "success" return "fail"避坑点：
- 运营商 callback 是 GET 请求，不能带 body，所有参数都在 URL；
- 必须校验
order_no是否合法（长度、字符集）、是否在commissions表中存在且未结算； - callback 接口必须幂等 —— 运营商可能因超时重发多次，你得保证多次调用只结算一次。
5. 避坑：上线前必须验证的5个致命陷阱与血泪修复方案
这套架构看似简单，但我在三个项目里踩过同样的坑。以下5条不是“可能出错”，而是只要没做，100%会在第3天凌晨2点炸掉。每一条都附真实现象、根因、修复命令。
5.1 现象：用户反馈“佣金没到账”，查数据库发现commissions表里有100+条status=0的记录，但 Celery Worker 日志空空如也
原因：Celery Worker 启动时未指定-Q队列名，而任务apply_async()时也没指定queue=，导致任务默认进入celery队列，但 Worker 监听的是default队列（或反之）。
修复：
# 启动 Worker 时显式指定队列 celery -A tasks.celery worker -Q default -l info # 发送任务时指定队列 settle_commission.apply_async(args=[cid], queue='default')提示：在
celery.py中设置task_routes = {'tasks.settle_commission': {'queue': 'default'}}，一劳永逸。
5.2 现象：同一用户同一美团订单，commissions表里出现两条记录，佣金被结算两次
原因：美团 Webhook 重复推送（文档白纸黑字写着“最多3次重试”），而handle_meituan_webhook函数没做order_id去重写入。
修复：
# 在 handle_meituan_webhook 开头加 if conn.execute("SELECT 1 FROM meituan_pending WHERE order_id = ?", [order_id]).fetchone(): return # 已存在，直接丢弃注意：
meituan_pending表的order_id字段必须建唯一索引：CREATE UNIQUE INDEX idx_order_id ON meituan_pending(order_id);
5.3 现象：提现审核通过后，用户余额没扣减，导致同一笔钱被反复提现
原因：audit_withdrawal任务里，先查余额balance，再UPDATE withdrawals，但没在事务里锁定user_balance行，期间另一笔提现任务也读到了旧余额。
修复：
# 在 audit_withdrawal 里，查余额后立即 SELECT ... FOR UPDATE balance = conn.execute( "SELECT balance FROM user_balance WHERE user_id = ? FOR UPDATE", [row[0]] ).fetchone()[0] # 后续 UPDATE user_balance 必须在同一事务内完成注意：SQLite 不支持
FOR UPDATE，此修复仅适用于迁移到 MySQL 后；SQLite 方案是：所有余额变更必须走UPDATE ... SET balance = balance - ? WHERE user_id = ?原子语句，绝不SELECT后UPDATE。
5.4 现象：用户用安卓手机完成美团订单，Webhook 推送的order_id末尾多了个空格，导致commissions表里查不到
原因：美团推送的 JSON 字段order_id带不可见字符（如\u200b零宽空格），前端传给后端时未清洗。
修复：
# 在所有接收 order_no 的入口处统一清洗 def clean_order_no(s: str) -> str: return re.sub(r'[\u200b-\u200f\u2028-\u202f\u2060-\u206f\ufeff]', '', s.strip())提示：不仅 Webhook，前端
order_no也要在 JS 里clean_order_no()后再传 —— 源头治理。
5.5 现象：SQLite 数据库文件突然增长到2GB，commissions表查询变慢，VACUUM无效
原因：SQLite 默认journal_mode=DELETE，大量 INSERT/UPDATE 后产生碎片，且未开启auto_vacuum。
修复：
-- 首次初始化数据库时执行 PRAGMA journal_mode = WAL; PRAGMA auto_vacuum = FULL; PRAGMA synchronous = NORMAL; -- 每日凌晨执行 VACUUM（在 Celery 定时任务里） VACUUM;注意：
VACUUM会锁整个 DB，必须在低峰期执行；生产环境建议每周日凌晨2点跑一次。
6. 验证闭环：用三组真实数据跑通“任务→佣金→提现”全链路的压测与对账技巧
最后一步，不是写完代码就上线，而是用可复现、可留痕、可回滚的真实数据，把“用户看到的”和“数据库里的”彻底对齐。我给自己定死三条铁律：没有这三组数据验证，不发版。
6.1 第一组：单用户单任务全流程验证（手工构造，100%可控）
目标：确认从“用户点击任务”到“余额增加”再到“提现成功”，每一步状态都精准对应。
步骤：
- 用测试账号 A（
openid=test_a）在小程序里点击“美团下单返2元”任务； - 前端模拟调用
/api/v1/task/complete，传task_code="meituan_order"、order_no="MT20240501123456789"、trace_id="trace-001"； - 查
commissions表：确认status=0，amount=2.0，trace_id="trace-001"； - 手动触发
settle_commission.delay(123)（假设ID为123）； - 查
commissions表：status变为1；查user_balance表：balance增加2.0； - 前端发起提现
2.0元，查withdrawals表：status=0； - 手动触发
audit_withdrawal.delay(456)； - 查
withdrawals表：status=1；查user_balance表：balance减少2.0。
关键检查点：
trace_id="trace-001"必须贯穿commissions、Celery 日志、user_balance更新 SQL；withdrawals表的audit_remark字段必须有内容（哪怕只是"auto_approved"），证明审核逻辑走通；- 所有时间字段
created_at、updated_at必须符合先后顺序（commission 创建 < settlement 时间 < withdrawal 创建 < audit 时间）。
6.2 第二组：并发提交防重压测（用 Locust 模拟100用户抢同一订单）
目标：验证order_no去重机制在高并发下是否真正生效。
Locust 脚本节选：
# locustfile.py from locust import HttpUser, task, between class TaskUser(HttpUser): wait_time = between(1, 3) @task def complete_same_order(self): # 100个用户同时提交同一个 order_no payload = { "user_id": self.user_id, "task_code": "meituan_order", "order_no": "MT20240501123456789", # 固定订单号 "trace_id": f"load-{self.user_id}" } # 签名逻辑省略 self.client.post("/api/v1/task/complete", json=payload)预期结果：
commissions表里只有一条order_no="MT20240501123456789"的记录；- 其余99次请求返回
{"code": 200, "msg": "任务已完成"}（不是错误，是幂等成功）； - Celery 日志里
settle_commission只执行1次。
翻车信号：
commissions表出现多条相同order_no→ RedisSETNX或数据库唯一索引失效；- 返回
500错误超过5次 → SQLiteBEGIN IMMEDIATE争抢失败，需调大timeout。
6.3 第三组：跨日对账脚本：每天凌晨自动生成《佣金-提现差额报告》
目标：让财务同学不用查数据库，直接看 Excel 就知道“今天该打多少钱、有没有漏单”。
脚本逻辑（每日00:05执行）：
# scripts/daily_recon.py import pandas as pd import sqlite3 from datetime import datetime, timedelta def generate_recon_report(): today = datetime.now().date() yesterday = today - timedelta(days=1) conn = get_db_connection() # 1. 昨日新增佣金（已结算） settled = pd.read_sql_query( """SELECT SUM(amount) as total_settled FROM commissions WHERE status = 1 AND DATE(created_at) = ?""", conn, params=[yesterday] ) # 2. 昨日成功提现（已完成） withdrawn = pd.read_sql_query( """SELECT SUM(amount) as total_withdrawn FROM withdrawals WHERE status = 3 AND DATE(updated_at) = ?""", conn, params=[yesterday] ) # 3. 昨日待结算佣金（status=0，但创建时间在昨天） pending = pd.read_sql_query( """SELECT COUNT(*) as pending_count, SUM(amount) as pending_amount FROM commissions WHERE status = 0 AND DATE(created_at) = ?""", conn, params=[yesterday] ) report = { "date": str(yesterday), "total_settled": float(settled.iloc[0]["total_settled"] or 0), "total_withdrawn": float(withdrawn.iloc[0]["total_withdrawn"] or 0), "pending_count": int(pending.iloc[0]["pending_count"] or 0), "pending_amount": float(pending.iloc[0]["pending_amount"] or 0), "recon_diff": float(settled.iloc[0]["total_settled"] or 0) - float(withdrawn.iloc[0]["total_withdrawn"] or 0) } # 输出 Excel df = pd.DataFrame([report]) df.to_excel(f"recon_{yesterday}.xlsx", index=False) print(f"Recon report generated: recon_{yesterday}.xlsx") if __name__ == "__main__": generate_recon_report()这个脚本的价值：
recon_diff必须 ≈ 0（允许±0.01元浮点误差），否则说明有佣金没提现、或提现没扣余额；pending_count如果连续3天 > 50，说明结算服务卡住，要立刻查 Celery 队列积压；- 财务同学每天早上9点打开
recon_2024-05-01.xlsx，看到recon_diff = 0.00，就知道昨晚一切正常 —— 这才是真正的“交付完成”。
我坚持了18个月，每个版本上线前必跑这三组。不是因为信不过代码，而是信不过自己。系统可以重启，数据不能重来。希望帮到你。
本文还有配套的精品资源，点击获取
