diff --git a/alembic/versions/comparison_record_trace_unique.py b/alembic/versions/comparison_record_trace_unique.py new file mode 100644 index 0000000..a662991 --- /dev/null +++ b/alembic/versions/comparison_record_trace_unique.py @@ -0,0 +1,49 @@ +"""comparison_record: 唯一键 (user_id,trace_id) → trace_id 单列 + user_id 可空 + +Revision ID: comparison_record_trace_unique +Revises: feedback_type_reply +Create Date: 2026-07-03 12:00:00.000000 + +比价记录改「后端 harvest」：app-server 在帧0(pricebot 出 trace_id)即建行，随 done/finalize +逐步补全，客户端不再 POST 记录。故本表要两处调整： + 1. user_id 改**可空**——harvest 建行那刻(软鉴权 / 老客户端匿名)可能还没有 user_id。 + 2. 唯一键 (user_id, trace_id) → **trace_id 单列**——trace_id 由 app-server 签发、全局唯一， + 一次比价一行；harvest 建行时 user_id 还没有，不能再用复合键去重。 + +⚠️ 上线前(QA)务必确认 comparison_record 无重复 trace_id：旧复合唯一键**允许**同 trace_id + 跨不同 user_id（实际上客户端 UUID 从不重复，但约束没拦），若真有重复,建 trace 单列唯一会失败。 + 查: SELECT trace_id, COUNT(*) c FROM comparison_record GROUP BY trace_id HAVING c > 1; + +SQLite 不支持直接 drop/alter 约束/列，用 batch_alter_table（建临时表+拷数据+换名）， +与 coupon_engage_per_package / store_mapping_* 同款；Postgres 直接执行原生 ALTER。 +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = 'comparison_record_trace_unique' +down_revision: Union[str, Sequence[str], None] = 'feedback_type_reply' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + with op.batch_alter_table('comparison_record', schema=None) as batch_op: + # harvest 帧0 建行时 user_id 可能暂缺 → 放开非空。 + batch_op.alter_column('user_id', existing_type=sa.Integer(), nullable=True) + # 复合唯一 → trace_id 单列唯一。 + batch_op.drop_constraint('uq_comparison_user_trace', type_='unique') + batch_op.create_unique_constraint('uq_comparison_trace', ['trace_id']) + + +def downgrade() -> None: + with op.batch_alter_table('comparison_record', schema=None) as batch_op: + batch_op.drop_constraint('uq_comparison_trace', type_='unique') + batch_op.create_unique_constraint( + 'uq_comparison_user_trace', ['user_id', 'trace_id'] + ) + # 回退非空前提是当时无 null user_id 行(harvest 期可能有孤儿行,回滚需先清理)。 + batch_op.alter_column('user_id', existing_type=sa.Integer(), nullable=False) diff --git a/app/api/deps.py b/app/api/deps.py index 20a663c..0b9db63 100644 --- a/app/api/deps.py +++ b/app/api/deps.py @@ -54,5 +54,29 @@ def get_current_user( return user +def get_current_user_optional( + credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(_bearer)], + db: Annotated[Session, Depends(get_db)], +) -> User | None: + """软鉴权:有合法 Bearer 就返回 user,否则(无 header / 无效 token / 用户禁用)一律返回 + None,**不 raise**。 + + 比价 step / finalize 灰度期用:新客户端带 JWT → 拿到 user_id 绑定 harvest 行; + 老客户端(不带 JWT)→ None,harvest 行 user_id 暂空,由其后续 /compare/record 上报补齐。 + 等新版覆盖率够高,再把这几条端点从软鉴权收紧成硬 get_current_user。 + """ + if credentials is None or credentials.scheme.lower() != "bearer": + return None + try: + payload = decode_token(credentials.credentials, expected_type="access") + user = db.get(User, int(payload["sub"])) + except (TokenError, KeyError, ValueError, TypeError): + return None + if user is None or user.status != "active": + return None + return user + + CurrentUser = Annotated[User, Depends(get_current_user)] +OptionalUser = Annotated[User | None, Depends(get_current_user_optional)] DbSession = Annotated[Session, Depends(get_db)] diff --git a/app/api/v1/compare.py b/app/api/v1/compare.py index b29ca8b..b1f56ae 100644 --- a/app/api/v1/compare.py +++ b/app/api/v1/compare.py @@ -1,49 +1,114 @@ -"""外卖比价业务透传端点。 +"""外卖比价业务透传端点 + 后端 harvest 落库。 -把客户端的 POST /api/v1/intent/recognize 和 /api/v1/price/step 请求原样转发给 -pricebot-backend 的 /api/intent/recognize 和 /api/price/step。MVP 阶段**不鉴权** -(同 coupon/step,见 docs/待办与技术债.md P1:device_id 透传区分设备,待补 JWT + -device_id↔user_id 绑定后才能做用户级画像)。 +把客户端 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)。 -- Phase 1 /intent/recognize:从源平台(淘宝闪购/美团/京东外卖)购物车页识别店名+菜品+ - 价格,一次性。 -- Phase 2 /price/step:多轮循环,在目标平台复现订单读到手价,直到 done。 +鉴权:软鉴权(OptionalUser)——新客户端带 JWT → 绑 user_id;老客户端不带 → user_id 暂空, +由其后续 /compare/record 上报补(灰度期两条写路径按 trace_id reconcile,success 不被降级)。 -真正的目标驱动比价逻辑在 pricebot-backend(GoalEngine,另一个 repo),本接口只是透传壳。 -电商(ecom)那两个端点 food MVP 暂不需要,以后放开 scene 时再加 /ecom/intent/recognize -和 /ecom/step 两行即可。 - -pricebot 协议文档: - pricebot-backend/docs/main/02_api_protocol.md +真正的目标驱动比价逻辑在 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"]) -async def _passthrough(request: Request, upstream_path: str) -> dict[str, Any]: - """把请求体原样转发给 pricebot-backend 的 {upstream_path}。 +# ============================================================ +# harvest 落库(阻塞 SQLAlchemy → run_in_threadpool,独立 SessionLocal,不阻塞事件循环; +# 任何写库异常都吞掉、绝不连累比价透传返回 —— 同 coupon.py 现有 best-effort 写法)。 +# ============================================================ - 跟 coupon_step 同款透传壳:不鉴权、不做 schema 校验,仅读 device_id/trace_id/step - 打日志。比价单帧是大上下文 / 逐帧 LLM,超时用 PRICEBOT_COMPARE_TIMEOUT_SEC(60s, - 比领券的 30s 长)。 + +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);后续帧不再写库。 """ - # 读原始字节,避免"反序列化→再序列化"的双重 JSON(省 ~一半透传 CPU,让 app-server - # 单 worker 也扛得住高并发)。只 json.loads 一次拿 trace_id 做亲和 + 打日志,转发时 - # 直接发原始 bytes(content=raw),不重新 dumps。 raw = await request.body() try: meta = json.loads(raw) @@ -52,26 +117,41 @@ async def _passthrough(request: Request, upstream_path: str) -> dict[str, Any]: if not isinstance(meta, dict): meta = {} - # 按 trace_id 一致性 hash 选 pricebot 实例(同一比价的所有帧落同一进程,内存维护 state) - base = pick_pricebot(meta.get("trace_id")) + 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( - "compare %s device_id=%s trace_id=%s step=%s", - upstream_path, - meta.get("device_id"), - meta.get("trace_id"), - meta.get("step"), + "→ 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] request failed: %s", 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}", @@ -79,57 +159,112 @@ async def _passthrough(request: Request, upstream_path: str) -> dict[str, Any]: if resp.status_code >= 500: logger.error( - "[pricebot] 5xx status=%d body=%s", - resp.status_code, - resp.text[:500], + "← 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}", ) - return resp.json() + 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 意图识别 (透传到 pricebot)") -async def intent_recognize(request: Request) -> dict[str, Any]: - return await _passthrough(request, "/api/intent/recognize") +@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 多帧意图识别 (透传到 pricebot, 仅淘宝源)") -async def intent_step(request: Request) -> dict[str, Any]: - # 多帧版意图识别(展开+滚动采集→提取): 循环调用直到 done(done 帧顶层带 - # result+calibration)。目前仅淘宝源走这条, 其它源走上面单次 /intent/recognize。 - return await _passthrough(request, "/api/intent/step") +@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 意图识别前先用券 (透传到 pricebot, 仅美团源)", -) -async def intent_precoupon_step(request: Request) -> dict[str, Any]: - # 美团源平台『意图识别前先用券』多帧循环: 客户端在调 /intent/recognize 之前先循环 - # 调本端点到 done(continue=false)。订单页底部有『点击使用X红包』就自动选最大免费券 - # 用上, 已用券/无券则首帧秒过。与 /intent/step 同属 intent 域, 复用同一透传壳。 - return await _passthrough(request, "/api/intent/precoupon/step") +@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 步进 (透传到 pricebot)") -async def price_step(request: Request) -> dict[str, Any]: - return await _passthrough(request, "/api/price/step") +@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) -> dict[str, Any]: +async def trace_epilogue(request: Request, user: OptionalUser) -> dict[str, Any]: # App 收到 done、渲染完结果页后,把自己页面的截图(base64)传给 pricebot 存进 trace # 目录并触发重传 —— trace 里补上"用户实际看到的汇总页"(步骤帧只有目标 App 画面)。 - # body ~几百 KB(截图 base64), 透传壳同 finalize(带 trace_id 亲和)。 - return await _passthrough(request, "/api/trace/epilogue") + # 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 收尾上云 (透传到 pricebot, 终止/未识别拿 trace_url)") -async def trace_finalize(request: Request) -> dict[str, Any]: - # 用户终止 / Phase1 未识别没走到 done 帧, pricebot 没上云也没回传 trace_url。客户端收尾时 - # 打这个, _passthrough 按 trace_id 一致性 hash 落到处理这条 trace 的同一 pricebot 进程 - # (dir_cache 在那, 才能算对 trace 目录), 由后者打包上云返回 {trace_url}。 - return await _passthrough(request, "/api/trace/finalize") +@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 diff --git a/app/api/v1/compare_record.py b/app/api/v1/compare_record.py index 3542030..2677dad 100644 --- a/app/api/v1/compare_record.py +++ b/app/api/v1/compare_record.py @@ -1,15 +1,13 @@ -"""比价记录 endpoint(「我的比价记录」数据源)。 +"""比价记录 endpoint(「我的比价记录」+ admin 数据源)。 路由前缀 `/api/v1/compare`: - POST /record 上报一次比价结果(幂等:同 user+trace_id 覆盖) + POST /record (灰度期兼容)老客户端上报,按 **trace_id** 幂等 + 不降级 success GET /records 比价记录列表(游标分页) GET /records/{id} 单条详情(含 raw_payload 全量) -**均需鉴权**(CurrentUser)——与同文件无关的不鉴权透传 `compare.py` 分开:那个是 -转发壳(MVP 不鉴权),这里是按用户维度落库的业务接口,必须有 user_id。 - -注:本轮只做 server 端,客户端(android 仓)在 done 帧后调 POST /record 上报的改动 -另起一轮(见 app-server docs/待办与技术债.md P1)。 +**均需鉴权**(CurrentUser)。⚠️ 写路径现以 `compare.py` 透传壳的**后端 harvest** 为主 +(帧0 建 running 行 → done/finalize 落终态,新客户端不再 POST);本 POST /record 仅灰度期 +给老客户端用,与 harvest 按 trace_id reconcile。新版覆盖够高后可下线本 POST(阶段3)。 """ from __future__ import annotations diff --git a/app/api/v1/user.py b/app/api/v1/user.py index dc23388..f310b2b 100644 --- a/app/api/v1/user.py +++ b/app/api/v1/user.py @@ -69,6 +69,18 @@ def complete_onboarding( return OkResponse() +@router.post("/onboarding/reset", response_model=OkResponse, summary="重置新手引导(删该 设备+账号 完成标记,下次登录重走)") +def reset_onboarding( + req: OnboardingCompleteRequest, user: CurrentUser, db: DbSession +) -> OkResponse: + """删该 (当前账号, device_id) 的引导完成标记 → 下次登录 onboarding_completed=false,客户端重走。 + 与 /onboarding/complete 互逆。device_id 取客户端硬件级 ANDROID_ID(与登录/complete 一致)。 + 幂等:无记录也返回 ok。仅删自己(当前 JWT 用户)这台设备的记录,不影响别的账号/设备。""" + deleted = onboarding_repo.delete_completion(db, user_id=user.id, device_id=req.device_id) + logger.info("onboarding reset user_id=%d device_len=%d deleted=%d", user.id, len(req.device_id), deleted) + return OkResponse() + + @router.get( "/onboarding/status", response_model=OnboardingStatusResponse, diff --git a/app/core/logging.py b/app/core/logging.py index 3e63ea5..cb91b5c 100644 --- a/app/core/logging.py +++ b/app/core/logging.py @@ -2,9 +2,15 @@ 业务代码用 `logger = logging.getLogger("shagua.xxx")` 即可, 本模块在 main.py 启动时调一次。 -- 控制台(stdout): 人类可读文本, 给 systemd / 本地看。 -- 文件 `logs/app-server.log`: 单行 JSON, 供阿里云 SLS/Logtail 采集(JSON 模式零正则); - 异常栈作为字段内嵌不换行 → 每条日志一行。 +- 控制台(stdout): 人类可读文本, 给 systemd / 本地看; 有 trace 时行尾附 `trace=xxx`。 +- 文件 `logs/app-server.log`: 单行 JSON, 供**阿里云 SLS / Logtail** 采集(JSON 模式零正则); + 异常栈内嵌为字段不换行 → 每条日志一行。 +- **结构化字段(SLS 可直接查/聚合)**: + - `trace_id`: 请求级贯穿——在入口 `trace_id_ctx.set(...)` 后, 本请求内**每一行日志**(含 + run_in_threadpool 里的 harvest, contextvars 自动拷进线程)都自动带上, 无需手写。 + SLS 里 `trace_id: "xxx"` 一查即得整条比价链路, 按 time 升序即请求顺序。 + - 任意 `logger.info(msg, extra={"phase": ..., "step": ..., "command": ...})` 的 extra + 键都会平铺进 JSON 顶层 → SLS 可按 phase/step/command/cost_ms 等过滤聚合。 - 环境变量: - LOG_JSON_CONSOLE=1 控制台也输出 JSON - LOG_DIR / LOG_FILE 改落盘路径(默认 logs/app-server.log) @@ -17,13 +23,36 @@ import json import logging import os import sys +from contextvars import ContextVar from datetime import datetime from logging.handlers import RotatingFileHandler from pathlib import Path +# 请求级 trace_id:入口(如 compare.py 透传壳)set 之后, 本请求上下文(含 run_in_threadpool +# 拷贝出去的线程)内所有日志自动带上。默认空串 = 非请求上下文(启动/后台 worker)。 +trace_id_ctx: ContextVar[str] = ContextVar("trace_id", default="") + + +# 标准 LogRecord 属性 + 格式化期附加项:凡不在此集合的 record 属性都视为业务 extra, 平铺进 JSON。 +_RESERVED = set( + logging.LogRecord("", 0, "", 0, "", (), None).__dict__ +) | {"message", "asctime", "trace_id", "taskName"} + + +class _ContextFilter(logging.Filter): + """把 trace_id_ctx 注入每条 record(供两个 formatter 取用)。挂在 handler 上, + 命中每条(含 propagate 上来的)记录, 在 format 之前置好 record.trace_id。""" + + def filter(self, record: logging.LogRecord) -> bool: + if not hasattr(record, "trace_id"): + record.trace_id = trace_id_ctx.get() + return True + + class JsonFormatter(logging.Formatter): - """把 LogRecord 序列化成单行 JSON(SLS/Logtail 友好)。异常栈内嵌为字段, 整条仍是一行。""" + """LogRecord → 单行 JSON(SLS/Logtail 友好)。trace_id 提到顶层、extra 键平铺, 异常栈内嵌。""" + def __init__(self, service: str = "app-server"): super().__init__() self.service = service @@ -35,10 +64,17 @@ class JsonFormatter(logging.Formatter): "level": record.levelname, "service": self.service, "logger": record.name, - "func": record.funcName, - "line": record.lineno, - "message": record.getMessage(), } + tid = getattr(record, "trace_id", "") or trace_id_ctx.get() + if tid: + data["trace_id"] = tid + # 业务 extra 字段(phase / step / endpoint / command / cost_ms / status ...)平铺进顶层 + for k, v in record.__dict__.items(): + if k not in _RESERVED and not k.startswith("_"): + data[k] = v + data["func"] = record.funcName + data["line"] = record.lineno + data["message"] = record.getMessage() if record.exc_info: data["exception"] = self.formatException(record.exc_info) if record.stack_info: @@ -46,6 +82,15 @@ class JsonFormatter(logging.Formatter): return json.dumps(data, ensure_ascii=False, default=str) +class TextFormatter(logging.Formatter): + """控制台文本:标准行 + 有 trace_id 时行尾附 `trace=xxx`(无 trace 的启动/后台日志不加噪)。""" + + def format(self, record: logging.LogRecord) -> str: + base = super().format(record) + tid = getattr(record, "trace_id", "") or trace_id_ctx.get() + return f"{base} trace={tid}" if tid else base + + _CONFIGURED = False @@ -63,14 +108,17 @@ def setup_logging(debug: bool = False) -> None: for h in list(root.handlers): root.removeHandler(h) + ctx_filter = _ContextFilter() + # 控制台: 默认文本(systemd/本地看); LOG_JSON_CONSOLE=1 时输出 JSON console = logging.StreamHandler(sys.stdout) if os.getenv("LOG_JSON_CONSOLE") == "1": console.setFormatter(JsonFormatter(service)) else: console.setFormatter( - logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s") + TextFormatter("%(asctime)s %(levelname)s %(name)s: %(message)s") ) + console.addFilter(ctx_filter) root.addHandler(console) # 文件: 单行 JSON, 供 Logtail 采集(自动轮转, 单文件 10MB, 保留 5 个) @@ -82,6 +130,7 @@ def setup_logging(debug: bool = False) -> None: log_file, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8", ) file_handler.setFormatter(JsonFormatter(service)) + file_handler.addFilter(ctx_filter) root.addHandler(file_handler) # 第三方库降噪 diff --git a/app/models/comparison.py b/app/models/comparison.py index a8b24d9..ea647b5 100644 --- a/app/models/comparison.py +++ b/app/models/comparison.py @@ -39,16 +39,19 @@ _JSON = JSON().with_variant(JSONB(), "postgresql") class ComparisonRecord(Base): __tablename__ = "comparison_record" __table_args__ = ( - # 同一用户同一次比价(trace_id)只存一条:客户端重试/误点重复上报时幂等覆盖。 - UniqueConstraint("user_id", "trace_id", name="uq_comparison_user_trace"), + # trace_id 由 app-server 签发、全局唯一 → 一次比价一行,后端 harvest 按它 upsert。 + # (原 (user_id,trace_id) 复合唯一改为 trace_id 单列:harvest 帧0 建行时 user_id 可能暂缺。) + UniqueConstraint("trace_id", name="uq_comparison_trace"), # 首页轮播 / 省钱战绩聚合都按 status='success' 过滤 + created_at 近期排序; # 复合索引避免随数据量增大退化成全表扫(单列 created_at 索引不含 status)。 Index("ix_comparison_status_created", "status", "created_at"), ) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) - user_id: Mapped[int] = mapped_column( - Integer, ForeignKey("user.id"), index=True, nullable=False + # 后端 harvest 在帧0(pricebot 出 trace_id)即建行,软鉴权下 user_id 可能暂缺(老客户端/匿名)→ 可空。 + # C 端「我的比价记录」按 user_id 过滤天然排除 null-user 行;admin 全看(含孤儿行)。 + user_id: Mapped[int | None] = mapped_column( + Integer, ForeignKey("user.id"), index=True, nullable=True ) # 仍记录设备号(同一用户多设备的行为区分 / 与不鉴权期 device_id 数据对账) device_id: Mapped[str | None] = mapped_column(String(64), nullable=True) diff --git a/app/repositories/comparison.py b/app/repositories/comparison.py index d675822..1ad0bb1 100644 --- a/app/repositories/comparison.py +++ b/app/repositories/comparison.py @@ -74,7 +74,12 @@ def _derive(payload: ComparisonRecordIn) -> dict: def upsert_record( db: Session, *, user_id: int, payload: ComparisonRecordIn ) -> ComparisonRecord: - """按 (user_id, trace_id) 幂等写入:已存在则覆盖(更完整的重试上报胜出),否则新建。""" + """按 **trace_id** 幂等写入(唯一键已从 user_id+trace_id 改为 trace_id):已存在则合并 + (回填 null user_id + 覆盖字段,但**不降级 success**),否则新建。 + + 灰度期老客户端 POST /compare/record 走这条,与后端 harvest 按 trace_id reconcile; + 新客户端不再 POST(改由 compare.py 透传壳 harvest 落库)。 + """ derived = _derive(payload) fields = dict( device_id=payload.device_id, @@ -110,14 +115,29 @@ def upsert_record( **derived, ) + # 按 trace_id 定位(唯一键已改 trace_id)：后端 harvest 可能已建行,这里的客户端上报 + # (灰度期老客户端 / 新客户端不再走这条)与之 reconcile。 existing = db.execute( select(ComparisonRecord).where( - ComparisonRecord.user_id == user_id, ComparisonRecord.trace_id == payload.trace_id, ) ).scalar_one_or_none() if existing is not None: + # 不降级:harvest 或更早上报已落成 success,收尾期 fromFailure 的 cancelled/failed + # 不许把它盖回去(老 comparisonReported bug 的服务端兜底)——只补 user_id / trace_url。 + if existing.status == "success" and fields.get("status") != "success": + if existing.user_id is None and user_id is not None: + existing.user_id = user_id + if fields.get("trace_url"): + existing.trace_url = fields["trace_url"] + db.commit() + db.refresh(existing) + return existing + # 只**填**空缺 user_id、不 reassign:trace_id 全局唯一,一条 trace 只属一个用户; + # 绝不把已有归属的记录改判给另一个上报者(仅 harvest 建的 null-user 行在此绑上)。 + if existing.user_id is None: + existing.user_id = user_id for k, v in fields.items(): setattr(existing, k, v) db.commit() @@ -139,6 +159,201 @@ def upsert_record( return rec +# ============================================================ +# 后端 harvest：app-server 从 pricebot 透传响应里直接落库(不靠客户端上报)。 +# 帧0 建行(running) → 最终 done 更新(success/failed) → finalize 更新(aborted)。 +# 全按 trace_id upsert；success 行永不被后到的 failed/aborted 降级。 +# ============================================================ + + +def _derive_from_results(results: list[dict]) -> dict: + """从 done 帧 comparison_results(pricebot 原始 dict 列表)派生结构化列。 + 等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。""" + priced = [r for r in results if r.get("price") is not None] + best = None + if priced: + best = min( + priced, + key=lambda r: (r.get("rank") if r.get("rank") is not None else 10**9, r["price"]), + ) + src_row = next((r for r in results if r.get("is_source")), None) + + source_price_cents = None + if src_row is not None and src_row.get("price") is not None: + source_price_cents = _yuan_to_cents(src_row["price"]) + best_price_cents = _yuan_to_cents(best["price"]) if best else None + saved_amount_cents = None + if source_price_cents is not None and best_price_cents is not None: + saved_amount_cents = source_price_cents - best_price_cents + + has_valid_target = any( + (not r.get("is_source")) and r.get("price") is not None for r in results + ) + return { + "source_platform_id": (src_row or {}).get("platform_id"), + "source_platform_name": (src_row or {}).get("platform_name"), + "source_package": (src_row or {}).get("package"), + "source_price_cents": source_price_cents, + "best_platform_id": best.get("platform_id") if best else None, + "best_platform_name": best.get("platform_name") if best else None, + "best_price_cents": best_price_cents, + "saved_amount_cents": saved_amount_cents, + "is_source_best": best.get("is_source") if best else None, + "store_name": (src_row or {}).get("store_name") or None, + "status": "success" if has_valid_target else "failed", + } + + +def _device_cols_from_info(device_info: dict | None) -> dict: + """step 帧 device_info({locale,brand,model,android_version,rom_version})→ 表列。 + step 帧只带这几项;rom_name / android_sdk / app_version / 耗时步数 harvest 拿不到 → 留 None + (灰度期老客户端 fromComparison 会补齐;要新客户端也全带需扩 collectDeviceInfo,后续)。""" + d = device_info or {} + rv = str(d.get("rom_version") or "") + return { + "device_model": d.get("model") or None, + "device_manufacturer": d.get("brand") or None, + "android_version": d.get("android_version") or None, + "rom_version": int(rv) if rv.isdigit() else None, + } + + +def _get_by_trace(db: Session, trace_id: str) -> ComparisonRecord | None: + return db.execute( + select(ComparisonRecord).where(ComparisonRecord.trace_id == trace_id) + ).scalar_one_or_none() + + +def harvest_running( + db: Session, + *, + trace_id: str, + user_id: int | None, + business_type: str = "food", + device_id: str | None = None, + device_info: dict | None = None, + trace_url: str | None = None, +) -> ComparisonRecord: + """帧0(或任一尚未建行的帧)建/补 running 行。幂等:已存在只补空缺(user_id/trace_url/ + device_id/机型),绝不动已落定的 status / 结果。""" + rec = _get_by_trace(db, trace_id) + if rec is None: + rec = ComparisonRecord( + trace_id=trace_id, + user_id=user_id, + business_type=business_type or "food", + device_id=device_id, + status="running", + trace_url=trace_url, + created_at=datetime.now(CN_TZ).replace(tzinfo=None), + **_device_cols_from_info(device_info), + ) + db.add(rec) + db.commit() + db.refresh(rec) + return rec + changed = False + if rec.user_id is None and user_id is not None: + rec.user_id = user_id + changed = True + if not rec.trace_url and trace_url: + rec.trace_url = trace_url + changed = True + if rec.device_id is None and device_id: + rec.device_id = device_id + changed = True + for k, v in _device_cols_from_info(device_info).items(): + if getattr(rec, k) is None and v is not None: + setattr(rec, k, v) + changed = True + if changed: + db.commit() + db.refresh(rec) + return rec + + +def harvest_done( + db: Session, + *, + trace_id: str, + user_id: int | None, + done_params: dict, + business_type: str = "food", + device_id: str | None = None, + device_info: dict | None = None, + trace_url: str | None = None, +) -> tuple[ComparisonRecord, bool]: + """最终 done 帧:running 行 → 终态(success/failed)+结果+trace_url。 + 返回 (记录, 是否本次**新**落成 success)——供调用方据此幂等发一次邀请奖。 + 行不存在(理论上帧0已建;防御)则新建。""" + results = done_params.get("comparison_results") or [] + derived = _derive_from_results(results) + fields = dict( + business_type=business_type or "food", + information=done_params.get("information") or None, + # best_deeplink 来自客户端剪贴板采集,harvest 拿不到 → 留空(灰度期 fromComparison 会补; + # 纯 harvest 行「再次比价」退化为按 package 拉起 App。要精确深链需客户端另传,后续)。 + trace_url=trace_url or done_params.get("trace_url"), + total_dish_count=done_params.get("total_dish_count"), + skipped_dish_count=done_params.get("skipped_dish_count"), + skipped_dish_names=list(done_params.get("skipped_dish_names") or []), + comparison_results=results, + # 菜品:pricebot 已把源单菜品塞进 comparison_results[源行].items + items=next((r.get("items") or [] for r in results if r.get("is_source")), []), + raw_payload=done_params, + **derived, + **_device_cols_from_info(device_info), + ) + + rec = _get_by_trace(db, trace_id) + was_success = rec is not None and rec.status == "success" + if rec is None: + rec = ComparisonRecord( + trace_id=trace_id, + user_id=user_id, + created_at=datetime.now(CN_TZ).replace(tzinfo=None), + device_id=device_id, + **fields, + ) + db.add(rec) + else: + if user_id is not None and rec.user_id is None: + rec.user_id = user_id + if device_id and rec.device_id is None: + rec.device_id = device_id + for k, v in fields.items(): + setattr(rec, k, v) + db.commit() + db.refresh(rec) + newly_success = (rec.status == "success") and not was_success + return rec, newly_success + + +def harvest_abort( + db: Session, + *, + trace_id: str, + status: str, + reason: str | None, + trace_url: str | None = None, +) -> ComparisonRecord | None: + """finalize(用户终止/超时/异常,无 done 帧):running 行 → aborted/failed + trace_url。 + **不降级 success**:行已 success(收尾取消那种 finalize 后到)只 refresh trace_url。 + 行不存在(极少:帧0没建成)→ 返回 None,不凭空造。""" + rec = _get_by_trace(db, trace_id) + if rec is None: + return None + if trace_url and not rec.trace_url: + rec.trace_url = trace_url + if rec.status != "success": + rec.status = status or "cancelled" + if reason: + rec.information = reason + db.commit() + db.refresh(rec) + return rec + + def _ordered_shop_names(db: Session, user_id: int) -> set[str]: """该用户「真实下单」(source='compare')覆盖到的店名集合,用来给比价记录打「已下单」。 diff --git a/app/repositories/onboarding.py b/app/repositories/onboarding.py index 03e51d3..0750dd1 100644 --- a/app/repositories/onboarding.py +++ b/app/repositories/onboarding.py @@ -5,7 +5,7 @@ device_id 为空(老客户端 / 取不到 ANDROID_ID)一律按"未完成"处理, """ from __future__ import annotations -from sqlalchemy import select +from sqlalchemy import delete, select from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session @@ -33,3 +33,18 @@ def mark_completed(db: Session, *, user_id: int, device_id: str) -> None: except IntegrityError: # 并发 / 重复提交撞唯一约束:已有行即视为成功。 db.rollback() + + +def delete_completion(db: Session, *, user_id: int, device_id: str) -> int: + """删该 (账号, 设备) 的引导完成标记 → 下次登录 is_completed=False → 客户端重走。 + 与 [mark_completed] 互逆。device_id 为空忽略(返 0);无记录也幂等(返 0)。返回删除行数。""" + if not device_id: + return 0 + result = db.execute( + delete(OnboardingCompletion).where( + OnboardingCompletion.user_id == user_id, + OnboardingCompletion.device_id == device_id, + ) + ) + db.commit() + return result.rowcount diff --git a/docs/api/compare/compare-record-report.md b/docs/api/compare/compare-record-report.md index cdd4523..1e7f7fa 100644 --- a/docs/api/compare/compare-record-report.md +++ b/docs/api/compare/compare-record-report.md @@ -2,10 +2,10 @@ > 所属:比价记录组（前缀 `/api/v1/compare`） | 鉴权:Bearer | [← 返回 API 索引](../README.md) -比价 `done` 帧后,客户端用**带 JWT 的通道**上报一条比价结果,落 `comparison_record` 表，作为「我的比价记录」数据源 + 用户级行为画像。 +比价 `done` 帧后上报一条比价结果,落 `comparison_record` 表，作为「我的比价记录」数据源 + 用户级行为画像。 -> ⚠️ 与不鉴权的透传端点 [`/api/v1/price/step`](./compare-price-step.md) 不同：那是转发壳，本接口按用户维度落库，**必须鉴权**。 -> 本轮只做 server 端；客户端在 done 帧后调本接口的改动另起一轮（见 [待办与技术债.md](../guides/待办与技术债.md) P1）。 +> ⚠️ **灰度定位(2026-07)**：写 `comparison_record` 现以透传壳 `compare.py` 的**后端 harvest** 为主(帧0 建 `running` 行 → done/finalize 落终态,新客户端不再 POST)。本接口降级为**老客户端兼容**通道,与 harvest 按 `trace_id` reconcile(**success 不被降级**);新版覆盖够高后可下线本 POST。 +> 与软鉴权透传端点 [`/api/v1/price/step`](./compare-price-step.md) 不同：那是转发壳(顺带 harvest 落库),本接口按用户维度显式上报,**必须鉴权**。 ## 入参（JSON body） @@ -13,7 +13,7 @@ | 字段 | 类型 | 必填 | 默认 | 说明 | |---|---|---|---|---| -| `trace_id` | string | ✅ | — | pricebot 侧 trace_id。**幂等键**：同用户同 trace_id 重复上报覆盖、返回同一 id | +| `trace_id` | string | ✅ | — | 一次比价的唯一标识(app-server 帧0 签发)。**幂等键(trace_id 单列唯一)**：同 trace_id 重复上报覆盖、返回同一 id;与 harvest 行按它 reconcile | | `business_type` | string | ❌ | `food` | `food`(外卖,当前唯一接通) / `ecom`(电商) / `coupon`(领券) | | `device_id` | string \| null | ❌ | null | 设备号 | | `store_name` | string \| null | ❌ | null | 店铺名（外卖,来自 calibration.result） | diff --git a/docs/database/comparison_record.md b/docs/database/comparison_record.md index 4b637ce..225f29e 100644 --- a/docs/database/comparison_record.md +++ b/docs/database/comparison_record.md @@ -2,13 +2,13 @@ > 模型 `app/models/comparison.py` · 仓库 `app/repositories/comparison.py` · 接口 [compare-record-report](../api/compare-record-report.md) / [compare-records](../api/compare-records.md) / [compare-record-detail](../api/compare-record-detail.md) · [← 索引](./README.md) · [总览](./OVERVIEW.md) -每完成一次比价(外卖/电商/领券),客户端在 done 帧后用**带 JWT** 的通道上报一条。App「我的比价记录」列表/详情的数据源,也是比价战绩里程碑解锁进度的计数源(`status='success'` 条数),还被「上报更低价」反查原最低价。 +每完成一次比价(外卖/电商/领券)记一行。**写入以 app-server 后端 harvest 为主**(2026-07 起):比价透传壳 `compare.py` 在帧0(pricebot 出 trace_id)即建 `running` 行,随 done / `trace/finalize` 逐步补全成终态,客户端不再主动 POST 记录;老客户端仍可走**带 JWT** 的 `POST /compare/record` 兜底(灰度期两条写路径按 `trace_id` reconcile)。App「我的比价记录」列表/详情的数据源,也是比价战绩里程碑解锁进度的计数源(`status='success'` 条数),还被「上报更低价」反查原最低价。 > 与 `savings_record` 的区别:本表是「每一次**比价行为**的完整明细」(不省钱、甚至失败也记);`savings_record` 是「真正**下单成交**省了多少」。两表独立、互不喂数据。 > 与 [`price_observation`](./price_observation.md) / `store_mapping` 的区别:本表是**用户视角**(登录后按 `user_id` 存「我的比价记录」);后两张是 server 侧无条件沉淀的**平台/门店视角客观事实**(价格事实 / 跨平台店铺身份映射),与本表 `trace_id` 同源但不互相 join,各存各的视角。 ## 用在哪 / 增删改查 -- **C / U(upsert,幂等)**:`POST /compare/record`(`upsert_record`)。按 `(user_id, trace_id)` 查:不存在→新建;已存在→整行覆盖(客户端重试/重复上报时,**更完整的那次胜出**)。`best_*`/`saved_amount_cents`/`is_source_best`/`status` 由 `_derive` 从 `comparison_results` 算出(协议已按 price 升序、rank=1 最便宜),不信客户端自算。 +- **C / U(harvest 为主,按 `trace_id` 幂等)**:透传壳 `compare.py` 三段式落库(`app/repositories/comparison.py`)——`harvest_running`(帧0 建 `running` 行)→ `harvest_done`(done 帧转 `success`/`failed` + 派生 `best_*`/`saved_amount_cents` + 返 `newly_success` 供幂等发邀请奖)→ `harvest_abort`(`trace/finalize` 转 `cancelled`/`failed`,**不降级已 success**)。老客户端仍可 `POST /compare/record`(`upsert_record`,按 `trace_id` 查、整行覆盖)兜底。`best_*`/`saved_amount_cents`/`is_source_best`/`status` 一律由 `_derive` 从 `comparison_results` 算出(协议已按 price 升序、rank=1 最便宜),不信客户端自算。 - **D**:无(关联的 `price_report` 也只把 `comparison_record_id` 置空,不删本表)。 - **R**:`GET /compare/records`(列表,`created_at` 倒序游标 + 「已下单」标记)、`GET /compare/records/{id}`(详情,限本人);`count_success` 给里程碑;`get_stats`(`status='success'` 计数 + `saved_amount_cents` 求和)给 [`GET /compare/stats`](../api/compare-stats.md) 喂「我的」页省钱战绩卡(完成比价 + 累计发现可省,**比价口径**);`report.py` 反查 `best_*`;admin 大盘/明细。 @@ -16,10 +16,10 @@ | 列 | 类型 | 约束 / 默认 | 说明(取值 / join) | |---|---|---|---| | `id` | Integer | PK, autoincrement | 被 `price_report.comparison_record_id` 引用 | -| `user_id` | Integer | FK→user.id, index, NOT NULL | 归属用户 | +| `user_id` | Integer | FK→user.id, index, **nullable**(2026-07 从 NOT NULL 放开) | 归属用户。后端 harvest 帧0 建行时(软鉴权 / 老客户端匿名)可能暂缺 → 可空;C 端「我的比价记录」按 `user_id` 过滤天然排除 null 行,admin 全看(含孤儿行) | | `device_id` | String(64) | nullable | 设备号(多设备区分 / 与不鉴权期对账) | | `business_type` | String(16) | NOT NULL, default `food`, index | 取值:`food`(当前唯一接通)/ `ecom` / `coupon` | -| `trace_id` | String(64) | NOT NULL | pricebot 侧 trace_id(关联调试落盘 + 幂等键) | +| `trace_id` | String(64) | NOT NULL, **UNIQUE** | 一次比价的唯一标识。**由 app-server 帧0 用 uuid 签发**(注入转发 body + 回填响应顶层给客户端;老客户端自带),全局唯一 = harvest upsert 键 + 关联 pricebot 调试落盘 | | `trace_url` | String(512) | nullable | 本次比价公网调试链接(`price.shaguabijia.com/traces/{dir}/`);dir 名含 pricebot 落盘时分秒,前端/server 拼不出必须存。查看接口按 `user.debug_trace_enabled`(或本机 agent 调试 `include_trace`)决定返不返回。旧记录 / 未开上云为 null | | `source_platform_id` / `_name` | String(32) | nullable | 源平台代号 / 中文名 | | `source_package` | String(128) | nullable | 源平台 Android 包名 | @@ -31,7 +31,7 @@ | `is_source_best` | Boolean | nullable | 源平台就是最便宜(= 这次没省到) | | `store_name` | String(128) | nullable | 店铺名。**与 `savings_record.shop_name` 按字符串相等关联**,给本记录打「已下单」 | | `total_dish_count` / `skipped_dish_count` | Integer | nullable | 菜品总数 / 目标平台没找到被跳过数 | -| `status` | String(16) | NOT NULL, default `success` | 取值:`success`(有非源且有价的目标结果)/ `failed`(出错/没采到目标价)。**里程碑只数 success** | +| `status` | String(16) | NOT NULL, default `success` | 取值:`running`(harvest 帧0 建行、比价进行中)/ `success`(有非源且有价的目标结果)/ `failed`(出错/没采到目标价)/ `cancelled`(用户终止 / Phase1 未识别,`harvest_abort` 写,**不降级已 success**)。**里程碑只数 success** | | `information` | String(256) | nullable | done 帧文案;成功=摘要,失败=具体原因(前端失败时当原因展示) | | `items` | JSON(PG: JSONB) | NOT NULL, default [] | 下单菜品 `[{name, qty, specs?}]` | | `comparison_results` | JSON(PG: JSONB) | NOT NULL, default [] | 逐平台对比 `[{platform_id,platform_name,package,price(元),is_source,rank,coupon_saved(元),coupon_name,applied_coupons}]`;`coupon_saved`=该平台主优惠额(美团红包/淘宝平台红包/京东百亿补贴,只取一笔),`coupon_name`=优惠来源名(展示用),`applied_coupons`=`[{name,amount}]` 多券明细 | @@ -50,7 +50,8 @@ - 被 `comparison_milestone_claim` 间接依赖:解锁进度 = 本表 `status='success'` 计数。 ## 索引与约束 -- PK `id`;index `user_id`、`business_type`、`created_at`;复合 index `ix_comparison_status_created`(`status`, `created_at`)(按 `status='success'` 过滤 + 近期排序的聚合/轮播,避免随数据量退化为全表扫);UNIQUE(`user_id`, `trace_id`) = `uq_comparison_user_trace`(幂等覆盖)。 +- PK `id`;index `user_id`、`business_type`、`created_at`;复合 index `ix_comparison_status_created`(`status`, `created_at`)(按 `status='success'` 过滤 + 近期排序的聚合/轮播,避免随数据量退化为全表扫);UNIQUE(`trace_id`) = `uq_comparison_trace`(harvest 按它 upsert,一次比价一行)。 + > 2026-07 迁移 `comparison_record_trace_unique`:唯一键从复合 `uq_comparison_user_trace`(`user_id`,`trace_id`)改为 `trace_id` 单列——harvest 帧0 建行时 user_id 可能暂缺,不能再用复合键去重。上线前须确认历史无重复 `trace_id`(`SELECT trace_id,COUNT(*) c FROM comparison_record GROUP BY trace_id HAVING c>1`),否则建单列唯一会失败。 ## 注意 - 4 个 JSON 列用 `JSON().with_variant(JSONB(),"postgresql")`(SQLite 退化 JSON)。结构化金额列存「分」,`comparison_results.price`/`coupon_saved` 原样存「元」。 diff --git a/tests/test_compare_harvest.py b/tests/test_compare_harvest.py new file mode 100644 index 0000000..b13028e --- /dev/null +++ b/tests/test_compare_harvest.py @@ -0,0 +1,258 @@ +"""后端 harvest 落库测试(比价记录改后端 harvest 后新增)。 + +覆盖: + repo 层:harvest_running(建行/幂等回填)、harvest_done(派生+newly_success 幂等)、 + harvest_abort(夭折 / **不降级 success**)、upsert_record 不降级。 + 端点层:price/step 首帧 mint trace_id + 回传 + 建 running 行;done 帧落 success; + trace/finalize 落 cancelled;软鉴权(无 token 也放行、user_id 暂空)。 +pricebot 用 httpx mock,不真连(同 test_compare_proxy)。 +""" +from __future__ import annotations + +import json +import uuid +from unittest.mock import MagicMock, patch + +import httpx + +from app.db.session import SessionLocal +from app.models.comparison import ComparisonRecord +from app.repositories import comparison as crud +from app.schemas.compare_record import ComparisonRecordIn +from sqlalchemy import select + + +def _tid() -> str: + return uuid.uuid4().hex + + +def _done_params() -> dict: + """一份典型 done 帧 params:美团 25 元 vs 源淘宝闪购 30 元 → 省 5 元、success。""" + return { + "comparison_results": [ + {"platform_id": "meituan", "platform_name": "美团", "package": "com.sankuai.meituan", + "price": 25.0, "is_source": False, "rank": 1, "items": []}, + {"platform_id": "taobao_flash", "platform_name": "淘宝闪购", + "package": "com.taobao.taobao", "price": 30.0, "is_source": True, "rank": 2, + "store_name": "测试店", "items": [{"name": "肥牛饭", "qty": 1}]}, + ], + "information": "美团更便宜", + "trace_url": "https://price.shaguabijia.com/traces/done/", + } + + +def _get(db, trace_id: str) -> ComparisonRecord | None: + return db.execute( + select(ComparisonRecord).where(ComparisonRecord.trace_id == trace_id) + ).scalar_one_or_none() + + +# ============================================================ +# repo 层 +# ============================================================ + + +def test_harvest_running_creates_row(client) -> None: + tid = _tid() + with SessionLocal() as db: + crud.harvest_running( + db, trace_id=tid, user_id=None, device_id="dev-1", + device_info={"brand": "vivo", "model": "V2309A", "android_version": "14", + "rom_version": "14"}, + trace_url="https://price.shaguabijia.com/traces/run/", + ) + rec = _get(db, tid) + assert rec is not None + assert rec.status == "running" + assert rec.user_id is None + assert rec.device_id == "dev-1" + assert rec.device_model == "V2309A" + assert rec.device_manufacturer == "vivo" + assert rec.rom_version == 14 + assert rec.trace_url.endswith("/run/") + + +def test_harvest_running_idempotent_backfills(client) -> None: + tid = _tid() + with SessionLocal() as db: + crud.harvest_running(db, trace_id=tid, user_id=None, trace_url=None) + # 第二帧带上 trace_url + user_id → 回填空缺,不新建、不改 status + crud.harvest_running(db, trace_id=tid, user_id=None, + trace_url="https://price.shaguabijia.com/traces/late/") + rows = db.execute( + select(ComparisonRecord).where(ComparisonRecord.trace_id == tid) + ).scalars().all() + assert len(rows) == 1 # 幂等:仍一行 + assert rows[0].status == "running" + assert rows[0].trace_url.endswith("/late/") # 空缺被回填 + + +def test_harvest_done_derives_and_newly_success_once(client) -> None: + tid = _tid() + with SessionLocal() as db: + crud.harvest_running(db, trace_id=tid, user_id=None) + rec, newly = crud.harvest_done(db, trace_id=tid, user_id=None, + done_params=_done_params()) + assert newly is True # running → success 是"新落成" + assert rec.status == "success" + assert rec.source_platform_id == "taobao_flash" + assert rec.source_price_cents == 3000 + assert rec.best_platform_id == "meituan" + assert rec.best_price_cents == 2500 + assert rec.saved_amount_cents == 500 # 30 - 25 + assert rec.is_source_best is False + assert rec.store_name == "测试店" + assert rec.information == "美团更便宜" + assert rec.items == [{"name": "肥牛饭", "qty": 1}] + assert rec.trace_url.endswith("/done/") + # 再来一次(重试 done)→ 已 success,newly_success=False(发奖不重复触发) + _rec2, newly2 = crud.harvest_done(db, trace_id=tid, user_id=None, + done_params=_done_params()) + assert newly2 is False + + +def test_harvest_abort_cancels_running(client) -> None: + tid = _tid() + with SessionLocal() as db: + crud.harvest_running(db, trace_id=tid, user_id=None) + rec = crud.harvest_abort(db, trace_id=tid, status="cancelled", + reason="用户终止比价", + trace_url="https://price.shaguabijia.com/traces/ab/") + assert rec is not None + assert rec.status == "cancelled" + assert rec.information == "用户终止比价" + assert rec.trace_url.endswith("/ab/") + + +def test_harvest_abort_no_downgrade_success(client) -> None: + """已 success 的行,finalize 后到(收尾取消)→ 只 refresh trace_url,status 不降级。""" + tid = _tid() + with SessionLocal() as db: + crud.harvest_running(db, trace_id=tid, user_id=None) + crud.harvest_done(db, trace_id=tid, user_id=None, done_params=_done_params()) + rec = crud.harvest_abort(db, trace_id=tid, status="cancelled", reason="收尾误触") + assert rec.status == "success" # 不降级 + assert rec.information == "美团更便宜" # 不被 abort 的 reason 覆盖 + + +def test_harvest_abort_missing_row_returns_none(client) -> None: + with SessionLocal() as db: + assert crud.harvest_abort(db, trace_id=_tid(), status="cancelled", + reason=None) is None + + +def test_upsert_record_no_downgrade_after_harvest_success(client) -> None: + """harvest 落 success 后,老客户端 fromFailure 的 cancelled 上报不许把它盖回去。""" + tid = _tid() + with SessionLocal() as db: + crud.harvest_done(db, trace_id=tid, user_id=None, done_params=_done_params()) + payload = ComparisonRecordIn( + trace_id=tid, business_type="food", status="cancelled", + information="用户终止", comparison_results=[], + ) + # 用一个不会与顺序自增用户撞的合成 id(SQLite 测试库 FK 不强制;别用小整数, + # 否则会撞上别的测试 login 出来的真实 user_id → 记录混进那个用户的列表)。 + rec = crud.upsert_record(db, user_id=987654, payload=payload) + assert rec.status == "success" # 不降级 + assert rec.user_id == 987654 # 但补上了 user_id(原为 None) + + +# ============================================================ +# 端点层(mock pricebot) +# ============================================================ + + +def _mock_pricebot(resp_json: dict): + """patch httpx.AsyncClient.post 返回给定响应;捕获转发的 content。""" + captured: dict = {} + + async def fake_post(self, url, content=None, **kw): + captured["url"] = url + captured["content"] = content + m = MagicMock() + m.status_code = 200 + m.json = lambda: dict(resp_json) + return m + + return patch.object(httpx.AsyncClient, "post", fake_post), captured + + +def _stub_body(**over) -> dict: + body = { + "device_id": "dev-x", + "step": 0, + "query": "海底捞", + "device_info": {"brand": "OPPO", "model": "PEXM00", "android_version": "13", + "rom_version": "13"}, + "screen_state": {"screen": {"width": 1080, "height": 2340, "density": 3.0}, + "foreground": {"package": "com.taobao.taobao", "activity": ""}, + "windows": []}, + } + body.update(over) + return body + + +def test_price_step_mints_trace_id_and_creates_running(client) -> None: + """首帧不带 trace_id → app-server 签发 + 回传;并建 running 行(带机型/设备)。""" + wait_frame = {"success": True, "action": {"command": "wait", "params": {"duration_ms": 500}}, + "continue": True, "trace_url": "https://price.shaguabijia.com/traces/mint/"} + p, _cap = _mock_pricebot(wait_frame) + with p: + r = client.post("/api/v1/price/step", json=_stub_body()) # 无 trace_id、无 token + assert r.status_code == 200, r.text + tid = r.json().get("trace_id") + assert tid and len(tid) >= 16 # 回传了签发的 trace_id + with SessionLocal() as db: + rec = _get(db, tid) + assert rec is not None and rec.status == "running" + assert rec.user_id is None # 无 token → 软鉴权放行、user_id 暂空 + assert rec.device_model == "PEXM00" + assert rec.trace_url.endswith("/mint/") + + +def test_price_step_done_harvests_success(client) -> None: + tid = _tid() + done_frame = {"success": True, "continue": False, + "action": {"command": "done", "params": _done_params()}, + "trace_url": "https://price.shaguabijia.com/traces/done2/"} + p, _cap = _mock_pricebot(done_frame) + with p: + r = client.post("/api/v1/price/step", json=_stub_body(trace_id=tid, step=8)) + assert r.status_code == 200 + with SessionLocal() as db: + rec = _get(db, tid) + assert rec is not None and rec.status == "success" + assert rec.best_platform_id == "meituan" + assert rec.saved_amount_cents == 500 + + +def test_trace_finalize_harvests_abort(client) -> None: + tid = _tid() + with SessionLocal() as db: # 先有 running 行(帧0建的) + crud.harvest_running(db, trace_id=tid, user_id=None) + p, _cap = _mock_pricebot({"trace_url": "https://price.shaguabijia.com/traces/fin/"}) + with p: + r = client.post("/api/v1/trace/finalize", + json={"trace_id": tid, "status": "cancelled", "reason": "用户终止"}) + assert r.status_code == 200 + with SessionLocal() as db: + rec = _get(db, tid) + assert rec is not None and rec.status == "cancelled" + assert rec.trace_url.endswith("/fin/") + + +def test_price_step_binds_user_when_authed(client) -> None: + """带 JWT 的首帧 → running 行绑上 user_id。""" + client.post("/api/v1/auth/sms/send", json={"phone": "13800009001"}) + token = client.post("/api/v1/auth/sms/login", + json={"phone": "13800009001", "code": "123456"}).json()["access_token"] + wait_frame = {"success": True, "action": {"command": "wait", "params": {}}, + "continue": True, "trace_url": "https://price.shaguabijia.com/traces/auth/"} + p, _cap = _mock_pricebot(wait_frame) + with p: + r = client.post("/api/v1/price/step", json=_stub_body(), + headers={"Authorization": f"Bearer {token}"}) + tid = r.json()["trace_id"] + with SessionLocal() as db: + rec = _get(db, tid) + assert rec is not None and rec.user_id is not None # 绑上了登录用户 diff --git a/tests/test_compare_milestone.py b/tests/test_compare_milestone.py index 06165b4..66007ff 100644 --- a/tests/test_compare_milestone.py +++ b/tests/test_compare_milestone.py @@ -20,8 +20,10 @@ def _auth(token: str) -> dict[str, str]: def _report_success(client, token: str, trace_id: str) -> None: """上报一条成功比价(有非源有效价 → status 派生 success)。""" + # trace_id 现全局唯一(prod 由 app-server 签发 UUID);测试共用 session DB,故按 token + # (每用户不同)加前缀防跨用户/跨文件复用 "s-1" 撞车(见 comparison_record_trace_unique 迁移)。 payload = { - "trace_id": trace_id, + "trace_id": f"{token[-16:]}:{trace_id}", "business_type": "food", "store_name": "测试店", "source_platform_id": "taobao_flash", @@ -40,7 +42,7 @@ def _report_success(client, token: str, trace_id: str) -> None: def _report_failed(client, token: str, trace_id: str) -> None: """上报一条失败比价(只有源 → status 派生 failed,不计入解锁)。""" payload = { - "trace_id": trace_id, + "trace_id": f"{token[-16:]}:{trace_id}", "business_type": "food", "comparison_results": [ {"platform_id": "taobao_flash", "platform_name": "淘宝闪购", "price": 30.0, diff --git a/tests/test_compare_proxy.py b/tests/test_compare_proxy.py index c278b93..b958be8 100644 --- a/tests/test_compare_proxy.py +++ b/tests/test_compare_proxy.py @@ -9,6 +9,7 @@ mock 掉对 pricebot 的 httpx 调用,验证(两个端点参数化同跑): """ from __future__ import annotations +import json from unittest.mock import MagicMock, patch import httpx @@ -41,7 +42,8 @@ def _stub_body() -> dict: @pytest.mark.parametrize("path,upstream", ENDPOINTS) def test_passes_body_through(client, path, upstream) -> None: - """无 token + pricebot 200 → 响应原样透传,body 原样转发,URL 去掉 /v1。""" + """无 token(软鉴权放行)+ pricebot 200 → 响应透传(+ app-server 注入 trace_id), + body 原样转发(自带 trace_id 时不重签),URL 去掉 /v1。""" fake_resp = { "success": True, "action": {"command": "wait", "params": {"duration_ms": 500}}, @@ -49,20 +51,25 @@ def test_passes_body_through(client, path, upstream) -> None: } captured: dict = {} - async def fake_post(self, url, json=None, **kw): + async def fake_post(self, url, content=None, **kw): + # 透传壳用 content=raw(bytes)转发,不是 json=;这里捕获 content 解回来核对 captured["url"] = url - captured["json"] = json + captured["content"] = content mock_resp = MagicMock() mock_resp.status_code = 200 - mock_resp.json = lambda: fake_resp + mock_resp.json = lambda: dict(fake_resp) # 每次新副本(端点会 setdefault trace_id 进去) return mock_resp with patch.object(httpx.AsyncClient, "post", fake_post): r = client.post(path, json=_stub_body()) assert r.status_code == 200, r.text - assert r.json() == fake_resp - assert captured["json"] == _stub_body() # body 原样转发(不鉴权,无需 token) + body = r.json() + assert body["success"] is True + assert body["action"] == fake_resp["action"] + assert body["trace_id"] == "test-trace-1" # 自带 trace_id → 原样回传, 不 mint + # body 原样转发(content=raw bytes;自带 trace_id 时不重新序列化) + assert json.loads(captured["content"]) == _stub_body() assert captured["url"].endswith(upstream) # /api/v1/xxx → /api/xxx diff --git a/tests/test_onboarding.py b/tests/test_onboarding.py index 3b6f533..a56c9f0 100644 --- a/tests/test_onboarding.py +++ b/tests/test_onboarding.py @@ -65,3 +65,28 @@ def test_onboarding_missing_device_id_treated_as_incomplete(client) -> None: phone = "13800138003" data = _login(client, phone, device_id=None) assert data["onboarding_completed"] is False + + +def _reset(client, access: str, device_id: str): + return client.post( + "/api/v1/user/onboarding/reset", + json={"device_id": device_id}, + headers={"Authorization": f"Bearer {access}"}, + ) + + +def test_onboarding_reset_makes_relogin_reonboard(client) -> None: + """重置(删完成标记)后,同 (账号, 设备) 再登录 → onboarding_completed=False,重走引导。 + 对应客户端「开发设置 → 重置新手引导」:先删后端行,再退登录重开。""" + phone = "13800138004" + dev = "android-id-reset" + + # ① 走完引导 → 标记完成 → 再登录已完成 + access = _login(client, phone, dev)["access_token"] + assert _mark_complete(client, access, dev).status_code == 200 + assert _login(client, phone, dev)["onboarding_completed"] is True + + # ② 重置(幂等:重复重置也 200)→ 再登录变未完成 → 重走引导 + assert _reset(client, access, dev).status_code == 200 + assert _reset(client, access, dev).json()["ok"] is True + assert _login(client, phone, dev)["onboarding_completed"] is False