Compare commits

..

7 Commits

Author SHA1 Message Date
guke 3146224944 test(auth): 补 jverify 绑号 401/502 错误分支测试
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-13 15:31:46 +08:00
guke 3621d606ad feat(auth): POST /wechat/bind-phone/jverify(极光本机号绑号)
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-13 15:27:00 +08:00
guke c1b00eca0c test(auth): 补 phone_occupied 的 avatar_url 断言(锁定 M2 冲突页契约)
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-13 15:23:22 +08:00
guke a5fdcb53d8 feat(auth): POST /wechat/bind-phone/sms(短信绑号:建号/占用)
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-13 15:14:42 +08:00
guke fe9b749154 style(auth): 微信登录 import 去前向未用 + isort 排序
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-13 15:10:22 +08:00
guke 88313cefdf feat(auth): POST /wechat-login(openid 命中即登入,否则发绑号令牌)
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-13 14:57:36 +08:00
guke 2a82b20365 feat(auth): 微信登录 bind_ticket 令牌 + 配置
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-13 14:40:47 +08:00
30 changed files with 498 additions and 1357 deletions
-7
View File
@@ -32,13 +32,6 @@ HEARTBEAT_MONITOR_ENABLED=true
HEARTBEAT_TIMEOUT_MINUTES=60
HEARTBEAT_SCAN_INTERVAL_SEC=60
# ===== OPPO 推送(厂商通道直连,提现审核通过等交易类通知)=====
# OPPO 开放平台注册应用后分配, 服务端鉴权用 AppKey + MasterSecret(≠极光 JG_*)。
# 缺任一 → oppo_push_configured=False, 推送静默跳过(不影响审核/打款)。
OPPO_PUSH_APP_KEY=
OPPO_PUSH_MASTER_SECRET=
# 可选(一般不改): OPPO_PUSH_ENABLED=true / OPPO_PUSH_REQUEST_TIMEOUT_SEC=10
# ===== 短信 (mock 模式) =====
# mock = true 时,任意 6 位数字均通过,且 /sms/send 不真发短信(只 log)。
# 后续接阿里云/腾讯云短信时,改成 false 并填供应商相关 key。
-33
View File
@@ -1,33 +0,0 @@
"""comparison_record: llm_cost_yuan + llm_price_snapshot(比价 LLM 调用成本 + 当时单价快照)
回填 llm_calls 时按「当时的价」逐模型算出本次比价 LLM 总成本(元),连同所用单价快照一起冻结到
记录上;admin 比价记录详情展示实际成本(旧记录 NULL → 前端回退估算)。见 services/llm_cost.py。
Revision ID: comparison_llm_cost
Revises: ad_ecpm_trace_id
Create Date: 2026-07-13
"""
from collections.abc import Sequence
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
from alembic import op
revision: str = "comparison_llm_cost"
down_revision: str | Sequence[str] | None = "ad_ecpm_trace_id"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
_JSONB = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
def upgrade() -> None:
# 均可空、无索引;SQLite 原生支持 ADD COLUMN,无需 batch_alter_table(同 comparison_debug_fields)。
op.add_column("comparison_record", sa.Column("llm_cost_yuan", sa.Float(), nullable=True))
op.add_column("comparison_record", sa.Column("llm_price_snapshot", _JSONB, nullable=True))
def downgrade() -> None:
op.drop_column("comparison_record", "llm_price_snapshot")
op.drop_column("comparison_record", "llm_cost_yuan")
@@ -1,34 +0,0 @@
"""device_liveness 加 oppo_register_id 列(OPPO 直连推送目标)
提现审核通过等交易类通知走 OPPO 官方推送直连(≠极光)。客户端集成 OPPO SDK 后拿到 OPPO
registerID, 经 device/register + heartbeat 上报, 存本列; 与极光 registration_id 并列,
一台 OPPO 设备两套 token 并存。
Revision ID: device_oppo_register_id
Revises: comparison_llm_cost
Create Date: 2026-07-15 00:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'device_oppo_register_id'
down_revision: Union[str, Sequence[str], None] = 'comparison_llm_cost'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
with op.batch_alter_table('device_liveness', schema=None) as batch_op:
batch_op.add_column(
sa.Column('oppo_register_id', sa.String(length=128), nullable=True)
)
def downgrade() -> None:
with op.batch_alter_table('device_liveness', schema=None) as batch_op:
batch_op.drop_column('oppo_register_id')
-7
View File
@@ -37,7 +37,6 @@ from app.integrations import wxpay
from app.models.admin import AdminUser
from app.repositories import app_config
from app.repositories import wallet as wallet_repo
from app.services import push_notify
router = APIRouter(
prefix="/admin/api/withdraws",
@@ -316,9 +315,6 @@ def bulk_approve_withdraws(
results.append(
WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=True, status=order.status)
)
# 审核通过 → OPPO 推送(fire-and-forget, 不阻塞批量审核; status=failed 已退款, 不推)
if order.status != "failed":
push_notify.notify_withdraw_approved_async(order.user_id)
except wallet_repo.WithdrawOrderNotFound:
db.rollback()
results.append(
@@ -442,9 +438,6 @@ def approve_withdraw_order(
},
ip=get_client_ip(request), commit=True,
)
# 审核通过 → OPPO 推送(fire-and-forget, 不阻塞审核响应; status=failed 是打款失败已退款, 不推)
if order.status != "failed":
push_notify.notify_withdraw_approved_async(order.user_id)
return WithdrawOrderOut.model_validate(order)
-4
View File
@@ -36,8 +36,6 @@ class AdminComparisonListItem(BaseModel):
retry_count: int | None = None
input_tokens: int | None = None # Σ usage.prompt_tokens(server 派生)
output_tokens: int | None = None # Σ usage.completion_tokens(server 派生)
# 本次比价 LLM 总成本(元,按当时价冻结);旧记录/未回填为 None → 前端「成本」列回退估算。见 services/llm_cost.py。
llm_cost_yuan: float | None = None
device_model: str | None = None
rom_vendor: str | None = None
rom_name: str | None = None
@@ -74,5 +72,3 @@ class AdminComparisonDetail(AdminComparisonListItem):
# 原始上报全量;「卡在哪一步」从 raw_payload.platform_results[*].status 读
# (store_not_found/items_not_found/below_minimum/unsupported = 卡在 找店/加菜/起送/读价)。
raw_payload: dict | None = None
# 算成本所用单价快照 {mode, prices:{model:{...}}}(llm_cost_yuan 继承自列表项)。见 services/llm_cost.py。
llm_price_snapshot: dict | None = None
+165 -1
View File
@@ -17,7 +17,14 @@ from fastapi import APIRouter, HTTPException, Request, status
from app.api.deps import CurrentUser, DbSession
from app.core import test_account
from app.core.ratelimit import enforce_rate_limit
from app.core.security import TokenError, decode_token, issue_token_pair
from app.core.security import (
TokenError,
create_bind_ticket,
decode_bind_ticket,
decode_token,
issue_token_pair,
)
from app.integrations import wxpay
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
@@ -25,6 +32,7 @@ from app.repositories import user as user_repo
from app.schemas.auth import (
JverifyLoginRequest,
LogoutResponse,
OccupiedAccountInfo,
RefreshRequest,
SmsLoginRequest,
SmsSendRequest,
@@ -32,6 +40,11 @@ from app.schemas.auth import (
TokenPair,
TokenWithUser,
UserOut,
WechatBindPhoneJverifyRequest,
WechatBindPhoneSmsRequest,
WechatBindResultResponse,
WechatLoginRequest,
WechatLoginResponse,
)
logger = logging.getLogger("shagua.auth")
@@ -166,6 +179,157 @@ def sms_login(req: SmsLoginRequest, request: Request, db: DbSession) -> TokenWit
return _login_response(user, onboarding_completed=completed)
# ===================== 微信登录 =====================
@router.post(
"/wechat-login",
response_model=WechatLoginResponse,
summary="微信登录(openid 命中即登入,否则发绑号令牌)",
)
def wechat_login(req: WechatLoginRequest, db: DbSession) -> WechatLoginResponse:
from app.core.config import settings # 局部 import,避免循环
# 微信登录只需 code→openid(sns/oauth2),不需要商户转账证书;故只校验 APP_ID/SECRET。
if not (settings.WECHAT_APP_ID and settings.WECHAT_APP_SECRET):
raise HTTPException(status_code=503, detail="wechat login not configured")
try:
info = wxpay.code_to_userinfo(req.code) # {openid, nickname, avatar_url, raw};失败抛 ValueError
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
openid = info["openid"]
user = user_repo.get_user_by_wechat_openid(db, openid)
if user is not None:
# openid 命中 → 直接登入(绝不套用提现 bind-wechat 的"撞号即 409"逻辑)
if user.status != "active":
raise HTTPException(status_code=403, detail="account disabled")
user_repo.touch_last_login(db, user)
completed = onboarding_repo.is_completed(db, user_id=user.id, device_id=req.device_id)
logger.info("wechat_login hit user_id=%d openid=%s*** onboarded=%s", user.id, openid[:6], completed)
return WechatLoginResponse(
status="logged_in",
token=_login_response(user, onboarding_completed=completed),
)
# 未命中 → 签发短时 bind_ticket,进手机号绑定流程(账号此刻还不建)
ticket = create_bind_ticket(
openid=openid,
wechat_nickname=info["nickname"],
wechat_avatar_url=info["avatar_url"],
)
logger.info("wechat_login new openid=%s*** issue bind_ticket", openid[:6])
return WechatLoginResponse(
status="need_bind_phone",
bind_ticket=ticket,
wechat_nickname=info["nickname"],
wechat_avatar_url=info["avatar_url"],
)
def _finish_wechat_bind(
db,
*,
openid: str,
wechat_nickname: str | None,
wechat_avatar_url: str | None,
phone: str,
device_id: str,
) -> WechatBindResultResponse:
"""绑手机建号的公共尾段:手机号被占用 → 返回 phone_occupied(M2 处理 3 选 1);
未占用 → 新建微信账号(channel=wechat,昵称头像取微信)→ 签 token 登入。"""
existing = user_repo.get_user_by_phone(db, phone)
if existing is not None:
logger.info("wechat bind phone occupied phone=%s by user_id=%d", mask_phone(phone), existing.id)
return WechatBindResultResponse(
status="phone_occupied",
occupied_account=OccupiedAccountInfo(
nickname=existing.nickname,
avatar_url=existing.avatar_url,
created_at=existing.created_at,
),
)
user = user_repo.create_wechat_user(
db,
phone=phone,
openid=openid,
wechat_nickname=wechat_nickname,
wechat_avatar_url=wechat_avatar_url,
)
completed = onboarding_repo.is_completed(db, user_id=user.id, device_id=device_id)
logger.info("wechat bind ok user_id=%d phone=%s openid=%s*** onboarded=%s",
user.id, mask_phone(phone), openid[:6], completed)
return WechatBindResultResponse(
status="logged_in",
token=_login_response(user, onboarding_completed=completed),
)
@router.post(
"/wechat/bind-phone/sms",
response_model=WechatBindResultResponse,
summary="微信登录·其他手机号(短信)绑定",
)
def wechat_bind_phone_sms(
req: WechatBindPhoneSmsRequest, request: Request, db: DbSession
) -> WechatBindResultResponse:
try:
claims = decode_bind_ticket(req.bind_ticket)
except TokenError as e:
raise HTTPException(status_code=401, detail="授权已过期,请重新用微信登录") from e
# 防刷:同 sms/login,按 设备+IP 每小时限流(放在验证码校验之前,失败也计数)
enforce_rate_limit(
request,
scope="wechat-bind-sms-device",
subject=req.device_id,
limit=SMS_LOGIN_MAX_PER_HOUR,
window_sec=3600,
detail="登录尝试过于频繁,请稍后再试",
)
if not verify_code(req.phone, req.code):
raise HTTPException(status_code=400, detail="invalid sms code")
return _finish_wechat_bind(
db,
openid=claims["openid"],
wechat_nickname=claims["wnk"],
wechat_avatar_url=claims["wav"],
phone=req.phone,
device_id=req.device_id,
)
@router.post(
"/wechat/bind-phone/jverify",
response_model=WechatBindResultResponse,
summary="微信登录·本机号(极光)绑定",
)
def wechat_bind_phone_jverify(
req: WechatBindPhoneJverifyRequest, db: DbSession
) -> WechatBindResultResponse:
try:
claims = decode_bind_ticket(req.bind_ticket)
except TokenError as e:
raise HTTPException(status_code=401, detail="授权已过期,请重新用微信登录") from e
try:
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,
openid=claims["openid"],
wechat_nickname=claims["wnk"],
wechat_avatar_url=claims["wav"],
phone=phone,
device_id=req.device_id,
)
# ===================== Refresh =====================
@router.post("/refresh", response_model=TokenPair, summary="用 refresh_token 换新 token 对")
-3
View File
@@ -27,7 +27,6 @@ from app.schemas.compare_record import (
ComparisonRecordOut,
ComparisonRecordPage,
)
from app.services.llm_cost import compute_llm_cost, get_llm_prices
from app.services.pricebot_llm_calls import fetch_llm_calls
logger = logging.getLogger("shagua.compare_record")
@@ -82,8 +81,6 @@ def _backfill_llm_calls(record_id: int, trace_id: str) -> None:
# error 的调用 usage 可能为 None,or {} 兜底)
rec.input_tokens = sum((c.get("usage") or {}).get("prompt_tokens") or 0 for c in calls)
rec.output_tokens = sum((c.get("usage") or {}).get("completion_tokens") or 0 for c in calls)
# 本次比价 LLM 成本(元)+ 当时单价快照:按 app_config 现价逐模型算好冻结(services/llm_cost.py)。
rec.llm_cost_yuan, rec.llm_price_snapshot = compute_llm_cost(calls, get_llm_prices(db))
db.commit()
logger.info(
"backfill llm_calls trace=%s n=%d in_tok=%d out_tok=%d",
-2
View File
@@ -40,7 +40,6 @@ def register_device(
user_id=user.id,
device_id=req.device_id,
registration_id=req.registration_id,
oppo_register_id=req.oppo_register_id,
platform=req.platform,
app_version=req.app_version,
)
@@ -65,7 +64,6 @@ def report_heartbeat(
device_id=req.device_id,
accessibility_enabled=req.accessibility_enabled,
registration_id=req.registration_id,
oppo_register_id=req.oppo_register_id,
)
return OkResponse()
-106
View File
@@ -1,106 +0,0 @@
"""微信服务号消息接收回调 /wx/mp/callback(裸路径, 非 /api/v1)。
对应公众平台"设置与开发 → 基本配置 → 服务器配置"里填的 URL。
GET : URL 接入验证 → 校验 signature → 原样返回 echostr(明文)
POST : 收用户消息(安全模式密文) → msg_signature 验签 → AES 解密 → 解析 XML
图片消息(MsgType=image): 打日志(openid/PicUrl/MediaId) + 用 PicUrl 下载落盘
其余消息/事件: 记一条日志忽略。一律返回 "success"(微信据此不再重试)。
MVP 只做"收到图片并落盘"证明链路通;后续接:识别平台+订单 → 打 pending_compare 标记
→ 心跳带回前端弹窗,以及 openid↔device 绑定。任何异常都返回 "success"、不让微信重试轰炸。
"""
from __future__ import annotations
import logging
import xml.etree.ElementTree as ET
from pathlib import Path
import httpx
from fastapi import APIRouter, Request
from fastapi.responses import PlainTextResponse
from app.core.config import settings
from app.integrations import wx_mp_crypto
logger = logging.getLogger("shagua.wx_mp")
router = APIRouter(prefix="/wx/mp", tags=["wx-mp"])
# 收到的截图暂存目录(MVP 验证用;接上识别后可改为不落盘或定期清理)
_INBOX = Path(settings.MEDIA_ROOT) / "wx_inbox"
@router.get("/callback", include_in_schema=False)
async def verify(
signature: str = "", timestamp: str = "", nonce: str = "", echostr: str = ""
) -> PlainTextResponse:
"""微信 URL 接入验证:校验 signature 通过则原样返回 echostr。"""
if not settings.WX_MP_TOKEN:
logger.warning("wx_mp verify: WX_MP_TOKEN 未配置")
return PlainTextResponse("", status_code=503)
if wx_mp_crypto.verify_url_signature(
settings.WX_MP_TOKEN, timestamp, nonce, signature
):
return PlainTextResponse(echostr)
logger.warning("wx_mp verify: signature 校验失败 ts=%s nonce=%s", timestamp, nonce)
return PlainTextResponse("invalid signature", status_code=403)
@router.post("/callback", include_in_schema=False)
async def receive(request: Request) -> PlainTextResponse:
"""收用户消息(安全模式)。任何异常都吞掉返回 success,不让微信重试轰炸,靠日志排查。"""
if not settings.wx_mp_callback_configured:
logger.warning("wx_mp receive: 回调凭证未配齐(Token/AESKey/AppID)")
return PlainTextResponse("success")
try:
body = (await request.body()).decode("utf-8")
qp = request.query_params
# 安全模式外层 XML 只有 ToUserName + Encrypt。来源=微信服务器(HTTPS)+ 下面验签,
# 且 stdlib ET 不扩展外部实体, XXE 不适用。
encrypt = ET.fromstring(body).findtext("Encrypt") or ""
if not wx_mp_crypto.verify_msg_signature(
settings.WX_MP_TOKEN,
qp.get("timestamp", ""),
qp.get("nonce", ""),
encrypt,
qp.get("msg_signature", ""),
):
logger.warning("wx_mp receive: msg_signature 校验失败")
return PlainTextResponse("success")
xml = wx_mp_crypto.decrypt_message(
settings.WX_MP_AES_KEY, settings.WX_MP_APPID, encrypt
)
await _handle_message(ET.fromstring(xml))
except Exception:
logger.exception("wx_mp receive: 处理异常")
return PlainTextResponse("success")
async def _handle_message(root: ET.Element) -> None:
openid = root.findtext("FromUserName") or ""
msg_type = root.findtext("MsgType") or ""
if msg_type == "image":
pic_url = root.findtext("PicUrl") or ""
media_id = root.findtext("MediaId") or ""
logger.info(
"wx_mp 收到图片: openid=%s media_id=%s pic_url=%s", openid, media_id, pic_url
)
await _download(pic_url, openid, media_id)
else:
logger.info("wx_mp 收到消息(暂忽略): openid=%s type=%s", openid, msg_type)
async def _download(pic_url: str, openid: str, media_id: str) -> None:
"""用 PicUrl 直接下载图片落盘。PicUrl 是临时公网链接,无需 access_token / IP 白名单。"""
if not pic_url:
return
try:
_INBOX.mkdir(parents=True, exist_ok=True)
async with httpx.AsyncClient(timeout=15) as client:
resp = await client.get(pic_url)
resp.raise_for_status()
dest = _INBOX / f"{openid[:12]}_{media_id[:16]}.jpg"
dest.write_bytes(resp.content)
logger.info("wx_mp 图片已落盘: %s (%d bytes)", dest, len(resp.content))
except Exception:
logger.exception("wx_mp 图片下载失败 pic_url=%s", pic_url)
+3 -39
View File
@@ -44,6 +44,9 @@ class Settings(BaseSettings):
JWT_ALGORITHM: str = "HS256"
JWT_ACCESS_TOKEN_EXPIRE_MINUTES: int = 120
JWT_REFRESH_TOKEN_EXPIRE_DAYS: int = 30
# 微信登录未命中 openid 时签发的"待绑手机"令牌有效期(分钟)。
# 见 auth.wechat_login / security.create_bind_ticket。
WECHAT_BIND_TICKET_EXPIRE_MINUTES: int = 10
# ===== Admin 后台 =====
# admin 用独立 JWT secret(≠ JWT_SECRET_KEY),App 用户 token 无法越权访问后台。
@@ -71,34 +74,6 @@ class Settings(BaseSettings):
HEARTBEAT_TIMEOUT_MINUTES: int = 60 # 多久没心跳算掉线(1 小时,避免短暂离线误判被杀)
HEARTBEAT_SCAN_INTERVAL_SEC: int = 60 # 扫描周期
# ===== OPPO 推送(厂商通道直连) =====
# 直连 OPPO PUSH 服务端 API(api.push.oppomobile.com), 不经极光。凭证来自 OPPO 开放平台
# 注册应用后分配(≠极光 JG_*)。两步鉴权: 先 /server/v1/auth 用 sha256(app_key+timestamp+
# master_secret) 换 auth_token(缓存 24h), 再带 auth_token 调 /message/notification/unicast
# 单推。提现审核通过等交易类通知走这条。未配凭证 → oppo_push_configured=False, 发送方静默
# 跳过(不连累审核/打款)。
OPPO_PUSH_ENABLED: bool = True # 总开关(配了凭证也能置 False 全局停发)
OPPO_PUSH_APP_KEY: str = ""
OPPO_PUSH_MASTER_SECRET: str = ""
OPPO_PUSH_AUTH_ENDPOINT: str = "https://api.push.oppomobile.com/server/v1/auth"
OPPO_PUSH_UNICAST_ENDPOINT: str = (
"https://api.push.oppomobile.com/server/v1/message/notification/unicast"
)
OPPO_PUSH_REQUEST_TIMEOUT_SEC: int = 10
# 通知点击行为(OPPO click_action_type; 0=启动应用首页)。交易类通知点开进首页即可。
OPPO_PUSH_CLICK_ACTION_TYPE: int = 0
# 通知渠道 id(Android 8+ NotificationChannel): 【客户端创建的渠道】+【OPPO 管理台登记】+
# 【服务端此值】三方必须一致, 否则真机可能不展示。客户端 OppoPushHelper.CHANNEL_ID 同值。
OPPO_PUSH_CHANNEL_ID: str = "push_oplus_category_content"
# OPPO 新消息分类(Android 12+ 触达; 提现属 ACCOUNT 账号资产变化)。默认空=不带该字段——
# 它依赖 OPPO 管理台的消息分类资质, 未开通就带上可能反被限制, 确认资质后再填(如 "ACCOUNT")。
OPPO_PUSH_CATEGORY: str = ""
@property
def oppo_push_configured(self) -> bool:
"""OPPO 推送凭证是否齐备(app_key + master_secret)。缺任一 → 发送方静默跳过。"""
return bool(self.OPPO_PUSH_APP_KEY and self.OPPO_PUSH_MASTER_SECRET)
# ===== 短信 =====
SMS_MOCK: bool = True
SMS_CODE_TTL_SEC: int = 300
@@ -164,12 +139,6 @@ class Settings(BaseSettings):
# (不跳授权、不拿 openid),整套微信代码保留。审核通过后 .env 置 true + 重启即启用,无需改代码。
WX_MP_OAUTH_ENABLED: bool = False
# ===== 微信服务号(消息接收回调) =====
# 用户发消息给服务号 → 微信 POST 到 /wx/mp/callback(安全模式:Token 验签 + AESKey 解密)。
# 区别于上面的网页授权(那是落地页拿 openid);这里是被动收用户主动发来的消息(截图比价入口)。
WX_MP_TOKEN: str = "" # 公众平台"服务器配置"的 Token(URL 验签 + 消息验签)
WX_MP_AES_KEY: str = "" # EncodingAESKey(43 位, 消息 AES-256-CBC 加解密)
@property
def wx_mp_configured(self) -> bool:
"""服务号网页授权凭证齐全(缺则落地页不发起授权,降级为无 openid)。"""
@@ -180,11 +149,6 @@ class Settings(BaseSettings):
"""落地页是否真正发起微信授权 = 凭证齐全 且 总开关开。"""
return self.wx_mp_configured and self.WX_MP_OAUTH_ENABLED
@property
def wx_mp_callback_configured(self) -> bool:
"""消息接收回调凭证齐全(Token + AESKey + AppID)。缺则回调端点拒绝处理消息。"""
return bool(self.WX_MP_TOKEN and self.WX_MP_AES_KEY and self.WX_MP_APPID)
# ===== 微信支付(商家转账到零钱 / 提现)=====
# 真实凭证放 .env(已 gitignore),证书 .pem 放 secrets/。WECHAT_APP_ID 同时用于
# 微信登录(code 换 openid)与转账,必须与 App 端开放平台 appid 一致。
-15
View File
@@ -96,19 +96,4 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
"group": "首页轮播", "type": "enum", "hidden": True,
"help": "mixed=真实优先+种子补位(默认);real=只用真实比价记录;seed=只用种子/合成(演示)。",
},
# 比价 LLM 调用成本计价。值是嵌套 JSON(非 str→int),借 dict_str_int 类型在配置页走原始 JSON
# 编辑框;set_value 不校验类型,嵌套 JSON 照存。
"llm_token_price": {
"default": {
"per_model": {"qwen3.5-flash": {"input_per_1m": 0.8, "output_per_1m": 2.0}},
"default": {"input_per_1m": 3.0, "output_per_1m": 15.0},
"currency": "CNY", "unit": "per_1m_tokens",
},
"label": "LLM 模型单价(元/百万 token)",
"group": "LLM 成本", "type": "dict_str_int",
"help": (
"比价 LLM 调用成本计价。JSON:per_model 按模型配 input/output 单价(元/1M token),"
"default 兜底未登记的模型。改价只影响之后回填的新记录,历史记录用当时价格快照。"
),
},
}
+39
View File
@@ -87,6 +87,45 @@ def issue_token_pair(user_id: int) -> dict[str, Any]:
}
def create_bind_ticket(
*, openid: str, wechat_nickname: str | None, wechat_avatar_url: str | None
) -> str:
"""微信登录未命中 openid 时,签发短时"待绑手机"令牌,承载 openid + 微信昵称头像。
typ='wechat_bind'sub=openid;有效期 settings.WECHAT_BIND_TICKET_EXPIRE_MINUTES 分钟
access/refresh 用同一 JWT_SECRET_KEY 签名, typ 区分,decode_bind_ticket 校验 typ
"""
now = _now()
expire = now + timedelta(minutes=settings.WECHAT_BIND_TICKET_EXPIRE_MINUTES)
payload: dict[str, Any] = {
"sub": openid,
"typ": "wechat_bind",
"wnk": wechat_nickname,
"wav": wechat_avatar_url,
"iat": int(now.timestamp()),
"exp": int(expire.timestamp()),
}
return jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM)
def decode_bind_ticket(token: str) -> dict[str, Any]:
"""解析"待绑手机"令牌,校验签名/过期/类型。失败抛 TokenError。
返回 {'openid': str, 'wnk': str|None, 'wav': str|None}
"""
try:
payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM])
except jwt.ExpiredSignatureError as e:
raise TokenError("bind ticket expired") from e
except jwt.InvalidTokenError as e:
raise TokenError(f"invalid bind ticket: {e}") from e
if payload.get("typ") != "wechat_bind":
raise TokenError(f"wrong token type: want=wechat_bind got={payload.get('typ')}")
if "sub" not in payload:
raise TokenError("bind ticket missing sub")
return {"openid": payload["sub"], "wnk": payload.get("wnk"), "wav": payload.get("wav")}
# ===================== 密码 hash(admin 后台账号用)=====================
# 用户侧是手机号+验证码登录,不存密码;仅 admin 账号用 username+password 登录。
-183
View File
@@ -1,183 +0,0 @@
"""OPPO 推送服务端直连(厂商通道)。
不经极光, 直接调 OPPO PUSH 服务端 API(api.push.oppomobile.com)提现审核通过等交易类
通知走这条凭证来自 OPPO 开放平台(app_key/master_secret, settings.OPPO_PUSH_*)
两步鉴权(OPPO 协议):
1. POST /server/v1/auth sign=sha256(app_key+timestamp+master_secret) auth_token
(有效期 24h)token 进程内缓存复用, 到期前自动重取
2. POST /server/v1/message/notification/unicast header auth_token, 按单个
registration_id(OPPO registerID) 定向推送一条通知栏消息
发送 best-effort: 未配凭证 / 网络失败 / OPPO 返错都抛 OppoPushError, 由调用方(push_notify)
吞掉只记日志, 绝不连累审核/打款主流程风格对齐 integrations/sms.py(httpx 同步 + 未配降级)
"""
from __future__ import annotations
import hashlib
import json
import logging
import time
from dataclasses import dataclass
from threading import Lock
import httpx
from app.core.config import settings
logger = logging.getLogger("shagua.oppo_push")
# OPPO target_type 枚举: 2 = 按 registration_id 单推
_TARGET_TYPE_REGISTRATION_ID = 2
# auth_token 官方有效期 24h; 提前 1h 视为过期重取, 留刷新余量(避免边界请求踩到刚过期)
_TOKEN_TTL_SEC = 24 * 3600
_TOKEN_REFRESH_MARGIN_SEC = 3600
class OppoPushError(Exception):
"""OPPO 推送失败(未配置 / 鉴权失败 / 网络错误 / OPPO 返错)。调用方 best-effort 吞。"""
@dataclass
class _CachedToken:
token: str
expires_at: float # epoch 秒(= 取到 token 的时刻 + 24h)
# 进程内存 token 缓存(单 worker 有效; 多 worker 各自缓存, 各自换 token, OPPO 侧无副作用)
_token_cache: _CachedToken | None = None
_token_lock = Lock()
def _sign(app_key: str, timestamp_ms: int, master_secret: str) -> str:
"""OPPO 鉴权签名: sha256(app_key + timestamp + master_secret) 十六进制小写。"""
raw = f"{app_key}{timestamp_ms}{master_secret}"
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
def _parse_oppo_response(resp: httpx.Response, phase: str) -> dict | None:
"""解析 OPPO 统一响应 {code, message, data}: code==0 返 data, 否则抛 OppoPushError。"""
try:
body = resp.json()
except Exception as e:
raise OppoPushError(
f"OPPO {phase} 响应非 JSON(http={resp.status_code}): {resp.text[:200]}"
) from e
if body.get("code") != 0:
raise OppoPushError(
f"OPPO {phase} 失败 code={body.get('code')} message={body.get('message')!r}"
)
return body.get("data")
def _fetch_auth_token() -> str:
"""调 /server/v1/auth 换 auth_token。失败抛 OppoPushError。"""
timestamp_ms = int(time.time() * 1000) # 毫秒时间戳(OPPO 允许 ±10 分钟误差)
sign = _sign(
settings.OPPO_PUSH_APP_KEY, timestamp_ms, settings.OPPO_PUSH_MASTER_SECRET
)
try:
resp = httpx.post(
settings.OPPO_PUSH_AUTH_ENDPOINT,
data={
"app_key": settings.OPPO_PUSH_APP_KEY,
"sign": sign,
"timestamp": timestamp_ms,
},
timeout=settings.OPPO_PUSH_REQUEST_TIMEOUT_SEC,
)
except httpx.HTTPError as e:
raise OppoPushError(f"OPPO auth 网络错误: {e}") from e
data = _parse_oppo_response(resp, "auth")
token = (data or {}).get("auth_token")
if not token:
raise OppoPushError(f"OPPO auth 未返回 auth_token: {resp.text[:200]}")
return token
def _get_auth_token(force_refresh: bool = False) -> str:
"""取 auth_token(缓存优先; 到期或 force_refresh 才重取)。线程安全。"""
global _token_cache
now = time.time()
with _token_lock:
cache = _token_cache
if (
not force_refresh
and cache is not None
and now < cache.expires_at - _TOKEN_REFRESH_MARGIN_SEC
):
return cache.token
token = _fetch_auth_token()
_token_cache = _CachedToken(token=token, expires_at=now + _TOKEN_TTL_SEC)
return token
def send_notification(
registration_id: str,
title: str,
content: str,
*,
click_action_type: int | None = None,
) -> str:
"""向单个 OPPO registerID 推一条通知栏消息, 返回 OPPO message_id。
失败(未配置 / 鉴权 / 网络 / OPPO 返错)一律抛 OppoPushError, 由调用方吞
unicast 首次失败时强刷一次 token 重试(覆盖 token OPPO 提前失效的情况); 网络错误不重试
"""
if not settings.OPPO_PUSH_ENABLED:
raise OppoPushError("OPPO 推送总开关关闭(OPPO_PUSH_ENABLED=false)")
if not settings.oppo_push_configured:
raise OppoPushError("OPPO 推送未配置(缺 OPPO_PUSH_APP_KEY/OPPO_PUSH_MASTER_SECRET)")
if not registration_id:
raise OppoPushError("registration_id 为空")
if click_action_type is None:
click_action_type = settings.OPPO_PUSH_CLICK_ACTION_TYPE
notification = {
"title": title,
"content": content,
"click_action_type": click_action_type,
}
# channel_id: Android 8+ 通知渠道(客户端已建同名渠道 + OPPO 管理台登记, 三方一致才展示)。
if settings.OPPO_PUSH_CHANNEL_ID:
notification["channel_id"] = settings.OPPO_PUSH_CHANNEL_ID
# category: OPPO 新消息分类(默认空不带; 依赖管理台资质, 见 config 注释)。
if settings.OPPO_PUSH_CATEGORY:
notification["category"] = settings.OPPO_PUSH_CATEGORY
message = {
"target_type": _TARGET_TYPE_REGISTRATION_ID,
"target_value": registration_id,
"notification": notification,
}
# OPPO unicast body 是 form 编码, message 字段为 JSON 字符串
payload = {"message": json.dumps(message, ensure_ascii=False)}
last_err: OppoPushError | None = None
for attempt in range(2):
token = _get_auth_token(force_refresh=(attempt == 1))
try:
resp = httpx.post(
settings.OPPO_PUSH_UNICAST_ENDPOINT,
data=payload,
headers={"auth_token": token},
timeout=settings.OPPO_PUSH_REQUEST_TIMEOUT_SEC,
)
except httpx.HTTPError as e:
# 网络错误与 token 无关, 不强刷重试
raise OppoPushError(f"OPPO unicast 网络错误: {e}") from e
try:
data = _parse_oppo_response(resp, "unicast")
except OppoPushError as e:
last_err = e
if attempt == 0:
logger.warning("[OPPO] unicast 失败(%s), 强刷 token 重试一次", e)
continue
raise
message_id = (data or {}).get("message_id", "")
logger.info(
"[OPPO] 推送成功 regId=%s… message_id=%s", registration_id[:12], message_id
)
return message_id
# 理论到不了(第二次失败已 raise); 防御性收尾
raise last_err or OppoPushError("OPPO unicast 重试后仍失败")
-69
View File
@@ -1,69 +0,0 @@
"""微信服务号消息接收回调的验签与解密(安全模式)。
服务号"服务器配置"选安全模式后:
- URL 接入验证(GET): sha1(sort(token, timestamp, nonce)) == signature 原样返回 echostr
- 消息(POST): body <Encrypt> 是密文;
msg_signature = sha1(sort(token, timestamp, nonce, encrypt))
密文 AES-256-CBC 解出: random(16B) + msg_len(4B big-endian) + msg + appid, PKCS7(=32) padding
只做验签 + 解密(收消息)被动回复(加密)MVP 不需要 收到后走客服消息异步回执
密钥/口令来自 settings(WX_MP_TOKEN / WX_MP_AES_KEY / WX_MP_APPID),本模块只做纯算法不读配置
"""
from __future__ import annotations
import base64
import hashlib
import hmac
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
def verify_url_signature(token: str, timestamp: str, nonce: str, signature: str) -> bool:
"""GET 接入验证:token/timestamp/nonce 三者字典序排序拼接后 sha1。"""
return _consteq(_sha1(token, timestamp, nonce), signature)
def verify_msg_signature(
token: str, timestamp: str, nonce: str, encrypt: str, msg_signature: str
) -> bool:
"""POST 消息验签:四者(含密文 encrypt)字典序排序拼接后 sha1。"""
return _consteq(_sha1(token, timestamp, nonce, encrypt), msg_signature)
def decrypt_message(aes_key_b64: str, expected_appid: str, encrypt_b64: str) -> str:
"""解密 <Encrypt> 密文, 返回明文消息 XML。appid 不符抛 ValueError。
aes_key_b64: EncodingAESKey(43 , 不含结尾 '='), '=' base64 解出 32 字节 AES-256 key
"""
aes_key = base64.b64decode(aes_key_b64 + "=") # 43 → 32 bytes
iv = aes_key[:16]
decryptor = Cipher(algorithms.AES(aes_key), modes.CBC(iv)).decryptor()
plain = decryptor.update(base64.b64decode(encrypt_b64)) + decryptor.finalize()
plain = _pkcs7_unpad(plain)
# random(16) + msg_len(4, big-endian) + msg(msg_len) + from_appid
content = plain[16:]
msg_len = int.from_bytes(content[:4], "big")
msg = content[4 : 4 + msg_len]
from_appid = content[4 + msg_len :].decode("utf-8")
if expected_appid and from_appid != expected_appid:
raise ValueError(f"appid mismatch: {from_appid!r} != {expected_appid!r}")
return msg.decode("utf-8")
def _sha1(*parts: str) -> str:
return hashlib.sha1("".join(sorted(parts)).encode("utf-8")).hexdigest()
def _consteq(a: str, b: str) -> bool:
return hmac.compare_digest(a, b)
def _pkcs7_unpad(data: bytes) -> bytes:
"""微信用块大小 32 的 PKCS7,末字节即 padding 长度(1..32)。越界则原样返回(容错)。"""
if not data:
return data
pad = data[-1]
if pad < 1 or pad > 32:
return data
return data[:-pad]
-3
View File
@@ -39,7 +39,6 @@ from app.api.v1.signin import router as signin_router
from app.api.v1.tasks import router as tasks_router
from app.api.v1.user import router as user_router
from app.api.v1.wallet import router as wallet_router
from app.api.v1.wx_mp import router as wx_mp_router
from app.api.v1.wxpay import router as wxpay_router
from app.core.config import settings
from app.core.daily_exchange_worker import (
@@ -141,8 +140,6 @@ app.include_router(internal_launch_confirm_router)
app.include_router(platform_router)
# CPS 群发短链跳转 /c/{code}(公网无鉴权:记点击 → 302 跳美团)
app.include_router(cps_redirect_router)
# 微信服务号消息接收回调 /wx/mp/callback(公网无鉴权:微信服务器验签, 截图比价入口)
app.include_router(wx_mp_router)
# 用户上传文件(头像)静态服务。生产可改由 nginx 直接 serve MEDIA_ROOT。
_media_root = Path(settings.MEDIA_ROOT)
-6
View File
@@ -137,12 +137,6 @@ class ComparisonRecord(Base):
# 每次 LLM 调用明细 [{scene,model,input_messages,output,usage,latency_ms,error}];
# server 收上报后按 trace_id 同机拉 pricebot 落库(见 compare_record 端点)。旧记录/未采集为 None。
llm_calls: Mapped[list | None] = mapped_column(_JSON, nullable=True)
# 本次比价 LLM 总成本(元):回填时按「当时的价」逐模型算好冻结(见 services/llm_cost.py)。
# 单次亚分级 → float「元」(不用 *_cents)。旧记录/未回填为 None,前端回退「估算成本」。
llm_cost_yuan: Mapped[float | None] = mapped_column(Float, nullable=True)
# 算成本所用单价快照 {mode, prices:{model:{input_per_1m,output_per_1m,_source}}}:app_config 只存
# 当前价、不留历史,故把当时价冻结进来供审计/复算。
llm_price_snapshot: Mapped[dict | None] = mapped_column(_JSON, nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
-3
View File
@@ -44,9 +44,6 @@ class DeviceLiveness(Base):
device_id: Mapped[str] = mapped_column(String(128), index=True, nullable=False)
# 极光推送 registration id;拿到才填(JCollectionAuth 同意后才下发)
registration_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
# OPPO 官方推送 SDK(HeytapPushManager)的 registerID; 直连 OPPO 厂商通道用(≠极光 registration_id)。
# 一台 OPPO 设备两套 token 并存: 极光走极光自有/其它厂商通道, 这个走 OPPO 直连(提现审核通过等交易通知)。
oppo_register_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
platform: Mapped[str] = mapped_column(String(16), nullable=False, default="android")
app_version: Mapped[str | None] = mapped_column(String(32), nullable=True)
+2 -29
View File
@@ -22,18 +22,16 @@ def register_or_update(
user_id: int,
device_id: str,
registration_id: str | None,
oppo_register_id: str | None = None,
platform: str = "android",
app_version: str | None = None,
) -> DeviceLiveness:
"""注册设备或更新其 registration_id / oppo_register_id / 元信息。upsert by (user_id, device_id)。"""
"""注册设备或更新其 registration_id / 元信息。upsert by (user_id, device_id)。"""
device = _get(db, user_id=user_id, device_id=device_id)
if device is None:
device = DeviceLiveness(
user_id=user_id,
device_id=device_id,
registration_id=registration_id,
oppo_register_id=oppo_register_id,
platform=platform or "android",
app_version=app_version,
)
@@ -41,8 +39,6 @@ def register_or_update(
else:
if registration_id:
device.registration_id = registration_id
if oppo_register_id:
device.oppo_register_id = oppo_register_id
if platform:
device.platform = platform
if app_version:
@@ -59,13 +55,11 @@ def touch_heartbeat(
device_id: str,
accessibility_enabled: bool,
registration_id: str | None,
oppo_register_id: str | None = None,
) -> DeviceLiveness:
"""处理一次心跳(心跳也能自注册)。
service 心跳或 accessibility_enabled=true ,刷新存活并把状态机重置回 alive
清掉 notified_at(掉线恢复 下次再断才会再推一条)registration_id / oppo_register_id
非空时顺带更新(客户端拿到 push token 后每次心跳带上, 后端最终一致)
清掉 notified_at(掉线恢复 下次再断才会再推一条)
"""
now = datetime.now(timezone.utc)
device = _get(db, user_id=user_id, device_id=device_id)
@@ -75,8 +69,6 @@ def touch_heartbeat(
if registration_id:
device.registration_id = registration_id
if oppo_register_id:
device.oppo_register_id = oppo_register_id
device.last_report_protection_on = accessibility_enabled
if accessibility_enabled:
@@ -126,25 +118,6 @@ def get_device(db: Session, *, user_id: int, device_id: str) -> DeviceLiveness |
return _get(db, user_id=user_id, device_id=device_id)
def list_oppo_register_ids_by_user(db: Session, *, user_id: int) -> list[str]:
"""取某用户所有设备已登记的 OPPO registerID(去重保序, 非空)。
提现审核通过推送用: 一个用户可能多台设备, 每台都推; OPPO token 的设备( OPPO /
未集成 OPPO SDK)自然不在列空列表 = 该用户没有可推的 OPPO 设备
"""
stmt = select(DeviceLiveness.oppo_register_id).where(
DeviceLiveness.user_id == user_id,
DeviceLiveness.oppo_register_id.is_not(None),
)
seen: set[str] = set()
out: list[str] = []
for rid in db.execute(stmt).scalars().all():
if rid and rid not in seen:
seen.add(rid)
out.append(rid)
return out
def ack_kill_alert(db: Session, *, user_id: int, device_id: str) -> None:
"""客户端已弹过「开启自启动」引导 → 清「待提醒」标记(幂等; 下次真掉线 worker 再置)。"""
device = _get(db, user_id=user_id, device_id=device_id)
+44
View File
@@ -86,6 +86,50 @@ def get_user_by_phone(db: Session, phone: str) -> User | None:
return db.execute(stmt).scalar_one_or_none()
def get_user_by_wechat_openid(db: Session, openid: str) -> User | None:
stmt = select(User).where(User.wechat_openid == openid)
return db.execute(stmt).scalar_one_or_none()
def touch_last_login(db: Session, user: User) -> User:
"""openid 命中登录时更新 last_login_at(手机号登录在 upsert_user_for_login 里已更新)。"""
user.last_login_at = datetime.now(timezone.utc)
db.commit()
db.refresh(user)
return user
def create_wechat_user(
db: Session,
*,
phone: str,
openid: str,
wechat_nickname: str | None,
wechat_avatar_url: str | None,
) -> User:
"""微信登录新建账号:register_channel='wechat';展示昵称/头像取微信(§10 新注册),
昵称缺失(微信隐私脱敏返回 None)时退默认昵称同时保存 wechat_* 源字段
openid 唯一约束是并发/重复绑定的最终防线(极罕见,openid wechat-login 刚查过为空)
"""
now = datetime.now(timezone.utc)
user = User(
phone=phone,
username=_gen_unique_username(db),
nickname=wechat_nickname or _gen_nickname(),
avatar_url=wechat_avatar_url,
register_channel="wechat",
wechat_openid=openid,
wechat_nickname=wechat_nickname,
wechat_avatar_url=wechat_avatar_url,
last_login_at=now,
)
db.add(user)
db.commit()
db.refresh(user)
return user
def upsert_user_for_login(
db: Session,
*,
+47
View File
@@ -102,3 +102,50 @@ class RefreshRequest(BaseModel):
class LogoutResponse(BaseModel):
ok: bool = True
# ===== 微信登录 =====
class WechatLoginRequest(BaseModel):
code: str = Field(..., min_length=1, description="微信 App 授权拿到的 code(单次有效)")
device_id: str = Field(
"", max_length=64,
description="硬件级设备标识(Android ANDROID_ID),用于新手引导按 设备+账号 去重;空=按未完成处理",
)
class WechatLoginResponse(BaseModel):
# status="logged_in" → openid 命中,token 有值;"need_bind_phone" → 未命中,bind_ticket 有值
status: str
token: TokenWithUser | None = None
bind_ticket: str | None = None
wechat_nickname: str | None = None
wechat_avatar_url: str | None = None
class OccupiedAccountInfo(BaseModel):
"""手机号被占用时返回的原账号脱敏展示信息(供 M2 冲突页;M1 客户端仅用于提示)。"""
nickname: str | None = None
avatar_url: str | None = None
created_at: datetime
class WechatBindResultResponse(BaseModel):
# status="logged_in" → 未占用,已建号登入,token 有值;
# "phone_occupied" → 手机号被占用,occupied_account 有值,token 为 None
status: str
token: TokenWithUser | None = None
occupied_account: OccupiedAccountInfo | None = None
class WechatBindPhoneSmsRequest(BaseModel):
bind_ticket: str = Field(..., min_length=1)
phone: str = Field(..., min_length=11, max_length=11, pattern=r"^1\d{10}$")
code: str = Field(..., min_length=4, max_length=8)
device_id: str = Field("", max_length=64)
class WechatBindPhoneJverifyRequest(BaseModel):
bind_ticket: str = Field(..., min_length=1)
login_token: str = Field(..., min_length=1, description="客户端 loginAuth 拿到的 loginToken")
device_id: str = Field("", max_length=64)
-3
View File
@@ -9,7 +9,6 @@ from pydantic import BaseModel, ConfigDict
class DeviceRegisterRequest(BaseModel):
device_id: str
registration_id: str | None = None
oppo_register_id: str | None = None
platform: str = "android"
app_version: str | None = None
@@ -19,7 +18,6 @@ class HeartbeatRequest(BaseModel):
source: str = "service" # service | app
accessibility_enabled: bool = True
registration_id: str | None = None
oppo_register_id: str | None = None
class DeviceOut(BaseModel):
@@ -28,7 +26,6 @@ class DeviceOut(BaseModel):
id: int
device_id: str
registration_id: str | None
oppo_register_id: str | None
ever_protected: bool
liveness_state: str
last_heartbeat_at: datetime | None
-53
View File
@@ -1,53 +0,0 @@
"""LLM 调用成本计算(纯逻辑,无 DB):按 model 分桶累加 token × 单价,返回总成本(元)+ 价格快照。
用量取自 comparison_record.llm_calls[].usage(pricebot 已归一为 prompt/completion_tokens);
error / usage 的调用跳过price_cfg = {per_model:{model:{input_per_1m,output_per_1m}}, default:{...}}
成本单位单次亚分级, float(不用 *_cents);snapshot 只含本次用到的模型的价(审计用,
不存整张价表)用到但没配价(既无 per_model 又无 default)的模型 快照标 unpriced,成本按 0
"""
from __future__ import annotations
_PRICE_KEY = "llm_token_price"
def get_llm_prices(db) -> dict:
"""读 LLM 单价配置(app_config;表内无则回退 CONFIG_DEFS 默认)。返回 compute_llm_cost 的 price_cfg。"""
from app.repositories import app_config # 延迟 import:compute_llm_cost 纯逻辑不牵连 DB 层
return app_config.get_value(db, _PRICE_KEY)
def compute_llm_cost(calls: list[dict], price_cfg: dict) -> tuple[float | None, dict | None]:
"""遍历 calls 按 model 分桶,cost = Σ(入/1e6*入价 + 出/1e6*出价);无有效调用 → (None, None)。"""
if not calls:
return None, None
per_model = price_cfg.get("per_model") or {}
default = price_cfg.get("default")
buckets: dict[str, list[int]] = {} # model -> [Σprompt_tokens, Σcompletion_tokens]
for c in calls:
if c.get("error"):
continue
usage = c.get("usage") or {}
model = c.get("model") or "unknown"
b = buckets.setdefault(model, [0, 0])
b[0] += usage.get("prompt_tokens") or 0
b[1] += usage.get("completion_tokens") or 0
if not buckets: # 全是 error / 无 usage
return None, None
total = 0.0
prices: dict[str, dict] = {}
for model, (tin, tout) in buckets.items():
price = per_model.get(model, default)
in_p = price.get("input_per_1m") if isinstance(price, dict) else None
out_p = price.get("output_per_1m") if isinstance(price, dict) else None
# 没配价 / 无 default / 单价残缺或非法(配置页手改 JSON 可能存出脏数据)→ 标记待补价、
# 不计入成本;绝不抛异常,以免连累同一回填里的 token/llm_calls 落库。
if not isinstance(in_p, (int, float)) or not isinstance(out_p, (int, float)):
prices[model] = {"input_per_1m": in_p, "output_per_1m": out_p, "unpriced": True}
continue
total += tin / 1e6 * in_p + tout / 1e6 * out_p
prices[model] = {
"input_per_1m": in_p,
"output_per_1m": out_p,
"_source": "per_model" if model in per_model else "default",
}
return round(total, 6), {"mode": "per_model", "prices": prices}
-74
View File
@@ -1,74 +0,0 @@
"""推送通知编排(查设备 push token → 调厂商推送)。
薄编排层: repositories( token) + integrations(发送) 的粘合, api / admin router 调用
所有发送 best-effort 失败只 log, 绝不抛给调用方(不连累审核/打款等主流程)
当前只有 OPPO 直连一条通道(提现审核通过通知)后续加华为/vivo 等厂商时, 在这里按设备
token 做通道分发, 上层业务函数(notify_withdraw_approved )不感知通道差异
"""
from __future__ import annotations
import logging
import threading
from sqlalchemy.orm import Session
from app.db.session import SessionLocal
from app.integrations import oppo_push
from app.integrations.oppo_push import OppoPushError
from app.repositories import device as device_repo
logger = logging.getLogger("shagua.push_notify")
_WITHDRAW_APPROVED_TITLE = "提现审核通过"
_WITHDRAW_APPROVED_CONTENT = "您的提现申请已审核通过,款项将很快到账,请留意微信零钱。"
def notify_withdraw_approved(db: Session, *, user_id: int) -> int:
"""提现审核通过 → 给该用户所有 OPPO 设备各推一条通知。返回成功推送的设备数。
best-effort: 未配 OPPO / OPPO 设备 / 单设备发送失败都不抛, log审核主流程绝不受影响
调用方仍应包一层 try/except 兜底本函数自身的意外异常( DB 查询失败)
"""
reg_ids = device_repo.list_oppo_register_ids_by_user(db, user_id=user_id)
if not reg_ids:
logger.info("[push] withdraw_approved user_id=%s 无 OPPO 设备, 跳过", user_id)
return 0
sent = 0
for rid in reg_ids:
try:
oppo_push.send_notification(
rid, _WITHDRAW_APPROVED_TITLE, _WITHDRAW_APPROVED_CONTENT
)
sent += 1
except OppoPushError as e:
logger.warning(
"[push] withdraw_approved user_id=%s regId=%s… 发送失败: %s",
user_id, rid[:12], e,
)
logger.info(
"[push] withdraw_approved user_id=%s 推送 %d/%d 台 OPPO 设备",
user_id, sent, len(reg_ids),
)
return sent
def notify_withdraw_approved_async(user_id: int) -> None:
"""fire-and-forget 版: 后台守护线程新开 session 发推送, 不阻塞审核响应(对齐 best-effort)。
审核 approve 已同步调微信转账, 推送不该再把外部 API 耗时叠进响应(多设备/OPPO 慢时最坏
N×timeout)线程内必须新开 SessionLocal 请求级 session 在响应返回后即关, 不能跨线程用
起线程失败也只 log, 绝不连累审核
"""
def _run() -> None:
try:
with SessionLocal() as db:
notify_withdraw_approved(db, user_id=user_id)
except Exception: # noqa: BLE001 - 后台线程绝不抛
logger.warning("[push] withdraw_approved async 异常 user_id=%s", user_id, exc_info=True)
try:
threading.Thread(target=_run, name=f"oppo-push-{user_id}", daemon=True).start()
except Exception: # noqa: BLE001 - 起线程失败也绝不连累审核
logger.warning("[push] withdraw_approved async 起线程失败 user_id=%s", user_id, exc_info=True)
-2
View File
@@ -40,8 +40,6 @@
| `raw_payload` | JSON(PG: JSONB) | nullable | 客户端原始上报全量(calibration + done.params),取数兜底 |
| `input_tokens` | Integer | nullable | 本次 LLM 累计输入 token = Σ `llm_calls[].usage.prompt_tokens`(server 收上报后从 `llm_calls` 累加;旧记录/未采集为 null) |
| `output_tokens` | Integer | nullable | 本次 LLM 累计输出 token = Σ `llm_calls[].usage.completion_tokens`(同上) |
| `llm_cost_yuan` | Float | nullable | 本次比价 LLM 总成本(元),回填时按「当时价」逐模型算好冻结(见 `services/llm_cost.py`);旧记录/未回填为 null → 前端回退「估算成本」 |
| `llm_price_snapshot` | JSON(PG: JSONB) | nullable | 算成本所用单价快照 `{mode, prices:{model:{input_per_1m,output_per_1m,_source}}}`;`app_config` 只存当前价、不留历史,故冻结当时价供审计/复算 |
| `created_at` | DateTime(tz) | server_default now(), index | 时间 |
> `ordered`(已下单)是**瞬态字段**,不在表里:`list_records` 读取时按 `store_name ∈ 该用户 source='compare' 的 savings_record.shop_name 集合` 现挂到实例上供出参用。
-248
View File
@@ -1,248 +0,0 @@
"""一次性 mock:造带 LLM token 成本的比价记录 + 配好 app_config 模型单价,用于测「管理后端」LLM 成本展示。
覆盖 admin比价记录详情抽屉的LLM 成本展示分支:
app_config.llm_token_price 写一条多模型单价(= 配置页LLM 成本卡片已改,get_llm_prices 读它)
comparison_record 5 ,逐条**复用生产的 compute_llm_cost + _backfill_llm_calls 同款派生**
(llm_call_count/retry_count/input_tokens/output_tokens/llm_cost_yuan/llm_price_snapshot),
确保 mock = 真实回填产出5 条刻意覆盖:
单模型真实样本(qwen3.5-flash ×4) ¥0.006184(核对精确值)
多模型(flash + plus) 快照含两个模型各自 _source=per_model
未登记模型(deepseek-v3) default,快照 _source=default
旧记录( token cost) llm_cost_yuan=NULL 前端回退估算成本
error 调用 error 那次跳过计费retry_count+1
记录挂到库里第一个真实用户(admin 列表能显示手机号);无用户则 user_id=NULL(孤儿行,admin 照样全看)
created_at 用北京 naive最近几分钟内错开,详情列表倒序即 置顶
幂等:重跑先按 trace_id 前缀MOCKLLM-清旧再建app_config 单价是 upsert(不随 --clean-only ,
因该 key 本就是本需求新增无历史真实值;要改价直接去配置页或重跑本脚本)
python -m scripts.seed_mock_llm_cost # 造价格 + 5 条记录
python -m scripts.seed_mock_llm_cost --clean-only # 只清 MOCKLLM- 记录(保留单价)
验收:admin比价记录 traceMOCKLLM- 5 点开详情看LLM 成本:
显示实际·当时价+ 价格快照; 显示估算
"""
from __future__ import annotations
import argparse
import sys
from datetime import datetime, timedelta, timezone
from sqlalchemy import delete, select
from app.db.session import SessionLocal
from app.models.comparison import ComparisonRecord
from app.models.user import User
from app.repositories import app_config
from app.services.llm_cost import compute_llm_cost
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8") # Windows 控制台输出中文/¥
_BJ = timezone(timedelta(hours=8))
ID_PREFIX = "MOCKLLM-"
# ── 写进 app_config 的模型单价(get_llm_prices 读它;配置页「LLM 成本」卡片可再改)──
PRICE_CFG = {
"per_model": {
"qwen3.5-flash": {"input_per_1m": 0.8, "output_per_1m": 2.0},
"qwen3.5-plus": {"input_per_1m": 4.0, "output_per_1m": 12.0},
},
"default": {"input_per_1m": 3.0, "output_per_1m": 15.0},
"currency": "CNY",
"unit": "per_1m_tokens",
}
def _c(scene: str, model: str, pin: int, cout: int, error: str | None = None) -> dict:
"""一条 llm_calls 明细,结构对齐真实 pricebot 归一后契约:
{scene, model, input_messages:[{role,content}], output, usage:{prompt/completion/total_tokens},
latency_ms, error}(详情抽屉会遍历 input_messages,缺了会崩)error 的调用无 usage/output"""
return {
"scene": scene,
"model": model,
"error": error,
"input_messages": [
{"role": "system", "content": f"你是比价助手,负责 {scene} 环节。"},
{"role": "user", "content": f"[mock] 请处理本次比价的 {scene} 任务。"},
],
"output": None if error else f"[mock] {scene} 环节完成。",
"usage": None if error else {
"prompt_tokens": pin, "completion_tokens": cout, "total_tokens": pin + cout,
},
"latency_ms": 780,
}
# ── 5 条记录蓝本:calls 决定成本;freeze=False 模拟旧记录(有 token 无 cost)──
RECORDS = [
{
"label": "①单模型·真实样本",
"source": ("美团外卖", 4280), "best": ("京东秒送", 3680),
"store": "肯德基(建国路店)", "product": "疯狂星期四全家桶",
"info": "在京东秒送找到同款,到手价 ¥36.80,省 ¥6.00",
"freeze": True,
"calls": [
_c("store_match", "qwen3.5-flash", 1512, 22),
_c("dish_match", "qwen3.5-flash", 2111, 160),
_c("dish_match", "qwen3.5-flash", 1940, 142),
_c("summary", "qwen3.5-flash", 1325, 13),
],
},
{
"label": "②多模型·flash+plus",
"source": ("淘宝闪购", 5900), "best": ("美团外卖", 5200),
"store": "瑞幸咖啡(国贸店)", "product": "生椰拿铁×2、丝绒拿铁",
"info": "在美团外卖找到同款,到手价 ¥52.00,省 ¥7.00",
"freeze": True,
"calls": [
_c("store_match", "qwen3.5-flash", 2000, 50),
_c("dish_match", "qwen3.5-flash", 1800, 40),
_c("reasoning", "qwen3.5-plus", 3000, 500),
],
},
{
"label": "③未登记模型走 default",
"source": ("京东秒送", 3100), "best": ("美团外卖", 2650),
"store": "麦当劳(soho店)", "product": "麦辣鸡腿堡套餐",
"info": "在美团外卖找到同款,到手价 ¥26.50,省 ¥4.50",
"freeze": True,
"calls": [
_c("store_match", "deepseek-v3", 5000, 800),
],
},
{
"label": "④旧记录·有token无成本(回退估算)",
"source": ("美团外卖", 3600), "best": ("淘宝闪购", 3200),
"store": "华莱士(双井店)", "product": "全鸡汉堡套餐",
"info": "在淘宝闪购找到同款,到手价 ¥32.00,省 ¥4.00",
"freeze": False, # 模拟本需求上线前的老记录:llm_cost_yuan=NULL → 前端回退估算
"calls": [
_c("store_match", "qwen3.5-flash", 2000, 100),
],
},
{
"label": "⑤含 error 调用(跳过计费)",
"source": ("淘宝闪购", 4100), "best": ("京东秒送", 3750),
"store": "海底捞(合生汇店)", "product": "番茄锅底、肥牛卷",
"info": "在京东秒送找到同款,到手价 ¥37.50,省 ¥3.50",
"freeze": True,
"calls": [
_c("store_match", "qwen3.5-flash", 0, 0, error="timeout"),
_c("store_match", "qwen3.5-flash", 1500, 30),
],
},
]
_PLATFORM_ID = { # 展示名 → 平台代号(comparison_results / source/best 列用)
"美团外卖": "meituan", "京东秒送": "jd", "淘宝闪购": "taobao",
}
def _naive_bj_now() -> datetime:
return datetime.now(_BJ).replace(tzinfo=None)
def clean(db) -> int:
n = db.execute(
delete(ComparisonRecord).where(ComparisonRecord.trace_id.like(f"{ID_PREFIX}%"))
).rowcount or 0
db.commit()
return n
def _build_record(spec: dict, owner_id: int | None, created_at: datetime) -> tuple[ComparisonRecord, float | None]:
"""按蓝本造一条记录,LLM 派生完全对齐 _backfill_llm_calls;返回 (记录, 冻结成本或 None)。"""
calls = spec["calls"]
src_name, src_cents = spec["source"]
best_name, best_cents = spec["best"]
# —— 与 _backfill_llm_calls 同款派生 ——
llm_call_count = len(calls)
retry_count = sum(1 for c in calls if c.get("error"))
input_tokens = sum((c.get("usage") or {}).get("prompt_tokens") or 0 for c in calls)
output_tokens = sum((c.get("usage") or {}).get("completion_tokens") or 0 for c in calls)
if spec["freeze"]:
cost, snapshot = compute_llm_cost(calls, PRICE_CFG) # 复用生产纯函数
else:
cost, snapshot = None, None # 旧记录:回填这段代码上线前就有,只有 token 没成本
rec = ComparisonRecord(
user_id=owner_id,
device_id=f"{ID_PREFIX.lower()}dev",
business_type="food",
trace_id=f"{ID_PREFIX}{spec['label'][0]}", # ①..⑤ 各一,唯一
status="success",
source_platform_id=_PLATFORM_ID.get(src_name), source_platform_name=src_name,
source_price_cents=src_cents,
best_platform_id=_PLATFORM_ID.get(best_name), best_platform_name=best_name,
best_price_cents=best_cents,
saved_amount_cents=src_cents - best_cents,
is_source_best=False,
store_name=spec["store"],
product_names=spec["product"],
information=spec["info"],
items=[{"name": spec["product"], "qty": 1}],
comparison_results=[
{"platform_id": _PLATFORM_ID.get(src_name), "platform_name": src_name,
"price": src_cents / 100, "is_source": True, "rank": 2},
{"platform_id": _PLATFORM_ID.get(best_name), "platform_name": best_name,
"price": best_cents / 100, "is_source": False, "rank": 1},
],
total_ms=90_000 + llm_call_count * 1000,
step_count=llm_call_count * 3,
llm_call_count=llm_call_count,
retry_count=retry_count,
input_tokens=input_tokens,
output_tokens=output_tokens,
llm_calls=calls,
llm_cost_yuan=cost,
llm_price_snapshot=snapshot,
created_at=created_at,
)
return rec, cost
def seed(db) -> list[tuple[str, float | None]]:
app_config.set_value(db, "llm_token_price", PRICE_CFG, admin_id=None) # upsert 单价
owner_id = db.execute(select(User.id).order_by(User.id).limit(1)).scalar()
base = _naive_bj_now()
out: list[tuple[str, float | None]] = []
for i, spec in enumerate(RECORDS):
rec, cost = _build_record(spec, owner_id, base - timedelta(minutes=i * 3))
db.add(rec)
out.append((spec["label"], cost))
db.commit()
return out, owner_id
def main() -> None:
parser = argparse.ArgumentParser(description="造带 LLM 成本的比价记录 + app_config 模型单价(测管理后端)")
parser.add_argument("--clean-only", action="store_true", help="只清 MOCKLLM- 记录,不重建(保留单价)")
args = parser.parse_args()
db = SessionLocal()
try:
removed = clean(db)
if removed:
print(f"🧹 已清理旧 mock 记录 {removed}")
if args.clean_only:
print("✅ 仅清理,已完成(app_config 单价保留)。")
return
results, owner_id = seed(db)
print(f"\n✅ 已写入 app_config.llm_token_price(单价)+ {len(results)} 条比价记录"
f"(挂 user_id={owner_id or 'NULL(孤儿行)'})")
print("\n📋 每条冻结成本(admin 详情「LLM 成本」应显示):")
for label, cost in results:
shown = "NULL → 前端回退「估算」" if cost is None else f"¥{cost}"
print(f" {label:<20} {shown}")
print("\n👉 验收:admin「比价记录」→ trace 搜「MOCKLLM-」→ 点开详情核对 LLM 成本 + 价格快照。")
print(" 配置页「系统配置」→「福利页」Tab →「LLM 成本」卡片,单价应为「已改」态。")
finally:
db.close()
if __name__ == "__main__":
main()
-78
View File
@@ -476,81 +476,3 @@ def test_withdraw_reconcile(admin_client: TestClient, finance_token: str, monkey
)
assert r.status_code == 200, r.text
assert "checked" in r.json() and "resolved" in r.json()
# ===== 提现审核通过 → OPPO 推送挂钩 =====
def test_approve_triggers_oppo_push(
admin_client: TestClient, finance_token: str, monkeypatch
) -> None:
"""审核通过(非 failed)后触发一次 OPPO 推送, 带被审核单的 user_id。"""
uid = _seed_user("13900000021")
db = SessionLocal()
try:
db.add(WithdrawOrder(
user_id=uid, out_bill_no="pushapprove0001", amount_cents=100, status="reviewing"
))
db.commit()
finally:
db.close()
from app.admin.routers import withdraw as withdraw_router
# 绕过真实微信转账: approve 直接把单置 pending 返回(不发 wxpay)
def _fake_approve(db, obn):
o = db.execute(
select(WithdrawOrder).where(WithdrawOrder.out_bill_no == obn)
).scalar_one()
o.status = "pending"
db.commit()
return o
monkeypatch.setattr(withdraw_router.wallet_repo, "approve_withdraw", _fake_approve)
pushed: list[int] = []
monkeypatch.setattr(
withdraw_router.push_notify, "notify_withdraw_approved_async",
lambda user_id: pushed.append(user_id),
)
r = admin_client.post(
"/admin/api/withdraws/pushapprove0001/approve", headers=_auth(finance_token)
)
assert r.status_code == 200, r.text
assert pushed == [uid]
def test_approve_failed_does_not_push(
admin_client: TestClient, finance_token: str, monkeypatch
) -> None:
"""打款直接失败(status=failed, 已退款)不推"审核通过""""
uid = _seed_user("13900000022")
db = SessionLocal()
try:
db.add(WithdrawOrder(
user_id=uid, out_bill_no="pushapprove0002", amount_cents=100, status="reviewing"
))
db.commit()
finally:
db.close()
from app.admin.routers import withdraw as withdraw_router
def _fake_approve_failed(db, obn):
o = db.execute(
select(WithdrawOrder).where(WithdrawOrder.out_bill_no == obn)
).scalar_one()
o.status = "failed"
o.fail_reason = "余额不足"
db.commit()
return o
monkeypatch.setattr(withdraw_router.wallet_repo, "approve_withdraw", _fake_approve_failed)
pushed: list[int] = []
monkeypatch.setattr(
withdraw_router.push_notify, "notify_withdraw_approved_async",
lambda user_id: pushed.append(user_id),
)
r = admin_client.post(
"/admin/api/withdraws/pushapprove0002/approve", headers=_auth(finance_token)
)
assert r.status_code == 200, r.text
assert pushed == []
-184
View File
@@ -1,184 +0,0 @@
"""LLM 调用成本计算 compute_llm_cost:按模型分桶累加 token × 单价;error/无 usage 跳过。"""
from __future__ import annotations
from app.services.llm_cost import compute_llm_cost
_PRICE = {
"per_model": {"qwen3.5-flash": {"input_per_1m": 0.8, "output_per_1m": 2.0}},
"default": {"input_per_1m": 3.0, "output_per_1m": 15.0},
}
def test_sums_per_model_single_model():
# 真实样本:4 次 qwen3.5-flash;Σprompt=6888、Σcompletion=337
calls = [
{"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1512, "completion_tokens": 22}},
{"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 2111, "completion_tokens": 160}},
{"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1940, "completion_tokens": 142}},
{"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1325, "completion_tokens": 13}},
]
cost, snapshot = compute_llm_cost(calls, _PRICE)
# 6888/1e6*0.8 + 337/1e6*2.0 = 0.0055104 + 0.000674 = 0.0061844 → round(6)
assert cost == 0.006184
assert snapshot == {
"mode": "per_model",
"prices": {
"qwen3.5-flash": {"input_per_1m": 0.8, "output_per_1m": 2.0, "_source": "per_model"},
},
}
def test_multi_model_prices_each_bucket_separately():
calls = [
{"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1_000_000, "completion_tokens": 0}},
{"model": "gpt-x", "error": None, "usage": {"prompt_tokens": 0, "completion_tokens": 1_000_000}},
]
price = {
"per_model": {
"qwen3.5-flash": {"input_per_1m": 0.8, "output_per_1m": 2.0},
"gpt-x": {"input_per_1m": 10.0, "output_per_1m": 30.0},
},
"default": {"input_per_1m": 3.0, "output_per_1m": 15.0},
}
cost, snap = compute_llm_cost(calls, price)
assert cost == 30.8 # qwen 1M入×0.8=0.8 + gpt-x 1M出×30=30.0
assert set(snap["prices"]) == {"qwen3.5-flash", "gpt-x"}
def test_unknown_model_falls_back_to_default():
calls = [{"model": "mystery", "error": None, "usage": {"prompt_tokens": 1_000_000, "completion_tokens": 0}}]
price = {"per_model": {}, "default": {"input_per_1m": 3.0, "output_per_1m": 15.0}}
cost, snap = compute_llm_cost(calls, price)
assert cost == 3.0
assert snap["prices"]["mystery"]["_source"] == "default"
def test_unpriced_model_marked_and_zero_cost():
calls = [{"model": "mystery", "error": None, "usage": {"prompt_tokens": 1_000_000, "completion_tokens": 999}}]
cost, snap = compute_llm_cost(calls, {"per_model": {}}) # 无 default
assert cost == 0.0
assert snap["prices"]["mystery"]["unpriced"] is True
def test_error_and_missing_usage_calls_skipped():
calls = [
{"model": "qwen3.5-flash", "error": "boom", "usage": {"prompt_tokens": 9_999_999, "completion_tokens": 9_999_999}},
{"model": "qwen3.5-flash", "error": None, "usage": None}, # 无 usage
{"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1_000_000, "completion_tokens": 0}},
]
cost, _ = compute_llm_cost(calls, _PRICE)
assert cost == 0.8 # 只有第 3 条计入
def test_empty_or_all_error_returns_none():
assert compute_llm_cost([], _PRICE) == (None, None)
assert compute_llm_cost(None, _PRICE) == (None, None)
all_error = [{"model": "x", "error": "boom", "usage": {"prompt_tokens": 100, "completion_tokens": 100}}]
assert compute_llm_cost(all_error, _PRICE) == (None, None)
def test_malformed_price_entry_is_treated_as_unpriced_not_raised():
# 手改配置页可能存出残缺/非法单价(缺 output_per_1m、非 dict);不能抛异常连累 token 回填。
calls = [
{"model": "bad-a", "error": None, "usage": {"prompt_tokens": 1_000_000, "completion_tokens": 5}},
{"model": "bad-b", "error": None, "usage": {"prompt_tokens": 1_000_000, "completion_tokens": 5}},
{"model": "ok", "error": None, "usage": {"prompt_tokens": 1_000_000, "completion_tokens": 0}},
]
price = {
"per_model": {
"bad-a": {"input_per_1m": 0.8}, # 缺 output_per_1m
"bad-b": 5, # 非 dict
"ok": {"input_per_1m": 3.0, "output_per_1m": 15.0},
},
}
cost, snap = compute_llm_cost(calls, price) # 不得抛异常
assert cost == 3.0 # 只有 ok(1M 入 × 3.0)计入;两个残缺项按 unpriced
assert snap["prices"]["bad-a"].get("unpriced") is True
assert snap["prices"]["bad-b"].get("unpriced") is True
def test_get_llm_prices_falls_back_to_default_then_uses_override():
from app.db.session import SessionLocal
from app.models.app_config import AppConfig
from app.repositories import app_config
from app.services.llm_cost import get_llm_prices
db = SessionLocal()
try:
# 无 override → CONFIG_DEFS 默认(含 per_model / default)
prices = get_llm_prices(db)
assert "per_model" in prices and "default" in prices
# 有 override → 用 DB 值
app_config.set_value(
db, "llm_token_price",
{"per_model": {"m": {"input_per_1m": 1.0, "output_per_1m": 2.0}},
"default": {"input_per_1m": 0.0, "output_per_1m": 0.0}},
admin_id=1,
)
assert get_llm_prices(db)["per_model"]["m"]["input_per_1m"] == 1.0
finally:
row = db.get(AppConfig, "llm_token_price")
if row is not None:
db.delete(row)
db.commit()
db.close()
def test_backfill_llm_calls_stores_cost_and_snapshot(monkeypatch):
from datetime import UTC, datetime
from app.api.v1 import compare_record
from app.db.session import SessionLocal
from app.models.app_config import AppConfig
from app.models.comparison import ComparisonRecord
from app.repositories import app_config
sample = [
{"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1512, "completion_tokens": 22}},
{"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 2111, "completion_tokens": 160}},
{"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1940, "completion_tokens": 142}},
{"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1325, "completion_tokens": 13}},
]
monkeypatch.setattr(compare_record, "fetch_llm_calls", lambda trace_id: sample)
db = SessionLocal()
try:
app_config.set_value(
db, "llm_token_price",
{"per_model": {"qwen3.5-flash": {"input_per_1m": 0.8, "output_per_1m": 2.0}},
"default": {"input_per_1m": 3.0, "output_per_1m": 15.0}},
admin_id=1,
)
rec = ComparisonRecord(
trace_id="llmcost-bf-1", status="success",
created_at=datetime.now(UTC).replace(tzinfo=None),
)
db.add(rec)
db.commit()
rid = rec.id
finally:
db.close()
compare_record._backfill_llm_calls(rid, "llmcost-bf-1") # 独立 session 内回填
db = SessionLocal()
try:
rec = db.get(ComparisonRecord, rid)
assert rec.llm_cost_yuan == 0.006184
assert rec.llm_price_snapshot["prices"]["qwen3.5-flash"]["input_per_1m"] == 0.8
assert rec.input_tokens == 6888 # 现有 token 派生仍在
finally:
db.delete(db.get(ComparisonRecord, rid))
row = db.get(AppConfig, "llm_token_price")
if row is not None:
db.delete(row)
db.commit()
db.close()
def test_admin_detail_schema_exposes_llm_cost_fields():
from app.admin.schemas.comparison import AdminComparisonDetail
fields = AdminComparisonDetail.model_fields
assert "llm_cost_yuan" in fields
assert "llm_price_snapshot" in fields
-101
View File
@@ -1,101 +0,0 @@
"""oppo_push 集成层单测:未配降级 / auth+unicast 正常流程 / token 缓存 / OPPO 错误码抛错。
httpx monkeypatch, 不发真实网络
"""
from __future__ import annotations
import json
import pytest
from app.integrations import oppo_push
from app.integrations.oppo_push import OppoPushError
class _FakeResp:
def __init__(self, status_code: int, payload: dict) -> None:
self.status_code = status_code
self._payload = payload
self.text = json.dumps(payload)
def json(self) -> dict:
return self._payload
def _configure(monkeypatch) -> None:
"""配齐凭证 + 清 token 缓存(避免跨用例污染)。"""
monkeypatch.setattr(oppo_push.settings, "OPPO_PUSH_ENABLED", True)
monkeypatch.setattr(oppo_push.settings, "OPPO_PUSH_APP_KEY", "ak")
monkeypatch.setattr(oppo_push.settings, "OPPO_PUSH_MASTER_SECRET", "ms")
oppo_push._token_cache = None
def test_not_configured_raises(monkeypatch) -> None:
monkeypatch.setattr(oppo_push.settings, "OPPO_PUSH_APP_KEY", "")
monkeypatch.setattr(oppo_push.settings, "OPPO_PUSH_MASTER_SECRET", "")
oppo_push._token_cache = None
with pytest.raises(OppoPushError):
oppo_push.send_notification("regid", "t", "c")
def test_disabled_raises(monkeypatch) -> None:
_configure(monkeypatch)
monkeypatch.setattr(oppo_push.settings, "OPPO_PUSH_ENABLED", False)
with pytest.raises(OppoPushError):
oppo_push.send_notification("regid", "t", "c")
def test_success_flow(monkeypatch) -> None:
_configure(monkeypatch)
calls: list[tuple] = []
def _fake_post(url, **kw):
calls.append((url, kw))
if url == oppo_push.settings.OPPO_PUSH_AUTH_ENDPOINT:
return _FakeResp(200, {"code": 0, "message": "success", "data": {"auth_token": "TOK"}})
return _FakeResp(200, {"code": 0, "message": "success", "data": {"message_id": "MID"}})
monkeypatch.setattr(oppo_push.httpx, "post", _fake_post)
mid = oppo_push.send_notification("regid123", "标题", "内容")
assert mid == "MID"
# 先 auth 再 unicast, unicast header 带 auth_token
assert calls[0][0] == oppo_push.settings.OPPO_PUSH_AUTH_ENDPOINT
assert calls[1][0] == oppo_push.settings.OPPO_PUSH_UNICAST_ENDPOINT
assert calls[1][1]["headers"]["auth_token"] == "TOK"
# unicast body 的 message 是 JSON 字符串, 含 target_value / notification
msg = json.loads(calls[1][1]["data"]["message"])
assert msg["target_value"] == "regid123"
assert msg["notification"]["title"] == "标题"
assert msg["notification"]["content"] == "内容"
# channel_id 必带(默认 shagua_wallet); category 默认空 → 不带
assert msg["notification"]["channel_id"] == "push_oplus_category_content"
assert "category" not in msg["notification"]
def test_auth_token_cached(monkeypatch) -> None:
_configure(monkeypatch)
auth_hits: list[int] = []
def _fake_post(url, **kw):
if url == oppo_push.settings.OPPO_PUSH_AUTH_ENDPOINT:
auth_hits.append(1)
return _FakeResp(200, {"code": 0, "data": {"auth_token": "TOK"}})
return _FakeResp(200, {"code": 0, "data": {"message_id": "MID"}})
monkeypatch.setattr(oppo_push.httpx, "post", _fake_post)
oppo_push.send_notification("r", "t", "c")
oppo_push.send_notification("r", "t", "c")
assert len(auth_hits) == 1 # 第二次复用缓存 token, 不再换 auth
def test_oppo_error_raises(monkeypatch) -> None:
_configure(monkeypatch)
def _fake_post(url, **kw):
if url == oppo_push.settings.OPPO_PUSH_AUTH_ENDPOINT:
return _FakeResp(200, {"code": 0, "data": {"auth_token": "TOK"}})
return _FakeResp(200, {"code": -2, "message": "invalid target"})
monkeypatch.setattr(oppo_push.httpx, "post", _fake_post)
with pytest.raises(OppoPushError):
oppo_push.send_notification("r", "t", "c")
-70
View File
@@ -1,70 +0,0 @@
"""push_notify.notify_withdraw_approved 单测:多 OPPO 设备全推 / 无设备返 0 / 发送失败吞错。"""
from __future__ import annotations
from app.db.session import SessionLocal
from app.integrations.oppo_push import OppoPushError
from app.repositories import device as device_repo
from app.repositories import user as user_repo
from app.services import push_notify
def _seed_user(phone: str) -> int:
db = SessionLocal()
try:
return user_repo.upsert_user_for_login(db, phone=phone, register_channel="sms").id
finally:
db.close()
def _add_oppo_device(user_id: int, device_id: str, oppo_reg: str) -> None:
db = SessionLocal()
try:
device_repo.register_or_update(
db, user_id=user_id, device_id=device_id,
registration_id=None, oppo_register_id=oppo_reg,
)
finally:
db.close()
def test_notify_no_oppo_device_returns_zero() -> None:
uid = _seed_user("13911100001")
db = SessionLocal()
try:
assert push_notify.notify_withdraw_approved(db, user_id=uid) == 0
finally:
db.close()
def test_notify_pushes_all_oppo_devices(monkeypatch) -> None:
uid = _seed_user("13911100002")
_add_oppo_device(uid, "dev_a", "OPPO_REG_A")
_add_oppo_device(uid, "dev_b", "OPPO_REG_B")
sent: list[str] = []
monkeypatch.setattr(
push_notify.oppo_push, "send_notification",
lambda rid, title, content, **kw: sent.append(rid) or "msgid",
)
db = SessionLocal()
try:
n = push_notify.notify_withdraw_approved(db, user_id=uid)
finally:
db.close()
assert n == 2
assert set(sent) == {"OPPO_REG_A", "OPPO_REG_B"}
def test_notify_swallows_send_error(monkeypatch) -> None:
uid = _seed_user("13911100003")
_add_oppo_device(uid, "dev_c", "OPPO_REG_C")
def _boom(rid, title, content, **kw):
raise OppoPushError("simulated failure")
monkeypatch.setattr(push_notify.oppo_push, "send_notification", _boom)
db = SessionLocal()
try:
# 发送失败被吞, 不抛; 成功数为 0
assert push_notify.notify_withdraw_approved(db, user_id=uid) == 0
finally:
db.close()
+198
View File
@@ -0,0 +1,198 @@
"""微信登录 M1 测试:bind_ticket 令牌、wechat-login(openid 命中/未命中)、
bind-phone(建号/占用/令牌过期)
沿用 tests/test_auth.py 风格:HTTP client fixture;微信 codeopenid monkeypatch
拦掉(conftest WECHAT_APP_ID/SECRET dummy,不真连微信);短信走 SMS_MOCK(任意 6 位通过)
"""
from __future__ import annotations
import pytest
from app.api.v1 import auth
from app.core import security
from app.integrations import wxpay
def _fake_userinfo(openid: str, nickname: str | None = "微信昵称", avatar: str | None = "http://x/a.png"):
"""返回一个可传给 monkeypatch 的假 code_to_userinfo(忽略 code,固定返回给定 openid)。"""
def _f(code: str) -> dict:
return {"openid": openid, "nickname": nickname, "avatar_url": avatar, "raw": {}}
return _f
# ===== Task 1: bind_ticket 令牌 =====
def test_bind_ticket_roundtrip() -> None:
token = security.create_bind_ticket(openid="oid1", wechat_nickname="", wechat_avatar_url="http://a")
claims = security.decode_bind_ticket(token)
assert claims["openid"] == "oid1"
assert claims["wnk"] == ""
assert claims["wav"] == "http://a"
def test_bind_ticket_wrong_type_rejected() -> None:
# 用 access token 冒充 bind_ticket → TokenError(typ 不匹配)
access, _ = security.create_token(user_id=1, token_type="access")
with pytest.raises(security.TokenError):
security.decode_bind_ticket(access)
def test_bind_ticket_expired_rejected(monkeypatch) -> None:
monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1)
token = security.create_bind_ticket(openid="oid", wechat_nickname=None, wechat_avatar_url=None)
with pytest.raises(security.TokenError):
security.decode_bind_ticket(token)
# ===== Task 2: wechat-login =====
def test_wechat_login_new_openid_returns_bind_ticket(client, monkeypatch) -> None:
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_new_1", "小明", "http://x/m.png"))
r = client.post("/api/v1/auth/wechat-login", json={"code": "wxcode1", "device_id": "devA"})
assert r.status_code == 200, r.text
body = r.json()
assert body["status"] == "need_bind_phone"
assert body["bind_ticket"]
assert body["wechat_nickname"] == "小明"
assert body["wechat_avatar_url"] == "http://x/m.png"
assert body["token"] is None
def test_wechat_login_invalid_code_returns_400(client, monkeypatch) -> None:
def _raise(code: str) -> dict:
raise ValueError("微信授权失败: invalid code")
monkeypatch.setattr(wxpay, "code_to_userinfo", _raise)
r = client.post("/api/v1/auth/wechat-login", json={"code": "bad", "device_id": "devA"})
assert r.status_code == 400, r.text
# ===== Task 3: bind-phone/sms =====
def test_wechat_bind_sms_creates_account_then_openid_logs_in(client, monkeypatch) -> None:
"""未占用 → 建微信账号(channel=wechat,昵称头像取微信);再次同 openid 登录 → 直接登入同一账号。"""
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_flow_2", "阿花", "http://x/h.png"))
phone = "13900139002"
# 1) 微信登录 → 未命中 → 拿 ticket
r = client.post("/api/v1/auth/wechat-login", json={"code": "c1", "device_id": "devB"})
ticket = r.json()["bind_ticket"]
assert ticket
# 2) 短信绑号(SMS_MOCK:任意 6 位通过)→ 建号 + 登入
r = client.post(
"/api/v1/auth/wechat/bind-phone/sms",
json={"bind_ticket": ticket, "phone": phone, "code": "123456", "device_id": "devB"},
)
assert r.status_code == 200, r.text
body = r.json()
assert body["status"] == "logged_in"
user = body["token"]["user"]
assert user["phone"] == phone
assert user["register_channel"] == "wechat"
assert user["nickname"] == "阿花"
assert user["avatar_url"] == "http://x/h.png"
uid = user["id"]
# 3) 再次微信登录(同 openid)→ 命中 → 直接登入同一账号
r = client.post("/api/v1/auth/wechat-login", json={"code": "c2", "device_id": "devB"})
assert r.status_code == 200, r.text
body = r.json()
assert body["status"] == "logged_in"
assert body["token"]["user"]["id"] == uid
def test_wechat_bind_sms_phone_occupied(client, monkeypatch) -> None:
"""手机号已被其他账号占用 → 返回 phone_occupied + 原账号信息(不建号)。"""
phone = "13900139003"
# 先用普通短信登录占用该手机号(register_channel=sms)
assert client.post("/api/v1/auth/sms/send", json={"phone": phone}).status_code == 200
r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
assert r.status_code == 200, r.text
occupied_nickname = r.json()["user"]["nickname"]
# 微信登录(新 openid)→ 未命中 → ticket
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_occ_3"))
ticket = client.post(
"/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devC"}
).json()["bind_ticket"]
# 绑同一手机号 → 占用
r = client.post(
"/api/v1/auth/wechat/bind-phone/sms",
json={"bind_ticket": ticket, "phone": phone, "code": "123456", "device_id": "devC"},
)
assert r.status_code == 200, r.text
body = r.json()
assert body["status"] == "phone_occupied"
assert body["token"] is None
assert body["occupied_account"]["nickname"] == occupied_nickname
assert body["occupied_account"]["avatar_url"] is None # 短信注册账号无头像 → 序列化为 null
assert body["occupied_account"]["created_at"]
def test_wechat_bind_sms_expired_ticket_returns_401(client, monkeypatch) -> None:
"""过期 bind_ticket → 401。"""
monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1)
expired = security.create_bind_ticket(openid="openid_exp", wechat_nickname="x", wechat_avatar_url=None)
r = client.post(
"/api/v1/auth/wechat/bind-phone/sms",
json={"bind_ticket": expired, "phone": "13900139009", "code": "123456", "device_id": "devD"},
)
assert r.status_code == 401, r.text
# ===== Task 4: bind-phone/jverify =====
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 ...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"}
).json()["bind_ticket"]
r = client.post(
"/api/v1/auth/wechat/bind-phone/jverify",
json={"bind_ticket": ticket, "login_token": "jgtoken", "device_id": "devE"},
)
assert r.status_code == 200, r.text
body = r.json()
assert body["status"] == "logged_in"
user = body["token"]["user"]
assert user["phone"] == phone
assert user["register_channel"] == "wechat"
assert user["nickname"] == "极光用户"
# 微信 userinfo 隐私脱敏 avatar=None → 头像为空(客户端兜底默认头像)
assert user["avatar_url"] is None
def test_wechat_bind_jverify_expired_ticket_returns_401(client, monkeypatch) -> None:
"""过期 bind_ticket → 401(极光绑号路径,decode 先于极光核验)。"""
monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1)
expired = security.create_bind_ticket(openid="openid_jv_exp", wechat_nickname="x", wechat_avatar_url=None)
r = client.post(
"/api/v1/auth/wechat/bind-phone/jverify",
json={"bind_ticket": expired, "login_token": "jgtoken", "device_id": "devE"},
)
assert r.status_code == 401, r.text
def test_wechat_bind_jverify_jiguang_error_returns_502(client, monkeypatch) -> None:
"""极光核验失败(JiguangError)→ 502。"""
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_jv_err"))
def _raise(token: str) -> str:
raise auth.JiguangError("mock jg failure")
monkeypatch.setattr(auth, "verify_and_get_phone", _raise)
ticket = client.post(
"/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devE"}
).json()["bind_ticket"]
r = client.post(
"/api/v1/auth/wechat/bind-phone/jverify",
json={"bind_ticket": ticket, "login_token": "badtoken", "device_id": "devE"},
)
assert r.status_code == 502, r.text