Compare commits

..

3 Commits

Author SHA1 Message Date
exinglang a0196b8d64 fix(push): 完善厂商推送排障日志 2026-07-29 19:52:06 +08:00
zuochenyong 53c3b7f60f feat(guide-video): 支持领券和比价独立视频奖励配置 (#196)
Co-authored-by: exinglang <exinglang@qq.com>
Reviewed-on: #196
Co-authored-by: zuochenyong <zuochenyong@wonderable.ai>
Co-committed-by: zuochenyong <zuochenyong@wonderable.ai>
2026-07-29 16:12:05 +08:00
marco 4bd4e66678 重构比价结果页取数据逻辑 (#195)
Reviewed-on: #195
2026-07-29 01:59:31 +08:00
25 changed files with 653 additions and 913 deletions
-5
View File
@@ -64,11 +64,6 @@ CHUANGLAN_SMS_TEMPLATE_ID=1022457679
CHUANGLAN_SMS_SIGNATURE=
CHUANGLAN_SMS_ENDPOINT=https://smssh.253.com/msg/sms/v2/tpl/send
CHUANGLAN_SMS_TIMEOUT_SEC=10
# --- 阿里云号码认证·一键登录(Dypnsapi GetMobile 换号)。同账号可填与 ALIYUN_SMS_ACCESS_KEY_* 相同的值 ---
ALIYUN_ONEKEY_ACCESS_KEY_ID=
ALIYUN_ONEKEY_ACCESS_KEY_SECRET=
ALIYUN_ONEKEY_ENDPOINT=dypnsapi.aliyuncs.com
ALIYUN_ONEKEY_TIMEOUT_SEC=15
# ===== 测试账号(release 包全流程联调用)=====
# 配一个固定测试手机号,专供无 SIM 卡 / 不走一键登录时打通全流程:该号登录【免短信验证码】
@@ -0,0 +1,26 @@
"""merge comparison platforms + fail_reason heads
Revision ID: 6d2309208549
Revises: comparison_platforms_col, comparison_record_fail_reason
Create Date: 2026-07-29 01:48:41.868083
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '6d2309208549'
down_revision: Union[str, Sequence[str], None] = ('comparison_platforms_col', 'comparison_record_fail_reason')
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
pass
def downgrade() -> None:
pass
@@ -0,0 +1,47 @@
"""add platforms unified array column to comparison_record
展示模型统一数组(pricebot done.params.platforms 原样存): 每平台一行、自带
status/is_best/display, 记录页据此直接渲染, 不再靠 comparison_results + 客户端合并 + 前端派生。
纯新增列, 老记录为空 → 前端回退老 comparison_results。
Revision ID: comparison_platforms_col
Revises: user_manual_risk_fields
"""
from __future__ import annotations
from collections.abc import Sequence
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
from alembic import op
revision: str = "comparison_platforms_col"
down_revision: str | None = "user_manual_risk_fields"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
def upgrade() -> None:
# 幂等: 线上为了提前给历史数据补 platforms(2026-07-29), 已手动
# `ALTER TABLE comparison_record ADD COLUMN IF NOT EXISTS platforms jsonb
# NOT NULL DEFAULT '[]'::jsonb`(与本 migration 定义一致)。列已存在时跳过,
# 否则上线 alembic upgrade head 会撞 DuplicateColumn 直接部署失败。
bind = op.get_bind()
cols = {c["name"] for c in sa.inspect(bind).get_columns("comparison_record")}
if "platforms" in cols:
return
with op.batch_alter_table("comparison_record") as batch_op:
batch_op.add_column(
sa.Column(
"platforms", _JSON, nullable=False,
server_default=sa.text("'[]'"),
)
)
def downgrade() -> None:
with op.batch_alter_table("comparison_record") as batch_op:
batch_op.drop_column("platforms")
@@ -0,0 +1,31 @@
"""guide video play count is independent for coupon and comparison
Revision ID: guide_video_scene_unique
Revises: 6d2309208549
"""
from alembic import op
revision = "guide_video_scene_unique"
down_revision = "6d2309208549"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.drop_index("uq_guide_video_play_user_seq", table_name="guide_video_play")
op.create_index(
"uq_guide_video_play_user_scene_seq",
"guide_video_play",
["user_id", "scene", "seq"],
unique=True,
)
def downgrade() -> None:
op.drop_index("uq_guide_video_play_user_scene_seq", table_name="guide_video_play")
op.create_index(
"uq_guide_video_play_user_seq",
"guide_video_play",
["user_id", "seq"],
unique=True,
)
+29 -14
View File
@@ -9,7 +9,7 @@ client_max_body_size,见 shaguabijia-admin-web/deploy/nginx/admin.shaguabijia.co
"""
from __future__ import annotations
from typing import Annotated
from typing import Annotated, Literal
from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile
@@ -27,14 +27,21 @@ router = APIRouter(
)
def _out(db: AdminDb) -> GuideVideoConfigOut:
GuideScene = Literal["coupon", "comparison"]
def _out(db: AdminDb, scene: GuideScene) -> GuideVideoConfigOut:
"""配置 + 播放统计合成响应(四个写接口都以最新状态返回,前端一次同步到位)。"""
return GuideVideoConfigOut(**guide_video.get_config(db), **guide_video.play_stats(db))
return GuideVideoConfigOut(
scene=scene,
**guide_video.get_config(db, scene),
**guide_video.play_stats(db, scene),
)
@router.get("", response_model=GuideVideoConfigOut, summary="新手引导视频配置(领券浮层)")
def get_config(db: AdminDb) -> GuideVideoConfigOut:
return _out(db)
def get_config(db: AdminDb, scene: GuideScene = "coupon") -> GuideVideoConfigOut:
return _out(db, scene)
@router.patch("", response_model=GuideVideoConfigOut, summary="改开关/次数/金币(带审计)")
@@ -43,21 +50,23 @@ def update_config(
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
scene: GuideScene = "coupon",
) -> GuideVideoConfigOut:
before, after = guide_video.update_config(
db,
enabled=body.enabled,
max_plays=body.max_plays,
reward_coin=body.reward_coin,
scene=scene,
admin_id=admin.id,
commit=False,
)
write_audit(
db, admin, action="guide_video.update", target_type="guide_video", target_id=None,
detail={"before": before, "after": after}, ip=get_client_ip(request), commit=False,
detail={"scene": scene, "before": before, "after": after}, ip=get_client_ip(request), commit=False,
)
db.commit()
return _out(db)
return _out(db, scene)
@router.post("/video", response_model=GuideVideoConfigOut, summary="上传新手引导视频(MP4,带审计)")
@@ -65,23 +74,26 @@ async def upload_video(
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
file: UploadFile = File(...),
file: Annotated[UploadFile, File()],
scene: GuideScene = "coupon",
) -> GuideVideoConfigOut:
data = await file.read()
try:
url = media.save_guide_video(data)
except media.MediaError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
before, after = guide_video.set_video(db, url, admin_id=admin.id, commit=False)
before, after = guide_video.set_video(
db, url, scene=scene, admin_id=admin.id, commit=False
)
write_audit(
db, admin, action="guide_video.set_video", target_type="guide_video", target_id=None,
detail={"before": before.get("video_url"), "after": url, "bytes": len(data)},
detail={"scene": scene, "before": before.get("video_url"), "after": url, "bytes": len(data)},
ip=get_client_ip(request), commit=False,
)
db.commit()
# 提交成功后再删旧片,避免新片没落库就把旧片丢了
media.delete_guide_video(before.get("video_url"))
return _out(db)
return _out(db, scene)
@router.delete("/video", response_model=GuideVideoConfigOut, summary="移除新手引导视频(带审计)")
@@ -89,13 +101,16 @@ def delete_video(
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
scene: GuideScene = "coupon",
) -> GuideVideoConfigOut:
"""移除后 /guide-video/start 一律返回 should_play=false,领券浮层回到「只放广告」。"""
before, after = guide_video.set_video(db, None, admin_id=admin.id, commit=False)
before, after = guide_video.set_video(
db, None, scene=scene, admin_id=admin.id, commit=False
)
write_audit(
db, admin, action="guide_video.delete_video", target_type="guide_video", target_id=None,
detail={"before": before.get("video_url")}, ip=get_client_ip(request), commit=False,
detail={"scene": scene, "before": before.get("video_url")}, ip=get_client_ip(request), commit=False,
)
db.commit()
media.delete_guide_video(before.get("video_url"))
return _out(db)
return _out(db, scene)
+1
View File
@@ -7,6 +7,7 @@ from app.repositories.guide_video import MAX_PLAYS_LIMIT, REWARD_COIN_LIMIT
class GuideVideoConfigOut(BaseModel):
scene: str
enabled: bool
video_url: str | None = None # 相对地址 /media/guide_video/xxx.mp4;未配片 = None
max_plays: int
+12 -13
View File
@@ -33,7 +33,7 @@ from app.core.security import (
issue_token_pair,
)
from app.integrations import wxpay
from app.integrations.oneclick import OneClickError, mask_phone, verify_and_get_phone
from app.integrations.jiguang import JiguangError, mask_phone, verify_and_get_phone
from app.integrations.sms import SmsError, send_code, verify_code
from app.repositories import onboarding as onboarding_repo
from app.repositories import phone_rebind as rebind_repo
@@ -110,15 +110,14 @@ def jverify_login(req: JverifyLoginRequest, request: Request, db: DbSession) ->
):
raise HTTPException(status_code=403, detail="当前设备环境异常,暂无法登录")
logger.info(
"jverify_login provider=%s operator=%s token_len=%d",
req.provider or "-",
"jverify_login operator=%s token_len=%d",
req.operator or "-",
len(req.login_token),
)
try:
phone = verify_and_get_phone(req.provider, req.login_token)
except OneClickError as e:
phone = verify_and_get_phone(req.login_token)
except JiguangError as e:
risk_repo.record_behavior_event(
db,
event_type=risk_repo.EVENT_ONECLICK_LOGIN,
@@ -129,11 +128,11 @@ def jverify_login(req: JverifyLoginRequest, request: Request, db: DbSession) ->
client_ip=_client_ip(request),
outcome="failed",
reason=str(e),
details={"operator": req.operator or None, "provider": req.provider or None},
details={"operator": req.operator or None},
evaluate_rule=risk_repo.RULE_ONECLICK_DAILY,
)
logger.error("[oneclick] verify+decrypt failed: %s", e, exc_info=True)
raise HTTPException(status_code=502, detail=f"oneclick verify failed: {e}") from e
logger.error("[JG] verify+decrypt failed: %s", e, exc_info=True)
raise HTTPException(status_code=502, detail=f"jiguang verify failed: {e}") from e
user = user_repo.upsert_user_for_login(db, phone=phone, register_channel="jverify")
risk_repo.record_behavior_event(
@@ -147,7 +146,7 @@ def jverify_login(req: JverifyLoginRequest, request: Request, db: DbSession) ->
phone=phone,
client_ip=_client_ip(request),
outcome="success",
details={"operator": req.operator or None, "provider": req.provider or None},
details={"operator": req.operator or None},
evaluate_rule=risk_repo.RULE_ONECLICK_DAILY,
)
if user.status != "active":
@@ -493,10 +492,10 @@ def wechat_bind_phone_jverify(
raise HTTPException(status_code=401, detail="授权已过期,请重新用微信登录") from e
try:
phone = verify_and_get_phone("jiguang", req.login_token)
except OneClickError as e:
logger.error("[oneclick] verify+decrypt failed: %s", e, exc_info=True)
raise HTTPException(status_code=502, detail=f"oneclick verify failed: {e}") from e
phone = verify_and_get_phone(req.login_token)
except JiguangError as e:
logger.error("[JG] verify+decrypt failed: %s", e, exc_info=True)
raise HTTPException(status_code=502, detail=f"jiguang verify failed: {e}") from e
return _finish_wechat_bind(
db,
-15
View File
@@ -165,14 +165,6 @@ class Settings(BaseSettings):
CHUANGLAN_SMS_ENDPOINT: str = "https://smssh.253.com/msg/sms/v2/tpl/send"
CHUANGLAN_SMS_TIMEOUT_SEC: int = 10 # httpx 读/连超时秒
# ===== 阿里云号码认证·一键登录(Dypnsapi GetMobile 换号)=====
# 与短信同属 dypnsapi 产品:同一阿里云账号可复用 ALIYUN_SMS_ACCESS_KEY_*,默认独立字段解耦。
# 缺凭证 → provider=aliyun 换号抛错→502,不启动崩(见 aliyun_oneclick_configured)。
ALIYUN_ONEKEY_ACCESS_KEY_ID: str = ""
ALIYUN_ONEKEY_ACCESS_KEY_SECRET: str = ""
ALIYUN_ONEKEY_ENDPOINT: str = "dypnsapi.aliyuncs.com"
ALIYUN_ONEKEY_TIMEOUT_SEC: int = 15 # 阿里云 API 读/连超时秒
@property
def aliyun_sms_configured(self) -> bool:
"""阿里云短信凭证齐全(缺则 SMS_PROVIDER=aliyun 时 /sms/* 返 503,而非启动崩)。"""
@@ -183,13 +175,6 @@ class Settings(BaseSettings):
and self.ALIYUN_SMS_TEMPLATE_CODE
)
@property
def aliyun_oneclick_configured(self) -> bool:
"""阿里云一键登录凭证齐全(缺则 provider=aliyun 换号抛错→502,而非启动崩)。"""
return bool(
self.ALIYUN_ONEKEY_ACCESS_KEY_ID and self.ALIYUN_ONEKEY_ACCESS_KEY_SECRET
)
@property
def chuanglan_sms_configured(self) -> bool:
"""创蓝短信凭证齐全(缺则 SMS_PROVIDER=chuanglan 时 /sms/send 返 503,而非启动崩)。"""
-89
View File
@@ -1,89 +0,0 @@
"""阿里云号码认证·一键登录服务端换号(Dypnsapi GetMobile)。
链路:
Android 阿里云 SDK getLoginToken → spToken(access_token)
→ 本服务调 Dypnsapi GetMobile(AccessToken=spToken)
→ 阿里云直接返回明文手机号(无需 RSA/AES,比极光/创蓝少一步解密)
与 jiguang 对齐:对外暴露 verify_and_get_phone(login_token)->str,失败抛 AliyunOneClickError,
由 oneclick.py 门面统一 catch。
SDK 交互隔离在 _call_get_mobile 薄封装(惰性 import + 惰性建 client,仿 sms/aliyun.py),
单测 monkeypatch 它即可,不触真 SDK/网络。凭证复用/独立见 config.ALIYUN_ONEKEY_*。
"""
from __future__ import annotations
import logging
from app.core.config import settings
logger = logging.getLogger("shagua.aliyun.onekey")
_client = None # 惰性构建的 dypnsapi client(模块级缓存)
class AliyunOneClickError(Exception):
"""阿里云取号失败的统一异常,由 oneclick 门面 catch 翻成 4xx/5xx。"""
def verify_and_get_phone(login_token: str) -> str:
"""对外唯一函数:loginToken(access_token) → 明文手机号。失败抛 AliyunOneClickError。"""
if not settings.aliyun_oneclick_configured:
raise AliyunOneClickError("ALIYUN_ONEKEY_ACCESS_KEY_ID/SECRET not configured")
result = _call_get_mobile(login_token)
if not (result["success"] and result["code"] == "OK"):
logger.error(
"[ALIYUN-ONEKEY] get_mobile failed code=%s msg=%s",
result["code"], result["message"],
)
raise AliyunOneClickError(f"aliyun get_mobile failed code={result['code']}")
phone = (result["mobile"] or "").strip()
if not (phone.isdigit() and len(phone) == 11):
logger.error("[ALIYUN-ONEKEY] unexpected mobile format: %r", phone)
raise AliyunOneClickError("aliyun get_mobile returned non-phone")
return phone
# ==================== SDK 接缝(单测 monkeypatch 这个)====================
def _get_client():
"""惰性构建 dypnsapi client(仿 sms/aliyun.py:jiguang-only 部署不加载 alibabacloud)。"""
global _client
if _client is None:
from alibabacloud_dypnsapi20170525.client import Client
from alibabacloud_tea_openapi import models as open_api_models
cfg = open_api_models.Config(
access_key_id=settings.ALIYUN_ONEKEY_ACCESS_KEY_ID,
access_key_secret=settings.ALIYUN_ONEKEY_ACCESS_KEY_SECRET,
read_timeout=settings.ALIYUN_ONEKEY_TIMEOUT_SEC * 1000, # SDK 单位 ms
connect_timeout=settings.ALIYUN_ONEKEY_TIMEOUT_SEC * 1000,
)
cfg.endpoint = settings.ALIYUN_ONEKEY_ENDPOINT
_client = Client(cfg)
return _client
def _call_get_mobile(login_token: str) -> dict:
"""调 GetMobile。返回归一化 {success, code, message, mobile};import/建 client/调用任一失败抛 AliyunOneClickError。
⚠️ GetMobile 响应体字段名以实际 SDK 为准(code=="OK"、get_mobile_result_dto.mobile)。接真号联调时
若字段不同,只需改本函数末尾归一化,verify_and_get_phone 及单测不动。
"""
try:
from alibabacloud_dypnsapi20170525 import models as dypns_models
req = dypns_models.GetMobileRequest(access_token=login_token)
body = _get_client().get_mobile(req).body
except Exception as e:
logger.exception("[ALIYUN-ONEKEY] get_mobile 调用异常")
raise AliyunOneClickError("aliyun get_mobile 调用异常") from e
dto = getattr(body, "get_mobile_result_dto", None)
mobile = getattr(dto, "mobile", None) if dto else None
return {
"success": (body.code == "OK"),
"code": body.code,
"message": body.message,
"mobile": mobile,
}
-41
View File
@@ -1,41 +0,0 @@
"""一键登录换号门面:按 provider 分派到极光/阿里云, 统一异常与手机号脱敏。
为什么要门面:一键登录 token 与「拉授权页的那家 SDK」强绑定 —— 客户端用极光 SDK 拉的
token 只能用极光换号, 阿里云的只能用阿里云换号。所以 provider 必须由客户端如实上报, 服务端
按此分派、不能猜。老客户端不带 provider → 默认极光(向后兼容)。
用法(api 层):
from app.integrations import oneclick
try:
phone = oneclick.verify_and_get_phone(req.provider, req.login_token)
except oneclick.OneClickError as e:
raise HTTPException(502, ...) from e
"""
from __future__ import annotations
# 用「模块属性访问」而非 from ... import 函数:保证 monkeypatch 各家实现时门面能拿到替身。
from app.integrations import aliyun_onekey, jiguang
from app.integrations.jiguang import mask_phone # 复用脱敏, 从门面转出供 api 层用
__all__ = ["OneClickError", "mask_phone", "verify_and_get_phone"]
PROVIDER_JIGUANG = "jiguang"
PROVIDER_ALIYUN = "aliyun"
class OneClickError(Exception):
"""换号失败统一异常(不区分厂商), api 层 catch → 502。"""
def verify_and_get_phone(provider: str, login_token: str) -> str:
"""按 provider 换取明文手机号。失败(任一厂商)抛 OneClickError。
provider 大小写不敏感;空 / 未知值兜底走极光(主家), 不因客户端传错值而拒登。
"""
p = (provider or PROVIDER_JIGUANG).strip().lower()
try:
if p == PROVIDER_ALIYUN:
return aliyun_onekey.verify_and_get_phone(login_token)
return jiguang.verify_and_get_phone(login_token)
except (jiguang.JiguangError, aliyun_onekey.AliyunOneClickError) as e:
raise OneClickError(f"[{p}] {e}") from e
+242 -35
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
@@ -69,6 +69,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 +127,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 +153,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(
@@ -159,13 +200,120 @@ def _require(value: str, name: str) -> str:
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(
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 +322,82 @@ 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
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(
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 +423,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,7 +434,7 @@ 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"))
@@ -270,6 +461,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 +472,7 @@ 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
@@ -294,6 +487,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 +498,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"))
@@ -333,6 +528,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 +537,7 @@ 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
@@ -357,6 +554,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 +565,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)
@@ -401,6 +600,8 @@ 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",
@@ -408,7 +609,7 @@ def _send_vivo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
},
)
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 +647,16 @@ 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
@@ -522,6 +725,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 +735,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)
@@ -573,6 +778,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 +787,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
+4
View File
@@ -113,6 +113,10 @@ class ComparisonRecord(Base):
items: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
# 逐平台对比 [{platform_id, platform_name, package, price, is_source, rank, coupon_saved, coupon_name, applied_coupons}](price/coupon_saved 单位:元,原样存;coupon_name=优惠来源名;applied_coupons=[{name,amount}] 多券明细)
comparison_results: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
# 展示模型统一数组(pricebot done.params.platforms 原样存): 每平台一行、自带
# status/is_best/display/display_order, 记录页据此直接渲染, 不再靠 comparison_results
# + 客户端合并 + 前端派生。老记录/旧客户端为空 → 前端回退老 comparison_results 渲染。
platforms: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
# 目标平台未找到、跳过的菜名
skipped_dish_names: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
# 客户端上报的原始 payload(calibration + done.params 全量),未来取数兜底
+1 -1
View File
@@ -38,7 +38,7 @@ class GuideVideoPlay(Base):
# start_play 捕获 IntegrityError 降级成"这次不放视频"。
# 用 unique Index 而非 UniqueConstraint:与迁移里的 create_index 对齐(SQLite 加约束
# 要整表重建),autogenerate 才不会每次报一条假 diff。
Index("uq_guide_video_play_user_seq", "user_id", "seq", unique=True),
Index("uq_guide_video_play_user_scene_seq", "user_id", "scene", "seq", unique=True),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+101 -12
View File
@@ -164,8 +164,9 @@ def _derive(payload: ComparisonRecordIn) -> dict:
is_source_best = best.is_source if best is not None else None
# status:客户端显式给了就用;否则有"非源且有价"的结果=success,否则 failed
status = payload.status
# status:优先 pricebot record_status(区分 below_minimum/store_closed) → 客户端显式 status
# → 兜底"非源且有价"=success/否则 failed。record_status 让"未满起送"不再塌缩成 failed。
status = payload.record_status or payload.status
if status is None:
has_valid_target = any(
(not r.is_source) and r.price is not None for r in results
@@ -197,16 +198,35 @@ def upsert_record(
灰度期老客户端 POST /compare/record 走这条,与后端 harvest 按 trace_id reconcile;
新客户端不再 POST(改由 compare.py 透传壳 harvest 落库)。
"""
derived = _derive(payload)
# 单源派生: 与 harvest_done 一致, payload 带 platforms 时从它派生(唯一真相源
# _derive_from_platforms), 老客户端不带 platforms 时回退 _derive(从 comparison_results)。
if payload.platforms:
derived = _derive_from_platforms(payload.platforms, payload.record_status)
# 对齐 _derive 返回键(#189 fail_reason): 两路径 fields 键集一致, 覆盖已有行时不残留旧值
derived["fail_reason"] = (
_derive_fail_display(payload.information, payload.platform_results or {})
if derived["status"] == "failed"
else None
)
# 单源派生取自 platforms 源行(常无源平台元数据/店名)→ 空则用 payload 兜底不丢字段。
# 下面 fields 不再显式写这四个键, 统一由 derived 提供(否则 dict(store_name=..., **derived)
# 与 _derive_from_platforms 同名键撞键 TypeError)。
for _k in ("store_name", "source_platform_id", "source_platform_name", "source_package"):
if not derived.get(_k):
derived[_k] = getattr(payload, _k)
else:
derived = _derive(payload)
# _derive 只从 comparison_results 派生, 不含源平台四件套 / store_name → 从 payload 补,
# 与上面 platforms 分支键集对齐(fields 统一靠 **derived 提供这些列)。
for _k in ("store_name", "source_platform_id", "source_platform_name", "source_package"):
derived[_k] = getattr(payload, _k)
items = [it.model_dump(exclude_none=True) for it in payload.items]
fields = dict(
device_id=payload.device_id,
business_type=payload.business_type,
store_name=payload.store_name,
product_names=_product_names_from_items(items),
source_platform_id=payload.source_platform_id,
source_platform_name=payload.source_platform_name,
source_package=payload.source_package,
# store_name / source_platform_id / source_platform_name / source_package 统一由
# derived 提供(见上方两分支补齐), 不在此显式写 —— 否则与 _derive_from_platforms 撞键。
information=payload.information,
best_deeplink=payload.best_deeplink,
trace_url=payload.trace_url,
@@ -214,6 +234,7 @@ def upsert_record(
skipped_dish_count=payload.skipped_dish_count,
items=items,
comparison_results=[r.model_dump() for r in payload.comparison_results],
platforms=list(payload.platforms or []),
skipped_dish_names=list(payload.skipped_dish_names),
# 客户端环境 / 性能(debug,客户端上报;旧客户端为 None)
device_model=payload.device_model,
@@ -286,7 +307,8 @@ def upsert_record(
def _derive_from_results(
results: list[dict], platform_results: dict | None = None
results: list[dict], platform_results: dict | None = None,
record_status: str | None = None,
) -> dict:
"""从 done 帧 comparison_results(pricebot 原始 dict 列表)派生结构化列。
等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。
@@ -334,7 +356,53 @@ def _derive_from_results(
"saved_amount_cents": saved_amount_cents,
"is_source_best": best.get("is_source") if best else None,
"store_name": (src_row or {}).get("store_name") or None,
"status": "success" if has_valid_target else "failed",
# 记录级结局: 优先用 pricebot 下发的 record_status(区分 below_minimum/store_closed,
# 不再把"未满起送"塌缩成 failed → 记录页不再误报"网络开小差"); 旧 pricebot 未下发时
# 回退老的 success/failed 二态派生, 向后兼容。
"status": record_status or ("success" if has_valid_target else "failed"),
}
def _derive_from_platforms(
platforms: list, record_status: str | None = None,
) -> dict:
"""从 done 帧 platforms(每平台一行、渲染就绪)派生结构化列——**单一真相源**。
best_* 直接取 platforms 里 is_best 的那一行、source_* 取 role=source 行,与前端读的
platforms 天然一致(不再像 _derive_from_results 那样从 comparison_results 二次评最优,
消除"标量列 vs platforms"双源不一致)。platforms 非空时优先走这里;老 pricebot 无
platforms 时调用方回退 _derive_from_results(向后兼容)。"""
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)
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
if source_price_cents is not None and best_price_cents is not None:
saved_amount_cents = source_price_cents - best_price_cents
has_valid_target = any(
p.get("role") != "source" and p.get("price") is not None for p in rows
)
# store_name: 优先源行; recompare 场景源平台自己当目标、源行被目标覆盖(pricebot
# _build_platform_rows 有意去重, platforms 无 role=source 行)→ 回退 best 行 → 首个有店名
# 的行(显示现场实际比到的店), 免得记录页店名空掉兜底显示成"比价"。正常比价有源行不走回退。
store_name = (
(src or {}).get("store_name")
or (best or {}).get("store_name")
or next((p.get("store_name") for p in rows if p.get("store_name")), None)
)
return {
"source_platform_id": (src or {}).get("platform_id"),
"source_platform_name": (src or {}).get("platform_name"),
"source_package": (src or {}).get("package"),
"source_price_cents": source_price_cents,
"best_platform_id": (best or {}).get("platform_id"),
"best_platform_name": (best or {}).get("platform_name"),
"best_price_cents": best_price_cents,
"saved_amount_cents": saved_amount_cents,
"is_source_best": (best.get("role") == "source") if best else None,
"store_name": store_name or None,
"status": record_status or ("success" if has_valid_target else "failed"),
}
@@ -493,7 +561,29 @@ def harvest_done(
返回 (记录, 是否本次**新**落成 success)——供调用方据此幂等发一次邀请奖。
行不存在(理论上帧0已建;防御)则新建。"""
results = done_params.get("comparison_results") or []
derived = _derive_from_results(results, done_params.get("platform_results"))
# 展示模型统一数组(pricebot 新增, 每平台一行自带 status/is_best): 原样存, 记录页据此直渲染。
# record_status: 记录级结局(success/below_minimum/store_closed/failed), 覆盖老二态派生。
platforms = done_params.get("platforms") or []
record_status = done_params.get("record_status")
# 单源派生: platforms(含 pricebot 权威 is_best)是唯一真相源, best_*/source_*/saved/status
# 全从它取 → 与前端读的 platforms 天然一致; 菜品也取 platforms 源行。老 pricebot 无
# platforms 时回退从 comparison_results 派生(向后兼容)。
if platforms:
derived = _derive_from_platforms(platforms, record_status)
# 菜品优先源行; recompare 无源行 → 回退 best 行 → 首个有菜品的行(同 store_name 回退)
_item_row = (
next((p for p in platforms if isinstance(p, dict) and p.get("role") == "source"), None)
or next((p for p in platforms if isinstance(p, dict) and p.get("is_best")), None)
or next((p for p in platforms if isinstance(p, dict) and p.get("items")), None)
)
items = (_item_row or {}).get("items") or []
else:
derived = _derive_from_results(
results, done_params.get("platform_results"), record_status
)
# pricebot 已把源单菜品塞进 comparison_results[源行].items
items = next((r.get("items") or [] for r in results if r.get("is_source")), [])
# 失败展示原因(#189): platforms / results 两个派生分支的 status 都可能 failed, 统一在此算
fail_reason = (
_derive_fail_display(
done_params.get("information"), done_params.get("platform_results")
@@ -501,8 +591,6 @@ def harvest_done(
if derived["status"] == "failed"
else None
)
# 菜品:pricebot 已把源单菜品塞进 comparison_results[源行].items
items = next((r.get("items") or [] for r in results if r.get("is_source")), [])
fields = dict(
business_type=business_type or "food",
information=done_params.get("information") or None,
@@ -514,6 +602,7 @@ def harvest_done(
skipped_dish_count=done_params.get("skipped_dish_count"),
skipped_dish_names=list(done_params.get("skipped_dish_names") or []),
comparison_results=results,
platforms=platforms,
items=items,
product_names=_product_names_from_items(items),
raw_payload=done_params,
+45 -22
View File
@@ -30,7 +30,11 @@ from app.models.app_config import AppConfig
from app.models.guide_video import GuideVideoPlay
from app.repositories import wallet as crud_wallet
_KEY = "coupon_guide_video"
SCENES = ("coupon", "comparison")
_KEY_BY_SCENE = {
"coupon": "coupon_guide_video",
"comparison": "comparison_guide_video",
}
#: 金币流水 biz_type。客户端收益明细按它显示「新手引导视频奖励」。
BIZ_TYPE = "guide_video"
@@ -41,7 +45,7 @@ _DEFAULTS: dict[str, Any] = {
"enabled": True,
"video_url": None, # None/空 = 未配片 → 不下发,浮层照旧放广告
"max_plays": 3, # 每个账号前 N 次浮层放引导视频
"reward_coin": 120, # 每次固定金币
"reward_coin": 100, # 每次固定金币
}
_FIELDS = tuple(_DEFAULTS.keys())
@@ -65,19 +69,28 @@ def _merge(raw: Any) -> dict[str, Any]:
return out
def get_config(db: Session) -> dict[str, Any]:
def _config_key(scene: str) -> str:
if scene not in _KEY_BY_SCENE:
raise ValueError(f"unsupported guide video scene: {scene}")
return _KEY_BY_SCENE[scene]
def get_config(db: Session, scene: str = "coupon") -> dict[str, Any]:
"""完整配置 + updated_at(admin 读 / 业务读共用)。"""
row = db.get(AppConfig, _KEY)
row = db.get(AppConfig, _config_key(scene))
cfg = _merge(row.value if row is not None else None)
cfg["updated_at"] = row.updated_at.isoformat() if row is not None and row.updated_at else None
return cfg
def _write(db: Session, value: dict[str, Any], *, admin_id: int, commit: bool) -> dict[str, Any]:
def _write(
db: Session, value: dict[str, Any], *, scene: str, admin_id: int, commit: bool
) -> dict[str, Any]:
"""整体覆写该行(value 须为完整字段 dict),返回合并后的完整配置(含 updated_at)。"""
row = db.get(AppConfig, _KEY)
key = _config_key(scene)
row = db.get(AppConfig, key)
if row is None:
row = AppConfig(key=_KEY, value=value, updated_by_admin_id=admin_id)
row = AppConfig(key=key, value=value, updated_by_admin_id=admin_id)
db.add(row)
else:
row.value = value # 整体重新赋值,SQLAlchemy 才侦测得到变更
@@ -98,11 +111,12 @@ def update_config(
enabled: bool | None = None,
max_plays: int | None = None,
reward_coin: int | None = None,
scene: str = "coupon",
admin_id: int,
commit: bool = True,
) -> tuple[dict[str, Any], dict[str, Any]]:
"""改开关 / 次数 / 金币(只改传了的字段;视频走 set_video)。返回 (before, after) 供审计。"""
row = db.get(AppConfig, _KEY)
row = db.get(AppConfig, _config_key(scene))
before = _merge(row.value if row is not None else None)
new_value = {k: before[k] for k in _FIELDS}
if enabled is not None:
@@ -111,45 +125,52 @@ def update_config(
new_value["max_plays"] = max(0, min(int(max_plays), MAX_PLAYS_LIMIT))
if reward_coin is not None:
new_value["reward_coin"] = max(0, min(int(reward_coin), REWARD_COIN_LIMIT))
after = _write(db, new_value, admin_id=admin_id, commit=commit)
after = _write(db, new_value, scene=scene, admin_id=admin_id, commit=commit)
return before, after
def set_video(
db: Session, video_url: str | None, *, admin_id: int, commit: bool = True
db: Session, video_url: str | None, *, scene: str = "coupon",
admin_id: int, commit: bool = True
) -> tuple[dict[str, Any], dict[str, Any]]:
"""设置/清空引导视频地址。返回 (before, after);before['video_url'] 供调用方删旧文件。"""
row = db.get(AppConfig, _KEY)
row = db.get(AppConfig, _config_key(scene))
before = _merge(row.value if row is not None else None)
new_value = {k: before[k] for k in _FIELDS}
new_value["video_url"] = video_url
after = _write(db, new_value, admin_id=admin_id, commit=commit)
after = _write(db, new_value, scene=scene, admin_id=admin_id, commit=commit)
return before, after
# ===== 播放计次 =====
def used_plays(db: Session, user_id: int) -> int:
def used_plays(db: Session, user_id: int, scene: str = "coupon") -> int:
"""该账号已用掉的引导视频次数(开播即算,含未发币的)。"""
return int(
db.execute(
select(func.count()).select_from(GuideVideoPlay).where(
GuideVideoPlay.user_id == user_id
GuideVideoPlay.user_id == user_id,
GuideVideoPlay.scene == scene,
)
).scalar_one()
)
def play_stats(db: Session) -> dict[str, int]:
def play_stats(db: Session, scene: str = "coupon") -> dict[str, int]:
"""全站播放统计(admin 页展示):总播放次数 / 其中已发币次数。"""
total = int(
db.execute(select(func.count()).select_from(GuideVideoPlay)).scalar_one()
db.execute(
select(func.count()).select_from(GuideVideoPlay).where(
GuideVideoPlay.scene == scene
)
).scalar_one()
)
granted = int(
db.execute(
select(func.count()).select_from(GuideVideoPlay).where(
GuideVideoPlay.status == "granted"
GuideVideoPlay.status == "granted",
GuideVideoPlay.scene == scene,
)
).scalar_one()
)
@@ -168,11 +189,12 @@ def start_play(
reward_coin 播完/中途关闭都发的固定金币
seq / remaining 第几次 / 发完这次还剩几次(仅展示与排查用)
"""
cfg = get_config(db)
_config_key(scene)
cfg = get_config(db, scene)
video_url = (cfg.get("video_url") or "").strip()
max_plays = int(cfg.get("max_plays") or 0)
reward_coin = int(cfg.get("reward_coin") or 0)
used = used_plays(db, user_id)
used = used_plays(db, user_id, scene)
def _miss(used_now: int) -> dict[str, Any]:
return {
@@ -194,7 +216,7 @@ def start_play(
scene=scene,
seq=seq,
video_url=video_url,
coin=0,
coin=reward_coin,
status="playing",
completed=0,
started_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
@@ -210,7 +232,7 @@ def start_play(
db.flush()
except IntegrityError:
db.rollback()
return _miss(used_plays(db, user_id))
return _miss(used_plays(db, user_id, scene))
return {
"should_play": True,
"video_url": video_url,
@@ -241,7 +263,8 @@ def grant_play(
"""
token = (play_token or "").strip()
# 金币额度以**服务端配置**为准,不信客户端(客户端只上报"播完/关闭")。
coin = int(get_config(db).get("reward_coin") or 0)
play = _find_play(db, user_id, token)
coin = int(play.coin if play is not None else 0)
# 幂等核心:把 status 放进 WHERE 做条件更新(compare-and-set),而不是"先读再判再写"。
# 「播完」与「✕ 关闭」抢跑、或客户端超时重试时,两个请求会都读到 status='playing',
-5
View File
@@ -67,11 +67,6 @@ class JverifyLoginRequest(BaseModel):
device_model: str = Field(
"", max_length=128, description="客户端设备型号快照,用于登录安全审计"
)
provider: str = Field(
"jiguang",
description="一键登录厂商:jiguang(默认)/aliyun。决定后端用哪家换号,"
"必须与客户端拉授权页的 SDK 一致。老客户端不带→默认极光(向后兼容)。",
)
# ===== 短信验证码 =====
+10
View File
@@ -107,6 +107,13 @@ class ComparisonRecordIn(BaseModel):
# 明细
items: list[ComparisonItemIn] = Field(default_factory=list)
comparison_results: list[ComparisonResultIn] = Field(default_factory=list)
# 展示模型统一数组(pricebot done.params.platforms 原样透传): 每平台一行、自带
# status/is_best/display/display_order,记录页据此直渲染。宽松 list[dict] 存(结构由
# pricebot 定,server 只原样落库),前端读它、老记录空时回退 comparison_results。
platforms: list[dict] = Field(default_factory=list)
# 记录级结局(pricebot 下发): success/below_minimum/store_closed/failed。让"未满起送"不再
# 被塌缩成 failed。_derive 优先用它、其次客户端 status、再兜底二态派生。
record_status: str | None = None
# 逐平台结局摘要(含失败平台的细分原因 status: store_not_found/items_not_found/below_minimum/
# unsupported/...)。来自 done.params.platform_results,客户端透传;落 raw_payload(不单列),
# admin「卡在哪一步」从这里读。dict{platform_id: {...}} 宽松存(结构由 pricebot 定——是
@@ -177,6 +184,9 @@ class ComparisonRecordOut(BaseModel):
fail_reason: str | None = None
items: list = []
comparison_results: list = []
# 展示模型统一数组(每平台一行、自带 status/is_best/display/display_order): 记录页据此
# 直渲染, 不再靠 comparison_results + 前端派生。老记录为空 → 前端回退 comparison_results。
platforms: list = []
skipped_dish_names: list = []
total_ms: int | None = None
# 「已下单」(店级):该店名在该用户真实下单(source='compare')里出现过即 True。
+3 -1
View File
@@ -1,13 +1,15 @@
"""新手引导视频(领券等候浮层前 N 次替代广告)的客户端请求/响应契约。"""
from __future__ import annotations
from typing import Literal
from pydantic import BaseModel, Field
class GuideVideoStartIn(BaseModel):
"""开播询问。scene 目前只有 coupon(领券浮层);预留给日后比价等场景。"""
scene: str = Field(default="coupon", max_length=16)
scene: Literal["coupon", "comparison"] = "coupon"
class GuideVideoStartOut(BaseModel):
@@ -1,479 +0,0 @@
# 阿里云一键登录 · 后端实施计划 (M1)
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:**`POST /api/v1/auth/jverify-login` 支持 `provider=aliyun`,用客户端 loginToken 调阿里云 Dypnsapi `GetMobile` 换明文手机号完成登录;极光/创蓝不受影响。
**Architecture:** 复用已存在的 `app/integrations/oneclick.py` 门面(按 provider 分派)。新增 `aliyun_onekey.py` 换号 provider(结构仿 `chuanglan_onekey.py`SDK 调用仿 `sms/aliyun.py` 的惰性 client)。把端点从直连 `jiguang` 改走门面并读 `req.provider`(这一步会让已存在但未接线的 `tests/test_jverify_login_endpoint.py` 转绿)。
**Tech Stack:** FastAPI + Pydantic v2 + pytest + `alibabacloud_dypnsapi20170525`(短信侧已引)。
**参考规范:** `../specs/2026-07-28-aliyun-oneclick-login-design.md`(在 android 仓)。
**运行测试:** `.venv/Scripts/python -m pytest tests/test_oneclick.py tests/test_aliyun_onekey.py tests/test_jverify_login_endpoint.py -q`。(conftest 用临时文件 SQLite,端点测试**无需 PG**`SGB_TEST_SKIP_DB` 未被 conftest 使用。)
> **执行状态(2026-07-28):M1 已实施并验证通过(未提交,待授权)。**
> 实施中发现原计划前提有误:`oneclick.py` 门面 / `chuanglan_onekey.py` / `test_oneclick.py` / `test_jverify_login_endpoint.py` 均为**未提交的创蓝评估 WIP,已清除,git 全程无记录**——门面并不存在。遂按「建小 facadejiguang+aliyun,弃创蓝)」重做,顺序:配置 → aliyun_onekey → **新建** oneclick 门面 → 接线 auth.py 两处 + provider 字段 → **新建**端点测试。
> 验证:新增 15 单测全绿(aliyun_onekey 4 / oneclick 8 / 端点 3+ 微信绑号回归绿;全量 `pytest` 与改动前逐一对照——同样 8 个既有失败(compare/coupon/invite/withdraw_tiers/notification,均与本次无关:SQLite/顺序依赖),本次 **0 新增失败**617→632 passed)。
> 下方 B1–B5 为实现参考(代码与落地一致);「已存在的门面/测试转绿」等措辞按本节修正理解,实际是先建 aliyun_onekey/facade 再接线。
---
### Task B1: 端点接入 oneclick 门面 + 加 `provider` 字段
已存在的 `tests/test_jverify_login_endpoint.py` 已按「门面已接线」写(monkeypatch `auth.verify_and_get_phone` 为 2 参、POST 带 `provider`),但线上 `auth.py` 仍直连 jiguang。本任务把它接上,使这些测试转绿。
**Files:**
- Modify: `app/schemas/auth.py:60-69``JverifyLoginRequest``provider`
- Modify: `app/api/v1/auth.py:36`import 改门面)、`:112-157``jverify_login` 调用/异常/埋点)、`:495-498`(微信 `bind-phone/jverify` 端点同一 `verify_and_get_phone`/`JiguangError`,一并改)
- Test: `tests/test_jverify_login_endpoint.py`(已存在,勿改,作为验收);回归 `tests/test_wechat_login.py``tests/test_wechat_conflict.py`(都覆盖微信绑号极光换号路径)
> ⚠️ `auth.py` 有**两处**用 `verify_and_get_phone` + `JiguangError``jverify_login`(119) 和微信 `bind-phone/jverify`(495)。换 import 会同时影响两处,必须都改;否则微信绑号编译/运行报错。
- [ ] **Step 1: 跑现有端点测试确认基线**
Run: `pytest tests/test_jverify_login_endpoint.py -v`
Expected: 3 条用例 **FAIL**(当前端点用 1 参 `verify_and_get_phone(req.login_token)`,被 monkeypatch 成 2 参后 `TypeError`)或在无 PG 时 **SKIP**。任一情况都说明「尚未接线」。
- [ ] **Step 2: schema 加 `provider` 字段**
`app/schemas/auth.py``JverifyLoginRequest`(当前 60-69 行)末尾加一个字段:
```python
class JverifyLoginRequest(BaseModel):
login_token: str = Field(..., description="客户端 loginAuth 拿到的 loginToken", min_length=1)
operator: str = Field("", description="CM/CU/CT,用于日志,可选")
device_id: str = Field(
"", max_length=64,
description="硬件级设备标识(Android ANDROID_ID),用于新手引导按 设备+账号 去重;空=按未完成处理",
)
device_model: str = Field(
"", max_length=128, description="客户端设备型号快照,用于登录安全审计"
)
provider: str = Field(
"jiguang",
description="一键登录厂商:jiguang(默认)/aliyun/chuanglan。决定后端用哪家换号,"
"必须与客户端拉授权页的 SDK 一致。老客户端不带→默认极光(向后兼容)。",
)
```
- [ ] **Step 3: 端点改走门面**
`app/api/v1/auth.py` 第 36 行的 import 从 jiguang 换成门面:
```python
# 旧: from app.integrations.jiguang import JiguangError, mask_phone, verify_and_get_phone
from app.integrations.oneclick import OneClickError, mask_phone, verify_and_get_phone
```
`jverify_login`(当前 102-157 行)里的换号调用与异常分支改为:
```python
try:
phone = verify_and_get_phone(req.provider, req.login_token)
except OneClickError as e:
risk_repo.record_behavior_event(
db,
event_type=risk_repo.EVENT_ONECLICK_LOGIN,
subject_type="device",
subject_id=subject_id,
device_id=req.device_id or None,
device_model=req.device_model or None,
client_ip=_client_ip(request),
outcome="failed",
reason=str(e),
details={"operator": req.operator or None, "provider": req.provider or None},
evaluate_rule=risk_repo.RULE_ONECLICK_DAILY,
)
logger.error("[oneclick] verify+decrypt failed: %s", e, exc_info=True)
raise HTTPException(status_code=502, detail=f"oneclick verify failed: {e}") from e
```
并把该函数里成功埋点的 `details={"operator": req.operator or None}`(约 149 行)同样补上 `"provider": req.provider or None`;日志行(约 112-116)改成打印 `req.provider`
换 import 后 `JiguangError` 不再被引用,从第 36 行 import 去掉(只留 `OneClickError, mask_phone, verify_and_get_phone`)。
- [ ] **Step 3b: 微信 `bind-phone/jverify` 端点同步改(否则编译/运行报错)**
`auth.py` 第 495-498 行的第二处换号(微信登录·本机号极光绑定)也走门面。该请求体 `WechatBindPhoneJverifyRequest` **无** `provider` 字段,M1 保持极光不变,显式传字面量 `"jiguang"`
```python
try:
phone = verify_and_get_phone("jiguang", req.login_token)
except OneClickError as e:
raise HTTPException(status_code=502, detail=f"oneclick verify failed: {e}") from e
```
> 说明:当客户端 `WECHAT_BIND` 皮肤改用阿里云后,此端点 + `WechatBindPhoneJverifyRequest` 也要加 `provider`(否则阿里云 token 被极光换号必失败)。这属**后续增强**,不在 M1;已在 spec 待办登记。
- [ ] **Step 4: 跑端点测试转绿**
Run: `pytest tests/test_jverify_login_endpoint.py -v`(有 PG
Expected: 3 条全 **PASS**(默认 provider 建号、provider 原样透传、OneClickError→502)。
- [ ] **Step 5: 跑门面 + 微信绑号测试不回归**
Run: `SGB_TEST_SKIP_DB=1 pytest tests/test_oneclick.py -v`(门面本就 2 参,未动 → PASS)
Run(有 PG: `pytest tests/test_wechat_login.py tests/test_wechat_conflict.py -v`
Expected: 全 **PASS**(微信绑号极光换号路径改走门面后行为不变)。
- [ ] **Step 6: Commit**
```bash
git add app/schemas/auth.py app/api/v1/auth.py
git commit -m "feat(auth): 一键登录/微信绑号接入 oneclick 门面并支持 provider 字段"
```
---
### Task B2: 阿里云一键登录配置项
**Files:**
- Modify: `app/core/config.py`(极光段 68-70 之后、或阿里云短信段 149-158 附近加一段)
- Modify: `.env.example`
- [ ] **Step 1: 加配置字段 + 已配置属性**
`app/core/config.py``Settings` 类里新增(放在阿里云短信段落之后,与 `aliyun_sms_configured` 属性相邻):
```python
# ===== 阿里云号码认证·一键登录(Dypnsapi GetMobile 换号)=====
# 与短信同属 dypnsapi 产品:同一阿里云账号时可复用 ALIYUN_SMS_ACCESS_KEY_*,
# 默认独立字段解耦(见 spec §2 决策)。缺凭证 → provider=aliyun 换号抛错→502,不启动崩。
ALIYUN_ONEKEY_ACCESS_KEY_ID: str = ""
ALIYUN_ONEKEY_ACCESS_KEY_SECRET: str = ""
ALIYUN_ONEKEY_ENDPOINT: str = "dypnsapi.aliyuncs.com"
ALIYUN_ONEKEY_TIMEOUT_SEC: int = 15
```
并在 `aliyun_sms_configured` 属性下方加:
```python
@property
def aliyun_oneclick_configured(self) -> bool:
"""阿里云一键登录凭证齐全(缺则 provider=aliyun 换号抛错→502,而非启动崩)。"""
return bool(
self.ALIYUN_ONEKEY_ACCESS_KEY_ID and self.ALIYUN_ONEKEY_ACCESS_KEY_SECRET
)
```
- [ ] **Step 2: 补 `.env.example`**
`.env.example` 里对应位置追加(值留空,注释说明可复用短信 AK/SK):
```dotenv
# 阿里云号码认证·一键登录(Dypnsapi GetMobile)。同账号可填与 ALIYUN_SMS_ACCESS_KEY_* 相同的值。
ALIYUN_ONEKEY_ACCESS_KEY_ID=
ALIYUN_ONEKEY_ACCESS_KEY_SECRET=
ALIYUN_ONEKEY_ENDPOINT=dypnsapi.aliyuncs.com
ALIYUN_ONEKEY_TIMEOUT_SEC=15
```
- [ ] **Step 3: 冒烟导入不报错**
Run: `SGB_TEST_SKIP_DB=1 python -c "from app.core.config import settings; print(settings.aliyun_oneclick_configured)"`
Expected: 打印 `False`(默认空凭证)。
- [ ] **Step 4: Commit**
```bash
git add app/core/config.py .env.example
git commit -m "feat(config): 增加阿里云一键登录换号配置项"
```
---
### Task B3: 实现 `aliyun_onekey` 换号 providerTDD
**Files:**
- Create: `app/integrations/aliyun_onekey.py`
- Test: `tests/test_aliyun_onekey.py`
- [ ] **Step 1: 写失败测试**
Create `tests/test_aliyun_onekey.py`
```python
"""阿里云一键登录换号 provider 单测:monkeypatch SDK 接缝 _call_get_mobile,不触真 SDK/网络。
SGB_TEST_SKIP_DB=1 pytest tests/test_aliyun_onekey.py
"""
from __future__ import annotations
import pytest
from app.core.config import settings
from app.integrations import aliyun_onekey
def _configure(monkeypatch):
monkeypatch.setattr(settings, "ALIYUN_ONEKEY_ACCESS_KEY_ID", "ak")
monkeypatch.setattr(settings, "ALIYUN_ONEKEY_ACCESS_KEY_SECRET", "sk")
def test_get_phone_success(monkeypatch):
_configure(monkeypatch)
monkeypatch.setattr(
aliyun_onekey, "_call_get_mobile",
lambda t: {"success": True, "code": "OK", "message": "OK", "mobile": "13800138000"},
)
assert aliyun_onekey.verify_and_get_phone("tok") == "13800138000"
def test_api_failure_raises(monkeypatch):
_configure(monkeypatch)
monkeypatch.setattr(
aliyun_onekey, "_call_get_mobile",
lambda t: {"success": False, "code": "MobileNumberIllegal", "message": "x", "mobile": None},
)
with pytest.raises(aliyun_onekey.AliyunOneClickError):
aliyun_onekey.verify_and_get_phone("tok")
def test_non_phone_result_raises(monkeypatch):
_configure(monkeypatch)
monkeypatch.setattr(
aliyun_onekey, "_call_get_mobile",
lambda t: {"success": True, "code": "OK", "message": "OK", "mobile": "not-a-phone"},
)
with pytest.raises(aliyun_onekey.AliyunOneClickError):
aliyun_onekey.verify_and_get_phone("tok")
def test_not_configured_raises(monkeypatch):
monkeypatch.setattr(settings, "ALIYUN_ONEKEY_ACCESS_KEY_ID", "")
monkeypatch.setattr(settings, "ALIYUN_ONEKEY_ACCESS_KEY_SECRET", "")
with pytest.raises(aliyun_onekey.AliyunOneClickError):
aliyun_onekey.verify_and_get_phone("tok")
```
- [ ] **Step 2: 跑测试确认失败**
Run: `SGB_TEST_SKIP_DB=1 pytest tests/test_aliyun_onekey.py -v`
Expected: FAIL`ModuleNotFoundError: app.integrations.aliyun_onekey`
- [ ] **Step 3: 实现 provider**
Create `app/integrations/aliyun_onekey.py`
```python
"""阿里云号码认证·一键登录服务端换号(Dypnsapi GetMobile)。
链路:
Android 阿里云 SDK getLoginToken → spToken(access_token)
→ 本服务调 Dypnsapi GetMobile(AccessToken=spToken)
→ 阿里云直接返回明文手机号(无需 RSA/AES,比极光/创蓝少一步解密)
与 jiguang/chuanglan 对齐:对外暴露 verify_and_get_phone(login_token)->str,
失败抛 AliyunOneClickError,由 oneclick.py 门面统一 catch。
SDK 交互隔离在 _call_get_mobile 薄封装(惰性 import + 惰性建 client,仿 sms/aliyun.py),
单测 monkeypatch 它即可。凭证复用/独立见 config.ALIYUN_ONEKEY_*。
"""
from __future__ import annotations
import logging
from app.core.config import settings
logger = logging.getLogger("shagua.aliyun.onekey")
_client = None # 惰性构建的 dypnsapi client(模块级缓存)
class AliyunOneClickError(Exception):
"""阿里云取号失败的统一异常,由 oneclick 门面 catch 翻成 4xx/5xx。"""
def verify_and_get_phone(login_token: str) -> str:
"""对外唯一函数:loginToken(access_token) → 明文手机号。失败抛 AliyunOneClickError。"""
if not settings.aliyun_oneclick_configured:
raise AliyunOneClickError("ALIYUN_ONEKEY_ACCESS_KEY_ID/SECRET not configured")
result = _call_get_mobile(login_token)
if not (result["success"] and result["code"] == "OK"):
logger.error(
"[ALIYUN-ONEKEY] get_mobile failed code=%s msg=%s",
result["code"], result["message"],
)
raise AliyunOneClickError(f"aliyun get_mobile failed code={result['code']}")
phone = (result["mobile"] or "").strip()
if not (phone.isdigit() and len(phone) == 11):
logger.error("[ALIYUN-ONEKEY] unexpected mobile format: %r", phone)
raise AliyunOneClickError("aliyun get_mobile returned non-phone")
return phone
# ==================== SDK 接缝(单测 monkeypatch 这个)====================
def _get_client():
"""惰性构建 dypnsapi client(仿 sms/aliyun.py:jiguang-only 部署不加载 alibabacloud)。"""
global _client
if _client is None:
from alibabacloud_dypnsapi20170525.client import Client
from alibabacloud_tea_openapi import models as open_api_models
cfg = open_api_models.Config(
access_key_id=settings.ALIYUN_ONEKEY_ACCESS_KEY_ID,
access_key_secret=settings.ALIYUN_ONEKEY_ACCESS_KEY_SECRET,
read_timeout=settings.ALIYUN_ONEKEY_TIMEOUT_SEC * 1000, # SDK 单位 ms
connect_timeout=settings.ALIYUN_ONEKEY_TIMEOUT_SEC * 1000,
)
cfg.endpoint = settings.ALIYUN_ONEKEY_ENDPOINT
_client = Client(cfg)
return _client
def _call_get_mobile(login_token: str) -> dict:
"""调 GetMobile。返回归一化 {success, code, message, mobile};import/建 client/调用任一失败抛 AliyunOneClickError。"""
try:
from alibabacloud_dypnsapi20170525 import models as dypns_models
req = dypns_models.GetMobileRequest(access_token=login_token)
body = _get_client().get_mobile(req).body
except Exception as e:
logger.exception("[ALIYUN-ONEKEY] get_mobile 调用异常")
raise AliyunOneClickError("aliyun get_mobile 调用异常") from e
dto = getattr(body, "get_mobile_result_dto", None)
mobile = getattr(dto, "mobile", None) if dto else None
return {
"success": (body.code == "OK"),
"code": body.code,
"message": body.message,
"mobile": mobile,
}
```
> 注:`GetMobile` 响应体字段名以实际 SDK 为准(`code`=="OK"、`get_mobile_result_dto.mobile`)。接真号联调时若字段名不同(如 `mobile` 在别的 dto 下),只需改 `_call_get_mobile` 的最后归一化,`verify_and_get_phone` 及测试不动。
- [ ] **Step 4: 跑测试转绿**
Run: `SGB_TEST_SKIP_DB=1 pytest tests/test_aliyun_onekey.py -v`
Expected: 4 条全 **PASS**
- [ ] **Step 5: Commit**
```bash
git add app/integrations/aliyun_onekey.py tests/test_aliyun_onekey.py
git commit -m "feat(auth): 新增阿里云一键登录换号 provider(Dypnsapi GetMobile)"
```
---
### Task B4: 门面注册 aliyun 分派(TDD
**Files:**
- Modify: `app/integrations/oneclick.py`
- Test: `tests/test_oneclick.py`(加两条)
- [ ] **Step 1: 加失败测试**
`tests/test_oneclick.py` 顶部 import 加 `aliyun_onekey`
```python
from app.integrations import aliyun_onekey, chuanglan_onekey, jiguang, oneclick
```
文件末尾追加:
```python
def test_dispatch_aliyun(monkeypatch):
seen = {}
def fake_al(t):
seen["al"] = t
return "13855555555"
monkeypatch.setattr(aliyun_onekey, "verify_and_get_phone", fake_al)
assert oneclick.verify_and_get_phone("aliyun", "tok") == "13855555555"
assert seen["al"] == "tok"
def test_aliyun_error_wrapped_as_oneclick_error(monkeypatch):
def boom(_t):
raise aliyun_onekey.AliyunOneClickError("aliyun down")
monkeypatch.setattr(aliyun_onekey, "verify_and_get_phone", boom)
with pytest.raises(oneclick.OneClickError):
oneclick.verify_and_get_phone("aliyun", "tok")
```
- [ ] **Step 2: 跑确认失败**
Run: `SGB_TEST_SKIP_DB=1 pytest tests/test_oneclick.py::test_dispatch_aliyun -v`
Expected: FAIL`aliyun` 未注册 → 走默认极光 → `13855555555` 不匹配 / `seen``al`)。
- [ ] **Step 3: 注册 aliyun 分派**
`app/integrations/oneclick.py` 三处改动:
```python
# import(第 17 行)加入 aliyun_onekey:
from app.integrations import aliyun_onekey, chuanglan_onekey, jiguang
# 常量段加:
PROVIDER_ALIYUN = "aliyun"
# verify_and_get_phone 分派与异常 catch:
def verify_and_get_phone(provider: str, login_token: str) -> str:
p = (provider or PROVIDER_JIGUANG).strip().lower()
try:
if p == PROVIDER_CHUANGLAN:
return chuanglan_onekey.verify_and_get_phone(login_token)
if p == PROVIDER_ALIYUN:
return aliyun_onekey.verify_and_get_phone(login_token)
return jiguang.verify_and_get_phone(login_token)
except (
jiguang.JiguangError,
chuanglan_onekey.ChuanglanError,
aliyun_onekey.AliyunOneClickError,
) as e:
raise OneClickError(f"[{p}] {e}") from e
```
- [ ] **Step 4: 跑门面测试全绿**
Run: `SGB_TEST_SKIP_DB=1 pytest tests/test_oneclick.py -v`
Expected: 全 **PASS**(含新增两条 + 原有创蓝/极光分派不回归)。
- [ ] **Step 5: Commit**
```bash
git add app/integrations/oneclick.py tests/test_oneclick.py
git commit -m "feat(auth): oneclick 门面注册阿里云 provider 分派"
```
---
### Task B5: 全量回归 + 真号联调冒烟
- [ ] **Step 1: 全量单测**
Run: `SGB_TEST_SKIP_DB=1 pytest tests/test_oneclick.py tests/test_aliyun_onekey.py -v` 及(有 PG`pytest tests/test_jverify_login_endpoint.py -v`
Expected: 全 PASS。
- [ ] **Step 2: 填真实凭证冒烟(联调期)**
`.env``ALIYUN_ONEKEY_ACCESS_KEY_ID/SECRET`(阿里云控制台号码认证方案对应账号 AK/SK),起服务,用客户端真机取到的 loginToken
Run:
```bash
curl -s -X POST http://127.0.0.1:8770/api/v1/auth/jverify-login \
-H 'Content-Type: application/json' \
-d '{"login_token":"<真机取到的token>","provider":"aliyun","device_id":"smoke"}'
```
Expected: 200 + `user.phone` 为真实手机号 + `access_token` 非空。失败看日志 `shagua.aliyun.onekey``GetMobile failed code=...`)。
- [ ] **Step 3: 无凭证降级冒烟**
清空 `ALIYUN_ONEKEY_*` 重启,`provider=aliyun` 请求应 **502**`OneClickError [aliyun] ... not configured`),且极光 `provider` 省略仍 200。
---
## 自检清单(实现者跑完对照)
- [ ] `provider` 缺省 = jiguang,老客户端不改也能登(B1)。
- [ ] `provider=aliyun` 端到端换号 200B5 Step2)。
- [ ] 任一厂商换号失败 → 502,不建号(B1 tests)。
- [ ] 缺阿里云凭证不 crash 启动,仅该 provider 502B2 property + B5 Step3)。
- [ ] 极光/创蓝分派与端点原行为不回归(B4 Step4 / B1 Step5)。
-51
View File
@@ -1,51 +0,0 @@
"""阿里云一键登录换号 provider 单测:monkeypatch SDK 接缝 _call_get_mobile,不触真 SDK/网络。
SGB_TEST_SKIP_DB=1 pytest tests/test_aliyun_onekey.py
"""
from __future__ import annotations
import pytest
from app.core.config import settings
from app.integrations import aliyun_onekey
def _configure(monkeypatch):
monkeypatch.setattr(settings, "ALIYUN_ONEKEY_ACCESS_KEY_ID", "ak")
monkeypatch.setattr(settings, "ALIYUN_ONEKEY_ACCESS_KEY_SECRET", "sk")
def test_get_phone_success(monkeypatch):
_configure(monkeypatch)
monkeypatch.setattr(
aliyun_onekey, "_call_get_mobile",
lambda t: {"success": True, "code": "OK", "message": "OK", "mobile": "13800138000"},
)
assert aliyun_onekey.verify_and_get_phone("tok") == "13800138000"
def test_api_failure_raises(monkeypatch):
_configure(monkeypatch)
monkeypatch.setattr(
aliyun_onekey, "_call_get_mobile",
lambda t: {"success": False, "code": "MobileNumberIllegal", "message": "x", "mobile": None},
)
with pytest.raises(aliyun_onekey.AliyunOneClickError):
aliyun_onekey.verify_and_get_phone("tok")
def test_non_phone_result_raises(monkeypatch):
_configure(monkeypatch)
monkeypatch.setattr(
aliyun_onekey, "_call_get_mobile",
lambda t: {"success": True, "code": "OK", "message": "OK", "mobile": "not-a-phone"},
)
with pytest.raises(aliyun_onekey.AliyunOneClickError):
aliyun_onekey.verify_and_get_phone("tok")
def test_not_configured_raises(monkeypatch):
monkeypatch.setattr(settings, "ALIYUN_ONEKEY_ACCESS_KEY_ID", "")
monkeypatch.setattr(settings, "ALIYUN_ONEKEY_ACCESS_KEY_SECRET", "")
with pytest.raises(aliyun_onekey.AliyunOneClickError):
aliyun_onekey.verify_and_get_phone("tok")
+1 -1
View File
@@ -201,7 +201,7 @@ def test_start_loses_seq_race_degrades_to_ad(client, guide_configured, monkeypat
assert first["should_play"] is True and first["seq"] == 1
# 本次请求读到的是过期计数 → 仍会算出 seq=1
monkeypatch.setattr(crud_guide, "used_plays", lambda db, user_id: 0)
monkeypatch.setattr(crud_guide, "used_plays", lambda db, user_id, scene="coupon": 0)
with SessionLocal() as db:
result = crud_guide.start_play(db, uid)
-50
View File
@@ -1,50 +0,0 @@
"""jverify-login 端点:provider 分派 + 错误映射(端到端:换号 → 建号 → 签 JWT)。
需要 DB(建号)。本机无 PG 时 client fixture 会自动 skip(SGB_TEST_SKIP_DB=1);
PG 环境(CI 或本地起 Docker)完整跑。换号一步 monkeypatch 掉, 只验端点编排:
provider 是否原样传给门面、成功建号、OneClickError → 502。
"""
from __future__ import annotations
from app.api.v1 import auth
from app.integrations.oneclick import OneClickError
def test_jverify_login_default_provider_builds_account(client, monkeypatch):
"""不带 provider 的老客户端 → 门面按默认(极光)换号, 建号登录成功。"""
monkeypatch.setattr(auth, "verify_and_get_phone", lambda provider, token: "13500000000")
r = client.post("/api/v1/auth/jverify-login", json={"login_token": "tok-old"})
assert r.status_code == 200
body = r.json()
assert body["user"]["phone"] == "13500000000"
assert body["access_token"]
def test_jverify_login_passes_provider_to_facade(client, monkeypatch):
"""provider=aliyun 必须原样传给门面(它决定用哪家换号)。"""
captured: dict = {}
def fake(provider, token):
captured["provider"] = provider
captured["token"] = token
return "13500000001"
monkeypatch.setattr(auth, "verify_and_get_phone", fake)
r = client.post(
"/api/v1/auth/jverify-login",
json={"login_token": "tok-al", "provider": "aliyun"},
)
assert r.status_code == 200
assert captured["provider"] == "aliyun"
assert captured["token"] == "tok-al"
def test_jverify_login_oneclick_error_maps_to_502(client, monkeypatch):
"""任一厂商换号失败(OneClickError) → 502, 不建号。"""
def boom(provider, token):
raise OneClickError("verify failed")
monkeypatch.setattr(auth, "verify_and_get_phone", boom)
r = client.post("/api/v1/auth/jverify-login", json={"login_token": "tok"})
assert r.status_code == 502
-74
View File
@@ -1,74 +0,0 @@
"""一键登录换号门面 oneclick 单测:按 provider 分派到极光/阿里云, 统一异常。
纯函数(monkeypatch 掉两家的 verify_and_get_phone), 不碰 DB。本机可跑:
SGB_TEST_SKIP_DB=1 pytest tests/test_oneclick.py
"""
from __future__ import annotations
import pytest
from app.integrations import aliyun_onekey, jiguang, oneclick
def test_dispatch_defaults_to_jiguang_when_blank(monkeypatch):
"""provider 为空 = 老客户端 → 走极光(向后兼容)。"""
seen = {}
def fake_jg(t):
seen["jg"] = t
return "13800000000"
monkeypatch.setattr(jiguang, "verify_and_get_phone", fake_jg)
assert oneclick.verify_and_get_phone("", "tok") == "13800000000"
assert seen["jg"] == "tok"
def test_dispatch_jiguang_explicit(monkeypatch):
monkeypatch.setattr(jiguang, "verify_and_get_phone", lambda t: "13811111111")
assert oneclick.verify_and_get_phone("jiguang", "tok") == "13811111111"
def test_dispatch_aliyun(monkeypatch):
seen = {}
def fake_al(t):
seen["al"] = t
return "13822222222"
monkeypatch.setattr(aliyun_onekey, "verify_and_get_phone", fake_al)
assert oneclick.verify_and_get_phone("aliyun", "tok") == "13822222222"
assert seen["al"] == "tok"
def test_provider_is_case_insensitive(monkeypatch):
monkeypatch.setattr(aliyun_onekey, "verify_and_get_phone", lambda t: "13844444444")
assert oneclick.verify_and_get_phone("Aliyun", "tok") == "13844444444"
def test_unknown_provider_falls_back_to_jiguang(monkeypatch):
"""未知 provider 兜底走极光(主家), 不因客户端传错值而拒登。"""
monkeypatch.setattr(jiguang, "verify_and_get_phone", lambda t: "13833333333")
assert oneclick.verify_and_get_phone("weird-value", "tok") == "13833333333"
def test_jiguang_error_wrapped_as_oneclick_error(monkeypatch):
def boom(_t):
raise jiguang.JiguangError("jg down")
monkeypatch.setattr(jiguang, "verify_and_get_phone", boom)
with pytest.raises(oneclick.OneClickError):
oneclick.verify_and_get_phone("jiguang", "tok")
def test_aliyun_error_wrapped_as_oneclick_error(monkeypatch):
def boom(_t):
raise aliyun_onekey.AliyunOneClickError("aliyun down")
monkeypatch.setattr(aliyun_onekey, "verify_and_get_phone", boom)
with pytest.raises(oneclick.OneClickError):
oneclick.verify_and_get_phone("aliyun", "tok")
def test_mask_phone_reexported():
"""auth 层只依赖 oneclick, 脱敏函数从门面转出。"""
assert oneclick.mask_phone("13800138000") == "138****00"
+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"
+5 -5
View File
@@ -147,8 +147,8 @@ def test_wechat_bind_jverify_creates_account(client, monkeypatch) -> None:
"""本机号(极光)绑定路径:verify_and_get_phone 拦掉,未占用 → 建号登入。"""
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_jv_5", "极光用户", None))
phone = "13900139005"
# loginToken→手机号 在 auth 模块命名空间打桩(auth.py 顶部 from ...oneclick import verify_and_get_phone,2 参 provider/token)
monkeypatch.setattr(auth, "verify_and_get_phone", lambda provider, token: phone)
# 极光 loginToken→手机号 在 auth 模块命名空间打桩(auth.py 顶部 from ...jiguang import verify_and_get_phone)
monkeypatch.setattr(auth, "verify_and_get_phone", lambda token: phone)
ticket = client.post(
"/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devE"}
@@ -181,11 +181,11 @@ def test_wechat_bind_jverify_expired_ticket_returns_401(client, monkeypatch) ->
def test_wechat_bind_jverify_jiguang_error_returns_502(client, monkeypatch) -> None:
"""换号失败(OneClickError)→ 502。"""
"""极光核验失败(JiguangError)→ 502。"""
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_jv_err"))
def _raise(provider: str, token: str) -> str:
raise auth.OneClickError("mock oneclick failure")
def _raise(token: str) -> str:
raise auth.JiguangError("mock jg failure")
monkeypatch.setattr(auth, "verify_and_get_phone", _raise)
ticket = client.post(