Compare commits

..

5 Commits

Author SHA1 Message Date
marco 3eb439b5ae feat(onboarding): 新增重置新手引导端点 /api/v1/user/onboarding/reset (#114)
Reviewed-on: #114
2026-07-04 02:39:12 +08:00
liujiahui 6b10e866e9 7-3 尾声帧: /api/v1/trace/epilogue 透传壳 (#112)
比价结果页"尾声帧"配套改动(透传壳一条)。

App 收到 done、渲染完比价结果页后, 把自己页面的截图(base64)经本壳透传给 pricebot
`/api/trace/epilogue`, 存进 trace 目录并触发重传 —— trace 里补上"用户实际看到的
汇总页"(trace 步骤帧只有目标 App 画面, App 自渲染的汇总页天然在镜头外)。

壳同 `/trace/finalize`: 不鉴权、按 trace_id 一致性 hash 亲和。

配套分支(同名 7-3-ljh-testlab): pricebot-backend(端点+viewer 实拍块, PR #160) /
shaguabijia-app-android(结果页拍屏上传)。

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: no_gen_mu <liujianhishen@gmail.com>
Reviewed-on: #112
Co-authored-by: liujiahui <liujiahui@wonderable.ai>
Co-committed-by: liujiahui <liujiahui@wonderable.ai>
2026-07-04 01:42:01 +08:00
liujiahui 13b143cd97 feat(invite): 发奖改「比价并下单」+ 好友 is_compared 状态 + DEV 造数/重置接口 (#113)
重开分支重提 PR #110 的邀请发奖改动,**不含任何测试文件**(`tests/` 3 个 + `scripts/invite_reward_e2e.*` 2 个 E2E 脚本本地自用,未入库)。

只含 8 个产品/部署文件:

- 发奖口径从「上报比价」改为「实际下单」:`order.py` / `compare_record.py` / `repositories/invite.py`
- 好友列表返回 `is_compared` 已比价状态:`schemas/invite.py` / `repositories/invite.py`
- 新增 DEV 内部接口(生产 `is_prod` 一律 404):`POST/DELETE /invite/seed-fake` 造/清虚拟好友;`POST /invite/dev-reset` 固定测试号打回出厂态
- fix(deploy):每日兑换定时器锁北京时区 `Asia/Shanghai`(`daily-exchange.service` / `.timer`)

对应 PR #110,源分支 `7-1ljh`。

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: no_gen_mu <liujianhishen@gmail.com>
Reviewed-on: #113
Co-authored-by: liujiahui <liujiahui@wonderable.ai>
Co-committed-by: liujiahui <liujiahui@wonderable.ai>
2026-07-03 19:01:37 +08:00
zhuzihao 93f052535e 精简短信登录风控:保留单号冷却 + 单设备频控两道 (#109)
按 mentor 要求精简 SMS 登录链路的防刷策略,只保留两道主策略:
- 单号 60s 冷却(发码,SMS_SEND_INTERVAL_SEC)
- 单设备(device_id + IP)每小时 5 次频控(发码 + 登录,enforce_rate_limit)

删除两道:
- 单号每日发送上限(SMS_DAILY_LIMIT_PER_PHONE)
- 登录接口纯 IP 20 次/分钟限流(rate_limit 依赖)

验证码安全底线保留(一次性使用 + 输错 SMS_MAX_VERIFY_ATTEMPTS 次即作废),
属验证码正确性、不算限流策略,不在精简范围。

顺带清理:sms.py 移除 _daily_count/_today 及 datetime 导入;config.py 删常量;
auth.py 清理变为无用的 Depends/rate_limit 导入;同步 docs/integrations/sms.md 与
docs/后端技术实现.md 的防刷说明(并修正 sms.md 里已过时的 /sms/send rate_limit 描述)。

影响:单号发码上限由 10 条/天放宽为仅受 60s 冷却约束;登录撞库防护改由设备频控
+ 验证码错误次数上限兜底。test_auth.py 相应用例已删除/更新,auth 测试全绿。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

---------

Co-authored-by: zzhyyyyy <2685922758@qq.com>
Reviewed-on: #109
Co-authored-by: zhuzihao <zhuzihao@wonderable.ai>
Co-committed-by: zhuzihao <zhuzihao@wonderable.ai>
2026-07-03 15:28:32 +08:00
guke a563c1ca4b @ (#111)
docs/api目录文档分类和补全

---------

Co-authored-by: guke <guke@autohome.com.cn>
Reviewed-on: #111
2026-07-03 15:00:37 +08:00
27 changed files with 963 additions and 182 deletions
@@ -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)
+24
View File
@@ -54,5 +54,29 @@ def get_current_user(
return 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)] CurrentUser = Annotated[User, Depends(get_current_user)]
OptionalUser = Annotated[User | None, Depends(get_current_user_optional)]
DbSession = Annotated[Session, Depends(get_db)] DbSession = Annotated[Session, Depends(get_db)]
+6 -7
View File
@@ -12,11 +12,11 @@ from __future__ import annotations
import logging import logging
from fastapi import APIRouter, Depends, HTTPException, Request, status from fastapi import APIRouter, HTTPException, Request, status
from app.api.deps import CurrentUser, DbSession from app.api.deps import CurrentUser, DbSession
from app.core import test_account from app.core import test_account
from app.core.ratelimit import enforce_rate_limit, rate_limit from app.core.ratelimit import enforce_rate_limit
from app.core.security import TokenError, decode_token, issue_token_pair from app.core.security import TokenError, decode_token, issue_token_pair
from app.integrations.jiguang import JiguangError, mask_phone, verify_and_get_phone from app.integrations.jiguang import JiguangError, mask_phone, verify_and_get_phone
from app.integrations.sms import SmsError, send_code, verify_code from app.integrations.sms import SmsError, send_code, verify_code
@@ -41,7 +41,7 @@ router = APIRouter(prefix="/api/v1/auth", tags=["auth"])
# 手机号登录防刷:同一设备(device_id) + 同一 IP 每小时最多的登录尝试次数(成功/失败都计)。 # 手机号登录防刷:同一设备(device_id) + 同一 IP 每小时最多的登录尝试次数(成功/失败都计)。
SMS_LOGIN_MAX_PER_HOUR = 5 SMS_LOGIN_MAX_PER_HOUR = 5
# 发码防刷:同一设备(device_id) + 同一 IP 每小时最多的发码次数。 # 发码防刷:同一设备(device_id) + 同一 IP 每小时最多的发码次数。
# 堵「换手机号绕开单号 60s 冷却 / 单号每日上限」的洞 —— 那两道是单号维度,一机换号能绕开。 # 堵「换手机号绕开单号 60s 冷却」的洞 —— 冷却是单号维度,一机换号能绕开。
SMS_SEND_MAX_PER_HOUR_PER_DEVICE = 5 SMS_SEND_MAX_PER_HOUR_PER_DEVICE = 5
@@ -100,8 +100,8 @@ def sms_send(req: SmsSendRequest, request: Request) -> SmsSendResponse:
return SmsSendResponse(sent=True, mock=True, cooldown_sec=0) return SmsSendResponse(sent=True, mock=True, cooldown_sec=0)
# 防刷:同一设备(device_id) + 同一 IP 每小时最多 SMS_SEND_MAX_PER_HOUR_PER_DEVICE 次发码。 # 防刷:同一设备(device_id) + 同一 IP 每小时最多 SMS_SEND_MAX_PER_HOUR_PER_DEVICE 次发码。
# 补「换手机号绕开单号 60s 冷却 / 单号每日上限」的洞(那两道是单号维度,一机换号能绕);设备维度按机器封顶, # 补「换手机号绕开单号 60s 冷却」的洞(冷却是单号维度,一机换号能绕);设备维度按机器封顶,
# 挡短信轰炸/烧钱。放在真发(send_code)之前 → 超限直接拦下、不真发短信。与路由上 IP 维度(10次/分钟)互补。 # 挡短信轰炸/烧钱。放在真发(send_code)之前 → 超限直接拦下、不真发短信。
enforce_rate_limit( enforce_rate_limit(
request, request,
scope="sms-send-device", scope="sms-send-device",
@@ -125,7 +125,6 @@ def sms_send(req: SmsSendRequest, request: Request) -> SmsSendResponse:
"/sms/login", "/sms/login",
response_model=TokenWithUser, response_model=TokenWithUser,
summary="手机号+验证码登录", summary="手机号+验证码登录",
dependencies=[Depends(rate_limit(20, 60, "sms-login"))], # 防撞库爆破(另有单码失败次数上限)
) )
def sms_login(req: SmsLoginRequest, request: Request, db: DbSession) -> TokenWithUser: def sms_login(req: SmsLoginRequest, request: Request, db: DbSession) -> TokenWithUser:
# 测试账号:免验证码登录 + 每日上限 + 每次都走新手引导(详见 app/core/test_account.py)。 # 测试账号:免验证码登录 + 每日上限 + 每次都走新手引导(详见 app/core/test_account.py)。
@@ -143,7 +142,7 @@ def sms_login(req: SmsLoginRequest, request: Request, db: DbSession) -> TokenWit
return _login_response(user, onboarding_completed=False, force_onboarding=True) return _login_response(user, onboarding_completed=False, force_onboarding=True)
# 防刷:同一设备(device_id) + 同一 IP 每小时最多 SMS_LOGIN_MAX_PER_HOUR 次登录尝试。放在验证码校验 # 防刷:同一设备(device_id) + 同一 IP 每小时最多 SMS_LOGIN_MAX_PER_HOUR 次登录尝试。放在验证码校验
# **之前** → 输错验证码的失败尝试也计数,才挡得住撞库/爆破。与路由上 IP 维度的 sms-login 限流(同 IP)互补 # **之前** → 输错验证码的失败尝试也计数,才挡得住撞库/爆破(另有单码失败 SMS_MAX_VERIFY_ATTEMPTS 次即作废兜底)
# ⚠️ 按设备而非手机号 → 一台机器换不同手机号刷登录也受限(防一机狂登多号);device_id 空(老客户端)时 # ⚠️ 按设备而非手机号 → 一台机器换不同手机号刷登录也受限(防一机狂登多号);device_id 空(老客户端)时
# 退化为该 IP 下所有空设备聚一桶,仍受限。 # 退化为该 IP 下所有空设备聚一桶,仍受限。
enforce_rate_limit( enforce_rate_limit(
+203 -60
View File
@@ -1,49 +1,114 @@
"""外卖比价业务透传端点。 """外卖比价业务透传端点 + 后端 harvest 落库
把客户端 POST /api/v1/intent/recognize 和 /api/v1/price/step 请求原样转发给 把客户端 POST /api/v1/intent/* 和 /api/v1/price/step 转发给 pricebot,同时**由 app-server
pricebot-backend 的 /api/intent/recognize 和 /api/price/step。MVP 阶段**不鉴权** 在透传里直接落库比价记录**(不再靠客户端 POST /compare/record):
(同 coupon/step,见 docs/待办与技术债.md P1:device_id 透传区分设备,待补 JWT + - 帧0(客户端首帧不带 trace_id)→ app-server 用 uuid 签发 trace_id、注入转发 body、回给
device_id↔user_id 绑定后才能做用户级画像)。 客户端;并按 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:从源平台(淘宝闪购/美团/京东外卖)购物车页识别店名+菜品+ 鉴权:软鉴权(OptionalUser)——新客户端带 JWT → 绑 user_id;老客户端不带 → user_id 暂空,
价格,一次性 由其后续 /compare/record 上报补(灰度期两条写路径按 trace_id reconcile,success 不被降级)
- Phase 2 /price/step:多轮循环,在目标平台复现订单读到手价,直到 done。
真正的目标驱动比价逻辑在 pricebot-backend(GoalEngine,另一个 repo),本接口是透传壳。 真正的目标驱动比价逻辑在 pricebot-backend(GoalEngine,另一个 repo),本接口是透传壳 + 落库
电商(ecom)那两个端点 food MVP 暂不需要,以后放开 scene 时再加 /ecom/intent/recognize pricebot 协议文档: pricebot-backend/docs/main/02_api_protocol.md
和 /ecom/step 两行即可。
pricebot 协议文档:
pricebot-backend/docs/main/02_api_protocol.md
""" """
from __future__ import annotations from __future__ import annotations
import json import json
import logging import logging
import time
import uuid
from typing import Any from typing import Any
import httpx import httpx
from fastapi import APIRouter, HTTPException, Request, status 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.config import settings
from app.core.logging import trace_id_ctx
from app.core.pricebot_client import get_pricebot_client from app.core.pricebot_client import get_pricebot_client
from app.core.pricebot_router import pick_pricebot 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") logger = logging.getLogger("shagua.compare")
router = APIRouter(prefix="/api/v1", tags=["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, def _harvest_running_blocking(
比领券的 30s 长)。 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() raw = await request.body()
try: try:
meta = json.loads(raw) meta = json.loads(raw)
@@ -52,26 +117,41 @@ async def _passthrough(request: Request, upstream_path: str) -> dict[str, Any]:
if not isinstance(meta, dict): if not isinstance(meta, dict):
meta = {} meta = {}
# 按 trace_id 一致性 hash 选 pricebot 实例(同一比价的所有帧落同一进程,内存维护 state) trace_id = meta.get("trace_id")
base = pick_pricebot(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}" url = f"{base.rstrip('/')}{upstream_path}"
timeout = settings.PRICEBOT_COMPARE_TIMEOUT_SEC timeout = settings.PRICEBOT_COMPARE_TIMEOUT_SEC
step = meta.get("step")
logger.info( logger.info(
"compare %s device_id=%s trace_id=%s step=%s", "→ pricebot %s step=%s%s", upstream_path, step, " [mint]" if minted else "",
upstream_path, extra={"phase": "forward", "endpoint": upstream_path, "step": step,
meta.get("device_id"), "device_id": meta.get("device_id"), "minted": minted,
meta.get("trace_id"), "query": meta.get("query"), "user_id": (user.id if user else None)},
meta.get("step"),
) )
t0 = time.monotonic()
try: try:
client = get_pricebot_client() client = get_pricebot_client()
resp = await client.post( resp = await client.post(
url, content=raw, headers={"Content-Type": "application/json"}, timeout=timeout url, content=raw, headers={"Content-Type": "application/json"}, timeout=timeout
) )
except httpx.RequestError as e: 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( raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY, status_code=status.HTTP_502_BAD_GATEWAY,
detail=f"pricebot upstream unreachable: {e}", detail=f"pricebot upstream unreachable: {e}",
@@ -79,49 +159,112 @@ async def _passthrough(request: Request, upstream_path: str) -> dict[str, Any]:
if resp.status_code >= 500: if resp.status_code >= 500:
logger.error( logger.error(
"[pricebot] 5xx status=%d body=%s", "pricebot %s 5xx=%d", upstream_path, resp.status_code,
resp.status_code, extra={"phase": "resp", "endpoint": upstream_path, "step": step,
resp.text[:500], "http_status": resp.status_code, "error": "upstream_5xx"},
) )
raise HTTPException( raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY, status_code=status.HTTP_502_BAD_GATEWAY,
detail=f"pricebot upstream returned {resp.status_code}", 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)") @router.post("/intent/recognize", summary="外卖比价 Phase 1 意图识别 (透传 + 建 running 行)")
async def intent_recognize(request: Request) -> dict[str, Any]: async def intent_recognize(request: Request, user: OptionalUser) -> dict[str, Any]:
return await _passthrough(request, "/api/intent/recognize") resp, _, _ = await _forward(request, "/api/intent/recognize", user)
return resp
@router.post("/intent/step", summary="外卖比价 Phase 1 多帧意图识别 (透传到 pricebot, 仅淘宝源)") @router.post("/intent/step", summary="外卖比价 Phase 1 多帧意图识别 (透传, 仅淘宝源)")
async def intent_step(request: Request) -> dict[str, Any]: async def intent_step(request: Request, user: OptionalUser) -> dict[str, Any]:
# 多帧版意图识别(展开+滚动采集→提取): 循环调用直到 done(done 帧顶层带 resp, _, _ = await _forward(request, "/api/intent/step", user)
# result+calibration)。目前仅淘宝源走这条, 其它源走上面单次 /intent/recognize。 return resp
return await _passthrough(request, "/api/intent/step")
@router.post( @router.post("/intent/precoupon/step", summary="外卖比价 Phase 0 识别前先用券 (透传, 仅美团源)")
"/intent/precoupon/step", async def intent_precoupon_step(request: Request, user: OptionalUser) -> dict[str, Any]:
summary="外卖比价 Phase 0 意图识别前先用券 (透传到 pricebot, 仅美团源)", resp, _, _ = await _forward(request, "/api/intent/precoupon/step", user)
) return resp
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("/price/step", summary="外卖比价 Phase 2 步进 (透传到 pricebot)") @router.post("/price/step", summary="外卖比价 Phase 2 步进 (透传 + done 落库)")
async def price_step(request: Request) -> dict[str, Any]: async def price_step(request: Request, user: OptionalUser) -> dict[str, Any]:
return await _passthrough(request, "/api/price/step") 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 收尾上云 (透传到 pricebot, 终止/未识别拿 trace_url)") @router.post("/trace/epilogue", summary="比价结果页尾声帧 (透传到 pricebot, 用户视角截图入 trace)")
async def trace_finalize(request: Request) -> dict[str, Any]: async def trace_epilogue(request: Request, user: OptionalUser) -> dict[str, Any]:
# 用户终止 / Phase1 未识别没走到 done 帧, pricebot 没上云也没回传 trace_url。客户端收尾时 # App 收到 done、渲染完结果页后,把自己页面的截图(base64)传给 pricebot 存进 trace
# 打这个, _passthrough 按 trace_id 一致性 hash 落到处理这条 trace 的同一 pricebot 进程 # 目录并触发重传 —— trace 里补上"用户实际看到的汇总页"(步骤帧只有目标 App 画面)。
# (dir_cache 在那, 才能算对 trace 目录), 由后者打包上云返回 {trace_url}。 # body ~几百 KB(截图 base64), 纯透传壳(harvest_first_frame=False:不建行/不落库,
return await _passthrough(request, "/api/trace/finalize") # 该 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
+9 -22
View File
@@ -1,15 +1,13 @@
"""比价记录 endpoint(「我的比价记录」数据源)。 """比价记录 endpoint(「我的比价记录」+ admin 数据源)。
路由前缀 `/api/v1/compare`: 路由前缀 `/api/v1/compare`:
POST /record 上报一次比价结果(幂等:同 user+trace_id 覆盖) POST /record (灰度期兼容)老客户端上报,按 **trace_id** 幂等 + 不降级 success
GET /records 比价记录列表(游标分页) GET /records 比价记录列表(游标分页)
GET /records/{id} 单条详情(含 raw_payload 全量) GET /records/{id} 单条详情(含 raw_payload 全量)
**均需鉴权**(CurrentUser)——与同文件无关的不鉴权透传 `compare.py` 分开:那个是 **均需鉴权**(CurrentUser)。⚠️ 写路径现以 `compare.py` 透传壳的**后端 harvest** 为主
转发壳(MVP 不鉴权),这里是按用户维度落库的业务接口,必须有 user_id。 (帧0 建 running 行 → done/finalize 落终态,新客户端不再 POST);本 POST /record 仅灰度期
给老客户端用,与 harvest 按 trace_id reconcile。新版覆盖够高后可下线本 POST(阶段3)。
注:本轮只做 server 端,客户端(android 仓)在 done 帧后调 POST /record 上报的改动
另起一轮(见 app-server docs/待办与技术债.md P1)。
""" """
from __future__ import annotations from __future__ import annotations
@@ -21,16 +19,15 @@ from app.api.deps import CurrentUser, DbSession
from app.db.session import SessionLocal from app.db.session import SessionLocal
from app.models.comparison import ComparisonRecord from app.models.comparison import ComparisonRecord
from app.repositories import comparison as crud_compare from app.repositories import comparison as crud_compare
from app.repositories import invite as crud_invite
from app.services.pricebot_llm_calls import fetch_llm_calls
from app.schemas.compare_record import ( from app.schemas.compare_record import (
CompareStatsOut, CompareStatsOut,
ComparisonRecordCreatedOut, ComparisonRecordCreatedOut,
ComparisonRecordDetailOut, ComparisonRecordDetailOut,
ComparisonRecordIn, ComparisonRecordIn,
ComparisonRecordPage,
ComparisonRecordOut, ComparisonRecordOut,
ComparisonRecordPage,
) )
from app.services.pricebot_llm_calls import fetch_llm_calls
logger = logging.getLogger("shagua.compare_record") logger = logging.getLogger("shagua.compare_record")
@@ -53,18 +50,8 @@ def report_record(
# 任务做,不阻塞上报响应(顺带给 pricebot 落盘留足余量)。upsert 已 commit,后台用 # 任务做,不阻塞上报响应(顺带给 pricebot 落盘留足余量)。upsert 已 commit,后台用
# 独立 session 按 record id 回填 llm_calls + 派生 llm_call_count/retry_count。 # 独立 session 按 record id 回填 llm_calls + 派生 llm_call_count/retry_count。
background_tasks.add_task(_backfill_llm_calls, rec.id, rec.trace_id) background_tasks.add_task(_backfill_llm_calls, rec.id, rec.trace_id)
# 邀请 v2 发奖:被邀请人完成一次【成功】比价 → 给邀请人发邀请奖励金(幂等,只发一次)。 # 注:邀请发奖已从"比价成功"挪到"实际下单"(见 api/v1/order.py report_order)——
# best-effort:发奖异常不影响比价上报本身(rec 已 commit),只 log;邀请人补偿靠后续对账 # 冰拍板:被邀请人完成比价并实际下单才算邀请成功,仅完成比价不再发奖
if rec.status == "success":
try:
reward = crud_invite.try_reward_on_compare(db, user.id)
if reward.status == "granted":
logger.info(
"invite compare reward granted inviter=%s invitee=%s cents=%s",
reward.inviter_user_id, user.id, reward.reward_cents,
)
except Exception as e: # noqa: BLE001 best-effort,发奖失败不阻塞上报
logger.warning("invite compare reward failed invitee=%s: %s", user.id, e)
logger.info( logger.info(
"compare record user=%s trace=%s biz=%s status=%s saved=%s (llm_calls backfill queued)", "compare record user=%s trace=%s biz=%s status=%s saved=%s (llm_calls backfill queued)",
user.id, user.id,
+19
View File
@@ -1,9 +1,14 @@
import logging
from fastapi import APIRouter from fastapi import APIRouter
from app.api.deps import CurrentUser, DbSession from app.api.deps import CurrentUser, DbSession
from app.repositories import invite as crud_invite
from app.repositories import savings as crud_savings from app.repositories import savings as crud_savings
from app.schemas.order import OrderReportOut, OrderReportRequest from app.schemas.order import OrderReportOut, OrderReportRequest
logger = logging.getLogger("shagua.order")
router = APIRouter(prefix="/api/v1/order", tags=["order"]) router = APIRouter(prefix="/api/v1/order", tags=["order"])
@@ -15,6 +20,20 @@ router = APIRouter(prefix="/api/v1/order", tags=["order"])
def report_order(req: OrderReportRequest, user: CurrentUser, db: DbSession) -> OrderReportOut: def report_order(req: OrderReportRequest, user: CurrentUser, db: DbSession) -> OrderReportOut:
# 记账唯一真相表是 savings_record(source='compare')。 # 记账唯一真相表是 savings_record(source='compare')。
rec, duplicated = crud_savings.create_from_report(db, user.id, req) rec, duplicated = crud_savings.create_from_report(db, user.id, req)
# 邀请 v2 发奖:被邀请人【实际下单】(而非仅完成比价)才给邀请人发邀请奖励金(幂等,只发一次)。
# 触发点从"比价成功上报"挪到这里(冰:完成比价并实际下单才算成功)。仅首次真实上报触发,
# 重复上报(duplicated)跳过。best-effort:发奖异常不影响订单上报本身(rec 已 commit),只 log;
# try_reward_on_compare 自带 compare_reward_granted 幂等闸,漏发靠后续对账补。
if not duplicated:
try:
reward = crud_invite.try_reward_on_compare(db, user.id)
if reward.status == "granted":
logger.info(
"invite order reward granted inviter=%s invitee=%s cents=%s",
reward.inviter_user_id, user.id, reward.reward_cents,
)
except Exception as e: # noqa: BLE001 best-effort,发奖失败不阻塞订单上报
logger.warning("invite order reward failed invitee=%s: %s", user.id, e)
return OrderReportOut( return OrderReportOut(
id=rec.id, id=rec.id,
platform=rec.platform or req.platform, platform=rec.platform or req.platform,
+12
View File
@@ -69,6 +69,18 @@ def complete_onboarding(
return OkResponse() 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( @router.get(
"/onboarding/status", "/onboarding/status",
response_model=OnboardingStatusResponse, response_model=OnboardingStatusResponse,
-1
View File
@@ -81,7 +81,6 @@ class Settings(BaseSettings):
SMS_SIGN_ID: int = 31729 # 极光短信签名 ID(非机密,可被 .env 覆盖) SMS_SIGN_ID: int = 31729 # 极光短信签名 ID(非机密,可被 .env 覆盖)
SMS_TEMPLATE_ID: int = 1 # 极光短信模板 ID(变量名 code,有效期 5 分钟) SMS_TEMPLATE_ID: int = 1 # 极光短信模板 ID(变量名 code,有效期 5 分钟)
SMS_CODE_LENGTH: int = 6 # 验证码位数(本服务生成;前端 code 字段 4-8 位兼容) SMS_CODE_LENGTH: int = 6 # 验证码位数(本服务生成;前端 code 字段 4-8 位兼容)
SMS_DAILY_LIMIT_PER_PHONE: int = 10 # 单手机号每日发送上限(防刷 + 控费)
SMS_MAX_VERIFY_ATTEMPTS: int = 5 # 单个验证码最多校验失败次数,超过即作废(防爆破) SMS_MAX_VERIFY_ATTEMPTS: int = 5 # 单个验证码最多校验失败次数,超过即作废(防爆破)
# ===== 测试账号(release 包全流程联调用)===== # ===== 测试账号(release 包全流程联调用)=====
+57 -8
View File
@@ -2,9 +2,15 @@
业务代码用 `logger = logging.getLogger("shagua.xxx")` 即可, 本模块在 main.py 启动时调一次 业务代码用 `logger = logging.getLogger("shagua.xxx")` 即可, 本模块在 main.py 启动时调一次
- 控制台(stdout): 人类可读文本, systemd / 本地看 - 控制台(stdout): 人类可读文本, systemd / 本地看; trace 时行尾附 `trace=xxx`
- 文件 `logs/app-server.log`: 单行 JSON, 供阿里云 SLS/Logtail 采集(JSON 模式零正则); - 文件 `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_JSON_CONSOLE=1 控制台也输出 JSON
- LOG_DIR / LOG_FILE 改落盘路径(默认 logs/app-server.log) - LOG_DIR / LOG_FILE 改落盘路径(默认 logs/app-server.log)
@@ -17,13 +23,36 @@ import json
import logging import logging
import os import os
import sys import sys
from contextvars import ContextVar
from datetime import datetime from datetime import datetime
from logging.handlers import RotatingFileHandler from logging.handlers import RotatingFileHandler
from pathlib import Path 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): class JsonFormatter(logging.Formatter):
"""LogRecord 序列化成单行 JSON(SLS/Logtail 友好)。异常栈内嵌为字段, 整条仍是一行""" """LogRecord 单行 JSON(SLS/Logtail 友好)。trace_id 提到顶层、extra 键平铺, 异常栈内嵌。"""
def __init__(self, service: str = "app-server"): def __init__(self, service: str = "app-server"):
super().__init__() super().__init__()
self.service = service self.service = service
@@ -35,10 +64,17 @@ class JsonFormatter(logging.Formatter):
"level": record.levelname, "level": record.levelname,
"service": self.service, "service": self.service,
"logger": record.name, "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: if record.exc_info:
data["exception"] = self.formatException(record.exc_info) data["exception"] = self.formatException(record.exc_info)
if record.stack_info: if record.stack_info:
@@ -46,6 +82,15 @@ class JsonFormatter(logging.Formatter):
return json.dumps(data, ensure_ascii=False, default=str) 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 _CONFIGURED = False
@@ -63,14 +108,17 @@ def setup_logging(debug: bool = False) -> None:
for h in list(root.handlers): for h in list(root.handlers):
root.removeHandler(h) root.removeHandler(h)
ctx_filter = _ContextFilter()
# 控制台: 默认文本(systemd/本地看); LOG_JSON_CONSOLE=1 时输出 JSON # 控制台: 默认文本(systemd/本地看); LOG_JSON_CONSOLE=1 时输出 JSON
console = logging.StreamHandler(sys.stdout) console = logging.StreamHandler(sys.stdout)
if os.getenv("LOG_JSON_CONSOLE") == "1": if os.getenv("LOG_JSON_CONSOLE") == "1":
console.setFormatter(JsonFormatter(service)) console.setFormatter(JsonFormatter(service))
else: else:
console.setFormatter( 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) root.addHandler(console)
# 文件: 单行 JSON, 供 Logtail 采集(自动轮转, 单文件 10MB, 保留 5 个) # 文件: 单行 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", log_file, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8",
) )
file_handler.setFormatter(JsonFormatter(service)) file_handler.setFormatter(JsonFormatter(service))
file_handler.addFilter(ctx_filter)
root.addHandler(file_handler) root.addHandler(file_handler)
# 第三方库降噪 # 第三方库降噪
+8 -25
View File
@@ -8,15 +8,16 @@
校验) 鉴权复用极光一键登录的 `JG_APP_KEY`/`JG_MASTER_SECRET`(同一极光应用) 校验) 鉴权复用极光一键登录的 `JG_APP_KEY`/`JG_MASTER_SECRET`(同一极光应用)
验证码存储:**进程内存**( worker uvicorn 够用)重启丢失(用户重发即可) 验证码存储:**进程内存**( worker uvicorn 够用)重启丢失(用户重发即可)
worker / 多机时内存不共享 冷却每日上限校验都会失效,届时迁移到 DB/Redis worker / 多机时内存不共享 冷却校验都会失效,届时迁移到 DB/Redis
docs/待办与技术债.md docs/待办与技术债.md
防刷(短信花钱 + `/sms/send` 在登录前无法 JWT 鉴权): 防刷(短信花钱 + `/sms/send` 在登录前无法 JWT 鉴权):
1. 单号 `SMS_SEND_INTERVAL_SEC` 冷却(本文件) 1. 单号 `SMS_SEND_INTERVAL_SEC` 冷却(本文件)
2. 号每日 `SMS_DAILY_LIMIT_PER_PHONE` 条上限(本文件) 2. 设备(device_id)每小时频控(api auth.sms_send enforce_rate_limit)+ 极光控制台 IP 白名单/防轰炸(运维侧)
3. 单设备(device_id)每小时频控(api auth.sms_send enforce_rate_limit)+ 极光控制台 IP 白名单/防轰炸(运维侧)
IP 频控(rate_limit 依赖)2026-06-26 按产品要求删除改设备维度; device_id 客户端可伪造/轮换, IP 频控(rate_limit 依赖)2026-06-26 按产品要求删除改设备维度; device_id 客户端可伪造/轮换,
脚本轮换 id 能绕过本层 挡脚本狂发主要靠极光控制台侧(+ 可选 nginx 限流) 脚本轮换 id 能绕过本层 挡脚本狂发主要靠极光控制台侧(+ 可选 nginx 限流)
单号每日上限2026-07-03 按精简要求删除(mentor :登录风控只留单号冷却 + 单设备频控);
单号维度现仅剩 60s 冷却,换号轰炸由单设备频控封顶
:单码校验失败 `SMS_MAX_VERIFY_ATTEMPTS` 次即作废(防爆破),验过即作废(一次性) :单码校验失败 `SMS_MAX_VERIFY_ATTEMPTS` 次即作废(防爆破),验过即作废(一次性)
""" """
from __future__ import annotations from __future__ import annotations
@@ -26,7 +27,6 @@ import logging
import secrets import secrets
import time import time
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime
from threading import Lock from threading import Lock
import httpx import httpx
@@ -56,22 +56,17 @@ class _CodeRecord:
# 进程内存(单 worker 有效;多 worker 不共享,见模块 docstring) # 进程内存(单 worker 有效;多 worker 不共享,见模块 docstring)
_codes: dict[str, _CodeRecord] = {} # phone -> 当前有效验证码 _codes: dict[str, _CodeRecord] = {} # phone -> 当前有效验证码
_last_sent: dict[str, float] = {} # phone -> 上次发送 epoch(冷却) _last_sent: dict[str, float] = {} # phone -> 上次发送 epoch(冷却)
_daily_count: dict[str, tuple[str, int]] = {} # phone -> (date_str, 当日发送数)
_lock = Lock() _lock = Lock()
_GC_THRESHOLD = 10000 # 任一内存 dict 超此阈值,send 时顺手清过期项(防无限增长,仿 ratelimit) _GC_THRESHOLD = 10000 # 任一内存 dict 超此阈值,send 时顺手清过期项(防无限增长,仿 ratelimit)
def _today() -> str:
return datetime.now().strftime("%Y-%m-%d")
def _gen_code() -> str: def _gen_code() -> str:
"""生成 N 位数字验证码(用 secrets 而非 random;允许前导 0)。""" """生成 N 位数字验证码(用 secrets 而非 random;允许前导 0)。"""
return "".join(secrets.choice("0123456789") for _ in range(settings.SMS_CODE_LENGTH)) return "".join(secrets.choice("0123456789") for _ in range(settings.SMS_CODE_LENGTH))
def _gc(now: float) -> None: def _gc(now: float) -> None:
"""顺手清理过期内存项,防个 dict 无限增长。仅在持锁时调用,且某 dict 超 """顺手清理过期内存项,防个 dict 无限增长。仅在持锁时调用,且某 dict 超
_GC_THRESHOLD 才扫它(低频,开销可忽略)""" _GC_THRESHOLD 才扫它(低频,开销可忽略)"""
if len(_codes) > _GC_THRESHOLD: if len(_codes) > _GC_THRESHOLD:
for p in [p for p, r in _codes.items() if now > r.expires_at]: for p in [p for p, r in _codes.items() if now > r.expires_at]:
@@ -80,10 +75,6 @@ def _gc(now: float) -> None:
cutoff = now - settings.SMS_SEND_INTERVAL_SEC cutoff = now - settings.SMS_SEND_INTERVAL_SEC
for p in [p for p, ts in _last_sent.items() if ts < cutoff]: for p in [p for p, ts in _last_sent.items() if ts < cutoff]:
_last_sent.pop(p, None) _last_sent.pop(p, None)
if len(_daily_count) > _GC_THRESHOLD:
today = _today()
for p in [p for p, (d, _c) in _daily_count.items() if d != today]:
_daily_count.pop(p, None)
def send_code(phone: str) -> int: def send_code(phone: str) -> int:
@@ -102,17 +93,9 @@ def send_code(phone: str) -> int:
remain = int(settings.SMS_SEND_INTERVAL_SEC - elapsed) remain = int(settings.SMS_SEND_INTERVAL_SEC - elapsed)
raise SmsError(f"发送过于频繁,请 {remain}s 后再试") raise SmsError(f"发送过于频繁,请 {remain}s 后再试")
today = _today()
day, cnt = _daily_count.get(phone, ("", 0))
if day != today:
cnt = 0
if cnt >= settings.SMS_DAILY_LIMIT_PER_PHONE:
raise SmsError("今日验证码发送次数已达上限,请明天再试")
code = _gen_code() code = _gen_code()
# 预占:先记冷却/计数/存码,释放锁后再发网络(发失败保留冷却+计数,见下) # 预占:先记冷却/存码,释放锁后再发网络(发失败保留冷却,见下)
_last_sent[phone] = now _last_sent[phone] = now
_daily_count[phone] = (today, cnt + 1)
_codes[phone] = _CodeRecord(code=code, expires_at=now + settings.SMS_CODE_TTL_SEC) _codes[phone] = _CodeRecord(code=code, expires_at=now + settings.SMS_CODE_TTL_SEC)
# --- lock 外:真正发送(网络 IO 不持锁)--- # --- lock 外:真正发送(网络 IO 不持锁)---
@@ -123,7 +106,7 @@ def send_code(phone: str) -> int:
_send_via_jiguang(phone, code) _send_via_jiguang(phone, code)
logger.info("[SMS] sent to %s****", phone[:3]) logger.info("[SMS] sent to %s****", phone[:3])
except Exception as e: except Exception as e:
# 发送失败:**保留冷却 + 每日计数**(失败也限速,挡住余额不足/签名失效时 # 发送失败:**保留冷却**(失败也限速,挡住余额不足/签名失效时
# 前端重试狂打极光),只清掉没发出去的码(用户收不到,留着无意义且占内存)。 # 前端重试狂打极光),只清掉没发出去的码(用户收不到,留着无意义且占内存)。
with _lock: with _lock:
_codes.pop(phone, None) _codes.pop(phone, None)
+7 -4
View File
@@ -39,16 +39,19 @@ _JSON = JSON().with_variant(JSONB(), "postgresql")
class ComparisonRecord(Base): class ComparisonRecord(Base):
__tablename__ = "comparison_record" __tablename__ = "comparison_record"
__table_args__ = ( __table_args__ = (
# 同一用户同一次比价(trace_id)只存一条:客户端重试/误点重复上报时幂等覆盖 # trace_id 由 app-server 签发、全局唯一 → 一次比价一行,后端 harvest 按它 upsert
UniqueConstraint("user_id", "trace_id", name="uq_comparison_user_trace"), # (原 (user_id,trace_id) 复合唯一改为 trace_id 单列:harvest 帧0 建行时 user_id 可能暂缺。)
UniqueConstraint("trace_id", name="uq_comparison_trace"),
# 首页轮播 / 省钱战绩聚合都按 status='success' 过滤 + created_at 近期排序; # 首页轮播 / 省钱战绩聚合都按 status='success' 过滤 + created_at 近期排序;
# 复合索引避免随数据量增大退化成全表扫(单列 created_at 索引不含 status)。 # 复合索引避免随数据量增大退化成全表扫(单列 created_at 索引不含 status)。
Index("ix_comparison_status_created", "status", "created_at"), Index("ix_comparison_status_created", "status", "created_at"),
) )
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
user_id: Mapped[int] = mapped_column( # 后端 harvest 在帧0(pricebot 出 trace_id)即建行,软鉴权下 user_id 可能暂缺(老客户端/匿名)→ 可空。
Integer, ForeignKey("user.id"), index=True, nullable=False # 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 数据对账)
device_id: Mapped[str | None] = mapped_column(String(64), nullable=True) device_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
+217 -2
View File
@@ -74,7 +74,12 @@ def _derive(payload: ComparisonRecordIn) -> dict:
def upsert_record( def upsert_record(
db: Session, *, user_id: int, payload: ComparisonRecordIn db: Session, *, user_id: int, payload: ComparisonRecordIn
) -> ComparisonRecord: ) -> 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) derived = _derive(payload)
fields = dict( fields = dict(
device_id=payload.device_id, device_id=payload.device_id,
@@ -110,14 +115,29 @@ def upsert_record(
**derived, **derived,
) )
# 按 trace_id 定位(唯一键已改 trace_id):后端 harvest 可能已建行,这里的客户端上报
# (灰度期老客户端 / 新客户端不再走这条)与之 reconcile。
existing = db.execute( existing = db.execute(
select(ComparisonRecord).where( select(ComparisonRecord).where(
ComparisonRecord.user_id == user_id,
ComparisonRecord.trace_id == payload.trace_id, ComparisonRecord.trace_id == payload.trace_id,
) )
).scalar_one_or_none() ).scalar_one_or_none()
if existing is not 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(): for k, v in fields.items():
setattr(existing, k, v) setattr(existing, k, v)
db.commit() db.commit()
@@ -139,6 +159,201 @@ def upsert_record(
return rec return rec
# ============================================================
# 后端 harvestapp-server 从 pricebot 透传响应里直接落库(不靠客户端上报)。
# 帧0 建行(running) → 最终 done 更新(success/failed) → finalize 更新(aborted)。
# 全按 trace_id upsertsuccess 行永不被后到的 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]: def _ordered_shop_names(db: Session, user_id: int) -> set[str]:
"""该用户「真实下单」(source='compare')覆盖到的店名集合,用来给比价记录打「已下单」。 """该用户「真实下单」(source='compare')覆盖到的店名集合,用来给比价记录打「已下单」。
+3
View File
@@ -304,6 +304,9 @@ def get_invitees(
"avatar_url": u.avatar_url or u.wechat_avatar_url or None, "avatar_url": u.avatar_url or u.wechat_avatar_url or None,
"coins": rel.inviter_coin, # v3 起恒 0(邀请人收益改走邀请奖励金) "coins": rel.inviter_coin, # v3 起恒 0(邀请人收益改走邀请奖励金)
"invited_at": rel.created_at, "invited_at": rel.created_at,
# 是否已完成过一次比价(= 已发过邀请奖励金)。客户端据此:好友列表分"去提醒/邀请成功"、
# 在途列表只取未比价(is_compared=False)、算在途好友数与在途收益(未比价数×2元)。
"is_compared": bool(rel.compare_reward_granted),
}) })
has_more = offset + len(rows) < int(total) has_more = offset + len(rows) < int(total)
return items, int(total), has_more return items, int(total), has_more
+16 -1
View File
@@ -5,7 +5,7 @@ device_id 为空(老客户端 / 取不到 ANDROID_ID)一律按"未完成"处理,
""" """
from __future__ import annotations from __future__ import annotations
from sqlalchemy import select from sqlalchemy import delete, select
from sqlalchemy.exc import IntegrityError from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
@@ -33,3 +33,18 @@ def mark_completed(db: Session, *, user_id: int, device_id: str) -> None:
except IntegrityError: except IntegrityError:
# 并发 / 重复提交撞唯一约束:已有行即视为成功。 # 并发 / 重复提交撞唯一约束:已有行即视为成功。
db.rollback() 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
+3
View File
@@ -296,6 +296,9 @@ def daily_auto_exchange(db: Session) -> dict:
- **逐用户独立事务**:单个用户异常 rollback 不影响其他人;exchange_coins_to_cash 内部按用户 commit - **逐用户独立事务**:单个用户异常 rollback 不影响其他人;exchange_coins_to_cash 内部按用户 commit
返回统计 dict(scanned/converted/skipped_done/skipped_dust/failed/total_cents) 返回统计 dict(scanned/converted/skipped_done/skipped_dust/failed/total_cents)
""" """
# ⚠️「今天」写死北京时(rewards.cn_today() = datetime.now(CN_TZ).date(), CN_TZ=+8),与服务器/用户
# 时区无关(用户 2026-07-01 硬约束:兑换一律北京 0 点日切)。**禁止**改成 date.today() / 裸
# datetime.now().date()——那会跟随进程本地时区,服务器非 CST 时会在错误的"天"兑/重复兑。
today = rewards.cn_today() today = rewards.cn_today()
stats = { stats = {
"scanned": 0, "converted": 0, "skipped_done": 0, "scanned": 0, "converted": 0, "skipped_done": 0,
+1
View File
@@ -69,6 +69,7 @@ class InviteeItem(BaseModel):
avatar_url: str | None = None # 头像 URL;null = 前端画默认色块 avatar_url: str | None = None # 头像 URL;null = 前端画默认色块
coins: int # 这次邀请给我(邀请人)发的金币 coins: int # 这次邀请给我(邀请人)发的金币
invited_at: datetime # 邀请绑定时间(前端转"今天/3天前") invited_at: datetime # 邀请绑定时间(前端转"今天/3天前")
is_compared: bool = False # 该好友是否已完成过一次比价(好友列表分"去提醒/邀请成功";在途列表只取 False)
class InviteeListOut(BaseModel): class InviteeListOut(BaseModel):
+3
View File
@@ -19,6 +19,9 @@ Type=oneshot
User=root User=root
WorkingDirectory=/opt/shaguabijia-app-server WorkingDirectory=/opt/shaguabijia-app-server
Environment="PATH=/opt/shaguabijia-app-server/.venv/bin:/usr/bin:/bin" Environment="PATH=/opt/shaguabijia-app-server/.venv/bin:/usr/bin:/bin"
# 写死北京时:兑换的"当天/0 点"一律按北京时,不随服务器 OS 时区漂(用户 2026-07-01 硬约束)。
# 脚本内的日期判断本就走 rewards.cn_today()(CN_TZ=+8),这里再把进程 TZ 也钉成北京,双保险。
Environment="TZ=Asia/Shanghai"
EnvironmentFile=/opt/shaguabijia-app-server/.env EnvironmentFile=/opt/shaguabijia-app-server/.env
ExecStart=/opt/shaguabijia-app-server/.venv/bin/python -m scripts.daily_auto_exchange --once ExecStart=/opt/shaguabijia-app-server/.venv/bin/python -m scripts.daily_auto_exchange --once
SyslogIdentifier=daily-exchange SyslogIdentifier=daily-exchange
+4 -2
View File
@@ -4,8 +4,10 @@
Description=Run daily auto-exchange coins->cash at midnight Description=Run daily auto-exchange coins->cash at midnight
[Timer] [Timer]
# 每天 0 点跑。客户端文案已注明「可能存在延迟」,可按需改 00:05 错开整点扎堆 # 每天【北京时】0 点跑,写死时区(用户 2026-07-01 硬约束:兑换一律北京 0 点,不随服务器本地时区漂)
OnCalendar=*-*-* 00:00:00 # systemd OnCalendar 支持尾缀时区;不写时区会按服务器 OS 本地时区触发 → 服务器非 CST 时会在错误时刻兑。
# 客户端文案已注明「可能存在延迟」,可按需改 00:05 错开整点扎堆。
OnCalendar=*-*-* 00:00:00 Asia/Shanghai
# 服务器宕机/重启后,补跑错过的那一轮(而不是干等次日)。 # 服务器宕机/重启后,补跑错过的那一轮(而不是干等次日)。
Persistent=true Persistent=true
AccuracySec=1min AccuracySec=1min
+4 -4
View File
@@ -2,10 +2,10 @@
> 所属:比价记录组(前缀 `/api/v1/compare` | 鉴权:Bearer | [← 返回 API 索引](../README.md) > 所属:比价记录组(前缀 `/api/v1/compare` | 鉴权:Bearer | [← 返回 API 索引](../README.md)
比价 `done` 帧后,客户端用**带 JWT 的通道**上报一条比价结果,落 `comparison_record` 表,作为「我的比价记录」数据源 + 用户级行为画像。 比价 `done` 帧后上报一条比价结果,落 `comparison_record` 表,作为「我的比价记录」数据源 + 用户级行为画像。
> ⚠️ 与不鉴权的透传端点 [`/api/v1/price/step`](./compare-price-step.md) 不同:那是转发壳,本接口按用户维度落库,**必须鉴权** > ⚠️ **灰度定位(2026-07)**:写 `comparison_record` 现以透传壳 `compare.py` 的**后端 harvest** 为主(帧0 建 `running` 行 → done/finalize 落终态,新客户端不再 POST)。本接口降级为**老客户端兼容**通道,与 harvest 按 `trace_id` reconcile(**success 不被降级**);新版覆盖够高后可下线本 POST
> 本轮只做 server 端;客户端在 done 帧后调本接口的改动另起一轮(见 [待办与技术债.md](../guides/待办与技术债.md) P1 > 与软鉴权透传端点 [`/api/v1/price/step`](./compare-price-step.md) 不同:那是转发壳(顺带 harvest 落库),本接口按用户维度显式上报,**必须鉴权**
## 入参(JSON body ## 入参(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`(领券) | | `business_type` | string | ❌ | `food` | `food`(外卖,当前唯一接通) / `ecom`(电商) / `coupon`(领券) |
| `device_id` | string \| null | ❌ | null | 设备号 | | `device_id` | string \| null | ❌ | null | 设备号 |
| `store_name` | string \| null | ❌ | null | 店铺名(外卖,来自 calibration.result | | `store_name` | string \| null | ❌ | null | 店铺名(外卖,来自 calibration.result |
+7 -6
View File
@@ -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) > 模型 `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` 是「真正**下单成交**省了多少」。两表独立、互不喂数据。 > 与 `savings_record` 的区别:本表是「每一次**比价行为**的完整明细」(不省钱、甚至失败也记);`savings_record` 是「真正**下单成交**省了多少」。两表独立、互不喂数据。
> 与 [`price_observation`](./price_observation.md) / `store_mapping` 的区别:本表是**用户视角**(登录后按 `user_id` 存「我的比价记录」);后两张是 server 侧无条件沉淀的**平台/门店视角客观事实**(价格事实 / 跨平台店铺身份映射),与本表 `trace_id` 同源但不互相 join,各存各的视角。 > 与 [`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` 置空,不删本表)。 - **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 大盘/明细。 - **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) | | 列 | 类型 | 约束 / 默认 | 说明(取值 / join) |
|---|---|---|---| |---|---|---|---|
| `id` | Integer | PK, autoincrement | 被 `price_report.comparison_record_id` 引用 | | `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 | 设备号(多设备区分 / 与不鉴权期对账) | | `device_id` | String(64) | nullable | 设备号(多设备区分 / 与不鉴权期对账) |
| `business_type` | String(16) | NOT NULL, default `food`, index | 取值:`food`(当前唯一接通)/ `ecom` / `coupon` | | `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 | | `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_platform_id` / `_name` | String(32) | nullable | 源平台代号 / 中文名 |
| `source_package` | String(128) | nullable | 源平台 Android 包名 | | `source_package` | String(128) | nullable | 源平台 Android 包名 |
@@ -31,7 +31,7 @@
| `is_source_best` | Boolean | nullable | 源平台就是最便宜(= 这次没省到) | | `is_source_best` | Boolean | nullable | 源平台就是最便宜(= 这次没省到) |
| `store_name` | String(128) | nullable | 店铺名。**与 `savings_record.shop_name` 按字符串相等关联**,给本记录打「已下单」 | | `store_name` | String(128) | nullable | 店铺名。**与 `savings_record.shop_name` 按字符串相等关联**,给本记录打「已下单」 |
| `total_dish_count` / `skipped_dish_count` | Integer | nullable | 菜品总数 / 目标平台没找到被跳过数 | | `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 帧文案;成功=摘要,失败=具体原因(前端失败时当原因展示) | | `information` | String(256) | nullable | done 帧文案;成功=摘要,失败=具体原因(前端失败时当原因展示) |
| `items` | JSON(PG: JSONB) | NOT NULL, default [] | 下单菜品 `[{name, qty, specs?}]` | | `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}]` 多券明细 | | `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'` 计数。 - 被 `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` 原样存「元」。 - 4 个 JSON 列用 `JSON().with_variant(JSONB(),"postgresql")`(SQLite 退化 JSON)。结构化金额列存「分」,`comparison_results.price`/`coupon_saved` 原样存「元」。
+8 -9
View File
@@ -16,9 +16,9 @@
## 函数 / 异常 ## 函数 / 异常
| 函数 | 行为 | | 函数 | 行为 |
|---|---| |---|---|
| `send_code(phone) -> int` | 防刷检查(冷却 + 每日上限)→ `secrets` 生成 N 位码 → **预占**(冷却/计数/存码)→ mock 打日志 / real 调极光 → 失败**回滚预占**。返回距下次可发秒数 | | `send_code(phone) -> int` | 防刷检查(冷却)→ `secrets` 生成 N 位码 → **预占**(冷却/存码)→ mock 打日志 / real 调极光 → 失败**回滚预占**。返回距下次可发秒数 |
| `verify_code(phone, code) -> bool` | mock 放行任意 6 位;real 比对存码,匹配即作废,失败累计到上限作废 | | `verify_code(phone, code) -> bool` | mock 放行任意 6 位;real 比对存码,匹配即作废,失败累计到上限作废 |
| `SmsError(msg, status_code)` | `status_code` 决定 HTTP 码:过频/每日超限 **429**、供应商失败 **503**、号码无效 **400** | | `SmsError(msg, status_code)` | `status_code` 决定 HTTP 码:过频 **429**、供应商失败 **503**、号码无效 **400** |
## 配置 ## 配置
| 配置项 | 默认 | 说明 | | 配置项 | 默认 | 说明 |
@@ -30,17 +30,16 @@
| `SMS_SIGN_ID` | 31729 | 极光签名 ID | | `SMS_SIGN_ID` | 31729 | 极光签名 ID |
| `SMS_TEMPLATE_ID` | 1 | 极光模板 ID(变量名 `code`) | | `SMS_TEMPLATE_ID` | 1 | 极光模板 ID(变量名 `code`) |
| `SMS_CODE_LENGTH` | 6 | 验证码位数(本服务生成;前端 code 4-8 位兼容) | | `SMS_CODE_LENGTH` | 6 | 验证码位数(本服务生成;前端 code 4-8 位兼容) |
| `SMS_DAILY_LIMIT_PER_PHONE` | 10 | 单号每日发送上限(防刷 + 控费) |
| `SMS_MAX_VERIFY_ATTEMPTS` | 5 | 单码最多校验失败次数,超过作废(防爆破) | | `SMS_MAX_VERIFY_ATTEMPTS` | 5 | 单码最多校验失败次数,超过作废(防爆破) |
> **鉴权复用极光一键登录**:`/v1/messages``base64(JG_APP_KEY:JG_MASTER_SECRET)` 做 HTTP Basic Auth——短信与一键登录是**同一个极光应用**(同 AppKey)。**上线不需要额外凭证,只需 `SMS_MOCK=false`**(`JG_*` 一键登录已配)。 > **鉴权复用极光一键登录**:`/v1/messages``base64(JG_APP_KEY:JG_MASTER_SECRET)` 做 HTTP Basic Auth——短信与一键登录是**同一个极光应用**(同 AppKey)。**上线不需要额外凭证,只需 `SMS_MOCK=false`**(`JG_*` 一键登录已配)。
## 防刷(短信花钱 + `/sms/send` 在登录前无法 JWT 鉴权) ## 防刷(短信花钱 + `/sms/send` 在登录前无法 JWT 鉴权)
1. 单号 `SMS_SEND_INTERVAL_SEC` 冷却 > 2026-07-03 精简:登录风控只留「单号冷却 + 单设备频控」两道主策略(删单号每日上限、删登录纯 IP `rate_limit`);单码失败上限属验证码安全底线,保留。
2. 单号每日 `SMS_DAILY_LIMIT_PER_PHONE` 条上限 1. 单号 `SMS_SEND_INTERVAL_SEC` 冷却(单号维度)
3. 单 IP 频控:`/sms/send` `rate_limit(10,60)``/sms/login``rate_limit(20,60)` 2. 单设备(`device_id` + IP)每小时频控:`/sms/send` `SMS_SEND_MAX_PER_HOUR_PER_DEVICE`(5)、`/sms/login``SMS_LOGIN_MAX_PER_HOUR`(5)——堵「换号绕开单号冷却」+ 挡登录撞库,在 `app/api/v1/auth.py``enforce_rate_limit`
4. 单码校验失败 `SMS_MAX_VERIFY_ATTEMPTS` 次即作废 3. 单码校验失败 `SMS_MAX_VERIFY_ATTEMPTS` 次即作废 + 验过即作废(一次性)
5. 运维侧建议在极光控制台叠加:**IP 白名单**(只许服务器 IP)+ **防轰炸设置** 4. 运维侧建议在极光控制台叠加:**IP 白名单**(只许服务器 IP)+ **防轰炸设置**
## 极光错误码(节选,映射在 `_send_via_jiguang`) ## 极光错误码(节选,映射在 `_send_via_jiguang`)
| code | 含义 | 处理 | | code | 含义 | 处理 |
@@ -56,4 +55,4 @@
3. 真机发一条验证:收到短信 + 能登录 3. 真机发一条验证:收到短信 + 能登录
## 已知局限 ## 已知局限
**验证码存进程内存**:单 worker uvicorn 够用;重启丢码(用户重发即可);**多 worker / 多机不共享 → 冷却 / 每日上限 / 校验失效**,扩 worker 前迁移到 DB/Redis。见 [待办与技术债](../guides/待办与技术债.md)。 **验证码存进程内存**:单 worker uvicorn 够用;重启丢码(用户重发即可);**多 worker / 多机不共享 → 冷却 / 校验失效**,扩 worker 前迁移到 DB/Redis。见 [待办与技术债](../guides/待办与技术债.md)。
+1 -1
View File
@@ -169,7 +169,7 @@ POST /api/v1/auth/sms/login { phone, code } → 任意 6 位通过 → upsert
短信冷却表存进程内存(`--workers 1` 下够用,多 worker/重启即失效)。 短信冷却表存进程内存(`--workers 1` 下够用,多 worker/重启即失效)。
**real 模式(`SMS_MOCK=false`,生产)**:自定义验证码——本服务 `secrets` 生成 6 位码 → 极光 `/v1/messages` 只负责发 → 本地校验(一次性 / 防爆破),鉴权**复用 `JG_APP_KEY`/`JG_MASTER_SECRET`**(短信与一键登录同一极光应用,**上线只需 `SMS_MOCK=false`**)。防刷四层(单号冷却 + 单号每日上限 + 单 IP `rate_limit` + 单码失败次数)+ 错误码 429/503/400。详见 [integrations/sms](./integrations/sms.md)。 **real 模式(`SMS_MOCK=false`,生产)**:自定义验证码——本服务 `secrets` 生成 6 位码 → 极光 `/v1/messages` 只负责发 → 本地校验(一次性 / 防爆破),鉴权**复用 `JG_APP_KEY`/`JG_MASTER_SECRET`**(短信与一键登录同一极光应用,**上线只需 `SMS_MOCK=false`**)。防刷(单号冷却 + 单设备频控 + 单码失败次数;2026-07-03 精简删单号每日上限与登录纯 IP 限流)+ 错误码 429/503/400。详见 [integrations/sms](./integrations/sms.md)。
### 4.3 Token 与刷新 ### 4.3 Token 与刷新
+2 -22
View File
@@ -17,7 +17,6 @@ def _reset(phone: str) -> None:
"""清该号的进程内存状态,隔离 real 模式用例。""" """清该号的进程内存状态,隔离 real 模式用例。"""
sms._codes.pop(phone, None) sms._codes.pop(phone, None)
sms._last_sent.pop(phone, None) sms._last_sent.pop(phone, None)
sms._daily_count.pop(phone, None)
class _OkResp: class _OkResp:
@@ -76,7 +75,7 @@ def test_sms_send_too_frequent(client) -> None:
def test_sms_send_device_ip_rate_limit(client, monkeypatch) -> None: def test_sms_send_device_ip_rate_limit(client, monkeypatch) -> None:
"""发码防刷:同一设备(device_id) + 同一 IP 每小时最多 N 次发码,超出 429。 """发码防刷:同一设备(device_id) + 同一 IP 每小时最多 N 次发码,超出 429。
用不同手机号(绕开单号 60s 冷却)证明限流按设备封顶 换号绕开单号冷却/每日上限的洞 用不同手机号(绕开单号 60s 冷却)证明限流按设备封顶 换号绕开单号冷却的洞
conftest 默认关限流;本用例临时打开 + 调小阈值便于测 + 清计数隔离""" conftest 默认关限流;本用例临时打开 + 调小阈值便于测 + 清计数隔离"""
from app.api.v1 import auth from app.api.v1 import auth
from app.core import ratelimit from app.core import ratelimit
@@ -155,21 +154,6 @@ def test_phone_format_validation(client) -> None:
assert r.status_code == 422 assert r.status_code == 422
def test_sms_daily_limit(monkeypatch) -> None:
"""单号每日上限(send 的防刷逻辑 mock/real 都跑;函数级绕开 60s 冷却)。"""
phone = "13511135000"
_reset(phone)
limit = 3
monkeypatch.setattr(sms.settings, "SMS_DAILY_LIMIT_PER_PHONE", limit)
for _ in range(limit):
sms.send_code(phone)
sms._last_sent.pop(phone, None) # 绕开冷却,只测每日上限
with pytest.raises(sms.SmsError, match="上限"):
sms.send_code(phone)
# ============================ real 路径(不真发)============================ # ============================ real 路径(不真发)============================
def test_sms_real_send_calls_jiguang(monkeypatch) -> None: def test_sms_real_send_calls_jiguang(monkeypatch) -> None:
@@ -255,24 +239,20 @@ def test_sms_real_balance_error_keeps_cooldown(monkeypatch) -> None:
def test_sms_gc_purges_stale_only(monkeypatch) -> None: def test_sms_gc_purges_stale_only(monkeypatch) -> None:
"""GC 清过期码 / 旧冷却 / 隔日计数,但不动今天有效的(阈值设 0 强制每次扫)。""" """GC 清过期码 / 旧冷却,但不动今天有效的(阈值设 0 强制每次扫)。"""
monkeypatch.setattr(sms, "_GC_THRESHOLD", 0) monkeypatch.setattr(sms, "_GC_THRESHOLD", 0)
sms._codes.clear() sms._codes.clear()
sms._last_sent.clear() sms._last_sent.clear()
sms._daily_count.clear()
now = time.time() now = time.time()
sms._codes["stale"] = sms._CodeRecord(code="111111", expires_at=now - 1) sms._codes["stale"] = sms._CodeRecord(code="111111", expires_at=now - 1)
sms._codes["fresh"] = sms._CodeRecord(code="222222", expires_at=now + 999) sms._codes["fresh"] = sms._CodeRecord(code="222222", expires_at=now + 999)
sms._last_sent["old"] = now - 99999 sms._last_sent["old"] = now - 99999
sms._last_sent["recent"] = now sms._last_sent["recent"] = now
sms._daily_count["yesterday"] = ("2000-01-01", 3)
sms._daily_count["today"] = (sms._today(), 1)
sms._gc(now) sms._gc(now)
assert "stale" not in sms._codes and "fresh" in sms._codes assert "stale" not in sms._codes and "fresh" in sms._codes
assert "old" not in sms._last_sent and "recent" in sms._last_sent assert "old" not in sms._last_sent and "recent" in sms._last_sent
assert "yesterday" not in sms._daily_count and "today" in sms._daily_count
# ============================ 用户名 / 默认昵称 ============================ # ============================ 用户名 / 默认昵称 ============================
+258
View File
@@ -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 # 绑上了登录用户
+4 -2
View File
@@ -20,8 +20,10 @@ def _auth(token: str) -> dict[str, str]:
def _report_success(client, token: str, trace_id: str) -> None: def _report_success(client, token: str, trace_id: str) -> None:
"""上报一条成功比价(有非源有效价 → status 派生 success)。""" """上报一条成功比价(有非源有效价 → status 派生 success)。"""
# trace_id 现全局唯一(prod 由 app-server 签发 UUID);测试共用 session DB,故按 token
# (每用户不同)加前缀防跨用户/跨文件复用 "s-1" 撞车(见 comparison_record_trace_unique 迁移)。
payload = { payload = {
"trace_id": trace_id, "trace_id": f"{token[-16:]}:{trace_id}",
"business_type": "food", "business_type": "food",
"store_name": "测试店", "store_name": "测试店",
"source_platform_id": "taobao_flash", "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: def _report_failed(client, token: str, trace_id: str) -> None:
"""上报一条失败比价(只有源 → status 派生 failed,不计入解锁)。""" """上报一条失败比价(只有源 → status 派生 failed,不计入解锁)。"""
payload = { payload = {
"trace_id": trace_id, "trace_id": f"{token[-16:]}:{trace_id}",
"business_type": "food", "business_type": "food",
"comparison_results": [ "comparison_results": [
{"platform_id": "taobao_flash", "platform_name": "淘宝闪购", "price": 30.0, {"platform_id": "taobao_flash", "platform_name": "淘宝闪购", "price": 30.0,
+13 -6
View File
@@ -9,6 +9,7 @@ mock 掉对 pricebot 的 httpx 调用,验证(两个端点参数化同跑):
""" """
from __future__ import annotations from __future__ import annotations
import json
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
import httpx import httpx
@@ -41,7 +42,8 @@ def _stub_body() -> dict:
@pytest.mark.parametrize("path,upstream", ENDPOINTS) @pytest.mark.parametrize("path,upstream", ENDPOINTS)
def test_passes_body_through(client, path, upstream) -> None: 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 = { fake_resp = {
"success": True, "success": True,
"action": {"command": "wait", "params": {"duration_ms": 500}}, "action": {"command": "wait", "params": {"duration_ms": 500}},
@@ -49,20 +51,25 @@ def test_passes_body_through(client, path, upstream) -> None:
} }
captured: dict = {} 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["url"] = url
captured["json"] = json captured["content"] = content
mock_resp = MagicMock() mock_resp = MagicMock()
mock_resp.status_code = 200 mock_resp.status_code = 200
mock_resp.json = lambda: fake_resp mock_resp.json = lambda: dict(fake_resp) # 每次新副本(端点会 setdefault trace_id 进去)
return mock_resp return mock_resp
with patch.object(httpx.AsyncClient, "post", fake_post): with patch.object(httpx.AsyncClient, "post", fake_post):
r = client.post(path, json=_stub_body()) r = client.post(path, json=_stub_body())
assert r.status_code == 200, r.text assert r.status_code == 200, r.text
assert r.json() == fake_resp body = r.json()
assert captured["json"] == _stub_body() # body 原样转发(不鉴权,无需 token) 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 assert captured["url"].endswith(upstream) # /api/v1/xxx → /api/xxx
+25
View File
@@ -65,3 +65,28 @@ def test_onboarding_missing_device_id_treated_as_incomplete(client) -> None:
phone = "13800138003" phone = "13800138003"
data = _login(client, phone, device_id=None) data = _login(client, phone, device_id=None)
assert data["onboarding_completed"] is False 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