Compare commits

..

8 Commits

Author SHA1 Message Date
exinglang 0124131b91 fix(推送): 隔离设备账号绑定并支持定向联调 2026-07-31 10:49:06 +08:00
guke ef0ab9d95a fix(短信): 号码无效提示由「手机号无效」改为「请输入有效的手机号」 (#206)
极光/阿里云/创蓝三处 provider 的号码无效错误统一改成更友好的用户提示。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: guke <guke@autohome.com.cn>
Reviewed-on: #206
2026-07-30 19:15:45 +08:00
zuochenyong 06cd718610 修复之前错误配置的端口 (#203)
Co-authored-by: exinglang <exinglang@qq.com>
Reviewed-on: #203
Co-authored-by: zuochenyong <zuochenyong@wonderable.ai>
Co-committed-by: zuochenyong <zuochenyong@wonderable.ai>
2026-07-30 16:32:32 +08:00
zuochenyong aa1a1240f2 fix(push): 完善厂商推送排障日志 (#202)
Co-authored-by: exinglang <exinglang@qq.com>
Reviewed-on: #202
Co-authored-by: zuochenyong <zuochenyong@wonderable.ai>
Co-committed-by: zuochenyong <zuochenyong@wonderable.ai>
2026-07-30 14:05:05 +08:00
guke b6ddb275f4 fix(comparison): 比价 best 无 is_best 时兜底回落「含源最低价」 (#201)
背景
_derive_from_platforms 是比价记录的单一真相源派生函数(pricebot done 帧带 platforms 时走它),正常路径直接取 platforms 里 is_best=true 的那行当最低价。

线上出现一类帧(如记录 id 3304):全平台 has_dish_diff(菜品「相似替换 / 价格仅供参考」),pricebot 认为无法认定权威最低价,于是一个 is_best 都不标。此时旧逻辑 best=None,导致 best_platform_id / best_price_cents / saved_amount_cents / is_source_best 整条落 NULL,连锁反应:

首页比价价显示 0.00
记录页无最低价红框
省额丢失、「累计发现可省」漏计
方案
无 is_best 时兜底:在有价行里取最低价当参考 best。

关键设计点 —— 候选池含源(而非仅目标):源平台常年全菜、价可信。若源本身最便宜(其余都是更贵的相似替换),则 best 回落到源、saved=0、is_source_best=True。这与仓库老派生函数 _derive 的既有语义(「全目标缺菜 → 回落源、不虚报省」)完全一致,三个派生函数行为对齐。

⚠️ 若像分支首个提交那样排除源、强选最低目标,当源最便宜时会选中更贵目标 → saved 变负,倒扣 get_stats 的「累计发现可省」(该聚合按 status='success' 求和、不带 >0 过滤)。第二个提交据此修正为含源。

影响面 / 兼容性
只影响「带 platforms 且无任何 is_best」这一条兜底分支;正常有 is_best 的路径不变。
老客户端不带 platforms → 走 _derive,不受影响。
纯派生逻辑,无 schema / 无迁移,回滚成本低。

---------

Co-authored-by: guke <guke@autohome.com.cn>
Reviewed-on: #201
2026-07-30 11:00:35 +08:00
zuochenyong ebacf01742 feat(新手引导): 支持十圈进度与逐圈发奖 (#200)
Co-authored-by: exinglang <exinglang@qq.com>
Reviewed-on: #200
Co-authored-by: zuochenyong <zuochenyong@wonderable.ai>
Co-committed-by: zuochenyong <zuochenyong@wonderable.ai>
2026-07-30 10:49:24 +08:00
zuochenyong 89f266419b fix(push): 完善厂商推送参数与消息通知 (#199)
Co-authored-by: exinglang <exinglang@qq.com>
Reviewed-on: #199
Co-authored-by: zuochenyong <zuochenyong@wonderable.ai>
Co-committed-by: zuochenyong <zuochenyong@wonderable.ai>
2026-07-30 10:49:08 +08:00
zuochenyong 675c7ecf81 fix(push): 完善厂商推送排障日志 (#198)
Co-authored-by: exinglang <exinglang@qq.com>
Reviewed-on: #198
Co-authored-by: zuochenyong <zuochenyong@wonderable.ai>
Co-committed-by: zuochenyong <zuochenyong@wonderable.ai>
2026-07-30 10:41:43 +08:00
23 changed files with 1320 additions and 128 deletions
+16 -1
View File
@@ -69,8 +69,10 @@ CHUANGLAN_SMS_TIMEOUT_SEC=10
# 配一个固定测试手机号,专供无 SIM 卡 / 不走一键登录时打通全流程:该号登录【免短信验证码】
# (real 模式下也跳过校验)、每次登录【都重走新手引导】,并有【每日登录上限】防被人猜到号后脚本刷。
# 逻辑见 app/core/test_account.py,与其他业务解耦。
# ⚠️ 留空 = 关闭整功能(生产默认);要启用才填号(如 11111111111)。改完重启生效,随时可清空停用
# ⚠️ 两项都留空 = 关闭整功能(生产默认)。多账号用英文逗号分隔,改完重启生效
TEST_ACCOUNT_PHONE=
# 推荐新配置;例如联调环境填 11111111111,22222222222。
TEST_ACCOUNT_PHONES=
# 该测试号每日最多登录次数,当日超过即拒绝(429),次日归零。
TEST_ACCOUNT_DAILY_LIMIT=500
@@ -175,3 +177,16 @@ PANGLE_REPORT_SITE_ID_TEST=5832303
# APPLOG_MAX_BATCH=500 # 单批最大条数(超 → 422;导入期常量,改需重启)
# APPLOG_MAX_BODY_BYTES=2097152 # 请求体上限 2MB(超 → 413;运行期可调)
# APPLOG_MAX_MSG_BYTES=8192 # 单条 msg 超此字节数截断
# ===== 荣耀 Push =====
HONOR_PUSH_APP_ID=
HONOR_PUSH_CLIENT_ID=
HONOR_PUSH_CLIENT_SECRET=
# 0=正式消息(默认);1=测试消息(仅开发联调,生产必须保持 0)
HONOR_PUSH_TARGET_USER_TYPE=0
# ===== 华为 Push =====
HUAWEI_PUSH_APP_ID=
HUAWEI_PUSH_APP_SECRET=
# 0=正式消息(默认);1=测试消息(仅开发联调,生产必须保持 0)
HUAWEI_PUSH_TARGET_USER_TYPE=0
@@ -0,0 +1,89 @@
"""isolate vendor push binding across user accounts
Revision ID: push_binding_isolation
Revises: guide_video_ten_circle_v2
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
revision = "push_binding_isolation"
down_revision = "guide_video_ten_circle_v2"
branch_labels = None
depends_on = None
def upgrade() -> None:
with op.batch_alter_table("device_liveness") as batch_op:
batch_op.add_column(sa.Column("push_binding_id", sa.String(length=128), nullable=True))
batch_op.add_column(
sa.Column(
"push_binding_revoked",
sa.Boolean(),
nullable=False,
server_default=sa.false(),
)
)
batch_op.create_index(
"ix_device_liveness_push_binding_id",
["push_binding_id"],
unique=False,
)
connection = op.get_bind()
device = sa.table(
"device_liveness",
sa.column("id", sa.Integer),
sa.column("push_vendor", sa.String),
sa.column("push_token", sa.String),
sa.column("updated_at", sa.DateTime),
)
connection.execute(
device.update()
.where(
sa.or_(
device.c.push_vendor == "",
device.c.push_token == "",
)
)
.values(push_vendor=None, push_token=None)
)
rows = connection.execute(
sa.select(device.c.id, device.c.push_vendor, device.c.push_token)
.where(
device.c.push_vendor.is_not(None),
device.c.push_token.is_not(None),
)
.order_by(device.c.updated_at.desc(), device.c.id.desc())
)
seen: set[tuple[str, str]] = set()
duplicate_ids: list[int] = []
for row in rows:
key = (row.push_vendor, row.push_token)
if key in seen:
duplicate_ids.append(row.id)
else:
seen.add(key)
if duplicate_ids:
connection.execute(
device.update()
.where(device.c.id.in_(duplicate_ids))
.values(push_vendor=None, push_token=None)
)
with op.batch_alter_table("device_liveness") as batch_op:
batch_op.create_unique_constraint(
"uq_device_liveness_vendor_token",
["push_vendor", "push_token"],
)
def downgrade() -> None:
with op.batch_alter_table("device_liveness") as batch_op:
batch_op.drop_constraint("uq_device_liveness_vendor_token", type_="unique")
batch_op.drop_index("ix_device_liveness_push_binding_id")
batch_op.drop_column("push_binding_revoked")
batch_op.drop_column("push_binding_id")
+2 -2
View File
@@ -71,8 +71,8 @@ admin_app = FastAPI(
# admin 前端独立部署。生产同域(nginx)无需 CORS;本地 next dev 跨域需放行开发源。
_dev_origins = [
"http://localhost:3002",
"http://127.0.0.1:3002",
"http://localhost:3001",
"http://127.0.0.1:3001",
"http://localhost:3000",
"http://127.0.0.1:3000",
]
+109 -6
View File
@@ -1,6 +1,7 @@
"""admin 反馈工单:列表筛选 + 审核采纳/拒绝(带金币发放与审计)。"""
from __future__ import annotations
import logging
from datetime import datetime
from typing import Annotated
@@ -25,6 +26,8 @@ from app.models.feedback import Feedback
from app.repositories import wallet as wallet_repo
from app.services import notification_events
logger = logging.getLogger("shagua.admin.feedback")
router = APIRouter(
prefix="/admin/api/feedbacks",
tags=["admin-feedback"],
@@ -46,10 +49,24 @@ def _approve_feedback(
*,
bulk: bool = False,
) -> FeedbackOut:
logger.info(
"feedback approve started feedback_id=%s admin_id=%s bulk=%s",
feedback_id, admin.id, bulk,
)
fb = db.get(Feedback, feedback_id, with_for_update=True)
if fb is None:
logger.warning(
"feedback approve rejected not found feedback_id=%s admin_id=%s bulk=%s",
feedback_id, admin.id, bulk,
)
raise HTTPException(status_code=404, detail="反馈不存在")
_ensure_pending(fb)
if fb.status not in {"pending", "new"}:
logger.warning(
"feedback approve rejected invalid status feedback_id=%s user_id=%s "
"admin_id=%s status=%s bulk=%s",
feedback_id, fb.user_id, admin.id, fb.status, bulk,
)
_ensure_pending(fb)
before = fb.status
mutations.review_feedback(
@@ -91,8 +108,18 @@ def _approve_feedback(
)
db.commit()
db.refresh(fb)
logger.info(
"feedback approve committed feedback_id=%s user_id=%s admin_id=%s "
"before=%s after=%s reward_coins=%s bulk=%s",
feedback_id, fb.user_id, admin.id, before, fb.status, payload.reward_coins, bulk,
)
out = FeedbackOut.model_validate(fb)
notification_events.notify_feedback_reward(db, fb)
logger.info(
"feedback approve notification dispatch returned feedback_id=%s user_id=%s "
"admin_id=%s notification_type=feedback_reward bulk=%s",
feedback_id, fb.user_id, admin.id, bulk,
)
return out
@@ -105,10 +132,24 @@ def _reject_feedback(
*,
bulk: bool = False,
) -> FeedbackOut:
logger.info(
"feedback reject started feedback_id=%s admin_id=%s bulk=%s",
feedback_id, admin.id, bulk,
)
fb = db.get(Feedback, feedback_id, with_for_update=True)
if fb is None:
logger.warning(
"feedback reject rejected not found feedback_id=%s admin_id=%s bulk=%s",
feedback_id, admin.id, bulk,
)
raise HTTPException(status_code=404, detail="反馈不存在")
_ensure_pending(fb)
if fb.status not in {"pending", "new"}:
logger.warning(
"feedback reject rejected invalid status feedback_id=%s user_id=%s "
"admin_id=%s status=%s bulk=%s",
feedback_id, fb.user_id, admin.id, fb.status, bulk,
)
_ensure_pending(fb)
before = fb.status
mutations.review_feedback(
@@ -142,8 +183,18 @@ def _reject_feedback(
)
db.commit()
db.refresh(fb)
logger.info(
"feedback reject committed feedback_id=%s user_id=%s admin_id=%s "
"before=%s after=%s bulk=%s",
feedback_id, fb.user_id, admin.id, before, fb.status, bulk,
)
out = FeedbackOut.model_validate(fb)
notification_events.notify_feedback_reply(db, fb)
logger.info(
"feedback reject notification dispatch returned feedback_id=%s user_id=%s "
"admin_id=%s notification_type=feedback_reply bulk=%s",
feedback_id, fb.user_id, admin.id, bulk,
)
return out
@@ -201,6 +252,10 @@ def bulk_approve_feedbacks(
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
) -> FeedbackBulkResult:
logger.info(
"feedback bulk approve started admin_id=%s item_count=%s",
admin.id, len(body.ids),
)
results: list[FeedbackBulkItemResult] = []
ip = get_client_ip(request)
for feedback_id in body.ids:
@@ -209,11 +264,24 @@ def bulk_approve_feedbacks(
results.append(FeedbackBulkItemResult(id=feedback_id, ok=True, status=out.status))
except HTTPException as exc:
db.rollback()
logger.warning(
"feedback bulk approve item failed feedback_id=%s admin_id=%s error=%s",
feedback_id, admin.id, exc.detail,
)
results.append(FeedbackBulkItemResult(id=feedback_id, ok=False, error=str(exc.detail)))
except Exception: # noqa: BLE001 - 单笔失败不打断整批
db.rollback()
logger.exception(
"feedback bulk approve item failed feedback_id=%s admin_id=%s",
feedback_id, admin.id,
)
results.append(FeedbackBulkItemResult(id=feedback_id, ok=False, error="系统异常"))
return _bulk_result(results)
result = _bulk_result(results)
logger.info(
"feedback bulk approve completed admin_id=%s total=%s success=%s failed=%s",
admin.id, result.total, result.success, result.failed,
)
return result
@router.post("/bulk/reject", response_model=FeedbackBulkResult, summary="批量拒绝采纳反馈")
@@ -223,6 +291,10 @@ def bulk_reject_feedbacks(
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
) -> FeedbackBulkResult:
logger.info(
"feedback bulk reject started admin_id=%s item_count=%s",
admin.id, len(body.ids),
)
results: list[FeedbackBulkItemResult] = []
ip = get_client_ip(request)
for feedback_id in body.ids:
@@ -231,11 +303,24 @@ def bulk_reject_feedbacks(
results.append(FeedbackBulkItemResult(id=feedback_id, ok=True, status=out.status))
except HTTPException as exc:
db.rollback()
logger.warning(
"feedback bulk reject item failed feedback_id=%s admin_id=%s error=%s",
feedback_id, admin.id, exc.detail,
)
results.append(FeedbackBulkItemResult(id=feedback_id, ok=False, error=str(exc.detail)))
except Exception: # noqa: BLE001 - 单笔失败不打断整批
db.rollback()
logger.exception(
"feedback bulk reject item failed feedback_id=%s admin_id=%s",
feedback_id, admin.id,
)
results.append(FeedbackBulkItemResult(id=feedback_id, ok=False, error="系统异常"))
return _bulk_result(results)
result = _bulk_result(results)
logger.info(
"feedback bulk reject completed admin_id=%s total=%s success=%s failed=%s",
admin.id, result.total, result.success, result.failed,
)
return result
@router.post("/{feedback_id}/handle", response_model=OkResponse, summary="标记反馈已处理")
@@ -258,7 +343,16 @@ def approve_feedback(
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
) -> FeedbackOut:
return _approve_feedback(db, admin, feedback_id, payload, get_client_ip(request))
try:
return _approve_feedback(db, admin, feedback_id, payload, get_client_ip(request))
except HTTPException:
raise
except Exception:
logger.exception(
"feedback approve failed feedback_id=%s admin_id=%s",
feedback_id, admin.id,
)
raise
@router.post("/{feedback_id}/reject", response_model=FeedbackOut, summary="拒绝采纳反馈")
@@ -269,4 +363,13 @@ def reject_feedback(
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
) -> FeedbackOut:
return _reject_feedback(db, admin, feedback_id, payload, get_client_ip(request))
try:
return _reject_feedback(db, admin, feedback_id, payload, get_client_ip(request))
except HTTPException:
raise
except Exception:
logger.exception(
"feedback reject failed feedback_id=%s admin_id=%s",
feedback_id, admin.id,
)
raise
+1 -1
View File
@@ -256,7 +256,7 @@ def sms_login(req: SmsLoginRequest, request: Request, db: DbSession) -> TokenWit
# 放在最前面:命中即不校验验证码;先扣当日额度,超限直接拒,挡住有人猜到号后脚本刷。
# 测试账号走自己的每日额度、不受下面 (设备+IP) 每小时限流约束(QA 需在一小时内反复登录联调)。
if test_account.is_test_account(req.phone):
if not test_account.try_consume_quota():
if not test_account.try_consume_quota(req.phone):
raise HTTPException(status_code=429, detail="测试账号今日使用次数已达上限,请明天再试")
user = user_repo.upsert_user_for_login(db, phone=req.phone, register_channel="sms")
risk_repo.record_behavior_event(
+33
View File
@@ -19,8 +19,10 @@ from app.api.deps import CurrentUser, DbSession
from app.integrations import vendor_push
from app.repositories import device as device_repo
from app.schemas.device import (
DeferredDeviceUnregisterRequest,
DeviceOut,
DeviceRegisterRequest,
DeviceUnregisterRequest,
HeartbeatRequest,
LivenessAckRequest,
LivenessOut,
@@ -78,6 +80,7 @@ def register_device(
registration_id=req.registration_id,
push_vendor=req.push_vendor,
push_token=req.push_token,
push_binding_id=req.push_binding_id,
platform=req.platform,
app_version=req.app_version,
)
@@ -106,6 +109,36 @@ def report_heartbeat(
registration_id=req.registration_id,
push_vendor=req.push_vendor,
push_token=req.push_token,
push_binding_id=req.push_binding_id,
)
return OkResponse()
@router.post("/unregister", response_model=OkResponse, summary="解绑当前用户的本机推送目标")
def unregister_device(
req: DeviceUnregisterRequest,
user: CurrentUser,
db: DbSession,
) -> OkResponse:
device_repo.unregister_push_binding(db, user_id=user.id, device_id=req.device_id)
logger.info("device unregister user_id=%d device_id=%s", user.id, req.device_id)
return OkResponse()
@router.post(
"/unregister/deferred",
response_model=OkResponse,
summary="离线退出后按登录会话凭据幂等解绑",
)
def unregister_device_deferred(
req: DeferredDeviceUnregisterRequest,
db: DbSession,
) -> OkResponse:
# 不返回是否命中,避免公开接口泄露设备绑定状态。
device_repo.unregister_push_binding_deferred(
db,
device_id=req.device_id,
push_binding_id=req.push_binding_id,
)
return OkResponse()
+20 -3
View File
@@ -79,6 +79,8 @@ class Settings(BaseSettings):
HONOR_PUSH_APP_ID: str = ""
HONOR_PUSH_CLIENT_ID: str = ""
HONOR_PUSH_CLIENT_SECRET: str = ""
# 0=正式消息(默认);1=测试消息(仅开发联调,勿用于生产)。
HONOR_PUSH_TARGET_USER_TYPE: int = Field(default=0, ge=0, le=1)
HONOR_PUSH_TOKEN_ENDPOINT: str = "https://iam.developer.honor.com/auth/token"
HONOR_PUSH_SEND_ENDPOINT_TEMPLATE: str = (
"https://push-api.cloud.honor.com/api/v1/{app_id}/sendMessage"
@@ -88,6 +90,8 @@ class Settings(BaseSettings):
# (OAuth 换 token 时 client_id 即 AppId)。发送走 v1 messages:send,成功码 80000000。
HUAWEI_PUSH_APP_ID: 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_SEND_ENDPOINT_TEMPLATE: str = (
"https://push-api.cloud.huawei.com/v1/{app_id}/messages:send"
@@ -189,14 +193,27 @@ class Settings(BaseSettings):
# (real 模式下也跳过校验)、每次登录【强制重走新手引导】,并设【每日使用次数上限】防被人
# 猜到号后脚本滥用。两个值都能随时改 .env。逻辑全在 app/core/test_account.py,与其他业务解耦。
# ⚠️ TEST_ACCOUNT_PHONE 留空 = 整个功能关闭(生产默认安全;要启用才显式填号)。
TEST_ACCOUNT_PHONE: str = "" # 测试手机号(11 位,如 11111111111);空=关闭整功能
TEST_ACCOUNT_PHONE: str = "" # 兼容旧配置:单个测试手机号
TEST_ACCOUNT_PHONES: str = "" # 多个测试手机号,英文逗号分隔
TEST_ACCOUNT_DAILY_LIMIT: int = 500 # 该测试号每日最多登录次数,当日超过即拒绝登录
@property
def test_account_phone(self) -> str:
"""规整后的测试手机号(去空白);空串=功能关闭"""
"""兼容旧调用:规整后的单个测试手机号。"""
return self.TEST_ACCOUNT_PHONE.strip()
@property
def test_account_phones(self) -> frozenset[str]:
"""所有测试手机号;新旧配置取并集,便于线上平滑迁移。"""
phones = {
phone.strip()
for phone in self.TEST_ACCOUNT_PHONES.split(",")
if phone.strip()
}
if self.test_account_phone:
phones.add(self.test_account_phone)
return frozenset(phones)
# ===== 美团联盟 CPS =====
# 未配置时所有 /api/v1/meituan/* 接口 200 返空(优雅降级),不影响登录/领券等其他业务。
MT_CPS_APP_KEY: str = ""
@@ -489,7 +506,7 @@ class Settings(BaseSettings):
return self.APP_ENV == "prod"
@model_validator(mode="after")
def _enforce_prod_secrets(self) -> "Settings":
def _enforce_prod_secrets(self) -> Settings:
"""prod 下强校验 JWT secret,弱/默认/空即启动报错(fail-fast,挡住 token 被伪造)。
只校验两个签发凭证:App 用户的 JWT_SECRET_KEY、后台的 ADMIN_JWT_SECRET——它们沿用默认值
+15 -16
View File
@@ -9,9 +9,8 @@
3. **每日使用次数上限**:防被人猜到这个号后写脚本一直刷。当天登录数超过上限即拒绝(429),
次日自动归零。
手机号与上限都在 .env 配(`TEST_ACCOUNT_PHONE` / `TEST_ACCOUNT_DAILY_LIMIT`),随时可改。
`TEST_ACCOUNT_PHONE` 留空 = 整个功能关闭(生产默认态),`is_test_account()` 对任何号都返回
False,登录/短信回到原逻辑,零影响。
手机号与上限都在 .env 配(`TEST_ACCOUNT_PHONES` / `TEST_ACCOUNT_DAILY_LIMIT`),随时可改。
兼容旧的单号配置 `TEST_ACCOUNT_PHONE`;两项都留空 = 整个功能关闭(生产默认态)
计数存**进程内存**(单 worker 够用,与 sms.py 同款约定):重启清零、多 worker 不共享。作为
一个测试号的粗粒度防滥用闸够用;且因所有登录都落同一个 phone → 同一个 user,滥用面天然只
@@ -27,9 +26,9 @@ from app.core.config import settings
logger = logging.getLogger("shagua.test_account")
# 进程内每日计数:(date_str, 当日已登录次数)。单 worker 有效,重启清零(见模块 docstring)
# 进程内每日计数:{手机号: (date_str, 当日已登录次数)}。单 worker 有效,重启清零。
_lock = Lock()
_usage: tuple[str, int] = ("", 0)
_usage: dict[str, tuple[str, int]] = {}
def _today() -> str:
@@ -37,16 +36,16 @@ def _today() -> str:
def is_enabled() -> bool:
"""功能总开关:配了 TEST_ACCOUNT_PHONE 才启用(空=关闭)"""
return bool(settings.test_account_phone)
"""功能总开关:至少配置了一个测试手机号才启用"""
return bool(settings.test_account_phones)
def is_test_account(phone: str) -> bool:
"""该手机号是否为配置的测试账号。功能关闭时对任何号都返回 False"""
return is_enabled() and phone == settings.test_account_phone
"""该手机号是否在测试账号集合中"""
return phone in settings.test_account_phones
def try_consume_quota() -> bool:
def try_consume_quota(phone: str) -> bool:
"""测试账号登录时调:当日计数 +1。
Returns:
@@ -59,17 +58,17 @@ def try_consume_quota() -> bool:
limit = settings.TEST_ACCOUNT_DAILY_LIMIT
with _lock:
today = _today()
day, cnt = _usage
day, cnt = _usage.get(phone, ("", 0))
if day != today: # 跨天归零
cnt = 0
if cnt >= limit:
_usage = (today, cnt) # 已满,保持不变
_usage[phone] = (today, cnt) # 已满,保持不变
logger.warning(
"测试账号 %s 今日登录数已达上限 %d,拒绝", settings.test_account_phone, limit
"测试账号 %s 今日登录数已达上限 %d,拒绝", phone, limit
)
return False
_usage = (today, cnt + 1)
logger.info("测试账号 %s%d/%d 次登录", settings.test_account_phone, cnt + 1, limit)
_usage[phone] = (today, cnt + 1)
logger.info("测试账号 %s%d/%d 次登录", phone, cnt + 1, limit)
return True
@@ -77,4 +76,4 @@ def _reset_for_test() -> None:
"""仅供单测:清空进程内计数,隔离用例间状态。"""
global _usage
with _lock:
_usage = ("", 0)
_usage = {}
+1 -1
View File
@@ -35,7 +35,7 @@ _GC_THRESHOLD = 10000 # 超此阈值,send 时顺手清老于
# 发码错误码 → (HTTP 码, 用户提示)。未列出的一律 503(供应商不可用)。
_SEND_ERRORS: dict[str, tuple[int, str]] = {
"MOBILE_NUMBER_ILLEGAL": (400, "手机号无效"),
"MOBILE_NUMBER_ILLEGAL": (400, "请输入有效的手机号"),
"BUSINESS_LIMIT_CONTROL": (429, "今日发送次数过多,请明天再试"),
"FREQUENCY_FAIL": (429, "发送过于频繁,请稍后再试"),
}
+1 -1
View File
@@ -53,7 +53,7 @@ _GC_THRESHOLD = 10000 # 任一内存 dict 超此阈值,send 时顺手清过期
# 发码错误码(创蓝 `code`)→ (HTTP 码, 用户提示)。未列出的一律 503(供应商不可用)。
_SEND_ERRORS: dict[str, tuple[int, str]] = {
"103": (429, "发送过于频繁,请稍后再试"), # 提交速度过快
"107": (400, "手机号无效"), # 手机号码错误
"107": (400, "请输入有效的手机号"), # 手机号码错误
}
# 需运维介入的配置/开通/余额类错误:打 critical 日志(仍归 503)。
_SEND_CRITICAL_CODES = frozenset({
+1 -1
View File
@@ -185,5 +185,5 @@ def _send_via_jiguang(phone: str, code: str) -> None:
if ecode == 50009: # 极光侧超频
raise SmsError("发送过于频繁,请稍后再试", status_code=429)
if ecode == 50006: # 手机号无效(schema 已挡格式,这里多是空号/停机)
raise SmsError("手机号无效", status_code=400)
raise SmsError("请输入有效的手机号", status_code=400)
raise SmsError(f"短信发送失败(code={ecode})", status_code=503)
+459 -47
View File
@@ -22,7 +22,7 @@ import uuid
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any
from urllib.parse import quote
from urllib.parse import quote, urlsplit
import httpx
@@ -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 → 中文名(测试/配置状态接口展示用)
@@ -69,6 +70,18 @@ class _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:
if not push_vendor:
@@ -115,8 +128,16 @@ def send_notification(
if mock:
logger.info(
"[mock push] vendor=%s token=%s... title=%s body=%s extras=%s",
vendor, token[:12], title, body, extras,
"vendor push mock vendor=%s request=%s",
vendor,
_log_summary(
{
"push_token": token,
"title": title,
"body": body,
"extras": extras,
}
),
)
return {
"mock": True,
@@ -133,7 +154,28 @@ def send_notification(
"xiaomi": _send_xiaomi,
"oppo": _send_oppo,
}
return dispatch[vendor](token, title, body, extras)
started = time.perf_counter()
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(
@@ -153,19 +195,191 @@ 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"}
# 小米服务端不再发送透传消息;通知栏消息仍照常发送,前台角标由客户端主动刷新。
if vendor == "xiaomi":
return {"skipped": True, "reason": "xiaomi data messages disabled"}
# 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,
}
return dispatch[vendor](token, payload)
def _require(value: str, name: str) -> str:
if not value:
raise VendorPushError(f"{name} not configured")
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 _vendor_response_failed(vendor: str, data: dict[str, Any]) -> bool:
"""识别 HTTP 200 中明确的厂商业务失败,避免将失败请求记录成 success。"""
if data.get("error") or data.get("success") is False:
return True
if vendor == "xiaomi":
code = data.get("code")
result = str(data.get("result", "ok")).lower()
return code not in (0, "0", None) or result not in ("ok", "success")
if vendor == "oppo" and data.get("code") is not None:
return int(data["code"]) != 0
if vendor == "vivo" and data.get("result") is not None:
return int(data["result"]) != 0
if vendor == "honor" and data.get("code") is not None:
return int(data["code"]) != 200
if vendor == "huawei" and data.get("code") is not None:
return str(data["code"]) != "80000000"
return False
def _request_json(
method: str,
url: str,
*,
vendor: str,
operation: str,
expected_status: tuple[int, ...] = (200,),
**kwargs: Any,
) -> dict[str, Any]:
started = time.perf_counter()
request_summary = _request_summary(kwargs)
try:
resp = httpx.request(
method,
@@ -174,41 +388,85 @@ def _request_json(
**kwargs,
)
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
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])
logger.error(
"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}")
try:
return resp.json()
data = resp.json()
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
vendor_failed = _vendor_response_failed(vendor, data)
log = logger.warning if vendor_failed else logger.info
log(
"vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
"outcome=%s http_status=%s request=%s response=%s elapsed_ms=%.1f",
vendor,
operation,
method,
_endpoint_for_log(url),
"vendor_error" if vendor_failed else "success",
resp.status_code,
request_summary,
_response_summary(resp, data),
_elapsed_ms(started),
)
return data
def _request_form(
method: str,
url: str,
*,
vendor: str,
operation: str,
expected_status: tuple[int, ...] = (200,),
**kwargs: Any,
) -> dict[str, Any]:
try:
resp = httpx.request(
method,
url,
timeout=settings.PUSH_REQUEST_TIMEOUT_SEC,
**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
return _request_json(
method,
url,
vendor=vendor,
operation=operation,
expected_status=expected_status,
**kwargs,
)
def _cache_get(key: str) -> str | None:
@@ -234,6 +492,8 @@ def _honor_access_token() -> str:
data = _request_form(
"POST",
settings.HONOR_PUSH_TOKEN_ENDPOINT,
vendor="honor",
operation="authenticate",
data={
"grant_type": "client_credentials",
"client_id": client_id,
@@ -243,25 +503,31 @@ def _honor_access_token() -> str:
)
token = data.get("access_token")
if not token:
raise VendorPushError(f"honor auth failed: {data}")
raise VendorPushError(f"honor auth failed: {_raw_log_summary(data)}")
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]:
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": {
"ttl": f"{settings.PUSH_TIME_TO_LIVE_SEC}s",
"targetUserType": 1,
"targetUserType": settings.HONOR_PUSH_TARGET_USER_TYPE,
"notification": {
"title": title,
"body": body,
"clickAction": {"type": 3},
"clickAction": click_action,
"importance": "NORMAL",
},
},
@@ -270,6 +536,8 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
data = _request_json(
"POST",
settings.HONOR_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
vendor="honor",
operation="send_notification",
json=payload,
headers={
"Content-Type": "application/json; charset=UTF-8",
@@ -279,7 +547,28 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
)
code = data.get("code")
if code is not None and int(code) != 200:
raise VendorPushError(f"honor push failed: {data}")
raise VendorPushError(f"honor push failed: {_raw_log_summary(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
@@ -294,6 +583,8 @@ def _huawei_access_token() -> str:
data = _request_form(
"POST",
settings.HUAWEI_PUSH_TOKEN_ENDPOINT,
vendor="huawei",
operation="authenticate",
data={
"grant_type": "client_credentials",
"client_id": app_id,
@@ -303,7 +594,7 @@ def _huawei_access_token() -> str:
)
token = data.get("access_token")
if not token:
raise VendorPushError(f"huawei auth failed: {data}")
raise VendorPushError(f"huawei auth failed: {_raw_log_summary(data)}")
return _cache_put(cache_key, str(token), data.get("expires_in"))
@@ -312,18 +603,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",
},
},
@@ -333,6 +630,8 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d
data = _request_json(
"POST",
settings.HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
vendor="huawei",
operation="send_notification",
json=payload,
headers={
"Content-Type": "application/json; charset=UTF-8",
@@ -340,7 +639,32 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d
},
)
if str(data.get("code", "")) != "80000000":
raise VendorPushError(f"huawei push failed: {data}")
raise VendorPushError(f"huawei push failed: {_raw_log_summary(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
@@ -357,6 +681,8 @@ def _vivo_auth_token() -> str:
data = _request_json(
"POST",
settings.VIVO_PUSH_AUTH_ENDPOINT,
vendor="vivo",
operation="authenticate",
json={
"appId": app_id,
"appKey": app_key,
@@ -366,10 +692,10 @@ def _vivo_auth_token() -> str:
headers={"Content-Type": "application/json"},
)
if int(data.get("result", -1)) != 0:
raise VendorPushError(f"vivo auth failed: {data}")
raise VendorPushError(f"vivo auth failed: {_raw_log_summary(data)}")
token = data.get("authToken")
if not token:
raise VendorPushError(f"vivo auth missing authToken: {data}")
raise VendorPushError(f"vivo auth missing authToken: {_raw_log_summary(data)}")
return _cache_put(cache_key, str(token), 24 * 3600)
@@ -385,15 +711,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. extrasOriginOS 才会把业务参数
# 原样交给 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:
@@ -401,14 +734,37 @@ def _send_vivo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
data = _request_json(
"POST",
settings.VIVO_PUSH_SEND_ENDPOINT,
vendor="vivo",
operation="send_notification",
json=payload,
headers={
"Content-Type": "application/json",
"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:
raise VendorPushError(f"vivo push failed: {data}")
raise VendorPushError(f"vivo push failed: {_raw_log_summary(data)}")
return data
@@ -446,14 +802,43 @@ def _send_xiaomi(token: str, title: str, body: str, extras: dict[str, str]) -> d
data = _request_form(
"POST",
settings.XIAOMI_PUSH_SEND_ENDPOINT,
vendor="xiaomi",
operation="send_notification",
data=form,
headers={"Authorization": f"key={app_secret}"},
)
code = data.get("code")
if code not in (0, "0", None):
raise VendorPushError(f"xiaomi push failed: {data}")
raise VendorPushError(f"xiaomi push failed: {_raw_log_summary(data)}")
if str(data.get("result", "ok")).lower() not in ("ok", "success"):
raise VendorPushError(f"xiaomi push failed: {data}")
raise VendorPushError(f"xiaomi push failed: {_raw_log_summary(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")
form = {
"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),
}
if settings.XIAOMI_PUSH_CHANNEL_ID.strip():
form["extra.channel_id"] = settings.XIAOMI_PUSH_CHANNEL_ID.strip()
data = _request_form(
"POST",
settings.XIAOMI_PUSH_SEND_ENDPOINT,
vendor="xiaomi",
operation="send_data_event",
data=form,
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
@@ -490,6 +875,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
@@ -522,6 +925,8 @@ def _oppo_auth_token() -> str:
data = _request_form(
"POST",
settings.OPPO_PUSH_AUTH_ENDPOINT,
vendor="oppo",
operation="authenticate",
data={
"app_key": app_key,
"timestamp": timestamp,
@@ -530,10 +935,10 @@ def _oppo_auth_token() -> str:
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
if int(data.get("code", -1)) != 0:
raise VendorPushError(f"oppo auth failed: {data}")
raise VendorPushError(f"oppo auth failed: {_raw_log_summary(data)}")
token = (data.get("data") or {}).get("auth_token") or data.get("auth_token")
if not token:
raise VendorPushError(f"oppo auth missing auth_token: {data}")
raise VendorPushError(f"oppo auth missing auth_token: {_raw_log_summary(data)}")
return _cache_put(cache_key, str(token), 24 * 3600)
@@ -548,6 +953,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 同款坑)。
@@ -573,6 +983,8 @@ def _send_oppo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
data = _request_form(
"POST",
settings.OPPO_PUSH_SEND_ENDPOINT,
vendor="oppo",
operation="send_notification",
data={
"auth_token": auth_token,
"message": json.dumps(message, ensure_ascii=False),
@@ -580,5 +992,5 @@ def _send_oppo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
if int(data.get("code", -1)) != 0:
raise VendorPushError(f"oppo push failed: {data}")
raise VendorPushError(f"oppo push failed: {_raw_log_summary(data)}")
return data
+10
View File
@@ -34,6 +34,11 @@ class DeviceLiveness(Base):
__tablename__ = "device_liveness"
__table_args__ = (
UniqueConstraint("user_id", "device_id", name="uq_device_liveness_user_device"),
UniqueConstraint(
"push_vendor",
"push_token",
name="uq_device_liveness_vendor_token",
),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
@@ -48,6 +53,11 @@ class DeviceLiveness(Base):
push_vendor: Mapped[str | None] = mapped_column(String(32), nullable=True)
# 厂商 push token / regId / registration_id;不同厂商命名不同,后端统一存这里。
push_token: Mapped[str | None] = mapped_column(String(256), nullable=True)
# 客户端每次登录生成的高熵随机值。离线退出后只凭此值精确撤销旧会话绑定,
# 新账号/新会话注册时会替换它,故旧解绑任务绝不能误解绑新登录。
push_binding_id: Mapped[str | None] = mapped_column(String(128), index=True, nullable=True)
# 被退出/转移的 binding 留作墓碑,阻止旧的在途 register/heartbeat 把 token 抢回。
push_binding_revoked: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
platform: Mapped[str] = mapped_column(String(16), nullable=False, default="android")
app_version: Mapped[str | None] = mapped_column(String(32), nullable=True)
+10
View File
@@ -375,6 +375,16 @@ def _derive_from_platforms(
rows = [p for p in (platforms or []) if isinstance(p, dict)]
src = next((p for p in rows if p.get("role") == "source"), None)
best = next((p for p in rows if p.get("is_best")), None)
if best is None:
# pricebot 没标 is_best(如全平台 has_dish_diff「相似替换/仅供参考」→ 不认定权威最低价)
# 但仍有有价行 → 兜底取有价行里最低价当参考 best,避免记录级 best_*/saved 整条落 NULL
# (否则首页价 0.00 / 记录页无最低红框 / 省额丢失 /「累计发现可省」漏计)。**候选含源**:
# 源常年全菜、价可信,源本身最便宜时 best 回落源(saved=0、is_source_best=True),与 _derive
# 「全目标缺菜回落源、不虚报省」同一语义;若排除源强选更贵目标,saved 会变负、倒扣「累计
# 发现可省」(get_stats 对 saved_amount_cents 求和不带 >0 过滤)。
priced = [p for p in rows if p.get("price") is not None]
if priced:
best = min(priced, key=lambda p: (p["price"], p.get("display_order") or 0))
source_price_cents = _yuan_to_cents(src.get("price")) if src else None
best_price_cents = _yuan_to_cents(best.get("price")) if best else None
saved_amount_cents = None
+117 -14
View File
@@ -1,9 +1,9 @@
"""device 表读写(设备注册 / 心跳 / 超时扫描)。"""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from datetime import UTC, datetime, timedelta
from sqlalchemy import select
from sqlalchemy import and_, or_, select
from sqlalchemy.orm import Session
from app.models.device import DeviceLiveness
@@ -24,13 +24,29 @@ def register_or_update(
registration_id: str | None = None,
push_vendor: str | None = None,
push_token: str | None = None,
push_binding_id: str | None = None,
platform: str = "android",
app_version: str | None = None,
) -> DeviceLiveness:
"""注册设备或更新其厂商 push token / 元信息。upsert by (user_id, device_id)"""
"""注册设备或更新推送目标;同一设备/token 的旧用户绑定在同一事务内失效"""
normalized_vendor = _normalize_push_vendor(push_vendor)
normalized_token = push_token.strip() if push_token else None
normalized_binding = push_binding_id.strip() if push_binding_id else None
device = _get(db, user_id=user_id, device_id=device_id)
stale_binding = bool(
device
and normalized_binding
and device.push_binding_id == normalized_binding
and device.push_binding_revoked
)
if not stale_binding:
_release_conflicting_push_bindings(
db,
user_id=user_id,
device_id=device_id,
push_vendor=normalized_vendor,
push_token=normalized_token,
)
if device is None:
device = DeviceLiveness(
user_id=user_id,
@@ -38,17 +54,22 @@ def register_or_update(
registration_id=registration_id,
push_vendor=normalized_vendor,
push_token=normalized_token,
push_binding_id=normalized_binding,
push_binding_revoked=False,
platform=platform or "android",
app_version=app_version,
)
db.add(device)
else:
if registration_id:
if registration_id and not stale_binding:
device.registration_id = registration_id
if normalized_vendor:
if normalized_vendor and not stale_binding:
device.push_vendor = normalized_vendor
if normalized_token:
if normalized_token and not stale_binding:
device.push_token = normalized_token
if normalized_binding and not stale_binding:
device.push_binding_id = normalized_binding
device.push_binding_revoked = False
if platform:
device.platform = platform
if app_version:
@@ -67,26 +88,45 @@ def touch_heartbeat(
registration_id: str | None = None,
push_vendor: str | None = None,
push_token: str | None = None,
push_binding_id: str | None = None,
) -> DeviceLiveness:
"""处理一次心跳(心跳也能自注册)。
service 心跳或 accessibility_enabled=true ,刷新存活并把状态机重置回 alive
清掉 notified_at(掉线恢复 下次再断才会再推一条)
"""
now = datetime.now(timezone.utc)
now = datetime.now(UTC)
device = _get(db, user_id=user_id, device_id=device_id)
normalized_vendor = _normalize_push_vendor(push_vendor)
normalized_token = push_token.strip() if push_token else None
normalized_binding = push_binding_id.strip() if push_binding_id else None
stale_binding = bool(
device
and normalized_binding
and device.push_binding_id == normalized_binding
and device.push_binding_revoked
)
if (normalized_token or normalized_binding) and not stale_binding:
_release_conflicting_push_bindings(
db,
user_id=user_id,
device_id=device_id,
push_vendor=normalized_vendor,
push_token=normalized_token,
)
if device is None:
device = DeviceLiveness(user_id=user_id, device_id=device_id)
db.add(device)
if registration_id:
if registration_id and not stale_binding:
device.registration_id = registration_id
normalized_vendor = _normalize_push_vendor(push_vendor)
normalized_token = push_token.strip() if push_token else None
if normalized_vendor:
if normalized_vendor and not stale_binding:
device.push_vendor = normalized_vendor
if normalized_token:
if normalized_token and not stale_binding:
device.push_token = normalized_token
if normalized_binding and not stale_binding:
device.push_binding_id = normalized_binding
device.push_binding_revoked = False
device.last_report_protection_on = accessibility_enabled
if accessibility_enabled:
@@ -107,7 +147,7 @@ def list_overdue(db: Session, *, timeout_minutes: int) -> list[DeviceLiveness]:
即使没有厂商 token 也要检出,后续由 kill_alert_pending 走客户端进 App 后兜底提醒
"""
cutoff = datetime.now(timezone.utc) - timedelta(minutes=timeout_minutes)
cutoff = datetime.now(UTC) - timedelta(minutes=timeout_minutes)
stmt = select(DeviceLiveness).where(
DeviceLiveness.ever_protected.is_(True),
DeviceLiveness.liveness_state == "alive",
@@ -126,7 +166,7 @@ def mark_notified(db: Session, *, device_id_pk: int) -> None:
device = db.get(DeviceLiveness, device_id_pk)
if device is not None:
device.liveness_state = "notified"
device.notified_at = datetime.now(timezone.utc)
device.notified_at = datetime.now(UTC)
device.kill_alert_pending = True
db.commit()
@@ -177,6 +217,69 @@ def list_push_targets(db: Session, *, user_id: int) -> list[DeviceLiveness]:
return targets
def unregister_push_binding(db: Session, *, user_id: int, device_id: str) -> None:
"""当前用户主动退出:清推送归属但保留设备存活历史。"""
device = _get(db, user_id=user_id, device_id=device_id)
if device is not None:
_clear_push_binding(device)
db.commit()
def unregister_push_binding_deferred(
db: Session, *, device_id: str, push_binding_id: str
) -> None:
"""离线退出补偿;只撤销完全匹配的旧登录会话,且始终幂等。"""
stmt = select(DeviceLiveness).where(
DeviceLiveness.device_id == device_id,
DeviceLiveness.push_binding_id == push_binding_id,
)
device = db.execute(stmt).scalar_one_or_none()
if device is not None:
_clear_push_binding(device)
db.commit()
def _release_conflicting_push_bindings(
db: Session,
*,
user_id: int,
device_id: str,
push_vendor: str | None,
push_token: str | None,
) -> None:
"""把设备或 token 从其他记录转移走,当前 (user,device) 行除外。"""
conflicts = [DeviceLiveness.device_id == device_id]
if push_vendor and push_token:
conflicts.append(
and_(
DeviceLiveness.push_vendor == push_vendor,
DeviceLiveness.push_token == push_token,
)
)
stmt = select(DeviceLiveness).where(
or_(*conflicts),
~and_(
DeviceLiveness.user_id == user_id,
DeviceLiveness.device_id == device_id,
),
)
released = False
for old in db.execute(stmt).scalars():
_clear_push_binding(old)
released = True
if released:
# 唯一约束下必须先落旧行清理,再把同一 token 写给新行。
db.flush()
def _clear_push_binding(device: DeviceLiveness) -> None:
device.registration_id = None
device.push_vendor = None
device.push_token = None
# binding_id 留作撤销墓碑;同一旧会话的在途请求不能重新认领 token。
device.push_binding_revoked = bool(device.push_binding_id)
def _normalize_push_vendor(push_vendor: str | None) -> str | None:
if not push_vendor:
return None
+15 -3
View File
@@ -7,9 +7,19 @@
"""
from __future__ import annotations
import re
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
from pydantic import BaseModel, ConfigDict, Field, field_validator
from app.core.config import settings
def _validate_login_phone(phone: str) -> str:
"""普通号保持大陆手机号格式;仅显式配置的测试号允许例外。"""
if re.fullmatch(r"^1\d{10}$", phone) or phone in settings.test_account_phones:
return phone
raise ValueError("invalid phone")
# ===== 用户对外信息 =====
@@ -72,7 +82,7 @@ class JverifyLoginRequest(BaseModel):
# ===== 短信验证码 =====
class SmsSendRequest(BaseModel):
phone: str = Field(..., min_length=11, max_length=11, pattern=r"^1\d{10}$")
phone: str = Field(..., min_length=11, max_length=11)
device_id: str = Field(
"", max_length=64,
description="硬件级设备标识(Android ANDROID_ID),用于发码防刷按 设备+IP 限流;空=按 IP 聚一桶",
@@ -80,6 +90,7 @@ class SmsSendRequest(BaseModel):
device_model: str = Field(
"", max_length=128, description="客户端设备型号快照,用于短信安全审计"
)
_valid_phone = field_validator("phone")(_validate_login_phone)
class SmsSendResponse(BaseModel):
@@ -89,7 +100,7 @@ class SmsSendResponse(BaseModel):
class SmsLoginRequest(BaseModel):
phone: str = Field(..., min_length=11, max_length=11, pattern=r"^1\d{10}$")
phone: str = Field(..., min_length=11, max_length=11)
code: str = Field(..., min_length=4, max_length=8)
device_id: str = Field(
"", max_length=64,
@@ -98,6 +109,7 @@ class SmsLoginRequest(BaseModel):
device_model: str = Field(
"", max_length=128, description="客户端设备型号快照,用于短信验证安全审计"
)
_valid_phone = field_validator("phone")(_validate_login_phone)
# ===== Refresh =====
+11
View File
@@ -12,6 +12,7 @@ class DeviceRegisterRequest(BaseModel):
registration_id: str | None = None
push_vendor: str | None = None
push_token: str | None = None
push_binding_id: str | None = Field(default=None, max_length=128)
platform: str = "android"
app_version: str | None = None
@@ -23,6 +24,7 @@ class HeartbeatRequest(BaseModel):
registration_id: str | None = None
push_vendor: str | None = None
push_token: str | None = None
push_binding_id: str | None = Field(default=None, max_length=128)
class DeviceOut(BaseModel):
@@ -33,6 +35,7 @@ class DeviceOut(BaseModel):
registration_id: str | None
push_vendor: str | None
push_token: str | None
push_binding_id: str | None
ever_protected: bool
liveness_state: str
last_heartbeat_at: datetime | None
@@ -43,6 +46,14 @@ class OkResponse(BaseModel):
ok: bool = True
class DeviceUnregisterRequest(BaseModel):
device_id: str = Field(min_length=1, max_length=128)
class DeferredDeviceUnregisterRequest(DeviceUnregisterRequest):
push_binding_id: str = Field(min_length=32, max_length=128)
class LivenessOut(BaseModel):
"""本机掉线告警状态(后置检测 pull 版)。客户端只需这一个布尔判断要不要弹「开启自启动」引导,
故只返回 kill_alert_pending(不暴露设备详情 / 内部 liveness_state )从未注册过 默认 False(无告警)"""
+143 -19
View File
@@ -83,8 +83,13 @@ def _dispatch(
extra: dict[str, str] | None = None,
dedup_key: str | None = None,
push_vars: dict[str, str] | None = None,
push_token_contains: str | None = None,
) -> Notification | None:
"""落一条站内消息并向该用户设备直推。返回落库行;去重命中/失败返回 None。"""
logger.info(
"notification dispatch started user_id=%s type=%s dedup_key=%s",
user_id, type_key, dedup_key,
)
try:
row = notif_repo.create_notification(
db,
@@ -111,11 +116,20 @@ def _dispatch(
logger.exception("rollback after notification failure also failed")
return None
_push_to_user_devices(db, row, push_vars)
logger.info(
"notification created user_id=%s type=%s notification_id=%s dedup_key=%s",
user_id, type_key, row.id, dedup_key,
)
_push_to_user_devices(db, row, push_vars, push_token_contains)
return row
def _push_to_user_devices(db: Session, row: Notification, push_vars: dict[str, str] | None) -> None:
def _push_to_user_devices(
db: Session,
row: Notification,
push_vars: dict[str, str] | None,
push_token_contains: str | None = None,
) -> None:
"""向消息归属用户的全部厂商推送目标直推(best-effort,单设备失败不影响其余)。"""
try:
title, body = catalog.render_push(row.type, push_vars)
@@ -124,29 +138,103 @@ def _push_to_user_devices(db: Session, row: Notification, push_vars: dict[str, s
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):
targets = device_repo.list_push_targets(db, user_id=row.user_id)
if push_token_contains:
normalized_filter = push_token_contains.casefold()
targets = [
target
for target in targets
if normalized_filter in (target.push_token or "").casefold()
]
logger.info(
"push targets resolved user_id=%s type=%s notification_id=%s "
"token_filter=%s target_count=%s",
row.user_id, row.type, row.id, bool(push_token_contains), len(targets),
)
if not targets:
logger.warning(
"push skipped no targets user_id=%s type=%s notification_id=%s",
row.user_id, row.type, row.id,
)
return
sent = failed = skipped = data_sent = data_failed = 0
for dev in targets:
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,
skipped += 1
logger.warning(
"push target skipped unsupported vendor user_id=%s type=%s "
"notification_id=%s device_id=%s raw_vendor=%s normalized_vendor=%s",
row.user_id, row.type, row.id, dev.device_id, dev.push_vendor, vendor,
)
continue
missing = vendor_push.missing_settings(vendor)
if missing:
skipped += 1
# 本地/测试环境凭据不齐 → 只落站内消息,不发真推送(与 push/vendors 的报缺口径一致)
logger.warning(
"push target skipped vendor not configured user_id=%s type=%s "
"notification_id=%s device_id=%s vendor=%s missing_settings=%s",
row.user_id, row.type, row.id, dev.device_id, vendor, missing,
)
continue
logger.info(
"push send started user_id=%s type=%s notification_id=%s "
"device_id=%s vendor=%s",
row.user_id, row.type, row.id, dev.device_id, vendor,
)
try:
vendor_push.send_notification(
response = vendor_push.send_notification(
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(
"push sent user_id=%s type=%s vendor=%s notification_id=%s",
row.user_id, row.type, vendor, row.id,
"push sent user_id=%s type=%s vendor=%s notification_id=%s device_id=%s",
row.user_id, row.type, vendor, row.id, dev.device_id,
)
sent += 1
except vendor_push.VendorPushError as e:
failed += 1
logger.warning(
"push failed user_id=%s type=%s vendor=%s notification_id=%s "
"device_id=%s error=%s",
row.user_id, row.type, vendor, row.id, dev.device_id, e,
)
try:
data_response = vendor_push.send_data_event(
vendor,
dev.push_token,
event=vendor_push.DATA_EVENT_NOTIFICATION_CREATED,
notification_id=str(row.id),
)
data_sent += 1
logger.info(
"push data event completed user_id=%s type=%s vendor=%s "
"notification_id=%s device_id=%s response=%s",
row.user_id, row.type, vendor, row.id, dev.device_id, data_response,
)
except vendor_push.VendorPushError as e:
data_failed += 1
# 透传只负责前台铃铛实时刷新,失败不能影响通知栏消息或站内消息。
logger.warning(
"push failed user_id=%s type=%s vendor=%s: %s", row.user_id, row.type, vendor, e
"push data event failed user_id=%s type=%s vendor=%s notification_id=%s "
"device_id=%s error=%s",
row.user_id, row.type, vendor, row.id, dev.device_id, e,
)
logger.info(
"push dispatch completed user_id=%s type=%s notification_id=%s targets=%s "
"sent=%s failed=%s skipped=%s data_sent=%s data_failed=%s",
row.user_id, row.type, row.id, len(targets),
sent, failed, skipped, data_sent, data_failed,
)
except Exception: # noqa: BLE001 — 渲染/查设备等意外失败同样不外抛
logger.exception("push notification failed user_id=%s type=%s", row.user_id, row.type)
@@ -156,7 +244,12 @@ def _push_to_user_devices(db: Session, row: Notification, push_vars: dict[str, s
# ---------------------------------------------------------------------------
def notify_withdraw_success(db: Session, order: WithdrawOrder) -> None:
def notify_withdraw_success(
db: Session,
order: WithdrawOrder,
*,
push_token_contains: str | None = None,
) -> None:
"""#3 提现成功:款项已存入微信零钱。点击无跳转仅消红点(extra 空)。"""
_dispatch(
db,
@@ -170,10 +263,16 @@ def notify_withdraw_success(db: Session, order: WithdrawOrder) -> None:
extra={},
dedup_key=order.out_bill_no,
push_vars={"amount": notif_repo.cash_yuan(order.amount_cents)},
push_token_contains=push_token_contains,
)
def notify_withdraw_failed(db: Session, order: WithdrawOrder) -> None:
def notify_withdraw_failed(
db: Session,
order: WithdrawOrder,
*,
push_token_contains: str | None = None,
) -> None:
"""#4 提现失败/退回:含微信侧失败、审核拒绝、解绑退回。点击跳提现页重新提现。
失败原因用 order.fail_reason( /withdraw/status 下发的用户可读原因同源)
@@ -191,10 +290,16 @@ def notify_withdraw_failed(db: Session, order: WithdrawOrder) -> None:
extra={"withdrawId": order.out_bill_no},
dedup_key=order.out_bill_no,
push_vars={"amount": notif_repo.cash_yuan(order.amount_cents), "reason": reason},
push_token_contains=push_token_contains,
)
def notify_feedback_reply(db: Session, feedback: Feedback) -> None:
def notify_feedback_reply(
db: Session,
feedback: Feedback,
*,
push_token_contains: str | None = None,
) -> None:
"""#9 官方回复:运营审核了反馈且未采纳(用户可见原因/留言落在反馈记录上)。
点击跳反馈历史页滚动高亮该条(extra.feedbackId)"""
_dispatch(
@@ -204,10 +309,16 @@ def notify_feedback_reply(db: Session, feedback: Feedback) -> None:
info_rows=[{"label": "说明文案", "value": "快去看看官方给您的回复吧~"}],
extra={"feedbackId": str(feedback.id)},
dedup_key=str(feedback.id),
push_token_contains=push_token_contains,
)
def notify_feedback_reward(db: Session, feedback: Feedback) -> None:
def notify_feedback_reward(
db: Session,
feedback: Feedback,
*,
push_token_contains: str | None = None,
) -> None:
"""#10 反馈奖励:反馈被采纳,金币已到账。PRD 约定发奖必带官方留言(admin_reply);
运营漏填时省略该信息行,不硬造文案"""
coins = int(feedback.reward_coins or 0)
@@ -225,10 +336,16 @@ def notify_feedback_reward(db: Session, feedback: Feedback) -> None:
extra={"feedbackId": str(feedback.id)},
dedup_key=str(feedback.id),
push_vars={"coins": str(coins)},
push_token_contains=push_token_contains,
)
def notify_report_approved(db: Session, report: PriceReport) -> None:
def notify_report_approved(
db: Session,
report: PriceReport,
*,
push_token_contains: str | None = None,
) -> None:
"""#11 爆料审核通过:上报的更低价过审,金币已到账。点击跳爆料记录页高亮该条。"""
coins = int(report.reward_coins or 0)
store = (report.store_name or "").strip() or "该店铺"
@@ -244,11 +361,17 @@ def notify_report_approved(db: Session, report: PriceReport) -> None:
extra={"reportId": str(report.id)},
dedup_key=str(report.id),
push_vars={"store": store, "coins": str(coins)},
push_token_contains=push_token_contains,
)
def notify_invite_order_reward(
db: Session, *, inviter_user_id: int, invitee_user_id: int, cash_cents: int
db: Session,
*,
inviter_user_id: int,
invitee_user_id: int,
cash_cents: int,
push_token_contains: str | None = None,
) -> None:
"""#12 好友下单到账:被邀请好友完成首次下单(比价),现金奖励已入邀请人账户。
通知发给邀请人;每个好友只发一次奖 dedup 按被邀请人"""
@@ -266,4 +389,5 @@ def notify_invite_order_reward(
extra={"inviteeNickname": nickname},
dedup_key=str(invitee_user_id),
push_vars={"nickname": nickname, "amount": _yuan_trim(cash_cents)},
push_token_contains=push_token_contains,
)
+1
View File
@@ -87,6 +87,7 @@ PRD §5 的 13 条 push 文案(标题固定 ≤11 字不带变量;正文 `{var}`
**真发注意**:
- 目标手机必须先装 App 且客户端已集成对应厂商 SDK、`/device/register` 上报过 token;
- 荣耀正式环境必须使用正式消息(`HONOR_PUSH_TARGET_USER_TYPE=0`);开发联调测试设备时才设为 `1`;
- vivo 未上架前走测试推送(`VIVO_PUSH_MODE=1`),目标手机需在 vivo 开放平台加入测试设备;
- 小米新设备需在开放平台把签名/包名配好,token 才有效。
+66 -10
View File
@@ -16,6 +16,10 @@ xiaomi/oppo/vivo)到该用户 device_liveness 里已注册的 push token。
.venv\\Scripts\\python.exe scripts\\fire_push_events.py --count 1 # 各 1 条(先小量验证通道)
.venv\\Scripts\\python.exe scripts\\fire_push_events.py --types withdraw_failed --count 3
.venv\\Scripts\\python.exe scripts\\fire_push_events.py --types all --count 2
.venv\\Scripts\\python.exe scripts\\fire_push_events.py --phone 22222222222 --token ABC123 --count 1
`--token` 是可选不区分大小写的 token 子串过滤器;会推送给该账号下所有 token 包含该
子串的设备没有匹配设备时脚本直接停止,不会生成站内消息
推送成败看输出里的 `shagua.vendor_push` 日志(push sent / push failed);2 台设备则每条各推 2
凭据缺失或 token 失效时 notification_events 只记日志不抛错(站内消息仍会落库)
@@ -58,33 +62,52 @@ _FAIL_REASONS = [
]
def _fire_one(db, uid: int, type_key: str, i: int) -> None:
def _fire_one(
db,
uid: int,
type_key: str,
i: int,
*,
push_token_contains: str | None = None,
) -> None:
"""构造一条该类型的瞬态业务对象(不落业务表,只为给 notify 函数读字段),触发真实推送。"""
if type_key == "withdraw_success":
order = WithdrawOrder(user_id=uid, out_bill_no=uuid.uuid4().hex, amount_cents=50, source="coin_cash")
notification_events.notify_withdraw_success(db, order)
notification_events.notify_withdraw_success(
db, order, push_token_contains=push_token_contains
)
elif type_key == "withdraw_failed":
order = WithdrawOrder(
user_id=uid, out_bill_no=uuid.uuid4().hex, amount_cents=350, source="coin_cash",
fail_reason=random.choice(_FAIL_REASONS),
)
notification_events.notify_withdraw_failed(db, order)
notification_events.notify_withdraw_failed(
db, order, push_token_contains=push_token_contains
)
elif type_key == "invite_order_reward":
# 假被邀请人 id(> 真实用户范围,避重):昵称回退「好友」。真实昵称请走 API 流程(见文末说明)。
fake_invitee = random.randint(900000, 999999)
notification_events.notify_invite_order_reward(
db, inviter_user_id=uid, invitee_user_id=fake_invitee, cash_cents=INVITE_COMPARE_REWARD_CENTS
db,
inviter_user_id=uid,
invitee_user_id=fake_invitee,
cash_cents=INVITE_COMPARE_REWARD_CENTS,
push_token_contains=push_token_contains,
)
elif type_key == "feedback_reward":
fb = Feedback(user_id=uid, content="(直发)", contact="", status="adopted",
reward_coins=300, admin_reply="感谢反馈,您说的问题已修复上线,金币请查收~")
fb.id = random.randint(900000, 999999)
notification_events.notify_feedback_reward(db, fb)
notification_events.notify_feedback_reward(
db, fb, push_token_contains=push_token_contains
)
elif type_key == "feedback_reply":
fb = Feedback(user_id=uid, content="(直发)", contact="", status="rejected",
admin_reply="您的建议我们记录啦,会在后续版本评估~")
fb.id = random.randint(900000, 999999)
notification_events.notify_feedback_reply(db, fb)
notification_events.notify_feedback_reply(
db, fb, push_token_contains=push_token_contains
)
elif type_key == "report_approved":
rep = PriceReport(
user_id=uid, reported_platform_id="jd", reported_platform_name="京东外卖",
@@ -92,7 +115,9 @@ def _fire_one(db, uid: int, type_key: str, i: int) -> None:
reward_coins=PRICE_REPORT_REWARD_COINS, store_name=f"测试火锅店{i:02d}",
)
rep.id = random.randint(900000, 999999)
notification_events.notify_report_approved(db, rep)
notification_events.notify_report_approved(
db, rep, push_token_contains=push_token_contains
)
else:
raise SystemExit(f"未知类型: {type_key}(可选: {', '.join(ALL_TYPES)})")
@@ -100,6 +125,11 @@ def _fire_one(db, uid: int, type_key: str, i: int) -> None:
def main() -> None:
parser = argparse.ArgumentParser(description="直接触发消息通知中心真实推送(后台驱动不了的事件用)")
parser.add_argument("--phone", default=DEFAULT_PHONE, help=f"目标用户手机号(默认 {DEFAULT_PHONE})")
parser.add_argument(
"--token",
default="",
help="可选的厂商 token 子串(不区分大小写);仅推送到该账号下所有匹配设备",
)
parser.add_argument("--count", type=int, default=10, help="每类发多少条(默认 10)")
parser.add_argument(
"--types", default=",".join(DEFAULT_TYPES),
@@ -122,14 +152,40 @@ def main() -> None:
uid = user.id
targets = device_repo.list_push_targets(db, user_id=uid)
print(f"目标用户 {args.phone}(id={uid});已注册推送设备 {len(targets)} 台:"
f"{[t.push_vendor for t in targets] or '无(手机收不到!先在 App 上报 push token)'}")
token_filter = args.token.strip()
normalized_filter = token_filter.casefold()
matched_targets = [
target
for target in targets
if not token_filter
or normalized_filter in (target.push_token or "").casefold()
]
if token_filter and not matched_targets:
print(
f"❌ 用户 {args.phone}{len(targets)} 台有效推送设备中,"
f"没有 token 包含 {token_filter!r};未生成站内消息,也未发送推送。"
)
return
print(
f"目标用户 {args.phone}(id={uid});有效推送设备 {len(targets)} 台;"
f"本次匹配 {len(matched_targets)} 台:"
f"{[target.push_vendor for target in matched_targets] or '无(手机收不到!先在 App 上报 push token)'}"
)
if token_filter:
print(f"token 子串过滤(不区分大小写): {token_filter!r}")
print(f"即将触发:{types},每类 {args.count} 条 → 共 {len(types) * args.count}\n")
for t in types:
print(f"── {t} ×{args.count} " + "" * 30)
for i in range(1, args.count + 1):
_fire_one(db, uid, t, i)
_fire_one(
db,
uid,
t,
i,
push_token_contains=token_filter or None,
)
print(f"\n✅ 已触发完。站内消息已落 notification 表(用 {args.phone} 登录 App 可在消息中心看到);"
"\n 手机推送成败见上方 `shagua.vendor_push` 日志(push sent=成功 / push failed=失败)。")
+36 -3
View File
@@ -15,6 +15,10 @@ services/notification_events.notify_invite_order_reward —— 与生产同一
.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 # 真实昵称
.venv\\Scripts\\python.exe scripts\\test_push_invite_order_reward.py --phone 22222222222 --token gkha
`--token` 是可选不区分大小写的 token 子串过滤器会推送给该账号下所有 token 包含该
子串的设备没有匹配设备时脚本直接停止不会生成站内消息
结果判读(看输出日志):
push sent = 厂商接口受理成功,手机应弹好友下单奖励到账通知
@@ -62,6 +66,11 @@ def _notif_count(db, uid: int) -> int:
def main() -> None:
parser = argparse.ArgumentParser(description="#12 好友下单到账 推送联调(每次 1 条,金额默认随机)")
parser.add_argument("--phone", default="11111111111", help="邀请人(收通知方)手机号,默认 11111111111")
parser.add_argument(
"--token",
default="",
help="可选的厂商 token 子串(不区分大小写);仅推送到该账号下所有匹配设备",
)
parser.add_argument("--cents", type=int, default=None,
help="奖励金额,单位分(默认随机 1~9999;线上真实值 200)")
parser.add_argument("--invitee-phone", default="",
@@ -87,13 +96,37 @@ def main() -> None:
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'}")
token_filter = args.token.strip()
normalized_filter = token_filter.casefold()
matched_targets = [
target
for target in targets
if not token_filter
or normalized_filter in (target.push_token or "").casefold()
]
if token_filter and not matched_targets:
print(
f"❌ 用户 {args.phone}{len(targets)} 台有效推送设备中,"
f"没有 token 包含 {token_filter!r};未生成站内消息,也未发送推送。"
)
return
print(
f"邀请人 {args.phone}(id={user.id});有效推送设备 {len(targets)} 台;"
f"本次匹配 {len(matched_targets)} 台:"
f"{[target.push_vendor for target in matched_targets] or '无 ← 手机收不到!先在 App 上报 push token'}"
)
if token_filter:
print(f"token 子串过滤(不区分大小写): {token_filter!r}")
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
db,
inviter_user_id=user.id,
invitee_user_id=invitee_id,
cash_cents=cents,
push_token_contains=token_filter or None,
)
created = _notif_count(db, user.id) - before
+69
View File
@@ -140,6 +140,75 @@ def test_harvest_done_failed_derives_fail_reason(client) -> None:
assert rec.information == "比价过程出错,请稍后重试" # 原文案仍留存
def _done_params_platforms_no_isbest() -> dict:
"""id 3304 型:done 帧 platforms 全平台「相似替换/仅供参考」(has_dish_diff),pricebot
一个 is_best 都没标,但有有价目标(美团 57.8 < 源淘宝闪购 60.8)"""
return {
"platforms": [
{"role": "source", "platform_id": "eleme", "platform_name": "淘宝闪购",
"package": "me.ele", "price": 60.8, "is_best": False, "has_dish_diff": False,
"store_name": "窑鸡王", "items": [{"name": "招牌窑鸡 整只-香辣", "qty": 1}]},
{"role": "target", "platform_id": "meituan_waimai", "platform_name": "美团外卖",
"package": "com.sankuai.meituan.takeoutnew", "price": 57.8, "is_best": False,
"has_dish_diff": True},
{"role": "target", "platform_id": "jd_waimai_standalone", "platform_name": "京东外卖",
"package": "com.jd.waimai", "price": 74.9, "is_best": False, "has_dish_diff": True},
],
"information": "美团更便宜(含相似商品替换)",
}
def test_harvest_done_platforms_no_isbest_falls_back_to_cheapest_target(client) -> None:
"""回归(id 3304):platforms 全无 is_best(全平台相似替换)但有有价目标 → best 应兜底取
有价目标里最低那家,不能让 best_*/saved 整条落 NULL(否则首页价 0.00 / 记录页无最低红框 /
省额丢失)"""
tid = _tid()
with SessionLocal() as db:
crud.harvest_running(db, trace_id=tid, user_id=None)
rec, _ = crud.harvest_done(db, trace_id=tid, user_id=None,
done_params=_done_params_platforms_no_isbest())
assert rec.status == "success"
assert rec.best_platform_id == "meituan_waimai" # 有价目标里最低
assert rec.best_price_cents == 5780 # 57.8 元
assert rec.source_price_cents == 6080 # 源淘宝闪购 60.8
assert rec.saved_amount_cents == 300 # 60.8 - 57.8
assert rec.is_source_best is False # 兜底选的是目标,非源
def _done_params_no_isbest_source_cheapest() -> dict:
"""无 is_best 且源本身最便宜:源淘宝闪购 50.0 < 全部 dish-diff 目标(美团 57.8 / 京东 74.9)。
此时不能强选更贵的目标当 best(否则 saved 变负污染累计发现可省),应回落源"""
return {
"platforms": [
{"role": "source", "platform_id": "eleme", "platform_name": "淘宝闪购",
"package": "me.ele", "price": 50.0, "is_best": False, "has_dish_diff": False,
"store_name": "窑鸡王", "items": [{"name": "招牌窑鸡 整只-香辣", "qty": 1}]},
{"role": "target", "platform_id": "meituan_waimai", "platform_name": "美团外卖",
"package": "com.sankuai.meituan.takeoutnew", "price": 57.8, "is_best": False,
"has_dish_diff": True},
{"role": "target", "platform_id": "jd_waimai_standalone", "platform_name": "京东外卖",
"package": "com.jd.waimai", "price": 74.9, "is_best": False, "has_dish_diff": True},
],
"information": "源平台已是最低(其余为相似替换)",
}
def test_harvest_done_no_isbest_source_cheapest_falls_back_to_source(client) -> None:
"""回归:无 is_best 且源最便宜 → best 回落源(与 _derive「全目标缺菜回落源、不虚报省」同语义),
saved=0is_source_best=True,绝不因强选更贵目标而让 saved 变负倒扣累计发现可省"""
tid = _tid()
with SessionLocal() as db:
crud.harvest_running(db, trace_id=tid, user_id=None)
rec, _ = crud.harvest_done(db, trace_id=tid, user_id=None,
done_params=_done_params_no_isbest_source_cheapest())
assert rec.status == "success"
assert rec.best_platform_id == "eleme" # 回落到源(源最便宜)
assert rec.best_price_cents == 5000 # 源 50.0
assert rec.source_price_cents == 5000
assert rec.saved_amount_cents == 0 # 没省到,绝不为负
assert rec.is_source_best is True # 源就是最便宜
def test_harvest_abort_cancels_running(client) -> None:
tid = _tid()
with SessionLocal() as db:
+95
View File
@@ -7,7 +7,9 @@ mock/真发/变量覆盖/站内联动/设备反查/参数校验。厂商 HTTP
from __future__ import annotations
import json
import logging
import httpx
import pytest
from fastapi.testclient import TestClient
@@ -110,6 +112,99 @@ def test_huawei_non_success_code_raises(monkeypatch) -> None:
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:
assert vendor_push.normalize_vendor("华为") == "huawei"
assert vendor_push.normalize_vendor("HMS") == "huawei"