"""一次性 mock:造几条不同状态的提现单,供运营后台「提现管理」页联调验收。 覆盖: - 全部 5 个状态 tab(待审核 / 打款中 / 已到账 / 已拒绝 / 打款失败) - 「累计提现」列(按该用户 success 单金额求和;有非零也有 0) - 空昵称(展示 "-")、提现实名、失败/拒绝原因 - 配套双分录现金流水(withdraw / withdraw_refund / exchange_in)+ 账户余额, 让顶部「账本校验」保持绿色、详情抽屉的现金余额/流水也真实。 约束:withdraw_order 有部分唯一索引(同一 user 在 reviewing/pending 最多 1 单), 本脚本每个 mock 用户至多 1 个活动单,满足约束。 幂等:每次运行先按固定 mock 手机号清掉旧 mock 再重建。仅清理用 --clean-only。 .venv\\Scripts\\python.exe scripts\\seed_mock_withdraws.py .venv\\Scripts\\python.exe scripts\\seed_mock_withdraws.py --clean-only """ from __future__ import annotations import argparse import sys import uuid from collections import defaultdict from datetime import datetime, timedelta, timezone from sqlalchemy import delete, select from app.core import rewards from app.db.session import SessionLocal from app.models.ad_feed_reward import AdFeedRewardRecord from app.models.ad_reward import AdRewardRecord from app.models.user import User from app.models.wallet import CashTransaction, CoinAccount, CoinTransaction, WithdrawOrder # Windows GBK 控制台下也能正常打印中文(避免 UnicodeEncodeError) if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") # 固定 mock 手机号:脚本只动这些号,便于幂等重建 / 清理。 MOCK_PHONES = [ "13800000001", "13900000002", "13700000003", "13600000004", "13500000005", "13312345678", ] REMARK = {"exchange_in": "金币兑入", "withdraw": "提现扣款", "withdraw_refund": "提现退款"} def _naive_utc_now() -> datetime: """与 func.now() 在 SQLite 的口径一致:naive UTC。前端按 UTC 解析再转北京时间。""" return datetime.now(timezone.utc).replace(tzinfo=None) def _mock_transfer_no() -> str: """模拟微信转账单号(长串数字)。""" return "1330" + str(uuid.uuid4().int)[:26] def clean(db) -> int: uids = list(db.execute(select(User.id).where(User.phone.in_(MOCK_PHONES))).scalars()) if not uids: return 0 for model in ( WithdrawOrder, CashTransaction, CoinTransaction, CoinAccount, AdRewardRecord, AdFeedRewardRecord, ): db.execute(delete(model).where(model.user_id.in_(uids))) db.execute(delete(User).where(User.id.in_(uids))) db.commit() return len(uids) def seed(db) -> list[WithdrawOrder]: now = _naive_utc_now() def ago(**kw) -> datetime: return now - timedelta(**kw) # 1) 建用户(手机号 / 昵称;U4 昵称留空测 "-" 展示) users_spec = [ ("13800000001", "省钱小能手"), ("13900000002", "薅羊毛达人"), ("13700000003", "比价狂魔"), ("13600000004", None), ("13500000005", "退款体验官"), ("13312345678", "已到账老用户"), ] users: dict[str, User] = {} for phone, nickname in users_spec: u = User( phone=phone, nickname=nickname, register_channel="sms", status="active", created_at=ago(days=12), last_login_at=ago(hours=2), ) db.add(u) users[phone] = u db.flush() # 拿 user.id # 2) 建提现单(覆盖各状态) def order( phone: str, cents: int, status: str, *, name: str | None = None, wechat: str | None = None, transfer: str | None = None, fail: str | None = None, created: datetime, updated: datetime | None = None, ) -> WithdrawOrder: return WithdrawOrder( user_id=users[phone].id, out_bill_no=uuid.uuid4().hex, # 与线上同格式:uuid4 的 32 位 hex amount_cents=cents, user_name=name, status=status, wechat_state=wechat, transfer_bill_no=transfer, fail_reason=fail, created_at=created, updated_at=updated or created, ) orders = [ # U1 省钱小能手:2 已到账 + 1 待审核 → 累计提现 ¥18.00 order("13800000001", 1000, "success", name="张伟", wechat="SUCCESS", transfer=_mock_transfer_no(), created=ago(days=3)), order("13800000001", 800, "success", name="张伟", wechat="SUCCESS", transfer=_mock_transfer_no(), created=ago(days=5)), order("13800000001", 500, "reviewing", name="张伟", created=ago(minutes=3)), # U2 薅羊毛达人:1 已到账 + 1 待审核 → 累计提现 ¥20.00 order("13900000002", 2000, "success", name="李娜", wechat="SUCCESS", transfer=_mock_transfer_no(), created=ago(days=2)), order("13900000002", 1250, "reviewing", name="李娜", created=ago(minutes=25)), # U3 比价狂魔:1 待审核 → 累计提现 ¥0 order("13700000003", 100, "reviewing", name="王芳", created=ago(hours=1)), # U4(空昵称):1 打款中 → 累计提现 ¥0 order("13600000004", 300, "pending", name="刘洋", wechat="ACCEPTED", transfer=_mock_transfer_no(), created=ago(minutes=40), updated=ago(minutes=20)), # U5 退款体验官:1 已拒绝 + 1 打款失败 → 累计提现 ¥0 order("13500000005", 5000, "rejected", name="陈静", fail="微信实名不匹配", created=ago(days=1)), order("13500000005", 3000, "failed", name="陈静", wechat="FAIL", transfer=_mock_transfer_no(), fail="收款账户异常，已退回余额", created=ago(hours=6), updated=ago(hours=5)), # U6 已到账老用户:2 已到账 → 累计提现 ¥40.00 order("13312345678", 1500, "success", name="赵强", wechat="SUCCESS", transfer=_mock_transfer_no(), created=ago(days=10)), order("13312345678", 2500, "success", name="赵强", wechat="SUCCESS", transfer=_mock_transfer_no(), created=ago(days=8)), ] db.add_all(orders) db.flush() # 3) 配套双分录现金流水 + 账户余额(让账本校验绿、详情抽屉真实): # 每单创建时扣款(withdraw, -amount, ref=out_bill_no);失败/拒绝再退款(withdraw_refund, +amount); # 每用户起始补一笔 exchange_in 把余额垫正;CoinAccount.cash 余额 = 该用户流水净额。 by_user: dict[int, list[WithdrawOrder]] = defaultdict(list) for o in orders: by_user[o.user_id].append(o) for uid, uorders in by_user.items(): initial = sum(o.amount_cents for o in uorders) + 2000 # 垫 ¥20 余额,确保全程非负 earliest = min(o.created_at for o in uorders) events: list[tuple[datetime, str, int, str | None]] = [ (earliest - timedelta(days=1), "exchange_in", initial, None) ] for o in uorders: events.append((o.created_at, "withdraw", -o.amount_cents, o.out_bill_no)) if o.status in ("failed", "rejected"): events.append((o.updated_at, "withdraw_refund", o.amount_cents, o.out_bill_no)) events.sort(key=lambda e: e[0]) bal = 0 for t, biz, amt, ref in events: bal += amt db.add(CashTransaction( user_id=uid, amount_cents=amt, balance_after_cents=bal, biz_type=biz, ref_id=ref, remark=REMARK[biz], created_at=t, )) db.add(CoinAccount(user_id=uid, coin_balance=0, cash_balance_cents=bal, total_coin_earned=0)) # 4) 给部分用户造看广告/签到记录(喂提现详情抽屉的「统计区」+「金币记录」表)。 # eCPM 原值单位=分/千次;金币用与线上同源的 rewards.calculate_ad_reward_coin 复算,真实偏小。 def _bj_date(t: datetime) -> str: return (t + timedelta(hours=8)).date().isoformat() coin_bal: dict[int, int] = defaultdict(int) def add_video(uid: int, ecpm_fen: int, created: datetime, nth: int, scene: str = "reward_video") -> None: db.add(AdRewardRecord( trans_id=uuid.uuid4().hex, user_id=uid, coin=rewards.calculate_ad_reward_coin(str(ecpm_fen), nth), status="granted", reward_scene=scene, ecpm_raw=str(ecpm_fen), reward_date=_bj_date(created), created_at=created, )) def add_feed(uid: int, ecpm_fen: int, units: int, created: datetime, scene: str | None = None) -> None: coin = sum(rewards.calculate_ad_reward_coin(str(ecpm_fen), i) for i in range(1, units + 1)) db.add(AdFeedRewardRecord( client_event_id=uuid.uuid4().hex, user_id=uid, reward_date=_bj_date(created), duration_seconds=units * 10, unit_count=units, ecpm_raw=str(ecpm_fen), feed_scene=scene, coin=coin, status="granted", created_at=created, )) def add_signin(uid: int, coin: int, created: datetime) -> None: coin_bal[uid] += coin db.add(CoinTransaction( user_id=uid, amount=coin, balance_after=coin_bal[uid], biz_type="signin", ref_id=_bj_date(created), remark="每日签到", created_at=created, )) u1 = users["13800000001"].id # 丰富:激励视频 + 信息流 + 签到 + 1 条签到膨胀 for i, e in enumerate([3000, 5000, 8000, 4000, 6000, 3500, 7000, 4500, 5500, 6500]): add_video(u1, e, ago(days=1, hours=i), nth=(i % 5) + 1) add_video(u1, 9000, ago(hours=2), nth=1, scene="signin_boost") # 信息流打 scene:比价×2 / 领券×1 / 未分类×1(演示金币记录「途径」分类) feed_scenes = ["comparison", "coupon", "comparison", None] for i, (e, units) in enumerate([(2000, 6), (3000, 4), (5000, 3), (2500, 5)]): add_feed(u1, e, units, ago(days=2, hours=i), scene=feed_scenes[i]) for i in range(5): add_signin(u1, 2000, ago(days=i + 1)) u2 = users["13900000002"].id # 较轻:几条激励视频 + 签到 for i, e in enumerate([3000, 4000, 5000]): add_video(u2, e, ago(days=1, hours=i), nth=i + 1) for i in range(3): add_signin(u2, 2000, ago(days=i + 1)) db.commit() return orders def _yuan(cents: int) -> str: return f"¥{cents / 100:.2f}" def main() -> None: parser = argparse.ArgumentParser(description="造提现 mock 数据(运营后台提现管理页联调用)") parser.add_argument("--clean-only", action="store_true", help="只清理 mock 数据,不重建") args = parser.parse_args() db = SessionLocal() try: removed = clean(db) if removed: print(f"🧹 已清理旧 mock:{removed} 个用户及其提现单/流水") if args.clean_only: print("✅ 仅清理,已完成。") return orders = seed(db) # 汇总打印(对照页面应显示的内容) cum: dict[int, int] = defaultdict(int) for o in orders: if o.status == "success": cum[o.user_id] += o.amount_cents uid2phone = dict(db.execute(select(User.id, User.phone).where(User.phone.in_(MOCK_PHONES))).all()) print(f"\n✅ 已生成 {len(orders)} 条提现单,分布:") by_status: dict[str, list[WithdrawOrder]] = defaultdict(list) for o in orders: by_status[o.status].append(o) label = {"reviewing": "待审核", "pending": "打款中", "success": "已到账", "rejected": "已拒绝", "failed": "打款失败"} for st in ("reviewing", "pending", "success", "rejected", "failed"): lst = by_status.get(st, []) total = sum(o.amount_cents for o in lst) print(f" {label[st]:<5} {len(lst)} 单 合计 {_yuan(total)}") print("\n 明细(手机号 | 提现金额 | 累计提现 | 状态 | 失败原因):") for o in sorted(orders, key=lambda x: (x.user_id, x.created_at)): phone = uid2phone.get(o.user_id, "?") print( f" {phone} {_yuan(o.amount_cents):>8} 累计 {_yuan(cum.get(o.user_id, 0)):>8} " f"{label[o.status]:<5} {o.fail_reason or ''}" ) print("\n👉 打开 http://localhost:3001 → 提现管理 查看。后端若没带 --reload 需重启才有新列。") finally: db.close() if __name__ == "__main__": main()