shaguabijia-app-server/scripts/test_push_invite_order_reward.py

"""#12 好友下单到账(invite_order_reward)推送联调脚本 —— 每次执行只发 1 条,金额随机。

后台没有驱动这个事件的入口(真实链路要好友注册 + 完成首次比价);本脚本直接调
services/notification_events.notify_invite_order_reward —— 与生产同一条下发链路:
落 notification 表(站内消息)+ 向【邀请人】全部已注册厂商 token 直推(honor/huawei/xiaomi/oppo/vivo)。

可重复性:
  默认用随机假「被邀请人 id」做 dedup_key → 永不去重,想跑多少次都行(昵称兜底显示「好友」)。
  金额默认每次随机(0.01 ~ 99.99 元)→ 手机上按金额认出这条通知;--cents 可固定(线上真实值 200 = 2 元)。
  带 --invitee-phone 指定真实用户(如种子好友 12000000001 柚子)→ 通知里显示真实昵称;
  ⚠️ 但 dedup_key = 被邀请人 id:上一条还未读时重复发会命中去重(日志出现 dedup hit,不落库不推送),
  在 App 里把那条读掉(或换号)即可再次触发。

用法(服务端进程无需在跑,脚本自己连库 + 直调厂商接口):
  .venv\\Scripts\\python.exe scripts\\test_push_invite_order_reward.py                        # 随机金额发 1 条
  .venv\\Scripts\\python.exe scripts\\test_push_invite_order_reward.py --cents 200            # 固定 2.00 元
  .venv\\Scripts\\python.exe scripts\\test_push_invite_order_reward.py --invitee-phone 12000000001  # 真实昵称

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

import argparse
import logging
import random
import sys

from sqlalchemy import func, select

from app.db.session import SessionLocal, engine
from app.models.notification import Notification
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 / dedup hit
# 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 = "invite_order_reward"


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="#12 好友下单到账 推送联调(每次 1 条,金额默认随机)")
    parser.add_argument("--phone", default="11111111111", help="邀请人(收通知方)手机号,默认 11111111111")
    parser.add_argument("--cents", type=int, default=None,
                        help="奖励金额,单位分(默认随机 1~9999;线上真实值 200)")
    parser.add_argument("--invitee-phone", default="",
                        help="被邀请人手机号(可选):显示真实昵称;不传用随机假 id,昵称显示「好友」")
    args = parser.parse_args()

    cents = args.cents if args.cents is not None else random.randint(1, 9999)

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

        if args.invitee_phone.strip():
            invitee = user_repo.get_user_by_phone(db, args.invitee_phone.strip())
            if invitee is None:
                print(f"❌ 被邀请人 {args.invitee_phone} 不存在(种子好友可用 12000000001 柚子 / 12000000003 小美)。")
                return
            invitee_id = invitee.id
        else:
            invitee_id = random.randint(900000, 999999)  # 假 id,昵称兜底「好友」,永不去重

        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)
        print(f"→ 本次奖励 【{cents / 100:.2f} 元】(invitee_user_id={invitee_id}),手机上按金额认领这条通知")
        notification_events.notify_invite_order_reward(
            db, inviter_user_id=user.id, invitee_user_id=invitee_id, cash_cents=cents
        )

        created = _notif_count(db, user.id) - before
        if created == 1:
            verdict = "✅ 已落库 1 条站内消息"
        else:
            verdict = "⚠️ 未落库(若日志有 dedup hit:该被邀请人上一条还未读,先在 App 里读掉再发)"
        print(f"\n{verdict};推送成败见上方 shagua.notification_events 日志。")
    finally:
        db.close()


if __name__ == "__main__":
    main()