shaguabijia-app-server/scripts/test_push_withdraw_failed.py

"""#4 提现失败(withdraw_failed)推送联调脚本 —— 每次执行只发 1 条,金额 + 失败原因随机。

后台虽能驱动提现拒绝,但受「同一用户同时只能有 1 张待审单」约束,批量测试凑不齐单子;
本脚本直接调 services/notification_events.notify_withdraw_failed —— 与生产同一条下发链路:
落 notification 表(站内消息)+ 向该用户全部已注册厂商 token 直推(honor/huawei/xiaomi/oppo/vivo)。

可重复性:每次执行用全新 uuid 单号做 dedup_key,永不命中「未读去重」,想跑多少次都行。
金额默认每次随机(0.01 ~ 99.99 元)、失败原因随机 → 手机上按金额/原因就能认出这条通知;
--cents / --reason 可固定。

用法(服务端进程无需在跑,脚本自己连库 + 直调厂商接口):
  .venv\\Scripts\\python.exe scripts\\test_push_withdraw_failed.py                       # 随机金额+原因发 1 条
  .venv\\Scripts\\python.exe scripts\\test_push_withdraw_failed.py --cents 350           # 固定 3.50 元
  .venv\\Scripts\\python.exe scripts\\test_push_withdraw_failed.py --reason "自定义原因"  # 固定失败原因

结果判读(看输出日志):
  push sent   = 厂商接口受理成功,手机应弹「提现失败」通知
  push failed = 厂商拒绝(原因见日志:token 失效/凭据错误等)
  skip push   = 该厂商凭据未配置,只落站内消息
站内消息用 11111111111 登录 App → 消息中心「提现助手」可见;点击应跳提现页(extra.withdrawId)。
"""
from __future__ import annotations

import argparse
import logging
import random
import sys
import uuid

from sqlalchemy import func, select

from app.db.session import SessionLocal, engine
from app.models.notification import Notification
from app.models.wallet import WithdrawOrder
from app.repositories import device as device_repo
from app.repositories import user as user_repo
from app.services import notification_events

# SQL 回显静音;shagua.* 开到 INFO 才看得到 push sent / push failed
# dev 下 APP_DEBUG=true → engine echo=True,echo 走 InstanceLogger 直写、无视 logger level,只能关 echo 本身
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)
engine.echo = False

if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(encoding="utf-8")

TYPE_KEY = "withdraw_failed"

# 失败原因池:不带 --reason 时随机抽一条,模拟不同失败场景(文案与 /withdraw/status 用户可读口径一致)
FAIL_REASONS = [
    "微信零钱未实名,款项已退回",
    "收款账户异常,款项已退回",
    "超出微信零钱收款限额,款项已退回",
    "微信实名与提现实名不一致,款项已原路退回现金余额",
]


def _notif_count(db, uid: int) -> int:
    return int(
        db.execute(
            select(func.count())
            .select_from(Notification)
            .where(Notification.user_id == uid, Notification.type == TYPE_KEY)
        ).scalar_one()
    )


def main() -> None:
    parser = argparse.ArgumentParser(description="#4 提现失败 推送联调(每次 1 条,金额/原因默认随机)")
    parser.add_argument("--phone", default="11111111111", help="目标用户手机号(默认 11111111111)")
    parser.add_argument("--cents", type=int, default=None, help="退回金额,单位分(默认随机 1~9999)")
    parser.add_argument("--reason", default="", help="失败原因(默认从内置原因池随机抽)")
    args = parser.parse_args()

    cents = args.cents if args.cents is not None else random.randint(1, 9999)
    reason = args.reason.strip() or random.choice(FAIL_REASONS)

    db = SessionLocal()
    try:
        user = user_repo.get_user_by_phone(db, args.phone)
        if user is None:
            print(f"❌ 用户 {args.phone} 不存在。先用该手机号在 App 登录一次再跑。")
            return

        targets = device_repo.list_push_targets(db, user_id=user.id)
        vendors = [t.push_vendor for t in targets]
        print(f"目标用户 {args.phone}(id={user.id});推送设备 {len(targets)} 台:{vendors or '无 ← 手机收不到!先在 App 上报 push token'}")

        before = _notif_count(db, user.id)
        order = WithdrawOrder(
            user_id=user.id, out_bill_no=uuid.uuid4().hex,
            amount_cents=cents, source="coin_cash", fail_reason=reason,
        )
        print(f"→ 本次金额 【{cents / 100:.2f} 元】,原因【{reason}】(单号 {order.out_bill_no[:8]}…)")
        notification_events.notify_withdraw_failed(db, order)

        created = _notif_count(db, user.id) - before
        verdict = "✅ 已落库 1 条站内消息" if created == 1 else "⚠️ 未落库(见上方日志)"
        print(f"\n{verdict};推送成败见上方 shagua.notification_events 日志。")
    finally:
        db.close()


if __name__ == "__main__":
    main()