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

"""外卖比价业务透传端点 + 后端 harvest 落库。

把客户端 POST /api/v1/intent/* 和 /api/v1/price/step 转发给 pricebot,同时**由 app-server
在透传里直接落库比价记录**(不再靠客户端 POST /compare/record):
  - 帧0(客户端首帧不带 trace_id)→ app-server 用 uuid 签发 trace_id、注入转发 body、回给
    客户端;并按 trace_id 建 running 行(harvest_running)。
  - 最终 done 帧 → 更新成 success/failed + 结果(harvest_done)+ **幂等发一次邀请奖**。
  - /trace/finalize(用户终止/Phase1 未识别,无 done)→ 更新成 cancelled/failed
    (harvest_abort,**不降级 success**)。
trace_url 从 pricebot 响应**顶层 trace_url** 取(pricebot 每帧都带,见其 goal_engine.process_step)。

鉴权:软鉴权(OptionalUser)——新客户端带 JWT → 绑 user_id;老客户端不带 → user_id 暂空,
由其后续 /compare/record 上报补(灰度期两条写路径按 trace_id reconcile,success 不被降级)。

真正的目标驱动比价逻辑在 pricebot-backend(GoalEngine,另一个 repo),本接口是透传壳 + 落库。
pricebot 协议文档: pricebot-backend/docs/main/02_api_protocol.md
"""
from __future__ import annotations

import json
import logging
import time
import uuid
from typing import Any

import httpx
from fastapi import APIRouter, HTTPException, Request, status
from fastapi.concurrency import run_in_threadpool

from app.api.deps import OptionalUser
from app.core.config import settings
from app.core.logging import trace_id_ctx
from app.core.pricebot_client import get_pricebot_client
from app.core.pricebot_router import pick_pricebot
from app.db.session import SessionLocal
from app.repositories import comparison as crud_compare
from app.repositories import invite as crud_invite

logger = logging.getLogger("shagua.compare")

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


# ============================================================
# harvest 落库(阻塞 SQLAlchemy → run_in_threadpool,独立 SessionLocal,不阻塞事件循环;
# 任何写库异常都吞掉、绝不连累比价透传返回 —— 同 coupon.py 现有 best-effort 写法)。
# ============================================================


def _harvest_running_blocking(
    trace_id: str, user_id: int | None, business_type: str,
    device_id: str | None, device_info: dict | None, trace_url: str | None,
) -> None:
    with SessionLocal() as db:
        crud_compare.harvest_running(
            db, trace_id=trace_id, user_id=user_id, business_type=business_type,
            device_id=device_id, device_info=device_info, trace_url=trace_url,
        )
    logger.info(
        "harvest running row (user=%s)", user_id,
        extra={"phase": "harvest_running", "status": "running", "user_id": user_id},
    )


def _harvest_done_blocking(
    trace_id: str, user_id: int | None, done_params: dict, business_type: str,
    device_id: str | None, device_info: dict | None, trace_url: str | None,
) -> None:
    with SessionLocal() as db:
        rec, newly_success = crud_compare.harvest_done(
            db, trace_id=trace_id, user_id=user_id, done_params=done_params,
            business_type=business_type, device_id=device_id,
            device_info=device_info, trace_url=trace_url,
        )
        logger.info(
            "harvest done → %s saved=%s newly=%s", rec.status,
            rec.saved_amount_cents, newly_success,
            extra={"phase": "harvest_done", "status": rec.status,
                   "saved_cents": rec.saved_amount_cents,
                   "best_platform": rec.best_platform_id, "newly_success": newly_success,
                   "user_id": user_id},
        )
        # 本次**新**落成 success + 有登录用户 → 发一次邀请奖(try_reward 自身幂等,双保险)。
        if newly_success and user_id is not None:
            try:
                crud_invite.try_reward_on_compare(db, user_id)
                logger.info(
                    "invite reward fired (user=%s)", user_id,
                    extra={"phase": "reward", "user_id": user_id},
                )
            except Exception as e:  # noqa: BLE001 发奖失败不连累比价
                logger.warning("harvest invite reward failed user=%s: %s", user_id, e)


