271 lines
12 KiB
Python
271 lines
12 KiB
Python
"""外卖比价业务透传端点 + 后端 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)。发奖不在这里:#113 已把
|
|
邀请奖口径从「比价」移到「实际下单」(order.py)。
|
|
- /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
|
|
|
|
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},
|
|
)
|
|
# 不在此处发邀请奖:#113 已把发奖口径从「比价」移到「实际下单」(order.py),harvest
|
|
# 只记录比价、不发奖。否则比价先于下单 + try_reward 幂等闸会让奖落在「比价」这步,
|
|
# 架空 #113 的「下单才发奖」防刷意图(newly_success 仅留作日志观测)。
|
|
|
|
|
|
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/epilogue", summary="比价结果页尾声帧 (透传到 pricebot, 用户视角截图入 trace)")
|
|
async def trace_epilogue(request: Request, user: OptionalUser) -> dict[str, Any]:
|
|
# App 收到 done、渲染完结果页后,把自己页面的截图(base64)传给 pricebot 存进 trace
|
|
# 目录并触发重传 —— trace 里补上"用户实际看到的汇总页"(步骤帧只有目标 App 画面)。
|
|
# body ~几百 KB(截图 base64), 纯透传壳(harvest_first_frame=False:不建行/不落库,
|
|
# 该 trace 的记录已由 done/finalize 落终态)。#112 原走 _passthrough,合并到 harvest
|
|
# 分支后统一走 _forward(_passthrough 已并入它)。
|
|
resp, _, _ = await _forward(
|
|
request, "/api/trace/epilogue", user, harvest_first_frame=False,
|
|
)
|
|
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
|