duobibi-app-mock/duobibi(mock)backend/duobibi-server/app/api/arena.py

"""POST /api/v1/arena-quote —— 多比比"比价擂台"接口。

用户输入一个商品名,后端用 LLM 估出该商品在主流电商各平台的「到手价」
以及「市场常见价」(中位)。客户端把估价铺成"多平台对比表",再用本机
真实记录覆盖对应平台,最低价高亮。

application/json:
  {"title": "iPhone 15 Pro 256GB", "platforms": ["淘宝","京东",...](可选)}

响应:
  {"title": "...(归一化)", "typical_price": 8299.0,
   "quotes": [{"platform":"淘宝","price":8099.0,"note":"..."}, ...],
   "lowest_platform": "拼多多"}

兜底:LLM 给了 typical 但缺平台报价 → 围绕 typical 确定性合成,保证表格每行有值;
完全估不出价格 → 422 llm_no_quote。
"""
from __future__ import annotations

import hashlib
import json
import logging
import re
from typing import Optional

from fastapi import APIRouter, HTTPException

from app.llm_client import MOCK_LLM, chat
from app.schemas import ArenaPlatformQuote, ArenaQuoteRequest, ArenaQuoteResponse

logger = logging.getLogger("app.arena")

router = APIRouter(prefix="/api/v1")

# 比价擂台默认对比 4 大主流电商(同一实体商品可跨平台买)。
# 美团 / 饿了么是本地生活 O2O,不进默认对比池,但客户端可显式传入 platforms 覆盖。
DEFAULT_PLATFORMS = ["淘宝", "京东", "拼多多", "抖音"]

SYSTEM_PROMPT = """你是一个电商多平台价格估算助手。

# 任务
用户给你一个商品标题和一组电商平台。请估算该商品在每个平台的「日常到手价」
(实付价,不含双 11 / 618 等大促),以及该商品的「市场常见价」(各平台中位值)。

# 估算原则
- 同一商品在不同平台价格通常有差异:拼多多 / 百亿补贴常偏低,京东自营偏高且稳,
  淘宝居中,抖音随直播波动。请给出**有合理区分度**的价格,不要每个平台都一样。
- 不熟悉的商品按品类 / 品牌 / 规格常识估算。完全陌生时给出符合常识的合理估值。
- 所有价格为正数(元),可带小数。

# 输出格式
严格 JSON,无 markdown,无解释:

{"title": "归一化后的商品标题", "typical_price": 8299.0,
 "quotes": [{"platform": "淘宝", "price": 8099.0, "note": "一句话(可空)"},
            {"platform": "京东", "price": 8299.0, "note": ""}]}

字段:
- title: 整理后的标题(纠错 / 统一规格写法,保留原意)
- typical_price: 市场常见价(正数,必填)
- quotes: 对**给定的每个平台**各给一条;price 正数必填,note 可空
"""


def _extract_json(s: str) -> Optional[dict]:
    s = s.strip()
    s = re.sub(r"^```(?:json)?\s*", "", s)
    s = re.sub(r"\s*```$", "", s)
    try:
        data = json.loads(s)
    except json.JSONDecodeError:
        m = re.search(r"\{.*\}", s, re.DOTALL)
        if not m:
            return None
        try:
            data = json.loads(m.group(0))
        except json.JSONDecodeError:
            return None
    return data if isinstance(data, dict) else None


def _num(v) -> Optional[float]:
    if isinstance(v, bool):
        return None
    if isinstance(v, (int, float)):
        return float(v)
    if isinstance(v, str):
        try:
            return float(v.strip())
        except ValueError:
            return None
    return None


def _synth_price(title: str, platform: str, typical: float) -> float:
    """给 LLM 没给报价的平台合成一个确定性估价(围绕 typical 的 0.93~1.06 倍)。"""
    seed = int(hashlib.sha256(f"{title}|{platform}".encode("utf-8")).hexdigest()[:6], 16)
    factor = 0.93 + (seed % 14) / 100.0
    return round(typical * factor, 2)


def _mock_raw(title: str) -> str:
    """占坑期 mock(DUOBIBI_MOCK_LLM):基于标题确定性造一个市场常见价,
    各平台报价交给下游 _synth_price 围绕 typical 合成。返回 LLM 同格式 JSON。"""
    seed = int(hashlib.sha256(title.encode("utf-8")).hexdigest()[:8], 16)
    typical = round(49 + seed % 9950 + (seed % 100) / 100.0, 2)
    return json.dumps({"title": title, "typical_price": typical}, ensure_ascii=False)


@router.post("/arena-quote", response_model=ArenaQuoteResponse)
def arena_quote(req: ArenaQuoteRequest) -> ArenaQuoteResponse:
    title_in = req.title.strip()
    if not title_in:
        raise HTTPException(status_code=422, detail="empty_title")
    platforms = [p.strip() for p in (req.platforms or DEFAULT_PLATFORMS) if p and p.strip()]
    if not platforms:
        platforms = list(DEFAULT_PLATFORMS)

    user_msg = f"商品标题: {title_in}\n平台: {', '.join(platforms)}"
    if MOCK_LLM:
        raw = _mock_raw(title_in)
    else:
        try:
            raw = chat(
                messages=[
                    {"role": "system", "content": SYSTEM_PROMPT},
                    {"role": "user", "content": user_msg},
                ]
            )
            logger.debug("arena-quote LLM raw: %s", raw[:400])
        except Exception as e:
            # LLM 不可用(限流/网络/超时等):降级为空响应,下方走 422 llm_no_quote 而非 500。
            logger.warning("arena-quote LLM call failed: %s", e)
            raw = ""

    data = _extract_json(raw) or {}
    title_out = data.get("title")
    title_out = title_out.strip() if isinstance(title_out, str) and title_out.strip() else title_in
    typical = _num(data.get("typical_price"))

    # 解析 quotes,仅保留请求的平台,price 必须 > 0
    by_platform: dict[str, ArenaPlatformQuote] = {}
    raw_quotes = data.get("quotes")
    if isinstance(raw_quotes, list):
        for q in raw_quotes:
            if not isinstance(q, dict):
                continue
            plat = q.get("platform")
            price = _num(q.get("price"))
            if not isinstance(plat, str) or plat.strip() not in platforms:
                continue
            if price is None or price <= 0:
                continue
            note = q.get("note")
            by_platform[plat.strip()] = ArenaPlatformQuote(
                platform=plat.strip(),
                price=round(price, 2),
                note=note.strip() if isinstance(note, str) else "",
            )

    # typical 兜底:用已解析报价的中位值
    if (typical is None or typical <= 0) and by_platform:
        prices = sorted(p.price for p in by_platform.values())
        typical = prices[len(prices) // 2]

    if typical is None or typical <= 0:
        logger.info("arena-quote no usable price for %r", title_in)
        raise HTTPException(status_code=422, detail="llm_no_quote")

    # 给缺失平台合成估价,保证表格每行都有值
    for plat in platforms:
        if plat not in by_platform:
            by_platform[plat] = ArenaPlatformQuote(
                platform=plat,
                price=_synth_price(title_out, plat, typical),
                note="估算",
            )

    quotes = [by_platform[p] for p in platforms]
    lowest = min(quotes, key=lambda q: q.price).platform if quotes else None

    logger.info(
        "arena-quote title=%r typical=%.2f platforms=%d lowest=%s",
        title_out, typical, len(quotes), lowest,
    )
    return ArenaQuoteResponse(
        title=title_out,
        typical_price=round(typical, 2),
        quotes=quotes,
        lowest_platform=lowest,
    )