Compare commits

..

3 Commits

Author SHA1 Message Date
marco e66f7fed8c feat(wx): 截图比价 PoC 后端 — 收图打信号 + 心跳下发 pending_compare
- wx_poc_signal 进程内存信号(device_id→source), 只对 WX_POC_TEST_DEVICE_ID 生效
- 收图 image → set_pending(测试设备, meituan)
- POST /device/heartbeat 返回 HeartbeatResponse, 该设备有信号则带回 pending_compare 并清除
- 心跳 schema OkResponse→HeartbeatResponse, 老前端不读 body 向后兼容
- WX_POC_TEST_DEVICE_ID 空=不触发任何设备(部署零风险, 待填测试机 id)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-16 01:30:41 +08:00
marco 5271d22d24 feat(wx): 微信服务号消息接收回调 /wx/mp/callback(截图比价入口第一步)
- GET 接入验证(校验 signature → 原样返回 echostr),供公众平台配置服务器 URL
- POST 收消息(安全模式:msg_signature 验签 + AES-256-CBC 解密),图片消息用 PicUrl 落盘
- config 加 WX_MP_TOKEN / WX_MP_AES_KEY + wx_mp_callback_configured
- 识别平台订单/pending标记/openid↔device 绑定留后续

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-16 00:48:13 +08:00
guke 5c6840dd71 feat(compare): 比价记录 LLM token 成本落库与展示(按当时价冻结) (#133)
- comparison_record 加 llm_cost_yuan(元/float)+ llm_price_snapshot(JSON)两列
- _backfill_llm_calls 回填时按 app_config 当时单价逐模型算成本、冻结成本+快照到记录
- app_config 新增 llm_token_price 配置(per_model + default 兜底,运营在系统配置页可改)
- services/llm_cost.py:compute_llm_cost 纯函数(按 model 分桶、error/无 usage 跳过、
  脏价格当 unpriced 不抛异常以免连累 token 回填)+ get_llm_prices reader
- admin schema 暴露成本:列表项带 llm_cost_yuan,详情另带价格快照
- tests/test_llm_cost.py(10 测试);scripts/seed_mock_llm_cost.py(mock seeder)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

---------

Co-authored-by: guke <guke@autohome.com.cn>
Reviewed-on: #133
2026-07-13 17:46:11 +08:00
16 changed files with 807 additions and 3 deletions
+33
View File
@@ -0,0 +1,33 @@
"""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")
+4
View File
@@ -36,6 +36,8 @@ 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
@@ -72,3 +74,5 @@ 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,6 +27,7 @@ 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")
@@ -81,6 +82,8 @@ 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",
+18 -3
View File
@@ -14,18 +14,24 @@ 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"])
@@ -52,12 +58,12 @@ def register_device(
return DeviceOut.model_validate(device)
@router.post("/heartbeat", response_model=OkResponse, summary="上报心跳")
@router.post("/heartbeat", response_model=HeartbeatResponse, summary="上报心跳")
def report_heartbeat(
req: HeartbeatRequest,
user: CurrentUser,
db: DbSession,
) -> OkResponse:
) -> HeartbeatResponse:
device_repo.touch_heartbeat(
db,
user_id=user.id,
@@ -65,7 +71,16 @@ def report_heartbeat(
accessibility_enabled=req.accessibility_enabled,
registration_id=req.registration_id,
)
return OkResponse()
# 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)
@router.get("/liveness", response_model=LivenessOut, summary="查询本机掉线告警(后置检测)")
+112
View File
@@ -0,0 +1,112 @@
"""微信服务号消息接收回调 /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)
+16
View File
@@ -136,6 +136,17 @@ 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 = ""
@property
def wx_mp_configured(self) -> bool:
"""服务号网页授权凭证齐全(缺则落地页不发起授权,降级为无 openid)。"""
@@ -146,6 +157,11 @@ 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,4 +96,19 @@ 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
@@ -0,0 +1,27 @@
"""微信截图比价 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)
+69
View File
@@ -0,0 +1,69 @@
"""微信服务号消息接收回调的验签与解密(安全模式)。
服务号"服务器配置"选安全模式后:
- 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,6 +39,7 @@ 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 (
@@ -140,6 +141,8 @@ 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,6 +137,12 @@ 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
+14
View File
@@ -36,6 +36,20 @@ 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
@@ -0,0 +1,53 @@
"""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,6 +40,8 @@
| `raw_payload` | JSON(PG: JSONB) | nullable | 客户端原始上报全量(calibration + done.params),取数兜底 |
| `input_tokens` | Integer | nullable | 本次 LLM 累计输入 token = Σ `llm_calls[].usage.prompt_tokens`(server 收上报后从 `llm_calls` 累加;旧记录/未采集为 null) |
| `output_tokens` | Integer | nullable | 本次 LLM 累计输出 token = Σ `llm_calls[].usage.completion_tokens`(同上) |
| `llm_cost_yuan` | Float | nullable | 本次比价 LLM 总成本(元),回填时按「当时价」逐模型算好冻结(见 `services/llm_cost.py`);旧记录/未回填为 null → 前端回退「估算成本」 |
| `llm_price_snapshot` | JSON(PG: JSONB) | nullable | 算成本所用单价快照 `{mode, prices:{model:{input_per_1m,output_per_1m,_source}}}`;`app_config` 只存当前价、不留历史,故冻结当时价供审计/复算 |
| `created_at` | DateTime(tz) | server_default now(), index | 时间 |
> `ordered`(已下单)是**瞬态字段**,不在表里:`list_records` 读取时按 `store_name ∈ 该用户 source='compare' 的 savings_record.shop_name 集合` 现挂到实例上供出参用。
+248
View File
@@ -0,0 +1,248 @@
"""一次性 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()
+184
View File
@@ -0,0 +1,184 @@
"""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