def _harvest_abort_blocking(
    trace_id: str, status_hint: str, reason: str | None, trace_url: str | None,
) -> None:
    with SessionLocal() as db:
        rec = crud_compare.harvest_abort(
            db, trace_id=trace_id, status=status_hint, reason=reason, trace_url=trace_url,
        )
    logger.info(
        "harvest abort → %s", (rec.status if rec else "no-row"),
        extra={"phase": "harvest_abort",
               "status": (rec.status if rec else None), "reason": reason},
    )


async def _forward(
    request: Request, upstream_path: str, user, *, harvest_first_frame: bool = True,
) -> tuple[dict[str, Any], str, dict[str, Any]]:
    """透传给 pricebot 的 {upstream_path},并 mint trace_id + 首帧建 running 行。
    返回 (resp_json, trace_id, meta)。

    - 客户端首帧不带 trace_id → app-server 用 uuid 签发、写进转发 body、回填进响应顶层
      `trace_id`(客户端据此拿到后续帧都带上)。带了(老客户端 / 后续帧)→ 原样用、走原始 bytes 快路。
    - 仅**本次 mint(=首帧)**建 running 行(idempotent);后续帧不再写库。
    """
    raw = await request.body()
    try:
        meta = json.loads(raw)
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"invalid json body: {e}") from e
    if not isinstance(meta, dict):
        meta = {}

    trace_id = meta.get("trace_id")
    minted = False
    if not trace_id:
        trace_id = str(uuid.uuid4())
        meta["trace_id"] = trace_id
        raw = json.dumps(meta).encode()  # 仅首帧重新序列化(注入 trace_id);后续帧走原始 bytes
        minted = True

    # 请求级 trace_id 贯穿:此后本请求(含 run_in_threadpool 里的 harvest)每行日志自动带 trace_id
    trace_id_ctx.set(trace_id)

    base = pick_pricebot(trace_id)
    url = f"{base.rstrip('/')}{upstream_path}"
    timeout = settings.PRICEBOT_COMPARE_TIMEOUT_SEC
    step = meta.get("step")

    logger.info(
        "→ pricebot %s step=%s%s", upstream_path, step, " [mint]" if minted else "",
        extra={"phase": "forward", "endpoint": upstream_path, "step": step,
               "device_id": meta.get("device_id"), "minted": minted,
               "query": meta.get("query"), "user_id": (user.id if user else None)},
    )

    t0 = time.monotonic()
    try:
        client = get_pricebot_client()
        resp = await client.post(
            url, content=raw, headers={"Content-Type": "application/json"}, timeout=timeout
        )
    except httpx.RequestError as e:
        logger.error(
            "← pricebot %s 不可达: %s", upstream_path, e,
            extra={"phase": "resp", "endpoint": upstream_path, "step": step,
                   "error": "upstream_unreachable"},
        )
        raise HTTPException(
            status_code=status.HTTP_502_BAD_GATEWAY,
            detail=f"pricebot upstream unreachable: {e}",
        ) from e

    if resp.status_code >= 500:
        logger.error(
            "← pricebot %s 5xx=%d", upstream_path, resp.status_code,
            extra={"phase": "resp", "endpoint": upstream_path, "step": step,
                   "http_status": resp.status_code, "error": "upstream_5xx"},
        )
        raise HTTPException(
            status_code=status.HTTP_502_BAD_GATEWAY,
            detail=f"pricebot upstream returned {resp.status_code}",
        )

    resp_json = resp.json()
    cost_ms = int((time.monotonic() - t0) * 1000)
    if not isinstance(resp_json, dict):
        return {}, trace_id, meta
    # 回传 app-server 签发的 trace_id(新客户端首帧据此拿到,后续帧都带上它)
    resp_json.setdefault("trace_id", trace_id)
    _action = resp_json.get("action") or {}
    logger.info(
        "← pricebot %s cmd=%s cont=%s %dms", upstream_path,
        _action.get("command"), resp_json.get("continue"), cost_ms,
        extra={"phase": "resp", "endpoint": upstream_path, "step": step,
               "command": _action.get("command"), "continue": resp_json.get("continue"),
               "cost_ms": cost_ms},
    )

    # 首帧建 running 行(仅本次 mint;老客户端自带 trace_id → 不 mint → 由 done / 其 POST 建行)
    if minted and harvest_first_frame:
        try:
            await run_in_threadpool(
                _harvest_running_blocking, trace_id,
                (user.id if user else None), "food",
                meta.get("device_id"), meta.get("device_info"),
                resp_json.get("trace_url"),
            )
        except Exception as e:  # noqa: BLE001
            logger.warning("harvest_running failed trace=%s: %s", trace_id, e)

    return resp_json, trace_id, meta


