Merge origin/main into feat/price-observation-pipeline
解决 app/core/config.py 冲突:同时保留 main 的 PRICEBOT_INSTANCES 多实例透传 (pricebot_instances 属性)与本分支的 INTERNAL_API_SECRET(价格观测内部端点密钥)。 .env.example 自动合并(两段并存)。main 未新增 alembic 迁移,price_observation_table 仍是唯一 head;app/main.py 未被 main 改动,内部路由注册不受影响。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
+18
-6
@@ -18,6 +18,7 @@ pricebot 协议文档:
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
@@ -25,6 +26,7 @@ import httpx
|
||||
from fastapi import APIRouter, HTTPException, Request, status
|
||||
|
||||
from app.core.config import settings
|
||||
from app.core.pricebot_router import pick_pricebot
|
||||
|
||||
logger = logging.getLogger("shagua.compare")
|
||||
|
||||
@@ -38,25 +40,35 @@ async def _passthrough(request: Request, upstream_path: str) -> dict[str, Any]:
|
||||
打日志。比价单帧是大上下文 / 逐帧 LLM,超时用 PRICEBOT_COMPARE_TIMEOUT_SEC(60s,
|
||||
比领券的 30s 长)。
|
||||
"""
|
||||
# 读原始字节,避免"反序列化→再序列化"的双重 JSON(省 ~一半透传 CPU,让 app-server
|
||||
# 单 worker 也扛得住高并发)。只 json.loads 一次拿 trace_id 做亲和 + 打日志,转发时
|
||||
# 直接发原始 bytes(content=raw),不重新 dumps。
|
||||
raw = await request.body()
|
||||
try:
|
||||
body = await request.json()
|
||||
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 = {}
|
||||
|
||||
url = f"{settings.PRICEBOT_BASE_URL.rstrip('/')}{upstream_path}"
|
||||
# 按 trace_id 一致性 hash 选 pricebot 实例(同一比价的所有帧落同一进程,内存维护 state)
|
||||
base = pick_pricebot(meta.get("trace_id"))
|
||||
url = f"{base.rstrip('/')}{upstream_path}"
|
||||
timeout = settings.PRICEBOT_COMPARE_TIMEOUT_SEC
|
||||
|
||||
logger.info(
|
||||
"compare %s device_id=%s trace_id=%s step=%s",
|
||||
upstream_path,
|
||||
body.get("device_id"),
|
||||
body.get("trace_id"),
|
||||
body.get("step"),
|
||||
meta.get("device_id"),
|
||||
meta.get("trace_id"),
|
||||
meta.get("step"),
|
||||
)
|
||||
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=timeout) as client:
|
||||
resp = await client.post(url, json=body)
|
||||
resp = await client.post(
|
||||
url, content=raw, headers={"Content-Type": "application/json"}
|
||||
)
|
||||
except httpx.RequestError as e:
|
||||
logger.error("[pricebot] request failed: %s", e)
|
||||
raise HTTPException(
|
||||
|
||||
+17
-6
@@ -10,6 +10,7 @@ pricebot 协议文档:
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
@@ -17,6 +18,7 @@ import httpx
|
||||
from fastapi import APIRouter, HTTPException, Request, status
|
||||
|
||||
from app.core.config import settings
|
||||
from app.core.pricebot_router import pick_pricebot
|
||||
|
||||
logger = logging.getLogger("shagua.coupon")
|
||||
|
||||
@@ -33,24 +35,33 @@ async def coupon_step(
|
||||
- 透传: 不做 schema 校验,pricebot 自己校验
|
||||
- 失败: 网络不可达 / pricebot 5xx → 502 + 友好 message
|
||||
"""
|
||||
# 读原始字节,避免"反序列化→再序列化"的双重 JSON(省透传 CPU)。只 loads 一次拿
|
||||
# trace_id 做亲和 + 打日志,转发时直接发原始 bytes,不重新 dumps。
|
||||
raw = await request.body()
|
||||
try:
|
||||
body = await request.json()
|
||||
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 = {}
|
||||
|
||||
url = f"{settings.PRICEBOT_BASE_URL.rstrip('/')}/api/coupon/step"
|
||||
# 按 trace_id 一致性 hash 选 pricebot 实例(同一领券任务的所有帧落同一进程)
|
||||
base = pick_pricebot(meta.get("trace_id"))
|
||||
url = f"{base.rstrip('/')}/api/coupon/step"
|
||||
timeout = settings.PRICEBOT_REQUEST_TIMEOUT_SEC
|
||||
|
||||
logger.info(
|
||||
"coupon_step device_id=%s trace_id=%s step=%s",
|
||||
body.get("device_id"),
|
||||
body.get("trace_id"),
|
||||
body.get("step"),
|
||||
meta.get("device_id"),
|
||||
meta.get("trace_id"),
|
||||
meta.get("step"),
|
||||
)
|
||||
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=timeout) as client:
|
||||
resp = await client.post(url, json=body)
|
||||
resp = await client.post(
|
||||
url, content=raw, headers={"Content-Type": "application/json"}
|
||||
)
|
||||
except httpx.RequestError as e:
|
||||
logger.error("[pricebot] request failed: %s", e)
|
||||
raise HTTPException(
|
||||
|
||||
Reference in New Issue
Block a user