shaguabijia-app-server/app/api/v1/compare_record.py

"""比价记录 endpoint(「我的比价记录」+ admin 数据源)。

路由前缀 `/api/v1/compare`:
  POST /record         (灰度期兼容)老客户端上报,按 **trace_id** 幂等 + 不降级 success
  GET  /records        比价记录列表(游标分页)
  GET  /records/{id}   单条详情(含 raw_payload 全量)

**均需鉴权**(CurrentUser)。⚠️ 写路径现以 `compare.py` 透传壳的**后端 harvest** 为主
(帧0 建 running 行 → done/finalize 落终态,新客户端不再 POST);本 POST /record 仅灰度期
给老客户端用,与 harvest 按 trace_id reconcile。新版覆盖够高后可下线本 POST(阶段3)。
"""
from __future__ import annotations

import logging

from fastapi import APIRouter, BackgroundTasks, HTTPException, Query, status

from app.api.deps import CurrentUser, DbSession
from app.db.session import SessionLocal
from app.models.comparison import ComparisonRecord
from app.repositories import comparison as crud_compare
from app.schemas.compare_record import (
    CompareStatsOut,
    ComparisonRecordCreatedOut,
    ComparisonRecordDetailOut,
    ComparisonRecordIn,
    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")

router = APIRouter(prefix="/api/v1/compare", tags=["compare-record"])


@router.post(
    "/record",
    response_model=ComparisonRecordCreatedOut,
    summary="上报一次比价结果(幂等)",
)
def report_record(
    payload: ComparisonRecordIn,
    user: CurrentUser,
    db: DbSession,
    background_tasks: BackgroundTasks,
) -> ComparisonRecordCreatedOut:
    rec = crud_compare.upsert_record(db, user_id=user.id, payload=payload)
    # LLM 调用明细异步回填:同机拉 pricebot llm_calls 最长 5s 且是 best-effort,放后台
    # 任务做,不阻塞上报响应(顺带给 pricebot 落盘留足余量)。upsert 已 commit,后台用
    # 独立 session 按 record id 回填 llm_calls + 派生 llm_call_count/retry_count。
    background_tasks.add_task(_backfill_llm_calls, rec.id, rec.trace_id)
    # 注:邀请发奖已从"比价成功"挪到"实际下单"(见 api/v1/order.py report_order)——
    # 冰拍板:被邀请人完成比价并实际下单才算邀请成功,仅完成比价不再发奖。
    logger.info(
        "compare record user=%s trace=%s biz=%s status=%s saved=%s (llm_calls backfill queued)",
        user.id,
        rec.trace_id,
        rec.business_type,
        rec.status,
        rec.saved_amount_cents,
    )
    return ComparisonRecordCreatedOut(id=rec.id)


def _backfill_llm_calls(record_id: int, trace_id: str) -> None:
    """后台回填本次比价的 LLM 调用明细 + 派生 llm_call_count/retry_count。
    独立 DB session(请求 session 此时已关);拉取/写库失败只 log,绝不影响已落库的上报。"""
    calls = fetch_llm_calls(trace_id)
    if not calls:
        return
    db = SessionLocal()
    try:
        rec = db.get(ComparisonRecord, record_id)
        if rec is None:
            return
        rec.llm_calls = calls
        rec.llm_call_count = len(calls)
        rec.retry_count = sum(1 for c in calls if c.get("error"))
        # token 累加(usage 已被 pricebot llm_client 归一为 prompt/completion_tokens;
        # 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",
            trace_id, len(calls), rec.input_tokens, rec.output_tokens,
        )
    except Exception as e:  # noqa: BLE001 best-effort
        logger.warning("backfill llm_calls failed trace=%s: %s", trace_id, e)
    finally:
        db.close()


@router.get(
    "/stats",
    response_model=CompareStatsOut,
    summary="比价口径战绩(完成比价数 + 累计发现可省)",
)
def stats(user: CurrentUser, db: DbSession) -> CompareStatsOut:
    count, saved = crud_compare.get_stats(db, user.id)
    return CompareStatsOut(compare_count=count, discovered_saved_cents=saved)


@router.get(
    "/records",
    response_model=ComparisonRecordPage,
    summary="比价记录列表(游标分页)",
)
def list_records(
    user: CurrentUser,
    db: DbSession,
    limit: int = Query(20, ge=1, le=100),
    cursor: int | None = Query(None, description="上一页末条 id"),
    ordered: bool | None = Query(
        None,
        description="true=只看「已下单」(店名命中本人真实下单)的记录;不传=全部",
    ),
    keyword: str | None = Query(
        None,
        max_length=64,
        description="按店名 / 菜名模糊搜索,忽略大小写;空白串等同不传",
    ),
    include_trace: bool = Query(
        False,
        description="客户端开了本机 agent 调试模式时带 true,放行本人记录的 trace_url",
    ),
) -> ComparisonRecordPage:
    items, next_cursor = crud_compare.list_records(
        db, user.id, limit=limit, cursor=cursor, ordered=ordered, keyword=keyword
    )
    outs = [ComparisonRecordOut.model_validate(it) for it in items]
    # 权限闸:未开 debug_trace_enabled 的用户不下发 trace_url(列表页「复制调试链接」靠它)。
    # include_trace=true 例外:客户端开了本机 agent 调试模式时带上,放行**本人记录**的 trace_url
    # ——list_records 只查 user.id 自己的记录,给本人看自己的调试链接无越权,与实时结果页
    # CompResultScreen「debug 权限 OR 本机 agent 调试」同口径(领导 2026-06-12 拍板)。
    if not (user.debug_trace_enabled or include_trace):
        for o in outs:
            o.trace_url = None
    return ComparisonRecordPage(items=outs, next_cursor=next_cursor)


@router.get(
    "/records/{record_id}",
    response_model=ComparisonRecordDetailOut,
    summary="比价记录详情(含 raw_payload)",
)
def get_record(
    record_id: int, user: CurrentUser, db: DbSession
) -> ComparisonRecordDetailOut:
    rec = crud_compare.get_record(db, user.id, record_id)
    if rec is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="record not found")
    out = ComparisonRecordDetailOut.model_validate(rec)
    # 权限闸:未开 debug_trace_enabled 的用户不下发 trace_url。
    # ⚠️ raw_payload 是上报体全量(model_dump),里面也藏着一份 trace_url,必须一并抹掉——
    # 否则无权限用户从详情接口的 raw_payload 绕过权限闸拿到 trace_url。
    if not user.debug_trace_enabled:
        out.trace_url = None
        if isinstance(out.raw_payload, dict):
            out.raw_payload.pop("trace_url", None)
    return out