shaguabijia-app-server/scripts/seed_mock_llm_cost.py

"""一次性 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()