Compare commits

..

3 Commits

5 changed files with 47 additions and 555 deletions
-6
View File
@@ -175,9 +175,3 @@ PANGLE_REPORT_SITE_ID_TEST=5832303
# APPLOG_MAX_BATCH=500 # 单批最大条数(超 → 422;导入期常量,改需重启) # APPLOG_MAX_BATCH=500 # 单批最大条数(超 → 422;导入期常量,改需重启)
# APPLOG_MAX_BODY_BYTES=2097152 # 请求体上限 2MB(超 → 413;运行期可调) # APPLOG_MAX_BODY_BYTES=2097152 # 请求体上限 2MB(超 → 413;运行期可调)
# APPLOG_MAX_MSG_BYTES=8192 # 单条 msg 超此字节数截断 # APPLOG_MAX_MSG_BYTES=8192 # 单条 msg 超此字节数截断
# ===== 华为 Push =====
HUAWEI_PUSH_APP_ID=
HUAWEI_PUSH_APP_SECRET=
# 0=正式消息(默认);1=测试消息(仅开发联调,生产必须保持 0)
HUAWEI_PUSH_TARGET_USER_TYPE=0
-2
View File
@@ -88,8 +88,6 @@ class Settings(BaseSettings):
# (OAuth 换 token 时 client_id 即 AppId)。发送走 v1 messages:send,成功码 80000000。 # (OAuth 换 token 时 client_id 即 AppId)。发送走 v1 messages:send,成功码 80000000。
HUAWEI_PUSH_APP_ID: str = "" HUAWEI_PUSH_APP_ID: str = ""
HUAWEI_PUSH_APP_SECRET: str = "" HUAWEI_PUSH_APP_SECRET: str = ""
# 0=正式消息(默认,受正式消息频控);1=测试消息(仅开发联调,勿用于生产)。
HUAWEI_PUSH_TARGET_USER_TYPE: int = Field(default=0, ge=0, le=1)
HUAWEI_PUSH_TOKEN_ENDPOINT: str = "https://oauth-login.cloud.huawei.com/oauth2/v3/token" HUAWEI_PUSH_TOKEN_ENDPOINT: str = "https://oauth-login.cloud.huawei.com/oauth2/v3/token"
HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE: str = ( HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE: str = (
"https://push-api.cloud.huawei.com/v1/{app_id}/messages:send" "https://push-api.cloud.huawei.com/v1/{app_id}/messages:send"
+46 -430
View File
@@ -22,7 +22,7 @@ import uuid
from collections.abc import Callable from collections.abc import Callable
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any from typing import Any
from urllib.parse import quote, urlsplit from urllib.parse import quote
import httpx import httpx
@@ -31,7 +31,6 @@ from app.core.config import settings
logger = logging.getLogger("shagua.vendor_push") logger = logging.getLogger("shagua.vendor_push")
TYPE_ACCESSIBILITY_DISABLED = "accessibility_disabled" TYPE_ACCESSIBILITY_DISABLED = "accessibility_disabled"
DATA_EVENT_NOTIFICATION_CREATED = "notification_created"
SUPPORTED_VENDORS = frozenset({"honor", "huawei", "vivo", "xiaomi", "oppo"}) SUPPORTED_VENDORS = frozenset({"honor", "huawei", "vivo", "xiaomi", "oppo"})
# vendor key → 中文名(测试/配置状态接口展示用) # vendor key → 中文名(测试/配置状态接口展示用)
@@ -70,18 +69,6 @@ class _CachedToken:
_token_cache: dict[str, _CachedToken] = {} _token_cache: dict[str, _CachedToken] = {}
_LOG_SUMMARY_MAX_CHARS = 1500
_LOG_STRING_MAX_CHARS = 200
_SENSITIVE_LOG_KEYS = {
"accesstoken",
"appkey",
"authtoken",
"authorization",
"clientsecret",
"mastersecret",
"sign",
}
def normalize_vendor(push_vendor: str | None) -> str | None: def normalize_vendor(push_vendor: str | None) -> str | None:
if not push_vendor: if not push_vendor:
@@ -128,16 +115,8 @@ def send_notification(
if mock: if mock:
logger.info( logger.info(
"vendor push mock vendor=%s request=%s", "[mock push] vendor=%s token=%s... title=%s body=%s extras=%s",
vendor, vendor, token[:12], title, body, extras,
_log_summary(
{
"push_token": token,
"title": title,
"body": body,
"extras": extras,
}
),
) )
return { return {
"mock": True, "mock": True,
@@ -154,28 +133,7 @@ def send_notification(
"xiaomi": _send_xiaomi, "xiaomi": _send_xiaomi,
"oppo": _send_oppo, "oppo": _send_oppo,
} }
started = time.perf_counter() return dispatch[vendor](token, title, body, extras)
try:
result = dispatch[vendor](token, title, body, extras)
except Exception:
logger.exception(
"vendor push dispatch failed vendor=%s token=%s notification_type=%s "
"elapsed_ms=%.1f",
vendor,
token,
extras.get("type", ""),
_elapsed_ms(started),
)
raise
logger.info(
"vendor push dispatch succeeded vendor=%s token=%s notification_type=%s "
"elapsed_ms=%.1f",
vendor,
token,
extras.get("type", ""),
_elapsed_ms(started),
)
return result
def send_accessibility_disabled( def send_accessibility_disabled(
@@ -195,169 +153,19 @@ 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=falseApp 在前台时必走
# 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: def _require(value: str, name: str) -> str:
if not value: if not value:
raise VendorPushError(f"{name} not configured") raise VendorPushError(f"{name} not configured")
return value return value
def _elapsed_ms(started: float) -> float:
return (time.perf_counter() - started) * 1000
def _redacted_value(value: Any) -> str:
raw = json.dumps(value, ensure_ascii=False, sort_keys=True, default=str)
digest = hashlib.sha256(raw.encode()).hexdigest()[:10]
return f"<redacted len={len(raw)} sha256={digest}>"
def _sanitize_for_log(value: Any, *, key: str = "", depth: int = 0) -> Any:
"""生成有排障价值、但不暴露服务端鉴权凭据的紧凑日志摘要。"""
normalized_key = "".join(char for char in key.lower() if char.isalnum())
if normalized_key in _SENSITIVE_LOG_KEYS or normalized_key.endswith("secret"):
return _redacted_value(value)
if depth >= 10:
return f"<{type(value).__name__}>"
if isinstance(value, dict):
items = list(value.items())
sanitized = {
str(item_key): _sanitize_for_log(item_value, key=str(item_key), depth=depth + 1)
for item_key, item_value in items[:30]
}
if len(items) > 30:
sanitized["<omitted_keys>"] = len(items) - 30
return sanitized
if isinstance(value, (list, tuple)):
items = list(value)
sanitized = [_sanitize_for_log(item, key=key, depth=depth + 1) for item in items[:20]]
if len(items) > 20:
sanitized.append(f"<omitted_items={len(items) - 20}>")
return sanitized
if isinstance(value, str):
if value.lstrip().startswith(("{", "[")):
try:
decoded = json.loads(value)
except ValueError:
pass
else:
return _sanitize_for_log(decoded, key=key, depth=depth + 1)
if len(value) > _LOG_STRING_MAX_CHARS:
return f"{value[:_LOG_STRING_MAX_CHARS]}…<len={len(value)}>"
return value
def _log_summary(value: Any) -> str:
rendered = json.dumps(
_sanitize_for_log(value),
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
default=str,
)
if len(rendered) > _LOG_SUMMARY_MAX_CHARS:
return f"{rendered[:_LOG_SUMMARY_MAX_CHARS]}…<len={len(rendered)}>"
return rendered
def _raw_log_summary(value: Any) -> str:
"""厂商响应摘要:保留原始字段和值,仅限制单条日志长度。"""
rendered = json.dumps(
value,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
default=str,
)
if len(rendered) > _LOG_SUMMARY_MAX_CHARS:
return f"{rendered[:_LOG_SUMMARY_MAX_CHARS]}…<len={len(rendered)}>"
return rendered
def _endpoint_for_log(url: str) -> str:
parsed = urlsplit(url)
return f"{parsed.scheme}://{parsed.netloc}{parsed.path}"
def _request_summary(kwargs: dict[str, Any]) -> str:
summary: dict[str, Any] = {}
if "json" in kwargs:
summary["body_type"] = "json"
summary["request_payload"] = kwargs["json"]
elif "data" in kwargs:
summary["body_type"] = "form"
summary["request_payload"] = kwargs["data"]
if kwargs.get("params"):
summary["params"] = kwargs["params"]
if kwargs.get("headers"):
summary["headers"] = kwargs["headers"]
return _log_summary(summary)
def _response_summary(resp: Any, parsed: Any | None = None) -> str:
if parsed is None:
try:
parsed = resp.json()
except ValueError:
pass
if parsed is not None:
return _raw_log_summary(parsed)
return _raw_log_summary(getattr(resp, "text", ""))
def _request_json( def _request_json(
method: str, method: str,
url: str, url: str,
*, *,
vendor: str,
operation: str,
expected_status: tuple[int, ...] = (200,), expected_status: tuple[int, ...] = (200,),
**kwargs: Any, **kwargs: Any,
) -> dict[str, Any]: ) -> dict[str, Any]:
started = time.perf_counter()
request_summary = _request_summary(kwargs)
try: try:
resp = httpx.request( resp = httpx.request(
method, method,
@@ -366,82 +174,41 @@ def _request_json(
**kwargs, **kwargs,
) )
except httpx.HTTPError as e: except httpx.HTTPError as e:
logger.error(
"vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
"outcome=network_error request=%s response=%s elapsed_ms=%.1f error=%s",
vendor,
operation,
method,
_endpoint_for_log(url),
request_summary,
"<no_response>",
_elapsed_ms(started),
type(e).__name__,
)
raise VendorPushError(f"push http error: {e}") from e raise VendorPushError(f"push http error: {e}") from e
if resp.status_code not in expected_status: if resp.status_code not in expected_status:
logger.error( logger.error("vendor push http failed url=%s http=%s body=%s", url, resp.status_code, resp.text[:500])
"vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
"outcome=http_error http_status=%s request=%s response=%s elapsed_ms=%.1f",
vendor,
operation,
method,
_endpoint_for_log(url),
resp.status_code,
request_summary,
_response_summary(resp),
_elapsed_ms(started),
)
raise VendorPushError(f"push http {resp.status_code}") raise VendorPushError(f"push http {resp.status_code}")
try: try:
data = resp.json() return resp.json()
except ValueError as e: except ValueError as e:
logger.error(
"vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
"outcome=invalid_json http_status=%s request=%s response=%s elapsed_ms=%.1f",
vendor,
operation,
method,
_endpoint_for_log(url),
resp.status_code,
request_summary,
_response_summary(resp),
_elapsed_ms(started),
)
raise VendorPushError(f"push invalid json: {resp.text[:200]}") from e raise VendorPushError(f"push invalid json: {resp.text[:200]}") from e
logger.info(
"vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
"outcome=success http_status=%s request=%s response=%s elapsed_ms=%.1f",
vendor,
operation,
method,
_endpoint_for_log(url),
resp.status_code,
request_summary,
_response_summary(resp, data),
_elapsed_ms(started),
)
return data
def _request_form( def _request_form(
method: str, method: str,
url: str, url: str,
*, *,
vendor: str,
operation: str,
expected_status: tuple[int, ...] = (200,), expected_status: tuple[int, ...] = (200,),
**kwargs: Any, **kwargs: Any,
) -> dict[str, Any]: ) -> dict[str, Any]:
return _request_json( try:
method, resp = httpx.request(
url, method,
vendor=vendor, url,
operation=operation, timeout=settings.PUSH_REQUEST_TIMEOUT_SEC,
expected_status=expected_status, **kwargs,
**kwargs, )
) except httpx.HTTPError as e:
raise VendorPushError(f"push http error: {e}") from e
if resp.status_code not in expected_status:
logger.error("vendor push http failed url=%s http=%s body=%s", url, resp.status_code, resp.text[:500])
raise VendorPushError(f"push http {resp.status_code}")
try:
return resp.json()
except ValueError as e:
raise VendorPushError(f"push invalid json: {resp.text[:200]}") from e
def _cache_get(key: str) -> str | None: def _cache_get(key: str) -> str | None:
@@ -467,8 +234,6 @@ def _honor_access_token() -> str:
data = _request_form( data = _request_form(
"POST", "POST",
settings.HONOR_PUSH_TOKEN_ENDPOINT, settings.HONOR_PUSH_TOKEN_ENDPOINT,
vendor="honor",
operation="authenticate",
data={ data={
"grant_type": "client_credentials", "grant_type": "client_credentials",
"client_id": client_id, "client_id": client_id,
@@ -478,22 +243,16 @@ def _honor_access_token() -> str:
) )
token = data.get("access_token") token = data.get("access_token")
if not token: if not token:
raise VendorPushError(f"honor auth failed: {_raw_log_summary(data)}") raise VendorPushError(f"honor auth failed: {data}")
return _cache_put(cache_key, str(token), data.get("expires_in")) return _cache_put(cache_key, str(token), data.get("expires_in"))
def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> dict[str, Any]: 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") app_id = _require(settings.HONOR_PUSH_APP_ID, "HONOR_PUSH_APP_ID")
access_token = _honor_access_token() 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 = { payload = {
# clickAction type=3(打开应用首页)时,荣耀点击会把 data JSON 的键值对注入启动 intent 的
# extras(与 HMS 同机制)→ MainActivity.consumeNavTarget 读 notif_id/notif_type 直达落地。
"data": json.dumps(_click_extras(extras), ensure_ascii=False), "data": json.dumps(_click_extras(extras), ensure_ascii=False),
"notification": {"title": title, "body": body}, "notification": {"title": title, "body": body},
"android": { "android": {
@@ -502,7 +261,7 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
"notification": { "notification": {
"title": title, "title": title,
"body": body, "body": body,
"clickAction": click_action, "clickAction": {"type": 3},
"importance": "NORMAL", "importance": "NORMAL",
}, },
}, },
@@ -511,8 +270,6 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
data = _request_json( data = _request_json(
"POST", "POST",
settings.HONOR_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id), settings.HONOR_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
vendor="honor",
operation="send_notification",
json=payload, json=payload,
headers={ headers={
"Content-Type": "application/json; charset=UTF-8", "Content-Type": "application/json; charset=UTF-8",
@@ -522,28 +279,7 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
) )
code = data.get("code") code = data.get("code")
if code is not None and int(code) != 200: if code is not None and int(code) != 200:
raise VendorPushError(f"honor push failed: {_raw_log_summary(data)}") raise VendorPushError(f"honor push failed: {data}")
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),
vendor="honor",
operation="send_data_event",
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: {_raw_log_summary(data)}")
return data return data
@@ -558,8 +294,6 @@ def _huawei_access_token() -> str:
data = _request_form( data = _request_form(
"POST", "POST",
settings.HUAWEI_PUSH_TOKEN_ENDPOINT, settings.HUAWEI_PUSH_TOKEN_ENDPOINT,
vendor="huawei",
operation="authenticate",
data={ data={
"grant_type": "client_credentials", "grant_type": "client_credentials",
"client_id": app_id, "client_id": app_id,
@@ -569,7 +303,7 @@ def _huawei_access_token() -> str:
) )
token = data.get("access_token") token = data.get("access_token")
if not token: if not token:
raise VendorPushError(f"huawei auth failed: {_raw_log_summary(data)}") raise VendorPushError(f"huawei auth failed: {data}")
return _cache_put(cache_key, str(token), data.get("expires_in")) return _cache_put(cache_key, str(token), data.get("expires_in"))
@@ -578,24 +312,18 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d
'80100000' 为部分成功(单 token 场景仍视为失败,错误里带原始响应便于排障)。""" '80100000' 为部分成功(单 token 场景仍视为失败,错误里带原始响应便于排障)。"""
app_id = _require(settings.HUAWEI_PUSH_APP_ID, "HUAWEI_PUSH_APP_ID") app_id = _require(settings.HUAWEI_PUSH_APP_ID, "HUAWEI_PUSH_APP_ID")
access_token = _huawei_access_token() 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 = { payload = {
"validate_only": False, "validate_only": False,
"message": { "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), "data": json.dumps(_click_extras(extras), ensure_ascii=False),
"android": { "android": {
"target_user_type": settings.HUAWEI_PUSH_TARGET_USER_TYPE,
"ttl": f"{settings.PUSH_TIME_TO_LIVE_SEC}s", "ttl": f"{settings.PUSH_TIME_TO_LIVE_SEC}s",
"notification": { "notification": {
"title": title, "title": title,
"body": body, "body": body,
"click_action": click_action, "click_action": {"type": 3},
"importance": "NORMAL", "importance": "NORMAL",
}, },
}, },
@@ -605,8 +333,6 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d
data = _request_json( data = _request_json(
"POST", "POST",
settings.HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id), settings.HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
vendor="huawei",
operation="send_notification",
json=payload, json=payload,
headers={ headers={
"Content-Type": "application/json; charset=UTF-8", "Content-Type": "application/json; charset=UTF-8",
@@ -614,32 +340,7 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d
}, },
) )
if str(data.get("code", "")) != "80000000": if str(data.get("code", "")) != "80000000":
raise VendorPushError(f"huawei push failed: {_raw_log_summary(data)}") raise VendorPushError(f"huawei push failed: {data}")
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),
vendor="huawei",
operation="send_data_event",
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: {_raw_log_summary(data)}")
return data return data
@@ -656,8 +357,6 @@ def _vivo_auth_token() -> str:
data = _request_json( data = _request_json(
"POST", "POST",
settings.VIVO_PUSH_AUTH_ENDPOINT, settings.VIVO_PUSH_AUTH_ENDPOINT,
vendor="vivo",
operation="authenticate",
json={ json={
"appId": app_id, "appId": app_id,
"appKey": app_key, "appKey": app_key,
@@ -667,10 +366,10 @@ def _vivo_auth_token() -> str:
headers={"Content-Type": "application/json"}, headers={"Content-Type": "application/json"},
) )
if int(data.get("result", -1)) != 0: if int(data.get("result", -1)) != 0:
raise VendorPushError(f"vivo auth failed: {_raw_log_summary(data)}") raise VendorPushError(f"vivo auth failed: {data}")
token = data.get("authToken") token = data.get("authToken")
if not token: if not token:
raise VendorPushError(f"vivo auth missing authToken: {_raw_log_summary(data)}") raise VendorPushError(f"vivo auth missing authToken: {data}")
return _cache_put(cache_key, str(token), 24 * 3600) return _cache_put(cache_key, str(token), 24 * 3600)
@@ -686,22 +385,15 @@ def _send_vivo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
"timeToLive": settings.PUSH_TIME_TO_LIVE_SEC, "timeToLive": settings.PUSH_TIME_TO_LIVE_SEC,
"requestId": uuid.uuid4().hex, "requestId": uuid.uuid4().hex,
"pushMode": settings.VIVO_PUSH_MODE, "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, "clientCustomMap": extras,
} }
# vivo Push SDK 480+ 已不再回调 skipType=3;必须使用 skipType=4 的完整 Intent URI。 # 点击落地:消息中心推送(带 notificationId)→ skipType=4 + skipContent=intent uri,由 vivo
# URI 同时包含 data/scheme、显式 component 和 S. extrasOriginOS 才会把业务参数 # 系统直启 MainActivity 并携带 S. extras(与小米 notify_effect=2 同机制)。不依赖客户端
# 原样交给 MainActivity(仅靠隐式 deeplink 会打开 App,但可能剥掉 extras)。 # VivoPushReceiver.onNotificationMessageClicked 里的后台 startActivity——Android 10+ BAL
# 会静默拦掉,receiver 路径仅作兜底。无 notificationId 的召回类保持 skipType=1 仅打开首页。
if extras.get("notificationId"): if extras.get("notificationId"):
payload["skipType"] = 4 payload["skipType"] = 4
payload["skipContent"] = _vivo_click_intent_uri(extras) payload["skipContent"] = _click_intent_uri(extras)
else: else:
payload["skipType"] = 1 payload["skipType"] = 1
if settings.VIVO_PUSH_CATEGORY: if settings.VIVO_PUSH_CATEGORY:
@@ -709,37 +401,14 @@ def _send_vivo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
data = _request_json( data = _request_json(
"POST", "POST",
settings.VIVO_PUSH_SEND_ENDPOINT, settings.VIVO_PUSH_SEND_ENDPOINT,
vendor="vivo",
operation="send_notification",
json=payload, json=payload,
headers={ headers={
"Content-Type": "application/json", "Content-Type": "application/json",
"authToken": auth_token, "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,
vendor="vivo",
operation="send_notification",
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: if int(data.get("result", -1)) != 0:
raise VendorPushError(f"vivo push failed: {_raw_log_summary(data)}") raise VendorPushError(f"vivo push failed: {data}")
return data return data
@@ -777,40 +446,14 @@ def _send_xiaomi(token: str, title: str, body: str, extras: dict[str, str]) -> d
data = _request_form( data = _request_form(
"POST", "POST",
settings.XIAOMI_PUSH_SEND_ENDPOINT, settings.XIAOMI_PUSH_SEND_ENDPOINT,
vendor="xiaomi",
operation="send_notification",
data=form, data=form,
headers={"Authorization": f"key={app_secret}"}, headers={"Authorization": f"key={app_secret}"},
) )
code = data.get("code") code = data.get("code")
if code not in (0, "0", None): if code not in (0, "0", None):
raise VendorPushError(f"xiaomi push failed: {_raw_log_summary(data)}") raise VendorPushError(f"xiaomi push failed: {data}")
if str(data.get("result", "ok")).lower() not in ("ok", "success"): if str(data.get("result", "ok")).lower() not in ("ok", "success"):
raise VendorPushError(f"xiaomi push failed: {_raw_log_summary(data)}") raise VendorPushError(f"xiaomi push failed: {data}")
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,
vendor="xiaomi",
operation="send_data_event",
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: {_raw_log_summary(data)}")
if str(data.get("result", "ok")).lower() not in ("ok", "success"):
raise VendorPushError(f"xiaomi data push failed: {_raw_log_summary(data)}")
return data return data
@@ -847,24 +490,6 @@ def _click_intent_uri(extras: dict[str, str]) -> str:
return ";".join(parts) 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: def _xiaomi_template_param(title: str, alert: str) -> str:
rendered = ( rendered = (
settings.XIAOMI_PUSH_TEMPLATE_PARAM_JSON settings.XIAOMI_PUSH_TEMPLATE_PARAM_JSON
@@ -897,8 +522,6 @@ def _oppo_auth_token() -> str:
data = _request_form( data = _request_form(
"POST", "POST",
settings.OPPO_PUSH_AUTH_ENDPOINT, settings.OPPO_PUSH_AUTH_ENDPOINT,
vendor="oppo",
operation="authenticate",
data={ data={
"app_key": app_key, "app_key": app_key,
"timestamp": timestamp, "timestamp": timestamp,
@@ -907,10 +530,10 @@ def _oppo_auth_token() -> str:
headers={"Content-Type": "application/x-www-form-urlencoded"}, headers={"Content-Type": "application/x-www-form-urlencoded"},
) )
if int(data.get("code", -1)) != 0: if int(data.get("code", -1)) != 0:
raise VendorPushError(f"oppo auth failed: {_raw_log_summary(data)}") raise VendorPushError(f"oppo auth failed: {data}")
token = (data.get("data") or {}).get("auth_token") or data.get("auth_token") token = (data.get("data") or {}).get("auth_token") or data.get("auth_token")
if not token: if not token:
raise VendorPushError(f"oppo auth missing auth_token: {_raw_log_summary(data)}") raise VendorPushError(f"oppo auth missing auth_token: {data}")
return _cache_put(cache_key, str(token), 24 * 3600) return _cache_put(cache_key, str(token), 24 * 3600)
@@ -925,11 +548,6 @@ def _send_oppo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
"off_line_ttl": ttl_hours, "off_line_ttl": ttl_hours,
"action_parameters": json.dumps(_click_extras(extras), ensure_ascii=False), "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 的 # 点击落地:OPPO SDK 没有点击回调,参数只能靠服务端点击动作配置送达——action_parameters 的
# 键值对仅在 click_action_type=1/4 时才会注入目标 Activity 的 intent extras(type=0「启动应用」 # 键值对仅在 click_action_type=1/4 时才会注入目标 Activity 的 intent extras(type=0「启动应用」
# 会忽略它,extras 全丢 → 点了没反应,与小米 notify_effect=1 同款坑)。 # 会忽略它,extras 全丢 → 点了没反应,与小米 notify_effect=1 同款坑)。
@@ -955,8 +573,6 @@ def _send_oppo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
data = _request_form( data = _request_form(
"POST", "POST",
settings.OPPO_PUSH_SEND_ENDPOINT, settings.OPPO_PUSH_SEND_ENDPOINT,
vendor="oppo",
operation="send_notification",
data={ data={
"auth_token": auth_token, "auth_token": auth_token,
"message": json.dumps(message, ensure_ascii=False), "message": json.dumps(message, ensure_ascii=False),
@@ -964,5 +580,5 @@ def _send_oppo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
headers={"Content-Type": "application/x-www-form-urlencoded"}, headers={"Content-Type": "application/x-www-form-urlencoded"},
) )
if int(data.get("code", -1)) != 0: if int(data.get("code", -1)) != 0:
raise VendorPushError(f"oppo push failed: {_raw_log_summary(data)}") raise VendorPushError(f"oppo push failed: {data}")
return data return data
+1 -22
View File
@@ -136,17 +136,9 @@ def _push_to_user_devices(db: Session, row: Notification, push_vars: dict[str, s
) )
continue continue
try: try:
response = vendor_push.send_notification( vendor_push.send_notification(
vendor, dev.push_token, title=title, body=body, extras=extras vendor, dev.push_token, title=title, body=body, extras=extras
) )
if vendor == "huawei" and row.type in {"feedback_reply", "feedback_reward"}:
# 华为反馈推送联调日志:保留厂商返回码/requestId 等排障信息,
# 请求本身的 token、Authorization 和应用密钥不会进入 response。
logger.info(
"huawei feedback push response user_id=%s type=%s "
"notification_id=%s response=%s",
row.user_id, row.type, row.id, response,
)
logger.info( logger.info(
"push sent user_id=%s type=%s vendor=%s notification_id=%s", "push sent user_id=%s type=%s vendor=%s notification_id=%s",
row.user_id, row.type, vendor, row.id, row.user_id, row.type, vendor, row.id,
@@ -155,19 +147,6 @@ def _push_to_user_devices(db: Session, row: Notification, push_vars: dict[str, s
logger.warning( logger.warning(
"push failed user_id=%s type=%s vendor=%s: %s", row.user_id, row.type, vendor, e "push failed user_id=%s type=%s vendor=%s: %s", row.user_id, row.type, vendor, e
) )
try:
vendor_push.send_data_event(
vendor,
dev.push_token,
event=vendor_push.DATA_EVENT_NOTIFICATION_CREATED,
notification_id=str(row.id),
)
except vendor_push.VendorPushError as e:
# 透传只负责前台铃铛实时刷新,失败不能影响通知栏消息或站内消息。
logger.warning(
"push data event failed user_id=%s type=%s vendor=%s: %s",
row.user_id, row.type, vendor, e,
)
except Exception: # noqa: BLE001 — 渲染/查设备等意外失败同样不外抛 except Exception: # noqa: BLE001 — 渲染/查设备等意外失败同样不外抛
logger.exception("push notification failed user_id=%s type=%s", row.user_id, row.type) logger.exception("push notification failed user_id=%s type=%s", row.user_id, row.type)
-95
View File
@@ -7,9 +7,7 @@ mock/真发/变量覆盖/站内联动/设备反查/参数校验。厂商 HTTP
from __future__ import annotations from __future__ import annotations
import json import json
import logging
import httpx
import pytest import pytest
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
@@ -112,99 +110,6 @@ def test_huawei_non_success_code_raises(monkeypatch) -> None:
vendor_push.send_notification("huawei", "bad-token", title="t", body="b") vendor_push.send_notification("huawei", "bad-token", title="t", body="b")
def test_vendor_http_logs_device_token_raw_response_and_elapsed(monkeypatch, caplog) -> None:
vendor_push._token_cache.clear()
def _fake_request(method, url, **kwargs): # noqa: ANN001
if url == vendor_push.settings.HUAWEI_PUSH_TOKEN_ENDPOINT:
return _Resp({"access_token": "raw-access-token", "expires_in": 3600})
return _Resp({"code": "80000000", "msg": "Success", "requestId": "vendor-request-1"})
monkeypatch.setattr(vendor_push.settings, "HUAWEI_PUSH_APP_ID", "10086001")
monkeypatch.setattr(vendor_push.settings, "HUAWEI_PUSH_APP_SECRET", "raw-app-secret")
monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
with caplog.at_level(logging.INFO, logger=vendor_push.logger.name):
vendor_push.send_notification(
"huawei",
"raw-device-token",
title="需要进入日志的标题",
body="需要进入日志的正文",
extras={"type": "withdraw_success", "notificationId": "90001"},
)
messages = "\n".join(record.getMessage() for record in caplog.records)
assert "operation=authenticate" in messages
assert "operation=send_notification" in messages
assert "request=" in messages
assert "response=" in messages
assert "elapsed_ms=" in messages
assert "vendor-request-1" in messages
assert "withdraw_success" in messages
assert "raw-access-token" in messages
assert "raw-device-token" in messages
assert "需要进入日志的标题" in messages
assert "需要进入日志的正文" in messages
assert "<redacted" in messages
assert "raw-app-secret" not in messages
send_http_log = next(
record.getMessage()
for record in caplog.records
if "operation=send_notification" in record.getMessage()
and "vendor push http completed" in record.getMessage()
)
assert "raw-device-token" in send_http_log
assert "raw-access-token" not in send_http_log # 请求头中的厂商鉴权 token 仍脱敏
def test_vendor_http_network_error_logs_request_context(monkeypatch, caplog) -> None:
def _fake_request(*args, **kwargs): # noqa: ANN002, ANN003
raise httpx.ConnectError("connection refused")
monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_APP_SECRET", "raw-xiaomi-secret")
monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
with (
caplog.at_level(logging.ERROR, logger=vendor_push.logger.name),
pytest.raises(vendor_push.VendorPushError, match="push http error"),
):
vendor_push.send_notification(
"xiaomi",
"raw-xiaomi-token",
title="title",
body="body",
extras={"type": "push_test"},
)
messages = "\n".join(record.getMessage() for record in caplog.records)
assert "outcome=network_error" in messages
assert "response=<no_response>" in messages
assert "operation=send_notification" in messages
assert "elapsed_ms=" in messages
assert "raw-xiaomi-secret" not in messages
assert "raw-xiaomi-token" in messages
@pytest.mark.parametrize(
("request_kwargs", "device_token"),
[
({"json": {"token": ["honor-device-token"]}}, "honor-device-token"),
({"json": {"message": {"token": ["huawei-device-token"]}}}, "huawei-device-token"),
({"json": {"regId": "vivo-device-token"}}, "vivo-device-token"),
({"data": {"registration_id": "xiaomi-device-token"}}, "xiaomi-device-token"),
(
{"data": {"message": '{"target_value":"oppo-device-token"}'}},
"oppo-device-token",
),
],
)
def test_all_vendor_device_token_fields_remain_visible_in_request_summary(
request_kwargs, device_token
) -> None:
summary = vendor_push._request_summary(request_kwargs)
assert device_token in summary
def test_vendor_aliases_normalize() -> None: def test_vendor_aliases_normalize() -> None:
assert vendor_push.normalize_vendor("华为") == "huawei" assert vendor_push.normalize_vendor("华为") == "huawei"
assert vendor_push.normalize_vendor("HMS") == "huawei" assert vendor_push.normalize_vendor("HMS") == "huawei"