Compare commits

..

6 Commits

Author SHA1 Message Date
no_gen_mu 35f24084ae feat(user): 注销账号数据级联清理 + 后台删除端点
C 端与后台复用同一套 soft_delete_account:物理删除个人内容/行为/PII 表(比价/省钱/
签到/领券/任务/设备/价格上报/反馈/引导/埋点/看广告时长/邀请指纹)+ 清零余额 + 匿名化
user 行释放唯一约束;保留资金流水/邀请关系/广告幂等+对账表(对账/幂等/邀请人侧归属)。

- api/v1/user.py: DELETE /api/v1/user 加资金前置闸(未提现现金/邀请金 / 在审提现单 → 409)
- repositories/user.py: soft_delete_account 落地分类删除(补 ad_watch_log 埋点删除)
- repositories/wallet.py: 新增 get_invite_cash_balance_cents(邀请金 409 校验用)
- admin/routers/users.py: DELETE /admin/api/users/{id} 复用清理 + 审计,仅 super_admin
- admin/schemas/user.py: DeleteUserRequest(必填 reason 入审计)
- tests: 级联删除/余额归零/保留表存活 6 测 + admin 删除/审计/403/400/409 4 测

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-04 15:00:40 +08:00
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
33 changed files with 1334 additions and 196 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)
+12 -1
View File
@@ -4,6 +4,7 @@
复用 App 的 DB/models/repositories/integrations;鉴权独立(admin JWT,见 app/admin/security.py)。 复用 App 的 DB/models/repositories/integrations;鉴权独立(admin JWT,见 app/admin/security.py)。
现有 app.main:app 不 import 本模块,两进程互不影响。 现有 app.main:app 不 import 本模块,两进程互不影响。
""" """
from __future__ import annotations from __future__ import annotations
import logging import logging
@@ -12,6 +13,7 @@ from contextlib import asynccontextmanager
from fastapi import FastAPI from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import PlainTextResponse, RedirectResponse
from app.admin.routers.ad_audit import router as ad_audit_router from app.admin.routers.ad_audit import router as ad_audit_router
from app.admin.routers.ad_config import router as ad_config_router from app.admin.routers.ad_config import router as ad_config_router
@@ -25,12 +27,12 @@ from app.admin.routers.coupon_data import router as coupon_data_router
from app.admin.routers.cps import router as cps_router from app.admin.routers.cps import router as cps_router
from app.admin.routers.dashboard import router as dashboard_router from app.admin.routers.dashboard import router as dashboard_router
from app.admin.routers.device_liveness import router as device_liveness_router from app.admin.routers.device_liveness import router as device_liveness_router
from app.admin.routers.ops_stat_config import router as ops_stat_config_router
from app.admin.routers.event_logs import router as event_logs_router from app.admin.routers.event_logs import router as event_logs_router
from app.admin.routers.feedback import router as feedback_router from app.admin.routers.feedback import router as feedback_router
from app.admin.routers.feedback_qr import router as feedback_qr_router from app.admin.routers.feedback_qr import router as feedback_qr_router
from app.admin.routers.onboarding import router as onboarding_router from app.admin.routers.onboarding import router as onboarding_router
from app.admin.routers.ops_marquee_seed import router as ops_marquee_seed_router from app.admin.routers.ops_marquee_seed import router as ops_marquee_seed_router
from app.admin.routers.ops_stat_config import router as ops_stat_config_router
from app.admin.routers.price_report import router as price_report_router from app.admin.routers.price_report import router as price_report_router
from app.admin.routers.users import router as users_router from app.admin.routers.users import router as users_router
from app.admin.routers.wallet import router as wallet_router from app.admin.routers.wallet import router as wallet_router
@@ -84,6 +86,15 @@ def health() -> dict[str, str]:
return {"status": "ok", "service": "admin"} return {"status": "ok", "service": "admin"}
@admin_app.get("/", include_in_schema=False)
def root():
"""根路径落脚点:直接访问 :8771 时,非 prod 跳 API 文档,prod 下给一句纯文本说明
(prod 关了 docs_url,不能跳一个不存在的页面,也不暴露文档)。避免浏览器打开根路径吃 404。"""
if not settings.is_prod:
return RedirectResponse(url="/admin/docs")
return PlainTextResponse("shaguabijia admin API. See /admin/api/*.")
admin_app.include_router(auth_router) admin_app.include_router(auth_router)
admin_app.include_router(dashboard_router) admin_app.include_router(dashboard_router)
admin_app.include_router(device_liveness_router) admin_app.include_router(device_liveness_router)
+35
View File
@@ -13,6 +13,7 @@ from app.admin.schemas.common import CursorPage, OkResponse
from app.admin.schemas.user import ( from app.admin.schemas.user import (
AdminUserListItem, AdminUserListItem,
AdminUserOverview, AdminUserOverview,
DeleteUserRequest,
GrantCashRequest, GrantCashRequest,
GrantCoinsRequest, GrantCoinsRequest,
SetDebugTraceRequest, SetDebugTraceRequest,
@@ -132,6 +133,40 @@ def set_user_status(
return OkResponse() return OkResponse()
@router.delete("/{user_id}", response_model=OkResponse, summary="注销/删除账号(清理个人数据 + 匿名化,带审计)")
def delete_user(
user_id: int,
body: DeleteUserRequest,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("super_admin"))],
db: AdminDb,
) -> OkResponse:
"""后台删除账号:复用 C 端自助注销同一套清理逻辑(soft_delete_account)——物理删除个人内容/
行为/PII 表 + 清零余额 + 匿名化 user 行释放唯一约束,保留资金流水/邀请关系/广告幂等+对账表。
仅 super_admin 可操作(最高危、不可逆)。资金安全同 C 端:有未提现现金/邀请金 / 在审提现单 → 409 拒绝。
"""
user = user_repo.get_user_by_id(db, user_id)
if user is None:
raise HTTPException(status_code=404, detail="用户不存在")
if user.status == "deleted":
raise HTTPException(status_code=400, detail="账号已注销,请勿重复操作")
# 资金前置校验(与 C 端 delete_account 端点一致):余额一旦清零拿不回,有活钱先拦下。
if wallet_repo.get_cash_balance_cents(db, user_id) > 0:
raise HTTPException(status_code=409, detail="该账户还有未提现的现金余额,请先处理后再删除")
if wallet_repo.get_invite_cash_balance_cents(db, user_id) > 0:
raise HTTPException(status_code=409, detail="该账户还有未提现的邀请奖励金,请先处理后再删除")
if wallet_repo.has_reviewing_withdraw(db, user_id):
raise HTTPException(status_code=409, detail="该账户有提现正在审核中,请等审核完成后再删除")
# 先写审计(commit=False,挂进 session),再由 soft_delete_account 做清理 + 最终 commit —— 一次 commit
# 同时落审计行与清理结果,保持"改了就有痕、有痕就改了"原子(soft_delete_account 内部 commit 会一并 flush)。
write_audit(
db, admin, action="user.delete", target_type="user", target_id=user_id,
detail={"reason": body.reason, "phone": user.phone}, ip=get_client_ip(request), commit=False,
)
user_repo.soft_delete_account(db, user)
return OkResponse()
@router.post("/{user_id}/debug-trace", response_model=OkResponse, summary="开关调试链接权限") @router.post("/{user_id}/debug-trace", response_model=OkResponse, summary="开关调试链接权限")
def set_user_debug_trace( def set_user_debug_trace(
user_id: int, user_id: int,
+6
View File
@@ -115,3 +115,9 @@ class SetUserStatusRequest(BaseModel):
class SetDebugTraceRequest(BaseModel): class SetDebugTraceRequest(BaseModel):
enabled: bool = Field(..., description="是否给该用户开「复制调试链接」权限") enabled: bool = Field(..., description="是否给该用户开「复制调试链接」权限")
class DeleteUserRequest(BaseModel):
reason: str = Field(..., min_length=1, max_length=128, description="删除原因(必填,入审计)")
_v_reason = field_validator("reason")(_strip_reason)
+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,
+19 -3
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,
@@ -86,15 +98,19 @@ def onboarding_status(
return OnboardingStatusResponse(completed=completed) return OnboardingStatusResponse(completed=completed)
@router.delete("", response_model=OkResponse, summary="注销账号(软删除)") @router.delete("", response_model=OkResponse, summary="注销账号(清理个人数据 + 匿名化)")
def delete_account(user: CurrentUser, db: DbSession) -> OkResponse: def delete_account(user: CurrentUser, db: DbSession) -> OkResponse:
# 资金前置校验:注销是软删 + 匿名化,余额一旦留在 deleted 行就锁死(既退不出、新号也 # 资金前置校验:注销会物理清理个人数据 + 清零余额 + 匿名化,余额一旦清零就再也拿不回。
# 拿不回)。有可提现现金 / 在审提现单 → 拒绝注销,引导用户先把钱处理掉。 # 有可提现现金 / 邀请奖励金 / 在审提现单 → 拒绝注销,引导用户先把钱处理掉。
# (pending 提现转账已在途、对账 worker 不依赖 user.status 照常到账,故不拦 pending。) # (pending 提现转账已在途、对账 worker 不依赖 user.status 照常到账,故不拦 pending。)
if wallet_repo.get_cash_balance_cents(db, user.id) > 0: if wallet_repo.get_cash_balance_cents(db, user.id) > 0:
raise HTTPException( raise HTTPException(
status_code=409, detail="账户还有未提现的现金余额,请先提现后再注销" status_code=409, detail="账户还有未提现的现金余额,请先提现后再注销"
) )
if wallet_repo.get_invite_cash_balance_cents(db, user.id) > 0:
raise HTTPException(
status_code=409, detail="账户还有未提现的邀请奖励金,请先提现后再注销"
)
if wallet_repo.has_reviewing_withdraw(db, user.id): if wallet_repo.has_reviewing_withdraw(db, user.id):
raise HTTPException( raise HTTPException(
status_code=409, detail="有提现正在审核中,请等审核完成后再注销" status_code=409, detail="有提现正在审核中,请等审核完成后再注销"
-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
+71 -8
View File
@@ -8,10 +8,29 @@ import secrets
import string import string
from datetime import datetime, timezone from datetime import datetime, timezone
from sqlalchemy import select from sqlalchemy import delete, select
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from app.models.ad_watch_log import AdWatchLog
from app.models.analytics_event import AnalyticsEvent
from app.models.comparison import ComparisonRecord
from app.models.comparison_milestone import ComparisonMilestoneClaim
from app.models.coupon_state import (
CouponClaimRecord,
CouponDailyCompletion,
CouponPromptEngagement,
CouponSession,
)
from app.models.device import DeviceLiveness
from app.models.feedback import Feedback
from app.models.invite_fingerprint import InviteFingerprint
from app.models.onboarding import OnboardingCompletion
from app.models.price_report import PriceReport
from app.models.savings import SavingsRecord
from app.models.signin import SigninBoostRecord, SigninRecord
from app.models.task import UserTask
from app.models.user import User from app.models.user import User
from app.models.wallet import CoinAccount
# ===== 创建时分配的标识:用户名(对外展示账号 ID)+ 默认昵称 ===== # ===== 创建时分配的标识:用户名(对外展示账号 ID)+ 默认昵称 =====
@@ -114,19 +133,63 @@ def set_avatar_url(db: Session, user: User, *, avatar_url: str) -> User:
def soft_delete_account(db: Session, user: User) -> None: def soft_delete_account(db: Session, user: User) -> None:
"""注销账号:软删除 + 匿名化 + 释放唯一约束/外部绑定 """注销账号:物理删除个人数据 + 清零余额 + 匿名化保留财务/关系 + 释放唯一约束。
phone 改成 `deleted_<id>` 释放手机号唯一约束,允许同号码重新注册成全新账号; 删除策略("删了是否破坏对账 / 影响别人 / 是否 PII"分类):
清空 PII(昵称/头像),保留行用于审计status=deleted 后将无法再登录该行 物理删除(个人内容/成绩/行为/PII,只涉本人删了不碰账目也不影响别人):比价/
(登录接口校验 status==active) 省钱/签到/领券/任务/里程碑领取/埋点/引导标记/设备活跃/价格上报/反馈/看广告时长日志
(ad_watch_log:前端上报时长不可信非发奖依据,纯行为埋点),以及
邀请指纹( IP/机型/UA PII, inviter_user_id = 本人作为邀请人留的)
清零保留:coin_account 的金币/现金/邀请金/累计收益归零(现金/邀请金已由端点
前置校验=0),行保留 避免碰资金流水外键,也留作审计
保留 + 随本行匿名化与身份解绑(动了会破坏对账/在途到账/发奖幂等/误伤邀请人):
三本资金流水(coin/cash/invite_cash_transaction)提现单(pending 在途照常到账)
微信转账授权(worker 对账 pending 要用)广告发奖幂等记录(ad_reward trans_id
ad_feed_reward_record client_event_id 唯一键,删了幂等键会致回调/重放重复发奖)
广告 eCPM 收益记录(ad_ecpm_record 是内部收益统计/对账口径,同资金流水按对账保留)
邀请关系(inviter/invitee 双向不含 PII,删了会破坏邀请人侧归属)这些表的 user_id
指向已匿名化的本行,身份已解绑
注销必须连同释放 user 其余唯一约束/外部身份,否则 deleted 行永久占着槽位, 匿名化必须连同释放 user 其余唯一约束/外部身份,否则 deleted 行永久占着槽位:
对应资源再也无法被复用:
- wechat_openid: 不清 这个微信再也无法被任何账号绑定(提现通道彻底锁死); - wechat_openid: 不清 这个微信再也无法被任何账号绑定(提现通道彻底锁死);
- invite_code: 不清 邀请码槽位被占 + 裸查仍命中死号(发奖已被 bind - invite_code: 不清 邀请码槽位被占 + 裸查仍命中死号(发奖已被 bind
status==active 校验兜住,清掉更干净) status==active 校验兜住,清掉更干净)
资金安全(未提现现金 / 在审提现单)由调用方 delete_account 端点前置校验,这里只匿名化 资金安全(未提现现金/邀请金 / 在审提现单)由调用方 delete_account 端点前置校验
""" """
uid = user.id
# 1) 物理删除个人内容/行为/成绩/PII 数据(不碰资金流水/账户/邀请关系/广告幂等)
personal_deletions = [
delete(ComparisonRecord).where(ComparisonRecord.user_id == uid),
delete(SavingsRecord).where(SavingsRecord.user_id == uid),
delete(SigninRecord).where(SigninRecord.user_id == uid),
delete(SigninBoostRecord).where(SigninBoostRecord.user_id == uid),
delete(CouponClaimRecord).where(CouponClaimRecord.user_id == uid),
delete(CouponDailyCompletion).where(CouponDailyCompletion.user_id == uid),
delete(CouponPromptEngagement).where(CouponPromptEngagement.user_id == uid),
delete(CouponSession).where(CouponSession.user_id == uid),
delete(UserTask).where(UserTask.user_id == uid),
delete(ComparisonMilestoneClaim).where(ComparisonMilestoneClaim.user_id == uid),
delete(DeviceLiveness).where(DeviceLiveness.user_id == uid),
delete(PriceReport).where(PriceReport.user_id == uid),
delete(Feedback).where(Feedback.user_id == uid),
delete(OnboardingCompletion).where(OnboardingCompletion.user_id == uid),
delete(AnalyticsEvent).where(AnalyticsEvent.user_id == uid),
delete(AdWatchLog).where(AdWatchLog.user_id == uid),
delete(InviteFingerprint).where(InviteFingerprint.inviter_user_id == uid),
]
for stmt in personal_deletions:
db.execute(stmt)
# 2) 清零余额账户(现金/邀请金已前置=0;金币/累计收益一并归零,行保留不碰流水外键)
acc = db.get(CoinAccount, uid)
if acc is not None:
acc.coin_balance = 0
acc.cash_balance_cents = 0
acc.invite_cash_balance_cents = 0
acc.total_coin_earned = 0
# 3) 匿名化本行 + 释放唯一约束/外部身份(保留行:审计锚 + 资金流水/邀请关系外键不断)
user.status = "deleted" user.status = "deleted"
user.phone = f"deleted_{user.id}" user.phone = f"deleted_{user.id}"
user.nickname = None user.nickname = None
+13
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,
@@ -380,6 +383,16 @@ def get_cash_balance_cents(db: Session, user_id: int) -> int:
return val or 0 return val or 0
def get_invite_cash_balance_cents(db: Session, user_id: int) -> int:
"""只读查邀请奖励金余额(分)。账户不存在返回 0,**不创建账户**(同 get_cash_balance_cents)。
注销前置校验用:邀请奖励金与金币兑换现金物理隔离,注销会把账户余额一并清零,若不单独
校验,用户没提现的返佣金会被直接清零吞掉"""
val = db.execute(
select(CoinAccount.invite_cash_balance_cents).where(CoinAccount.user_id == user_id)
).scalar_one_or_none()
return val or 0
def has_reviewing_withdraw(db: Session, user_id: int) -> bool: def has_reviewing_withdraw(db: Session, user_id: int) -> bool:
"""用户是否有「待审核(reviewing)」的提现单。 """用户是否有「待审核(reviewing)」的提现单。
+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 与刷新
+97
View File
@@ -476,3 +476,100 @@ def test_withdraw_reconcile(admin_client: TestClient, finance_token: str, monkey
) )
assert r.status_code == 200, r.text assert r.status_code == 200, r.text
assert "checked" in r.json() and "resolved" in r.json() assert "checked" in r.json() and "resolved" in r.json()
# ===== 删除账号(super_admin,复用 soft_delete_account + 审计) =====
def test_admin_delete_user_purges_and_audits(
admin_client: TestClient, super_token: str
) -> None:
"""super_admin 删除:user 行匿名化 + 个人数据被清 + 落审计(action=user.delete,含 reason)。"""
uid = _seed_user("13900000031")
db = SessionLocal()
try:
db.add(Feedback(user_id=uid, content="待清理", contact="wx", status="pending"))
db.commit()
finally:
db.close()
r = admin_client.request(
"DELETE", f"/admin/api/users/{uid}",
json={"reason": "违规账号"}, headers=_auth(super_token),
)
assert r.status_code == 200, r.text
db = SessionLocal()
try:
u = db.get(User, uid)
assert u is not None and u.status == "deleted"
assert u.phone == f"deleted_{uid}"
# 个人表被清(复用 C 端 soft_delete_account 同一套级联删除)
fbs = db.execute(
select(Feedback).where(Feedback.user_id == uid)
).scalars().all()
assert fbs == []
# 审计:业务 + 审计同一 commit 一起落(改了就有痕)
logs = db.execute(
select(AdminAuditLog).where(
AdminAuditLog.action == "user.delete", AdminAuditLog.target_id == str(uid)
)
).scalars().all()
assert len(logs) == 1 and logs[0].detail["reason"] == "违规账号"
finally:
db.close()
def test_admin_delete_user_forbidden_for_operator(
admin_client: TestClient, operator_token: str
) -> None:
"""删除是最高危操作:operator/finance 都不可,仅 super_admin。且拒绝后账号不被删。"""
uid = _seed_user("13900000032")
r = admin_client.request(
"DELETE", f"/admin/api/users/{uid}",
json={"reason": "x"}, headers=_auth(operator_token),
)
assert r.status_code == 403
db = SessionLocal()
try:
assert db.get(User, uid).status == "active" # 未被删
finally:
db.close()
def test_admin_delete_user_blocked_when_cash_positive(
admin_client: TestClient, super_token: str, finance_token: str
) -> None:
"""有未提现现金 → 409 拒绝(同 C 端资金前置闸),账号不被删。"""
uid = _seed_user("13900000033")
# 用 finance 给该用户灌现金
admin_client.post(
f"/admin/api/users/{uid}/cash", json={"amount_cents": 100, "reason": ""},
headers=_auth(finance_token),
)
r = admin_client.request(
"DELETE", f"/admin/api/users/{uid}",
json={"reason": "x"}, headers=_auth(super_token),
)
assert r.status_code == 409, r.text
assert "现金" in r.json()["detail"]
db = SessionLocal()
try:
assert db.get(User, uid).status == "active"
finally:
db.close()
def test_admin_delete_user_rejects_already_deleted(
admin_client: TestClient, super_token: str
) -> None:
uid = _seed_user("13900000034")
r = admin_client.request(
"DELETE", f"/admin/api/users/{uid}",
json={"reason": "首删"}, headers=_auth(super_token),
)
assert r.status_code == 200, r.text
# 再删已注销账号 → 400
r = admin_client.request(
"DELETE", f"/admin/api/users/{uid}",
json={"reason": "重复"}, headers=_auth(super_token),
)
assert r.status_code == 400
+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
+133 -2
View File
@@ -1,14 +1,18 @@
"""注销账号(DELETE /api/v1/user):软删 + 释放唯一约束 + 资金前置校验。 """注销账号(DELETE /api/v1/user):数据级联清理 + 软删 + 释放唯一约束 + 资金前置校验。
覆盖: 覆盖:
- 干净注销:status=deleted + phone 匿名化 + wechat_openid/invite_code 等唯一槽全部释放 - 干净注销:status=deleted + phone 匿名化 + wechat_openid/invite_code 等唯一槽全部释放
- 回归原始 bug:注销释放 openid ,别的账号能绑同一个微信 - 回归原始 bug:注销释放 openid ,别的账号能绑同一个微信
- 资金前置闸:有可提现现金 / 在审提现单 409 拒绝注销,账号不被删 - 资金前置闸:有可提现现金 / 在审提现单 409 拒绝注销,账号不被删
- 注销后旧 token 即时失效(status==active ) - 注销后旧 token 即时失效(status==active )
- 级联清理:个人内容/成绩/行为/PII 表被物理删空余额清零;财务流水/邀请关系/广告幂等
+对账表按策略保留(soft_delete_account 的分类落地断言)
""" """
from __future__ import annotations from __future__ import annotations
from sqlalchemy import select from datetime import date, datetime, timezone
from sqlalchemy import func, select
from app.db.session import SessionLocal from app.db.session import SessionLocal
from app.models.user import User from app.models.user import User
@@ -139,3 +143,130 @@ def test_deleted_user_old_token_is_rejected(client) -> None:
# 注销后旧 token 立即失效(get_current_user 校验 status==active) # 注销后旧 token 立即失效(get_current_user 校验 status==active)
r = client.get("/api/v1/user/onboarding/status", params={"device_id": "d"}, headers=_auth(token)) r = client.get("/api/v1/user/onboarding/status", params={"device_id": "d"}, headers=_auth(token))
assert r.status_code == 401, r.text assert r.status_code == 401, r.text
# ===== 级联清理:个人表删空 / 余额清零 / 保留表存活 =====
def _count(model, col, uid: int) -> int:
db = SessionLocal()
try:
return db.execute(
select(func.count()).select_from(model).where(col == uid)
).scalar_one()
finally:
db.close()
def test_delete_purges_personal_data_zeroes_balance_keeps_ledgers(client) -> None:
"""注销后:① 个人内容/行为/PII 表按 user_id 被物理删空;② 金币等余额清零;
资金流水 / 邀请关系 / 广告幂等+对账表保留(soft_delete_account 分类策略的端到端断言)"""
from app.models.ad_ecpm import AdEcpmRecord
from app.models.ad_feed_reward import AdFeedRewardRecord
from app.models.ad_watch_log import AdWatchLog
from app.models.analytics_event import AnalyticsEvent
from app.models.comparison import ComparisonRecord
from app.models.comparison_milestone import ComparisonMilestoneClaim
from app.models.coupon_state import (
CouponClaimRecord,
CouponDailyCompletion,
CouponPromptEngagement,
CouponSession,
)
from app.models.device import DeviceLiveness
from app.models.feedback import Feedback
from app.models.invite import InviteRelation
from app.models.invite_fingerprint import InviteFingerprint
from app.models.onboarding import OnboardingCompletion
from app.models.price_report import PriceReport
from app.models.savings import SavingsRecord
from app.models.signin import SigninBoostRecord, SigninRecord
from app.models.task import UserTask
from app.models.wallet import CoinTransaction
phone = "13900000007"
token = _login(client, phone)
client.get("/api/v1/wallet/account", headers=_auth(token)) # 建 coin_account
uid = _get_user(phone).id
db = SessionLocal()
try:
# 余额:给金币 + 累计收益(现金/邀请金保持 0,否则命中 409 前置闸)
acc = db.get(CoinAccount, uid)
acc.coin_balance = 500
acc.total_coin_earned = 500
# 每张"应删"表各灌 1 行(user_id 维度;指纹按 inviter_user_id)
db.add_all([
ComparisonRecord(user_id=uid, trace_id="del_test_comparison"),
SavingsRecord(user_id=uid, order_amount_cents=1, saved_amount_cents=1),
SigninRecord(user_id=uid, signin_date=date(2026, 7, 4), cycle_day=1, streak=1, coin_awarded=1),
SigninBoostRecord(user_id=uid, signin_date=date(2026, 7, 4), coin_awarded=1),
CouponClaimRecord(user_id=uid, device_id="del_cc", coupon_id="c1", claim_date=date(2026, 7, 4), status="success"),
CouponDailyCompletion(user_id=uid, device_id="del_cd", complete_date=date(2026, 7, 4)),
CouponPromptEngagement(user_id=uid, device_id="del_ce", engage_date=date(2026, 7, 4), engage_type="shown"),
CouponSession(user_id=uid, trace_id="del_test_couponsession", device_id="d1", status="started",
started_at=datetime(2026, 7, 4, tzinfo=timezone.utc), started_date=date(2026, 7, 4)),
UserTask(user_id=uid, task_key="del_test_usertask"),
ComparisonMilestoneClaim(user_id=uid, milestone=1),
DeviceLiveness(user_id=uid, device_id="del_test_dl"),
PriceReport(user_id=uid, reported_platform_id="p1", reported_platform_name="x", reported_price_cents=1),
Feedback(user_id=uid, content="x", contact="x"),
OnboardingCompletion(user_id=uid, device_id="del_test_ob"),
AnalyticsEvent(user_id=uid, event="x", device_id="del_test_an", client_ts=1),
AdWatchLog(user_id=uid, watch_date="2026-07-04"),
InviteFingerprint(inviter_user_id=uid, ip="1.2.3.4"),
])
# 每张"应保留"表各灌 1 行
db.add_all([
CoinTransaction(user_id=uid, amount=1, balance_after=1, biz_type="signin"),
InviteRelation(inviter_user_id=uid, invitee_user_id=999999),
AdEcpmRecord(user_id=uid, ad_type="reward_video", ecpm_raw="100", report_date="2026-07-04"),
AdFeedRewardRecord(user_id=uid, client_event_id="del_test_afr", ecpm_raw="100", reward_date="2026-07-04"),
])
db.commit()
finally:
db.close()
# 灌完先确认确实各有 1 行(否则"删空"断言可能因根本没灌进去而假过)
deleted_specs = [
(ComparisonRecord, ComparisonRecord.user_id),
(SavingsRecord, SavingsRecord.user_id),
(SigninRecord, SigninRecord.user_id),
(SigninBoostRecord, SigninBoostRecord.user_id),
(CouponClaimRecord, CouponClaimRecord.user_id),
(CouponDailyCompletion, CouponDailyCompletion.user_id),
(CouponPromptEngagement, CouponPromptEngagement.user_id),
(CouponSession, CouponSession.user_id),
(UserTask, UserTask.user_id),
(ComparisonMilestoneClaim, ComparisonMilestoneClaim.user_id),
(DeviceLiveness, DeviceLiveness.user_id),
(PriceReport, PriceReport.user_id),
(Feedback, Feedback.user_id),
(OnboardingCompletion, OnboardingCompletion.user_id),
(AnalyticsEvent, AnalyticsEvent.user_id),
(AdWatchLog, AdWatchLog.user_id),
(InviteFingerprint, InviteFingerprint.inviter_user_id),
]
for model, col in deleted_specs:
assert _count(model, col, uid) == 1, f"seed missing for {model.__name__}"
r = client.delete("/api/v1/user", headers=_auth(token))
assert r.status_code == 200, r.text
# ① 应删表全部删空
for model, col in deleted_specs:
assert _count(model, col, uid) == 0, f"{model.__name__} 未被删空(残留 PII/行为)"
# ② 余额清零(行保留,不碰流水外键)
acc = SessionLocal().get(CoinAccount, uid)
assert acc is not None
assert acc.coin_balance == 0
assert acc.total_coin_earned == 0
assert acc.cash_balance_cents == 0
assert acc.invite_cash_balance_cents == 0
# ③ 保留表存活(资金流水 / 邀请关系 / 广告幂等+对账)
assert _count(CoinTransaction, CoinTransaction.user_id, uid) == 1
assert _count(InviteRelation, InviteRelation.inviter_user_id, uid) == 1
assert _count(AdEcpmRecord, AdEcpmRecord.user_id, uid) == 1
assert _count(AdFeedRewardRecord, AdFeedRewardRecord.user_id, uid) == 1
+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