@router.post("/intent/recognize", summary="外卖比价 Phase 1 意图识别 (透传 + 建 running 行)")
async def intent_recognize(request: Request, user: OptionalUser) -> dict[str, Any]:
    resp, _, _ = await _forward(request, "/api/intent/recognize", user)
    return resp


@router.post("/intent/step", summary="外卖比价 Phase 1 多帧意图识别 (透传, 仅淘宝源)")
async def intent_step(request: Request, user: OptionalUser) -> dict[str, Any]:
    resp, _, _ = await _forward(request, "/api/intent/step", user)
    return resp


@router.post("/intent/precoupon/step", summary="外卖比价 Phase 0 识别前先用券 (透传, 仅美团源)")
async def intent_precoupon_step(request: Request, user: OptionalUser) -> dict[str, Any]:
    resp, _, _ = await _forward(request, "/api/intent/precoupon/step", user)
    return resp


@router.post("/price/step", summary="外卖比价 Phase 2 步进 (透传 + done 落库 + 发奖)")
async def price_step(request: Request, user: OptionalUser) -> dict[str, Any]:
    resp, trace_id, meta = await _forward(request, "/api/price/step", user)
    # 最终 done 帧(command=done 且 continue=false)→ harvest 更新成终态 + 幂等发奖。
    # (单平台中途 done 被 pricebot 改写成 wait+continue=true,不会命中这里,同 coupon 语义。)
    action = resp.get("action") or {}
    if action.get("command") == "done" and not resp.get("continue", True):
        done_params = action.get("params") or {}
        try:
            await run_in_threadpool(
                _harvest_done_blocking, trace_id, (user.id if user else None),
                done_params, "food",
                meta.get("device_id"), meta.get("device_info"),
                resp.get("trace_url") or done_params.get("trace_url"),
            )
        except Exception as e:  # noqa: BLE001
            logger.warning("harvest_done failed trace=%s: %s", trace_id, e)
    return resp


@router.post("/trace/finalize", summary="比价 trace 收尾上云 (透传 + 夭折落库)")
async def trace_finalize(request: Request, user: OptionalUser) -> dict[str, Any]:
    # 用户终止 / Phase1 未识别没到 done 帧: pricebot 打包半截上云返回 {trace_url};
    # app-server 顺手把该 trace 的 running 行更新成夭折终态(**不降级 success**)。
    # 客户端 finalize body 带 status(cancelled/failed)+ reason;老客户端只带 trace_id →
    # 默认 cancelled,其后续 /compare/record 上报再补精确态。
    resp, trace_id, meta = await _forward(
        request, "/api/trace/finalize", user, harvest_first_frame=False,
    )
    try:
        await run_in_threadpool(
            _harvest_abort_blocking, trace_id,
            (meta.get("status") or "cancelled"),
            (meta.get("reason") or meta.get("information")),
            (resp.get("trace_url") if isinstance(resp, dict) else None),
        )
    except Exception as e:  # noqa: BLE001
        logger.warning("harvest_abort failed trace=%s: %s", trace_id, e)
    return resp