Compare commits

..

1 Commits

Author SHA1 Message Date
no_gen_mu 2d74693c21 fix(withdraw): allow multiple active applications 2026-07-12 19:27:43 +08:00
34 changed files with 71 additions and 1403 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')
@@ -0,0 +1,32 @@
"""allow multiple active withdraw orders per user
Revision ID: withdraw_allow_multiple_active
Revises: ad_ecpm_trace_id
Create Date: 2026-07-12 18:00:00.000000
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
revision: str = "withdraw_allow_multiple_active"
down_revision: str | Sequence[str] | None = "ad_ecpm_trace_id"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
op.drop_index("ux_withdraw_order_user_active", table_name="withdraw_order")
def downgrade() -> None:
op.create_index(
"ux_withdraw_order_user_active",
"withdraw_order",
["user_id"],
unique=True,
sqlite_where=sa.text("status IN ('reviewing', 'pending')"),
postgresql_where=sa.text("status IN ('reviewing', 'pending')"),
)
-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
-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()
+1 -6
View File
@@ -201,7 +201,7 @@ def withdraw_info(
"/withdraw",
response_model=WithdrawResultOut,
summary="发起提现(扣款建单,待人工审核;审核通过后才打款)",
dependencies=[Depends(rate_limit(5, 60, "withdraw"))], # IP 级粗限流;用户级未完成单限制在仓库层
dependencies=[Depends(rate_limit(5, 60, "withdraw"))], # IP 级粗限流;余额/档位/幂等限制在仓库层
)
def withdraw(req: WithdrawRequest, user: CurrentUser, db: DbSession) -> WithdrawResultOut:
# 提现发起本身不调微信(打款在审核通过后),但仍要求微信支付已配置——否则审核通过也打不了款,提前拦
@@ -222,11 +222,6 @@ def withdraw(req: WithdrawRequest, user: CurrentUser, db: DbSession) -> Withdraw
) from e
except crud_wallet.WechatNotBoundError as e:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="请先绑定微信") from e
except crud_wallet.WithdrawTooFrequentError as e:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="已有提现申请正在审核或打款中,请处理完成后再申请",
) from e
except crud_wallet.WithdrawTierUnavailableError as e:
# 福利页档位闸(7-9):次数满/已选其他额度。正常客户端已按 tiers 预拦,此处兜底防绕过。
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="今日额度已达上限") from e
-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)
-39
View File
@@ -71,34 +71,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 +136,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 +146,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 兜底未登记的模型。改价只影响之后回填的新记录,历史记录用当时价格快照。"
),
},
}
-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)
-10
View File
@@ -96,16 +96,6 @@ class WithdrawOrder(Base):
"""
__tablename__ = "withdraw_order"
__table_args__ = (
Index(
"ux_withdraw_order_user_active",
"user_id",
unique=True,
sqlite_where=text("status IN ('reviewing', 'pending')"),
postgresql_where=text("status IN ('reviewing', 'pending')"),
),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
user_id: Mapped[int] = mapped_column(
Integer, ForeignKey("user.id"), index=True, nullable=False
+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)
-22
View File
@@ -33,7 +33,6 @@ from app.models.wallet import (
_WX_STATE_SUCCESS = "SUCCESS"
_WX_STATE_FAILED = {"FAIL", "CANCELLED", "CLOSED"}
_WX_STATE_WAIT_CONFIRM = "WAIT_USER_CONFIRM" # 用户还没在微信确认页确认
_WITHDRAW_ACTIVE_STATUSES = {"reviewing", "pending"}
# 免确认收款授权状态
_WX_AUTH_ACTIVE = "TAKING_EFFECT" # 已生效,可免确认转账
_WX_AUTH_CLOSED = "CLOSED" # 已关闭(用户/商户/风控),需重新开启
@@ -63,10 +62,6 @@ class InsufficientCashError(Exception):
"""现金余额不足。"""
class WithdrawTooFrequentError(Exception):
"""提现申请过于频繁,或已有未完成提现单。"""
class WithdrawTierUnavailableError(Exception):
"""该档位今日不可提:次数已满,或今天已选了其他额度(7-9 福利页档位规则)。"""
@@ -738,15 +733,6 @@ def create_withdraw(
else:
out_bill_no = uuid.uuid4().hex
active_order_id = db.execute(
select(WithdrawOrder.id).where(
WithdrawOrder.user_id == user_id,
WithdrawOrder.status.in_(_WITHDRAW_ACTIVE_STATUSES),
).limit(1)
).scalar_one_or_none()
if active_order_id is not None:
raise WithdrawTooFrequentError
# 福利页档位闸(7-9):coin_cash 只能提预设档位,且该档今日可提(服务端权威口径,防绕过
# 客户端刷)。放在幂等返回/在途互斥之后:同号重试仍原样返回旧单,不被档位闸误杀。
# allow_sub_min(0.01 调试直发)保持原样放行,不受档位约束;invite_cash 本轮无档位概念不校验。
@@ -805,14 +791,6 @@ def create_withdraw(
).scalar_one_or_none()
if existing is not None:
return existing
active_order_id = db.execute(
select(WithdrawOrder.id).where(
WithdrawOrder.user_id == user_id,
WithdrawOrder.status.in_(_WITHDRAW_ACTIVE_STATUSES),
).limit(1)
).scalar_one_or_none()
if active_order_id is not None:
raise WithdrawTooFrequentError from None
raise
db.refresh(order)
return order # 待管理员审核;**不在此处打款**
-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 集合` 现挂到实例上供出参用。
+2 -2
View File
@@ -45,9 +45,9 @@ reviewing ──admin 审核拒绝──▶ rejected(已退款)
## 索引与约束
- PK `id`;UNIQUE+index `out_bill_no`;index `user_id``created_at`
- 部分唯一索引 `ux_withdraw_order_user_active`(`user_id`),条件 `status IN ('reviewing', 'pending')`:每个用户同时只能有一笔在途(待审核 / 打款中)提现单,DB 层挡并发重复提现
- 同一用户允许同时存在多笔 `reviewing` / `pending` 提现单;每笔以唯一 `out_bill_no` 独立审核、打款和对账
## 注意
- **资金安全**:原子扣款(`WHERE cash_balance_cents >= amount`)+ `out_bill_no` 幂等 + 结果不明时**先查单再决定,绝不盲目退款**(防退款后又到账)+ 孤儿 pending 单 `reconcile_pending_withdraws` 对账兜底。
- **资金安全**:允许多笔在途不等于重复扣款。仍由原子扣款(`WHERE cash_balance_cents >= amount`)+ `out_bill_no` 幂等 + 结果不明时**先查单再决定,绝不盲目退款**(防退款后又到账)+ 孤儿 pending 单 `reconcile_pending_withdraws` 对账兜底。
- `WITHDRAW_MIN_CENTS=10`(0.1 元,微信商家转账地板价),可经 `app_config.withdraw_min_cents` 后台调。
- "待审核期间钱已扣减",防用户拿同一笔余额重复发起多笔提现。
@@ -134,7 +134,7 @@ App 用户主表。两种登录(极光一键 / 短信验证码)都映射到
## `withdraw_order` — 提现单
现金 → 微信零钱。状态机:reviewing(待审核)→ pending(打款在途)→ success / failedreviewing →(拒绝)→ rejected(已退款)。同一用户同时只能有一笔进行中
现金 → 微信零钱。状态机:reviewing(待审核)→ pending(打款在途)→ success / failedreviewing →(拒绝)→ rejected(已退款)。同一用户同时存在多笔进行中的提现单,每次提交通过唯一 `out_bill_no` 保证幂等
| 字段 | 类型 | 说明 |
|---|---|---|
-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「比价记录」→ 找 trace「MOCKLLM-」的 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()
+1 -1
View File
@@ -7,7 +7,7 @@
- 配套双分录现金流水(withdraw / withdraw_refund / exchange_in)+ 账户余额,
让顶部「账本校验」保持绿色、详情抽屉的现金余额/流水也真实。
约束:withdraw_order 有部分唯一索引(同一 user reviewing/pending 最多 1 单),
约束:同一 user 可有多笔 reviewing/pending 单,每笔 out_bill_no 唯一且独立对账,
本脚本每个 mock 用户至多 1 个活动单,满足约束。
幂等:每次运行先按固定 mock 手机号清掉旧 mock 再重建。仅清理用 --clean-only。
-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 == []
+4 -5
View File
@@ -128,8 +128,7 @@ def test_invite_cash_reject_refunds_invite_account(client, monkeypatch) -> None:
def test_two_accounts_withdraw_independent(client, monkeypatch) -> None:
"""两账户各提各的不串:先提 invite_cash(拒绝结清),再提 cash,各扣各账户。
注:一个用户同一时间只能一个活跃提现单(跨账户),故第二笔需先结清第一笔。"""
"""两账户各提各的不串:invite_cash 与 coin_cash 可同时保留 reviewing 单。"""
_patch_userinfo(monkeypatch, "openid_ic_3")
token = _login(client, "13800004003")
_seed_balances(client, token, "13800004003", cash=400, invite_cash=500)
@@ -140,17 +139,17 @@ def test_two_accounts_withdraw_independent(client, monkeypatch) -> None:
json={"amount_cents": 200, "source": "invite_cash"},
headers=_auth(token),
)
_reject(r1.json()["out_bill_no"]) # 退回 invite_cash + 结清活跃单
assert r1.status_code == 200 and r1.json()["status"] == "reviewing", r1.text
r2 = client.post(
"/api/v1/wallet/withdraw",
# 50 分 = 0.5 元档(7-9 起 coin_cash 只能提预设档位)
json={"amount_cents": 50, "source": "coin_cash"},
headers=_auth(token),
)
assert r2.json()["status"] == "reviewing"
assert r2.status_code == 200 and r2.json()["status"] == "reviewing", r2.text
cash, invite_cash = _balances(client, token)
assert invite_cash == 500 # 已退回
assert invite_cash == 300 # invite_cash 单仍待审核,已扣 200
assert cash == 350 # 扣了 cash 50
-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()
+28
View File
@@ -268,6 +268,34 @@ def test_withdraw_idempotent_same_bill_no(client, monkeypatch) -> None:
assert sum(1 for t in r.json()["items"] if t["biz_type"] == "withdraw") == 1
def test_withdraw_allows_multiple_active_orders_with_distinct_bill_no(client, monkeypatch) -> None:
"""不同 out_bill_no 是不同提现意图,允许同时处于 reviewing;每笔分别扣款建单。"""
_patch_userinfo(monkeypatch, "openid_multi_active")
token = _login(client, "13800002017")
_seed_cash(client, token, "13800002017", 200)
client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
r1 = client.post(
"/api/v1/wallet/withdraw",
json={"amount_cents": 50, "out_bill_no": "multi_active_bill_1"},
headers=_auth(token),
)
r2 = client.post(
"/api/v1/wallet/withdraw",
json={"amount_cents": 50, "out_bill_no": "multi_active_bill_2"},
headers=_auth(token),
)
assert r1.status_code == 200 and r1.json()["status"] == "reviewing", r1.text
assert r2.status_code == 200 and r2.json()["status"] == "reviewing", r2.text
assert r1.json()["out_bill_no"] != r2.json()["out_bill_no"]
account = client.get("/api/v1/wallet/account", headers=_auth(token)).json()
assert account["cash_balance_cents"] == 100
txns = client.get("/api/v1/wallet/cash-transactions", headers=_auth(token)).json()["items"]
assert sum(1 for t in txns if t["biz_type"] == "withdraw") == 2
def test_withdraw_ambiguous_timeout_then_success_no_refund(client, monkeypatch) -> None:
"""#3 转账调用超时(异常),但查单确认已 SUCCESS → 不退款,单置 success。"""
monkeypatch.setattr("app.integrations.wxpay.code_to_userinfo", lambda code: {"openid": "openid_amb", "nickname": None, "avatar_url": None, "raw": {}})