marco
5ae4c474e4
- integrations/oppo_push.py: OPPO PUSH 服务端直连(sha256 鉴权 + auth_token 24h 缓存 + unicast 单推) - device_liveness 加 oppo_register_id 列 + 迁移; register/heartbeat 透传 + list 查询 - services/push_notify: 审核通过 fire-and-forget 推送(best-effort, 不阻塞审核响应) - admin withdraw approve(单笔+批量)挂钩(status=failed 已退款不推) - 单测 10 passed; 真调 OPPO 生产 auth/unicast 验证签名与请求构造 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
75 lines
3.2 KiB
Python
75 lines
3.2 KiB
Python
"""推送通知编排(查设备 push token → 调厂商推送)。
|
||
|
||
薄编排层: repositories(取 token) + integrations(发送) 的粘合, 供 api / admin router 调用。
|
||
所有发送 best-effort —— 失败只 log, 绝不抛给调用方(不连累审核/打款等主流程)。
|
||
|
||
当前只有 OPPO 直连一条通道(提现审核通过通知)。后续加华为/vivo 等厂商时, 在这里按设备
|
||
token 做通道分发, 上层业务函数(notify_withdraw_approved 等)不感知通道差异。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import threading
|
||
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.db.session import SessionLocal
|
||
from app.integrations import oppo_push
|
||
from app.integrations.oppo_push import OppoPushError
|
||
from app.repositories import device as device_repo
|
||
|
||
logger = logging.getLogger("shagua.push_notify")
|
||
|
||
_WITHDRAW_APPROVED_TITLE = "提现审核通过"
|
||
_WITHDRAW_APPROVED_CONTENT = "您的提现申请已审核通过，款项将很快到账，请留意微信零钱。"
|
||
|
||
|
||
def notify_withdraw_approved(db: Session, *, user_id: int) -> int:
|
||
"""提现审核通过 → 给该用户所有 OPPO 设备各推一条通知。返回成功推送的设备数。
|
||
|
||
best-effort: 未配 OPPO / 无 OPPO 设备 / 单设备发送失败都不抛, 只 log。审核主流程绝不受影响。
|
||
调用方仍应包一层 try/except 兜底本函数自身的意外异常(如 DB 查询失败)。
|
||
"""
|
||
reg_ids = device_repo.list_oppo_register_ids_by_user(db, user_id=user_id)
|
||
if not reg_ids:
|
||
logger.info("[push] withdraw_approved user_id=%s 无 OPPO 设备, 跳过", user_id)
|
||
return 0
|
||
|
||
sent = 0
|
||
for rid in reg_ids:
|
||
try:
|
||
oppo_push.send_notification(
|
||
rid, _WITHDRAW_APPROVED_TITLE, _WITHDRAW_APPROVED_CONTENT
|
||
)
|
||
sent += 1
|
||
except OppoPushError as e:
|
||
logger.warning(
|
||
"[push] withdraw_approved user_id=%s regId=%s… 发送失败: %s",
|
||
user_id, rid[:12], e,
|
||
)
|
||
logger.info(
|
||
"[push] withdraw_approved user_id=%s 推送 %d/%d 台 OPPO 设备",
|
||
user_id, sent, len(reg_ids),
|
||
)
|
||
return sent
|
||
|
||
|
||
def notify_withdraw_approved_async(user_id: int) -> None:
|
||
"""fire-and-forget 版: 后台守护线程新开 session 发推送, 不阻塞审核响应(对齐 best-effort)。
|
||
|
||
审核 approve 已同步调微信转账, 推送不该再把外部 API 耗时叠进响应(多设备/OPPO 慢时最坏
|
||
N×timeout)。线程内必须新开 SessionLocal —— 请求级 session 在响应返回后即关, 不能跨线程用。
|
||
起线程失败也只 log, 绝不连累审核。
|
||
"""
|
||
def _run() -> None:
|
||
try:
|
||
with SessionLocal() as db:
|
||
notify_withdraw_approved(db, user_id=user_id)
|
||
except Exception: # noqa: BLE001 - 后台线程绝不抛
|
||
logger.warning("[push] withdraw_approved async 异常 user_id=%s", user_id, exc_info=True)
|
||
|
||
try:
|
||
threading.Thread(target=_run, name=f"oppo-push-{user_id}", daemon=True).start()
|
||
except Exception: # noqa: BLE001 - 起线程失败也绝不连累审核
|
||
logger.warning("[push] withdraw_approved async 起线程失败 user_id=%s", user_id, exc_info=True)
|