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
27 changed files with 72 additions and 1161 deletions
-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")
@@ -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')"),
)
-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",
+3 -18
View File
@@ -14,24 +14,18 @@ import logging
from fastapi import APIRouter
from app.api.deps import CurrentUser, DbSession
from app.core import wx_poc_signal
from app.repositories import device as device_repo
from app.schemas.device import (
DeviceOut,
DeviceRegisterRequest,
HeartbeatRequest,
HeartbeatResponse,
LivenessAckRequest,
LivenessOut,
OkResponse,
PendingCompare,
)
logger = logging.getLogger("shagua.device")
# PoC:源平台代号 → Android 包名(前端 launch 用)。当前只用美团。
_SOURCE_PACKAGES = {"meituan": "com.sankuai.meituan"}
router = APIRouter(prefix="/api/v1/device", tags=["device"])
@@ -58,12 +52,12 @@ def register_device(
return DeviceOut.model_validate(device)
@router.post("/heartbeat", response_model=HeartbeatResponse, summary="上报心跳")
@router.post("/heartbeat", response_model=OkResponse, summary="上报心跳")
def report_heartbeat(
req: HeartbeatRequest,
user: CurrentUser,
db: DbSession,
) -> HeartbeatResponse:
) -> OkResponse:
device_repo.touch_heartbeat(
db,
user_id=user.id,
@@ -71,16 +65,7 @@ def report_heartbeat(
accessibility_enabled=req.accessibility_enabled,
registration_id=req.registration_id,
)
# PoC:该设备有"待比价"信号则带回(只对写死的测试设备生效, 取走即清、只弹一次)
source = wx_poc_signal.pop_pending(req.device_id)
pending = None
if source:
pending = PendingCompare(
source_platform=source,
source_package=_SOURCE_PACKAGES.get(source, ""),
)
logger.info("heartbeat 下发比价信号 device=%s source=%s", req.device_id, source)
return HeartbeatResponse(pending_compare=pending)
return OkResponse()
@router.get("/liveness", response_model=LivenessOut, summary="查询本机掉线告警(后置检测)")
+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
-159
View File
@@ -1,159 +0,0 @@
"""微信客服(企业微信)消息接收回调 /wx/kf/callback(裸路径, 非 /api/v1)。
对应企业微信「微信客服」→ 客服账号 → 接收消息回调里填的 URL。是服务号版(wx_mp.py)的
平行实现:下游(set_pending → 心跳下发 → 前端弹窗)完全一致, 只有上游"如何拿到用户消息"不同 ——
服务号是微信把消息直接 POST 给你;微信客服是 POST 一个事件通知, 你再用 <Token> 主动 sync_msg 拉。
GET : URL 接入验证 → msg_signature 验签(含密文 echostr) → 解密 echostr → 返回明文。
⚠️ 与公众号明文模式不同:企业微信 echostr 是密文, 必须解密后返回。
POST : 收事件(密文) → msg_signature 验签 → AES 解密 → 解析 XML。
Event=kf_msg_or_event → 取 <Token>/<OpenKfId> → sync_msg 增量拉消息 →
客户(origin=3)发来的图片(image) → set_pending(测试设备, "meituan") + 下载落盘。
其余消息/事件记日志忽略。一律返回 "success"(异常也吞掉, 不让企业微信重试轰炸)。
回调加密与公众号「安全模式」同源, 直接复用 integrations/wx_mp_crypto(验签 + AES-256-CBC 解密),
差别只在 receiveid 用 corpid(公众号是 appid)。MVP 只做"收图打信号 + 落盘证明链路通"
"""
from __future__ import annotations
import logging
import xml.etree.ElementTree as ET
from pathlib import Path
from typing import Any
from fastapi import APIRouter, Request
from fastapi.responses import PlainTextResponse
from app.core import wx_poc_signal
from app.core.config import settings
from app.integrations import wx_kf_client, wx_mp_crypto
logger = logging.getLogger("shagua.wx_kf")
router = APIRouter(prefix="/wx/kf", tags=["wx-kf"])
# 收到的截图暂存目录(MVP 验证用;接上识别后可改为不落盘或定期清理)
_INBOX = Path(settings.MEDIA_ROOT) / "wx_kf_inbox"
@router.get("/callback", include_in_schema=False)
async def verify(
msg_signature: str = "", timestamp: str = "", nonce: str = "", echostr: str = ""
) -> PlainTextResponse:
"""企业微信 URL 接入验证:验签(密文 echostr 参与)→ 解密 echostr → 返回明文。"""
if not settings.wx_kf_callback_configured:
logger.warning("wx_kf verify: 回调凭证未配齐(CorpId/Token/AESKey/Secret)")
return PlainTextResponse("", status_code=503)
if not wx_mp_crypto.verify_msg_signature(
settings.WX_KF_TOKEN, timestamp, nonce, echostr, msg_signature
):
logger.warning("wx_kf verify: signature 校验失败 ts=%s nonce=%s", timestamp, nonce)
return PlainTextResponse("invalid signature", status_code=403)
try:
plain = wx_mp_crypto.decrypt_message(
settings.WX_KF_AES_KEY, settings.WX_KF_CORP_ID, echostr
)
except Exception:
logger.exception("wx_kf verify: echostr 解密失败")
return PlainTextResponse("decrypt failed", status_code=403)
return PlainTextResponse(plain)
@router.post("/callback", include_in_schema=False)
async def receive(request: Request) -> PlainTextResponse:
"""收事件通知(安全模式)。任何异常都吞掉返回 success, 不让企业微信重试轰炸, 靠日志排查。"""
if not settings.wx_kf_callback_configured:
logger.warning("wx_kf receive: 回调凭证未配齐")
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_KF_TOKEN,
qp.get("timestamp", ""),
qp.get("nonce", ""),
encrypt,
qp.get("msg_signature", ""),
):
logger.warning("wx_kf receive: msg_signature 校验失败")
return PlainTextResponse("success")
xml = wx_mp_crypto.decrypt_message(
settings.WX_KF_AES_KEY, settings.WX_KF_CORP_ID, encrypt
)
await _handle_event(ET.fromstring(xml))
except Exception:
logger.exception("wx_kf receive: 处理异常")
return PlainTextResponse("success")
async def _handle_event(root: ET.Element) -> None:
"""客服事件:MsgType=event 且 Event=kf_msg_or_event → 用 Token 增量拉消息。"""
event = root.findtext("Event") or ""
if event != "kf_msg_or_event":
logger.info("wx_kf 收到事件(暂忽略): event=%s", event)
return
token = root.findtext("Token") or ""
open_kfid = root.findtext("OpenKfId") or ""
msgs = await wx_kf_client.sync_messages(token, open_kfid)
logger.info("wx_kf sync_msg 拉到 %d 条 open_kfid=%s", len(msgs), open_kfid)
for msg in msgs:
await _handle_message(msg)
async def _handle_message(msg: dict[str, Any]) -> None:
"""处理单条客服消息。只关心【客户发来的】(origin=3)图片 / 小程序卡片:打比价信号(+ 落盘)。
origin:3=客户发的 / 4=系统推送 / 5=接待人员发的。接待人员或系统消息一律不触发比价。
两种触发物对应场景:image=用户发结算页截图;miniprogram=用户分享美团小程序卡片(见需求截图)。
"""
if msg.get("origin") != 3:
return
msgtype = msg.get("msgtype") or ""
external_userid = msg.get("external_userid") or ""
if msgtype == "image":
media_id = (msg.get("image") or {}).get("media_id", "")
logger.info(
"wx_kf 收到图片: external_userid=%s media_id=%s", external_userid, media_id
)
_fire_pending("image")
await _save_media(media_id, external_userid)
elif msgtype == "miniprogram":
mp = msg.get("miniprogram") or {}
logger.info(
"wx_kf 收到小程序卡片: external_userid=%s title=%s appid=%s",
external_userid,
mp.get("title"),
mp.get("appid"),
)
# 卡片无可下载媒体, 只打信号;源平台/订单识别(可据 appid 判平台)留后续, 现写死美团。
_fire_pending("miniprogram")
else:
logger.info(
"wx_kf 收到消息(暂忽略): external_userid=%s type=%s", external_userid, msgtype
)
def _fire_pending(kind: str) -> None:
"""PoC:给写死的测试设备打"从美团比价"信号(下次心跳带回前端弹窗, 与服务号版完全一致)。"""
poc_dev = settings.WX_POC_TEST_DEVICE_ID
if not poc_dev:
return
wx_poc_signal.set_pending(poc_dev, "meituan")
logger.info("wx_kf PoC: 已给测试设备 %s 打比价信号 source=meituan (触发=%s)", poc_dev, kind)
async def _save_media(media_id: str, external_userid: str) -> None:
"""图片消息只给 media_id, 走临时素材下载接口拉回落盘(PoC 验证用)。"""
content = await wx_kf_client.download_media(media_id)
if not content:
return
try:
_INBOX.mkdir(parents=True, exist_ok=True)
dest = _INBOX / f"{external_userid[:16]}_{media_id[:16]}.jpg"
dest.write_bytes(content)
logger.info("wx_kf 图片已落盘: %s (%d bytes)", dest, len(content))
except Exception:
logger.exception("wx_kf 图片落盘失败 media_id=%s", media_id)
-112
View File
@@ -1,112 +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 import wx_poc_signal
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
)
# PoC: 写死"该测试设备要从美团比价", 下次心跳带回前端弹选平台窗
poc_dev = settings.WX_POC_TEST_DEVICE_ID
if poc_dev:
wx_poc_signal.set_pending(poc_dev, "meituan")
logger.info("wx_mp PoC: 已给测试设备 %s 打比价信号 source=meituan", poc_dev)
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)
-35
View File
@@ -136,26 +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 加解密)
# ===== 微信截图比价 PoC(临时验证链路, 验证后删) =====
# 只对这台写死的测试设备下发"从美团比价"信号;空=不触发任何设备(部署上线零风险)。
# 收图 → 给此 device_id 打 pending → 该设备下次心跳带回 → 前端弹选平台窗。
WX_POC_TEST_DEVICE_ID: str = ""
# ===== 微信客服(企业微信;截图比价上游入口 —— 服务号版的平行实现) =====
# 用户在「微信客服」会话发消息 → 企业微信 POST 事件到 /wx/kf/callback → 后端 sync_msg 拉消息。
# 回调加密与公众号「安全模式」同源(复用 wx_mp_crypto), 但 receiveid 用 corpid 且 echostr 需解密。
# 拉消息 / 下载图片走 qyapi, 需 access_token(corpid + secret 换取)。全空=回调端点拒收, 零风险。
WX_KF_CORP_ID: str = "" # 企业微信 corpid(回调解密 receiveid + 换 access_token)
WX_KF_SECRET: str = "" # 「微信客服」Secret(换 access_token)
WX_KF_TOKEN: str = "" # 客服「接收消息」回调的 Token(URL 验签 + 消息验签)
WX_KF_AES_KEY: str = "" # 客服「接收消息」回调的 EncodingAESKey(AES-256-CBC 解密)
@property
def wx_mp_configured(self) -> bool:
"""服务号网页授权凭证齐全(缺则落地页不发起授权,降级为无 openid)。"""
@@ -166,21 +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)
@property
def wx_kf_callback_configured(self) -> bool:
"""微信客服回调凭证齐全(CorpId + Token + AESKey + Secret)。缺则回调端点拒绝处理。"""
return bool(
self.WX_KF_CORP_ID
and self.WX_KF_TOKEN
and self.WX_KF_AES_KEY
and self.WX_KF_SECRET
)
# ===== 微信支付(商家转账到零钱 / 提现)=====
# 真实凭证放 .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 兜底未登记的模型。改价只影响之后回填的新记录,历史记录用当时价格快照。"
),
},
}
-27
View File
@@ -1,27 +0,0 @@
"""微信截图比价 PoC:进程内存的"待比价"信号(单 worker 够用, 重启即失效, PoC 可接受)。
收图端点 set_pending(device_id, source) → 该设备下次心跳 pop_pending 取走并清除 →
心跳响应带回 → 前端弹选平台窗。只对 settings.WX_POC_TEST_DEVICE_ID 写入, 不碰其他设备。
后续接真识别 / openid↔device 绑定时整体替换本模块。
"""
from __future__ import annotations
import threading
_lock = threading.Lock()
_pending: dict[str, str] = {} # device_id -> source_platform(如 "meituan")
def set_pending(device_id: str, source_platform: str) -> None:
if not device_id:
return
with _lock:
_pending[device_id] = source_platform
def pop_pending(device_id: str) -> str | None:
"""取出并清除该设备的待比价信号;无则 None。心跳每帧调, 取到即消费(只弹一次)。"""
if not device_id:
return None
with _lock:
return _pending.pop(device_id, None)
-126
View File
@@ -1,126 +0,0 @@
"""微信客服(企业微信)API 客户端:access_token 缓存 + sync_msg 拉消息 + 临时素材下载。
对应 api/v1/wx_kf.py:收到 `kf_msg_or_event` 回调后, 用这里的函数增量拉取用户消息
- access_token(7200s)进程内缓存, 提前 300s 刷新( worker 够用, 重启即失效)
- sync_msg 游标 next_cursor open_kfid 进程内存续(PoC 重启后从最近 3 天重拉, 可接受)
凭证来自 settings(WX_KF_CORP_ID / WX_KF_SECRET);纯外部 HTTP, 不含业务逻辑
"""
from __future__ import annotations
import asyncio
import logging
import time
from typing import Any
import httpx
from app.core.config import settings
logger = logging.getLogger("shagua.wx_kf")
_QYAPI = "https://qyapi.weixin.qq.com/cgi-bin"
# access_token 进程内缓存(单 worker)。token 空 / 未到刷新点直接用。
_token_lock = asyncio.Lock()
_token_cache: dict[str, Any] = {"token": "", "expire_at": 0.0}
# sync_msg 游标:open_kfid -> next_cursor(PoC 进程内存;重启从最近 3 天重拉)
_cursors: dict[str, str] = {}
async def _get_access_token() -> str:
"""取企业微信 access_token, 进程内缓存, 提前 300s 过期刷新。失败返回空串(调用方降级)。"""
now = time.time()
if _token_cache["token"] and _token_cache["expire_at"] - 300 > now:
return _token_cache["token"]
async with _token_lock:
now = time.time() # 拿锁后复检, 避免并发重复刷新
if _token_cache["token"] and _token_cache["expire_at"] - 300 > now:
return _token_cache["token"]
try:
async with httpx.AsyncClient(timeout=10) as client:
resp = await client.get(
f"{_QYAPI}/gettoken",
params={
"corpid": settings.WX_KF_CORP_ID,
"corpsecret": settings.WX_KF_SECRET,
},
)
data = resp.json()
except Exception:
logger.exception("wx_kf gettoken 请求失败")
return ""
if data.get("errcode"):
logger.warning(
"wx_kf gettoken errcode=%s errmsg=%s", data.get("errcode"), data.get("errmsg")
)
return ""
_token_cache["token"] = data.get("access_token", "")
_token_cache["expire_at"] = now + int(data.get("expires_in", 7200))
return _token_cache["token"]
async def sync_messages(token: str, open_kfid: str) -> list[dict[str, Any]]:
"""收到 kf_msg_or_event 回调后调用:用回调带的一次性 token 增量拉消息, has_more 循环拉完。
token: 回调事件里的 <Token>(消息拉取凭证, 非配置 Token;不传会有严格频控)
open_kfid:客服账号 id(回调里的 <OpenKfId>)
返回本次新拉到的 msg_list(已合并多页);游标按 open_kfid 进程内存续
"""
access_token = await _get_access_token()
if not access_token:
return []
out: list[dict[str, Any]] = []
cursor = _cursors.get(open_kfid, "")
try:
async with httpx.AsyncClient(timeout=15) as client:
for _ in range(20): # 最多 20 页护栏, 防异常时死循环
body: dict[str, Any] = {"cursor": cursor, "token": token, "limit": 1000}
if open_kfid:
body["open_kfid"] = open_kfid
resp = await client.post(
f"{_QYAPI}/kf/sync_msg",
params={"access_token": access_token},
json=body,
)
data = resp.json()
if data.get("errcode"):
logger.warning(
"wx_kf sync_msg errcode=%s errmsg=%s",
data.get("errcode"),
data.get("errmsg"),
)
break
out.extend(data.get("msg_list", []))
cursor = data.get("next_cursor", cursor)
_cursors[open_kfid] = cursor
if not data.get("has_more"):
break
except Exception:
logger.exception("wx_kf sync_msg 请求失败 open_kfid=%s", open_kfid)
return out
async def download_media(media_id: str) -> bytes | None:
"""临时素材下载(图片消息只给 media_id)。成功返回二进制, 失败返回 None。"""
if not media_id:
return None
access_token = await _get_access_token()
if not access_token:
return None
try:
async with httpx.AsyncClient(timeout=20) as client:
resp = await client.get(
f"{_QYAPI}/media/get",
params={"access_token": access_token, "media_id": media_id},
)
resp.raise_for_status()
# 成功=二进制文件流;失败=JSON(errcode)。据 Content-Type 区分。
ctype = resp.headers.get("Content-Type", "")
if "application/json" in ctype or "text/plain" in ctype:
logger.warning("wx_kf media/get 非文件响应: %s", resp.text[:200])
return None
return resp.content
except Exception:
logger.exception("wx_kf media/get 下载失败 media_id=%s", media_id)
return None
-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]
-6
View File
@@ -39,8 +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_kf import router as wx_kf_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 (
@@ -142,10 +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)
# 微信客服(企业微信)消息接收回调 /wx/kf/callback(服务号版的平行实现, 收事件→sync_msg 拉图)
app.include_router(wx_kf_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
-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
-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 # 待管理员审核;**不在此处打款**
-14
View File
@@ -36,20 +36,6 @@ class OkResponse(BaseModel):
ok: bool = True
class PendingCompare(BaseModel):
"""PoC:后端通过心跳下发的"用户要从某源平台比价"信号。前端据此弹选平台窗 + launch 源平台。"""
source_platform: str # pricebot 源平台代号(PoC 写死 "meituan")
source_package: str # 源平台 Android 包名(前端 launch 用)
class HeartbeatResponse(BaseModel):
"""心跳响应。常规只回 ok;PoC 期带回 pending_compare 触发比价(exclude_none 省略 null)。"""
model_config = ConfigDict()
ok: bool = True
pending_compare: PendingCompare | None = None
class LivenessOut(BaseModel):
"""本机掉线告警状态(后置检测 pull 版)。客户端只需这一个布尔判断要不要弹「开启自启动」引导,
故只返回 kill_alert_pending(不暴露设备详情 / 内部 liveness_state )从未注册过 默认 False(无告警)"""
-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}
-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比价记录 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()
+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
+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
+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": {}})