@@ -31,6 +31,7 @@ from app.core.config import settings
logger = logging . getLogger ( " shagua.vendor_push " )
TYPE_ACCESSIBILITY_DISABLED = " accessibility_disabled "
DATA_EVENT_NOTIFICATION_CREATED = " notification_created "
SUPPORTED_VENDORS = frozenset ( { " honor " , " huawei " , " vivo " , " xiaomi " , " oppo " } )
# vendor key → 中文名(测试/配置状态接口展示用)
@@ -153,6 +154,49 @@ def send_accessibility_disabled(
)
def send_data_event (
push_vendor : str ,
push_token : str ,
* ,
event : str ,
notification_id : str ,
mock : bool = False ,
) - > dict [ str , Any ] :
""" 发送无界面的轻量事件；当前只允许站内消息创建事件。 """
vendor = normalize_vendor ( push_vendor )
token = push_token . strip ( ) if push_token else " "
if not vendor or vendor not in SUPPORTED_VENDORS :
raise VendorPushError ( f " unsupported push vendor: { push_vendor } " )
if not token :
raise VendorPushError ( " push token is empty " )
if event != DATA_EVENT_NOTIFICATION_CREATED :
raise VendorPushError ( f " unsupported data event: { event } " )
if not notification_id :
raise VendorPushError ( " notification_id is empty " )
payload = { " event " : event , " notificationId " : str ( notification_id ) }
if mock :
return { " mock " : True , " vendor " : vendor , " payload " : payload }
# vivo 的通知已设置 foregroundShow=false：
# onForegroundMessageArrived，正好就是本事件需要的刷新信号；不重复占用一次推送配额。
if vendor == " vivo " :
return { " skipped " : True , " reason " : " foreground notification callback " }
# OPush 当前只支持通知栏消息，没有服务端透传单推接口。
# 客户端在 OPPO 首页可见时轮询未读数；这里必须安全跳过，不能请求不存在的
# /message/transparent/unicast（该地址会稳定返回 HTTP 404）。
if vendor == " oppo " :
return { " skipped " : True , " reason " : " oppo does not support data messages " }
dispatch : dict [ str , Callable [ [ str , dict [ str , str ] ] , dict [ str , Any ] ] ] = {
" honor " : _send_honor_data ,
" huawei " : _send_huawei_data ,
" xiaomi " : _send_xiaomi_data ,
}
return dispatch [ vendor ] ( token , payload )
def _require ( value : str , name : str ) - > str :
if not value :
raise VendorPushError ( f " { name } not configured " )
@@ -250,9 +294,15 @@ def _honor_access_token() -> str:
def _send_honor ( token : str , title : str , body : str , extras : dict [ str , str ] ) - > dict [ str , Any ] :
app_id = _require ( settings . HONOR_PUSH_APP_ID , " HONOR_PUSH_APP_ID " )
access_token = _honor_access_token ( )
click_action : dict [ str , Any ]
if extras . get ( " notificationId " ) :
# type=3 只负责打开首页，不保证 data 会变成目标 Activity 的 extras。消息中心通知必须
# 用自定义页面(type=1)+ intent URI，把 notificationId/type/feedbackId 等直接带给
# MainActivity；否则华为/荣耀系统能展示通知，但用户点击后客户端收不到任何导航参数。
click_action = { " type " : 1 , " intent " : _click_intent_uri ( extras ) }
else :
click_action = { " type " : 3 }
payload = {
# clickAction type=3(打开应用首页)时,荣耀点击会把 data JSON 的键值对注入启动 intent 的
# extras(与 HMS 同机制)→ MainActivity.consumeNavTarget 读 notif_id/notif_type 直达落地。
" data " : json . dumps ( _click_extras ( extras ) , ensure_ascii = False ) ,
" notification " : { " title " : title , " body " : body } ,
" android " : {
@@ -261,7 +311,7 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
" notification " : {
" title " : title ,
" body " : body ,
" clickAction " : { " type " : 3 } ,
" clickAction " : click_action ,
" importance " : " NORMAL " ,
} ,
} ,
@@ -283,6 +333,25 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
return data
def _send_honor_data ( token : str , payload : dict [ str , str ] ) - > dict [ str , Any ] :
app_id = _require ( settings . HONOR_PUSH_APP_ID , " HONOR_PUSH_APP_ID " )
access_token = _honor_access_token ( )
data = _request_json (
" POST " ,
settings . HONOR_PUSH_SEND_ENDPOINT_TEMPLATE . format ( app_id = app_id ) ,
json = { " data " : json . dumps ( payload , ensure_ascii = False ) , " token " : [ token ] } ,
headers = {
" Content-Type " : " application/json; charset=UTF-8 " ,
" Authorization " : f " Bearer { access_token } " ,
" timestamp " : str ( int ( time . time ( ) * 1000 ) ) ,
} ,
)
code = data . get ( " code " )
if code is not None and int ( code ) != 200 :
raise VendorPushError ( f " honor data push failed: { data } " )
return data
def _huawei_access_token ( ) - > str :
""" 华为 OAuth2 client_credentials 换 access_token(client_id 即 AGC 应用的 AppId)。 """
cache_key = " huawei "
@@ -312,18 +381,24 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d
' 80100000 ' 为部分成功(单 token 场景仍视为失败,错误里带原始响应便于排障)。 """
app_id = _require ( settings . HUAWEI_PUSH_APP_ID , " HUAWEI_PUSH_APP_ID " )
access_token = _huawei_access_token ( )
click_action : dict [ str , Any ]
if extras . get ( " notificationId " ) :
# type=3 只打开首页，Mate 20/EMUI 不会把 message.data 自动拆成启动 Intent extras。
# type=1 的自定义 intent 才能稳定携带反馈记录 id，且冷启动/onNewIntent 都走同一路由。
click_action = { " type " : 1 , " intent " : _click_intent_uri ( extras ) }
else :
click_action = { " type " : 3 }
payload = {
" validate_only " : False ,
" message " : {
# click_action type=3(打开应用首页)时,HMS 点击会把 data JSON 的键值对注入启动 intent
# 的 extras → MainActivity.consumeNavTarget 读 notif_id/notif_type 直达落地。
" data " : json . dumps ( _click_extras ( extras ) , ensure_ascii = False ) ,
" android " : {
" target_user_type " : settings . HUAWEI_PUSH_TARGET_USER_TYPE ,
" ttl " : f " { settings . PUSH_TIME_TO_LIVE_SEC } s " ,
" notification " : {
" title " : title ,
" body " : body ,
" click_action " : { " type " : 3 } ,
" click_action " : click_action ,
" importance " : " NORMAL " ,
} ,
} ,
@@ -344,6 +419,29 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d
return data
def _send_huawei_data ( token : str , payload : dict [ str , str ] ) - > dict [ str , Any ] :
app_id = _require ( settings . HUAWEI_PUSH_APP_ID , " HUAWEI_PUSH_APP_ID " )
access_token = _huawei_access_token ( )
data = _request_json (
" POST " ,
settings . HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE . format ( app_id = app_id ) ,
json = {
" validate_only " : False ,
" message " : {
" data " : json . dumps ( payload , ensure_ascii = False ) ,
" token " : [ token ] ,
} ,
} ,
headers = {
" Content-Type " : " application/json; charset=UTF-8 " ,
" Authorization " : f " Bearer { access_token } " ,
} ,
)
if str ( data . get ( " code " , " " ) ) != " 80000000 " :
raise VendorPushError ( f " huawei data push failed: { data } " )
return data
def _vivo_auth_token ( ) - > str :
cache_key = " vivo "
cached = _cache_get ( cache_key )
@@ -385,15 +483,22 @@ def _send_vivo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
" timeToLive " : settings . PUSH_TIME_TO_LIVE_SEC ,
" requestId " : uuid . uuid4 ( ) . hex ,
" pushMode " : settings . VIVO_PUSH_MODE ,
# vivo SDK 4.1.5 只有在前台展示关闭时才调用
# OpenClientPushMessageReceiver.onForegroundMessageArrived。
# App 在该回调中刷新服务端未读数并补发一条本地通知；后台/锁屏时仍由
# vivo 系统展示，因而任何场景都只会出现一条通知。
" foregroundShow " : False ,
# vivo 官方 VPush 角标字段：离线收到通知时先由桌面自动 +1；
# App 启动/同步后再由统一未读数通过系统 API 精确校准。
" addBadge " : True ,
" clientCustomMap " : extras ,
}
# 点击落地:消息中心推送(带 notificationId)→ skipType=4 + skipContent=intent uri,由 vivo
# 系统直启 MainActivity 并携带 S. extras(与小米 notify_effect=2 同机制)。不依赖客户端
# VivoPushReceiver.onNotificationMessageClicked 里的后台 startActivity——Android 10+ BAL
# 会静默拦掉,receiver 路径仅作兜底。无 notificationId 的召回类保持 skipType=1 仅打开首页。
# vivo Push SDK 480+ 已不再回调 skipType=3；必须使用 skipType=4 的完整 Intent URI。
# URI 同时包含 data/scheme、显式 component 和 S. extras，
# 原样交给 MainActivity（仅靠隐式 deeplink 会打开 App，但可能剥掉 extras）。
if extras . get ( " notificationId " ) :
payload [ " skipType " ] = 4
payload [ " skipContent " ] = _click_intent_uri ( extras )
payload [ " skipContent " ] = _vivo_click_intent_uri ( extras )
else :
payload [ " skipType " ] = 1
if settings . VIVO_PUSH_CATEGORY :
@@ -407,6 +512,25 @@ def _send_vivo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
" authToken " : auth_token ,
} ,
)
# 10089 = 当前 vivo 应用尚未开通 VPush「离线自动角标」能力。
# 角标只是通知中心未读数的镜像，不能因此让整条通知发送失败：去掉 addBadge
# 原样重试，通知到达/应用同步后仍由客户端系统 API 按统一未读数精确设置。
if int ( data . get ( " result " , - 1 ) ) == 10089 :
logger . warning ( " vivo VPush badge not enabled (10089); retrying without addBadge " )
fallback_payload = dict ( payload )
fallback_payload . pop ( " addBadge " , None )
fallback_payload [ " requestId " ] = uuid . uuid4 ( ) . hex
data = _request_json (
" POST " ,
settings . VIVO_PUSH_SEND_ENDPOINT ,
json = fallback_payload ,
headers = {
" Content-Type " : " application/json " ,
" authToken " : auth_token ,
} ,
)
if int ( data . get ( " result " , - 1 ) ) == 0 :
data = { * * data , " badgeFallback " : True }
if int ( data . get ( " result " , - 1 ) ) != 0 :
raise VendorPushError ( f " vivo push failed: { data } " )
return data
@@ -457,6 +581,28 @@ def _send_xiaomi(token: str, title: str, body: str, extras: dict[str, str]) -> d
return data
def _send_xiaomi_data ( token : str , payload : dict [ str , str ] ) - > dict [ str , Any ] :
app_secret = _require ( settings . XIAOMI_PUSH_APP_SECRET , " XIAOMI_PUSH_APP_SECRET " )
data = _request_form (
" POST " ,
settings . XIAOMI_PUSH_SEND_ENDPOINT ,
data = {
" registration_id " : token ,
" restricted_package_name " : settings . ANDROID_PACKAGE_NAME ,
" payload " : json . dumps ( payload , ensure_ascii = False ) ,
" pass_through " : " 1 " ,
" time_to_live " : str ( settings . PUSH_TIME_TO_LIVE_SEC * 1000 ) ,
} ,
headers = { " Authorization " : f " key= { app_secret } " } ,
)
code = data . get ( " code " )
if code not in ( 0 , " 0 " , None ) :
raise VendorPushError ( f " xiaomi data push failed: { data } " )
if str ( data . get ( " result " , " ok " ) ) . lower ( ) not in ( " ok " , " success " ) :
raise VendorPushError ( f " xiaomi data push failed: { data } " )
return data
def _click_extras ( extras : dict [ str , str ] ) - > dict [ str , str ] :
""" 点击落地参数:消息中心推送(extras 带 notificationId)补 notif_id/notif_type 别名——
客户端 MainActivity.consumeNavTarget 首选这两个键(厂商 receiver 路径的历史约定),原始键
@@ -490,6 +636,24 @@ def _click_intent_uri(extras: dict[str, str]) -> str:
return " ; " . join ( parts )
def _vivo_click_intent_uri ( extras : dict [ str , str ] ) - > str :
""" vivo skipType=4 使用官方要求的完整 intent deeplink 形态。
`?#Intent` 中的问号不能省；OriginOS 对缺少它的 URI 会退化为“仅打开应用”，
不把 S. 参数交给目标 Activity。
"""
pkg = settings . ANDROID_PACKAGE_NAME
parts = [
" intent://push/detail?#Intent " ,
" scheme=shaguabijia " ,
f " component= { pkg } / { pkg } .MainActivity " ,
" launchFlags=0x14000000 " ,
]
parts + = [ f " S. { key } = { quote ( str ( value ) , safe = ' ' ) } " for key , value in _click_extras ( extras ) . items ( ) ]
parts . append ( " end " )
return " ; " . join ( parts )
def _xiaomi_template_param ( title : str , alert : str ) - > str :
rendered = (
settings . XIAOMI_PUSH_TEMPLATE_PARAM_JSON
@@ -548,6 +712,11 @@ def _send_oppo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
" off_line_ttl " : ttl_hours ,
" action_parameters " : json . dumps ( _click_extras ( extras ) , ensure_ascii = False ) ,
}
if extras . get ( " notificationId " ) :
# 只有已落库的站内消息才计入角标；保活提醒等临时推送不应留下无法消除的未读数。
# 客户端打开/已读后会按服务端真实未读总数覆盖，避免长期漂移。
notification [ " badge_operation_type " ] = 1
notification [ " badge_message_count " ] = 1
# 点击落地:OPPO SDK 没有点击回调,参数只能靠服务端点击动作配置送达——action_parameters 的
# 键值对仅在 click_action_type=1/4 时才会注入目标 Activity 的 intent extras(type=0「启动应用」
# 会忽略它,extras 全丢 → 点了没反应,与小米 notify_effect=1 同款坑)。
exinglang