"""消息通知中心:业务事件 → 站内消息 + 厂商 push 的统一下发口。 PRD《消息通知中心》真实业务触发在此收口(替代 /push/test 的样例数据),已接入: #3 withdraw_success 提现到账(repositories/wallet 各「pending→success」转换点) #4 withdraw_failed 提现失败/退回(repositories/wallet._refund_withdraw,含审核拒绝) #9 feedback_reply 官方回复(admin 反馈审核「拒绝」,带用户可见原因/留言) #10 feedback_reward 反馈奖励(admin 反馈审核「采纳」发金币,必带官方留言) #11 report_approved 爆料审核通过(admin 上报更低价「通过」发金币) #12 invite_order_reward 好友下单到账(repositories/invite.try_reward_on_compare 发奖后) 行为约定(调用方唯一需要知道的两条): 1. **绝不抛异常**——通知只是业务的副产物,站内消息落库失败/推送失败只 log, 绝不让提现退款、审核发奖等主流程回滚或报错。 2. **必须在业务事务 commit 之后调用**——内部会再 commit(写 notification 表); 若在业务半途调用,会把调用方未提交的脏状态一并提交。 去重:各事件用业务主键做 dedup_key(提现单号/反馈 id/爆料 id/被邀请人 id),配合 notification 表的部分唯一索引,同一事件并发重复触发时未读期间只落一条、只推一次。 推送:向该用户所有已上报厂商 token 的设备直推(integrations/vendor_push); 厂商凭据未配置(本地/测试环境)时自动跳过推送、只落站内消息。extras 按 PRD §4 约定带 {type, notificationId, ...跳转参数},客户端点击 push 深链落地 并调 POST /notifications/read 同步置读。 """ from __future__ import annotations import logging from datetime import datetime from typing import TYPE_CHECKING from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from app.core import notification_catalog as catalog from app.core.rewards import CN_TZ from app.integrations import vendor_push from app.models.user import User from app.repositories import device as device_repo from app.repositories import notification as notif_repo if TYPE_CHECKING: from app.models.feedback import Feedback from app.models.notification import Notification from app.models.price_report import PriceReport from app.models.wallet import WithdrawOrder logger = logging.getLogger("shagua.notification_events") def _fmt_time(dt: datetime) -> str: """信息行「到账时间」的展示格式(与 repositories/notification 样例卡一致)。""" return dt.strftime("%Y-%m-%d %H:%M") def _yuan_trim(cents: int) -> str: """分 → 元,去掉多余的 0(200→"2"、1280→"12.80")。push 正文用(PRD §5 示例口径: 「{2}元现金已到账」);卡片数值仍走 cash_cents 由前端按两位小数渲染。""" yuan = cents / 100 return f"{yuan:.2f}".rstrip("0").rstrip(".") def _display_name(user: User | None) -> str: """好友昵称展示:昵称 → 微信昵称 → 手机尾号,全无则「好友」。""" name = ((user.nickname if user else None) or (user.wechat_nickname if user else None) or "").strip() if not name and user and user.phone: name = f"用户{user.phone[-4:]}" return name or "好友" # --------------------------------------------------------------------------- # 内核:落站内消息 + 厂商推送(全程吞异常) # --------------------------------------------------------------------------- def _dispatch( db: Session, *, user_id: int, type_key: str, coins: int | None = None, cash_cents: int | None = None, info_rows: list[dict[str, str]] | None = None, extra: dict[str, str] | None = None, dedup_key: str | None = None, push_vars: dict[str, str] | None = None, ) -> Notification | None: """落一条站内消息并向该用户设备直推。返回落库行;去重命中/失败返回 None。""" try: row = notif_repo.create_notification( db, user_id=user_id, type_key=type_key, coins=coins, cash_cents=cash_cents, info_rows=info_rows, extra=extra, dedup_key=dedup_key, ) except IntegrityError: # 同 (user, type, dedup_key) 已有未读消息 = 同一事件并发/重复触发 → 不重复落、不重复推 db.rollback() logger.info( "notification dedup hit user_id=%s type=%s dedup_key=%s", user_id, type_key, dedup_key ) return None except Exception: # noqa: BLE001 — 通知失败绝不影响业务主流程 logger.exception("create notification failed user_id=%s type=%s", user_id, type_key) try: db.rollback() except Exception: # noqa: BLE001 — 回滚失败也不外抛,session 由请求生命周期兜底 logger.exception("rollback after notification failure also failed") return None _push_to_user_devices(db, row, push_vars) return row def _push_to_user_devices(db: Session, row: Notification, push_vars: dict[str, str] | None) -> None: """向消息归属用户的全部厂商推送目标直推(best-effort,单设备失败不影响其余)。""" try: title, body = catalog.render_push(row.type, push_vars) # PRD §4 push 联动:extras 至少带 type + notificationId,外加该类型的跳转参数(extra 列) extras: dict[str, str] = {"type": row.type} extras.update({str(k): str(v) for k, v in (row.extra or {}).items()}) extras["notificationId"] = str(row.id) for dev in device_repo.list_push_targets(db, user_id=row.user_id): vendor = vendor_push.normalize_vendor(dev.push_vendor) if not vendor or vendor not in vendor_push.SUPPORTED_VENDORS: continue if vendor_push.missing_settings(vendor): # 本地/测试环境凭据不齐 → 只落站内消息,不发真推送(与 push/vendors 的报缺口径一致) logger.info( "skip push (vendor %s not configured) user_id=%s type=%s", vendor, row.user_id, row.type, ) continue try: vendor_push.send_notification( vendor, dev.push_token, title=title, body=body, extras=extras ) logger.info( "push sent user_id=%s type=%s vendor=%s notification_id=%s", row.user_id, row.type, vendor, row.id, ) except vendor_push.VendorPushError as e: logger.warning( "push failed user_id=%s type=%s vendor=%s: %s", row.user_id, row.type, vendor, e ) except Exception: # noqa: BLE001 — 渲染/查设备等意外失败同样不外抛 logger.exception("push notification failed user_id=%s type=%s", row.user_id, row.type) # --------------------------------------------------------------------------- # 六个业务事件(PRD §1/§3/§5 编号见文件头) # --------------------------------------------------------------------------- def notify_withdraw_success(db: Session, order: WithdrawOrder) -> None: """#3 提现成功:款项已存入微信零钱。点击无跳转仅消红点(extra 空)。""" _dispatch( db, user_id=order.user_id, type_key="withdraw_success", cash_cents=order.amount_cents, info_rows=[ {"label": "到账账户", "value": "微信钱包"}, {"label": "到账时间", "value": _fmt_time(datetime.now(CN_TZ))}, ], extra={}, dedup_key=order.out_bill_no, push_vars={"amount": notif_repo.cash_yuan(order.amount_cents)}, ) def notify_withdraw_failed(db: Session, order: WithdrawOrder) -> None: """#4 提现失败/退回:含微信侧失败、审核拒绝、解绑退回。点击跳提现页重新提现。 失败原因用 order.fail_reason(与 /withdraw/status 下发的用户可读原因同源)。 """ reason = (order.fail_reason or "").strip() or "提现未成功" _dispatch( db, user_id=order.user_id, type_key="withdraw_failed", cash_cents=order.amount_cents, info_rows=[ {"label": "失败原因", "value": reason}, {"label": "退回说明", "value": "款项已原路退回现金余额"}, ], extra={"withdrawId": order.out_bill_no}, dedup_key=order.out_bill_no, push_vars={"amount": notif_repo.cash_yuan(order.amount_cents), "reason": reason}, ) def notify_feedback_reply(db: Session, feedback: Feedback) -> None: """#9 官方回复:运营审核了反馈且未采纳(用户可见原因/留言落在反馈记录上)。 点击跳反馈历史页滚动高亮该条(extra.feedbackId)。""" _dispatch( db, user_id=feedback.user_id, type_key="feedback_reply", info_rows=[{"label": "说明文案", "value": "快去看看官方给您的回复吧～"}], extra={"feedbackId": str(feedback.id)}, dedup_key=str(feedback.id), ) def notify_feedback_reward(db: Session, feedback: Feedback) -> None: """#10 反馈奖励:反馈被采纳,金币已到账。PRD 约定发奖必带官方留言(admin_reply); 运营漏填时省略该信息行,不硬造文案。""" coins = int(feedback.reward_coins or 0) info_rows = [{"label": "奖励说明", "value": "感谢您的反馈，您的金币奖励已到账"}] reply = (feedback.admin_reply or "").strip() if reply: info_rows.append({"label": "官方留言", "value": reply}) info_rows.append({"label": "到账时间", "value": _fmt_time(datetime.now(CN_TZ))}) _dispatch( db, user_id=feedback.user_id, type_key="feedback_reward", coins=coins, info_rows=info_rows, extra={"feedbackId": str(feedback.id)}, dedup_key=str(feedback.id), push_vars={"coins": str(coins)}, ) def notify_report_approved(db: Session, report: PriceReport) -> None: """#11 爆料审核通过:上报的更低价过审,金币已到账。点击跳爆料记录页高亮该条。""" coins = int(report.reward_coins or 0) store = (report.store_name or "").strip() or "该店铺" _dispatch( db, user_id=report.user_id, type_key="report_approved", coins=coins, info_rows=[ {"label": "奖励说明", "value": f"您爆料的「{store}」更低价已通过审核，金币奖励已到账"}, {"label": "到账时间", "value": _fmt_time(datetime.now(CN_TZ))}, ], extra={"reportId": str(report.id)}, dedup_key=str(report.id), push_vars={"store": store, "coins": str(coins)}, ) def notify_invite_order_reward( db: Session, *, inviter_user_id: int, invitee_user_id: int, cash_cents: int ) -> None: """#12 好友下单到账:被邀请好友完成首次下单(比价),现金奖励已入邀请人账户。 通知发给【邀请人】;每个好友只发一次奖 → dedup 按被邀请人。""" invitee = db.get(User, invitee_user_id) nickname = _display_name(invitee) _dispatch( db, user_id=inviter_user_id, type_key="invite_order_reward", cash_cents=cash_cents, info_rows=[ {"label": "奖励说明", "value": f"好友「{nickname}」完成首次下单"}, {"label": "到账时间", "value": _fmt_time(datetime.now(CN_TZ))}, ], extra={"inviteeNickname": nickname}, dedup_key=str(invitee_user_id), push_vars={"nickname": nickname, "amount": _yuan_trim(cash_cents)}, )