Compare commits
6 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 35f24084ae | |||
| 3eb439b5ae | |||
| 6b10e866e9 | |||
| 13b143cd97 | |||
| 93f052535e | |||
| a563c1ca4b |
@@ -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
@@ -4,6 +4,7 @@
|
||||
复用 App 的 DB/models/repositories/integrations;鉴权独立(admin JWT,见 app/admin/security.py)。
|
||||
现有 app.main:app 不 import 本模块,两进程互不影响。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
@@ -12,6 +13,7 @@ from contextlib import asynccontextmanager
|
||||
|
||||
from fastapi import FastAPI
|
||||
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_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.dashboard import router as dashboard_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.feedback import router as feedback_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.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.users import router as users_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"}
|
||||
|
||||
|
||||
@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(dashboard_router)
|
||||
admin_app.include_router(device_liveness_router)
|
||||
|
||||
@@ -13,6 +13,7 @@ from app.admin.schemas.common import CursorPage, OkResponse
|
||||
from app.admin.schemas.user import (
|
||||
AdminUserListItem,
|
||||
AdminUserOverview,
|
||||
DeleteUserRequest,
|
||||
GrantCashRequest,
|
||||
GrantCoinsRequest,
|
||||
SetDebugTraceRequest,
|
||||
@@ -132,6 +133,40 @@ def set_user_status(
|
||||
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="开关调试链接权限")
|
||||
def set_user_debug_trace(
|
||||
user_id: int,
|
||||
|
||||
@@ -115,3 +115,9 @@ class SetUserStatusRequest(BaseModel):
|
||||
|
||||
class SetDebugTraceRequest(BaseModel):
|
||||
enabled: bool = Field(..., description="是否给该用户开「复制调试链接」权限")
|
||||
|
||||
|
||||
class DeleteUserRequest(BaseModel):
|
||||
reason: str = Field(..., min_length=1, max_length=128, description="删除原因(必填,入审计)")
|
||||
|
||||
_v_reason = field_validator("reason")(_strip_reason)
|
||||
|
||||
@@ -54,5 +54,29 @@ def get_current_user(
|
||||
return user
|
||||
|
||||
|
||||
def get_current_user_optional(
|
||||
credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(_bearer)],
|
||||
db: Annotated[Session, Depends(get_db)],
|
||||
) -> User | None:
|
||||
"""软鉴权:有合法 Bearer 就返回 user,否则(无 header / 无效 token / 用户禁用)一律返回
|
||||
None,**不 raise**。
|
||||
|
||||
比价 step / finalize 灰度期用:新客户端带 JWT → 拿到 user_id 绑定 harvest 行;
|
||||
老客户端(不带 JWT)→ None,harvest 行 user_id 暂空,由其后续 /compare/record 上报补齐。
|
||||
等新版覆盖率够高,再把这几条端点从软鉴权收紧成硬 get_current_user。
|
||||
"""
|
||||
if credentials is None or credentials.scheme.lower() != "bearer":
|
||||
return None
|
||||
try:
|
||||
payload = decode_token(credentials.credentials, expected_type="access")
|
||||
user = db.get(User, int(payload["sub"]))
|
||||
except (TokenError, KeyError, ValueError, TypeError):
|
||||
return None
|
||||
if user is None or user.status != "active":
|
||||
return None
|
||||
return user
|
||||
|
||||
|
||||
CurrentUser = Annotated[User, Depends(get_current_user)]
|
||||
OptionalUser = Annotated[User | None, Depends(get_current_user_optional)]
|
||||
DbSession = Annotated[Session, Depends(get_db)]
|
||||
|
||||
+6
-7
@@ -12,11 +12,11 @@ from __future__ import annotations
|
||||
|
||||
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.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.integrations.jiguang import JiguangError, mask_phone, verify_and_get_phone
|
||||
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 每小时最多的登录尝试次数(成功/失败都计)。
|
||||
SMS_LOGIN_MAX_PER_HOUR = 5
|
||||
# 发码防刷:同一设备(device_id) + 同一 IP 每小时最多的发码次数。
|
||||
# 堵「换手机号绕开单号 60s 冷却 / 单号每日上限」的洞 —— 那两道是单号维度,一机换号能绕开。
|
||||
# 堵「换手机号绕开单号 60s 冷却」的洞 —— 冷却是单号维度,一机换号能绕开。
|
||||
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)
|
||||
|
||||
# 防刷:同一设备(device_id) + 同一 IP 每小时最多 SMS_SEND_MAX_PER_HOUR_PER_DEVICE 次发码。
|
||||
# 补「换手机号绕开单号 60s 冷却 / 单号每日上限」的洞(那两道是单号维度,一机换号能绕);设备维度按机器封顶,
|
||||
# 挡短信轰炸/烧钱。放在真发(send_code)之前 → 超限直接拦下、不真发短信。与路由上 IP 维度(10次/分钟)互补。
|
||||
# 补「换手机号绕开单号 60s 冷却」的洞(冷却是单号维度,一机换号能绕);设备维度按机器封顶,
|
||||
# 挡短信轰炸/烧钱。放在真发(send_code)之前 → 超限直接拦下、不真发短信。
|
||||
enforce_rate_limit(
|
||||
request,
|
||||
scope="sms-send-device",
|
||||
@@ -125,7 +125,6 @@ def sms_send(req: SmsSendRequest, request: Request) -> SmsSendResponse:
|
||||
"/sms/login",
|
||||
response_model=TokenWithUser,
|
||||
summary="手机号+验证码登录",
|
||||
dependencies=[Depends(rate_limit(20, 60, "sms-login"))], # 防撞库爆破(另有单码失败次数上限)
|
||||
)
|
||||
def sms_login(req: SmsLoginRequest, request: Request, db: DbSession) -> TokenWithUser:
|
||||
# 测试账号:免验证码登录 + 每日上限 + 每次都走新手引导(详见 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)
|
||||
|
||||
# 防刷:同一设备(device_id) + 同一 IP 每小时最多 SMS_LOGIN_MAX_PER_HOUR 次登录尝试。放在验证码校验
|
||||
# **之前** → 输错验证码的失败尝试也计数,才挡得住撞库/爆破。与路由上 IP 维度的 sms-login 限流(同 IP)互补。
|
||||
# **之前** → 输错验证码的失败尝试也计数,才挡得住撞库/爆破(另有单码失败 SMS_MAX_VERIFY_ATTEMPTS 次即作废兜底)。
|
||||
# ⚠️ 按设备而非手机号 → 一台机器换不同手机号刷登录也受限(防一机狂登多号);device_id 空(老客户端)时
|
||||
# 退化为该 IP 下所有空设备聚一桶,仍受限。
|
||||
enforce_rate_limit(
|
||||
|
||||
+203
-60
@@ -1,49 +1,114 @@
|
||||
"""外卖比价业务透传端点。
|
||||
"""外卖比价业务透传端点 + 后端 harvest 落库。
|
||||
|
||||
把客户端的 POST /api/v1/intent/recognize 和 /api/v1/price/step 请求原样转发给
|
||||
pricebot-backend 的 /api/intent/recognize 和 /api/price/step。MVP 阶段**不鉴权**
|
||||
(同 coupon/step,见 docs/待办与技术债.md P1:device_id 透传区分设备,待补 JWT +
|
||||
device_id↔user_id 绑定后才能做用户级画像)。
|
||||
把客户端 POST /api/v1/intent/* 和 /api/v1/price/step 转发给 pricebot,同时**由 app-server
|
||||
在透传里直接落库比价记录**(不再靠客户端 POST /compare/record):
|
||||
- 帧0(客户端首帧不带 trace_id)→ app-server 用 uuid 签发 trace_id、注入转发 body、回给
|
||||
客户端;并按 trace_id 建 running 行(harvest_running)。
|
||||
- 最终 done 帧 → 更新成 success/failed + 结果(harvest_done)。发奖不在这里:#113 已把
|
||||
邀请奖口径从「比价」移到「实际下单」(order.py)。
|
||||
- /trace/finalize(用户终止/Phase1 未识别,无 done)→ 更新成 cancelled/failed
|
||||
(harvest_abort,**不降级 success**)。
|
||||
trace_url 从 pricebot 响应**顶层 trace_url** 取(pricebot 每帧都带,见其 goal_engine.process_step)。
|
||||
|
||||
- Phase 1 /intent/recognize:从源平台(淘宝闪购/美团/京东外卖)购物车页识别店名+菜品+
|
||||
价格,一次性。
|
||||
- Phase 2 /price/step:多轮循环,在目标平台复现订单读到手价,直到 done。
|
||||
鉴权:软鉴权(OptionalUser)——新客户端带 JWT → 绑 user_id;老客户端不带 → user_id 暂空,
|
||||
由其后续 /compare/record 上报补(灰度期两条写路径按 trace_id reconcile,success 不被降级)。
|
||||
|
||||
真正的目标驱动比价逻辑在 pricebot-backend(GoalEngine,另一个 repo),本接口只是透传壳。
|
||||
电商(ecom)那两个端点 food MVP 暂不需要,以后放开 scene 时再加 /ecom/intent/recognize
|
||||
和 /ecom/step 两行即可。
|
||||
|
||||
pricebot 协议文档:
|
||||
pricebot-backend/docs/main/02_api_protocol.md
|
||||
真正的目标驱动比价逻辑在 pricebot-backend(GoalEngine,另一个 repo),本接口是透传壳 + 落库。
|
||||
pricebot 协议文档: pricebot-backend/docs/main/02_api_protocol.md
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
import uuid
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
from fastapi import APIRouter, HTTPException, Request, status
|
||||
from fastapi.concurrency import run_in_threadpool
|
||||
|
||||
from app.api.deps import OptionalUser
|
||||
from app.core.config import settings
|
||||
from app.core.logging import trace_id_ctx
|
||||
from app.core.pricebot_client import get_pricebot_client
|
||||
from app.core.pricebot_router import pick_pricebot
|
||||
from app.db.session import SessionLocal
|
||||
from app.repositories import comparison as crud_compare
|
||||
|
||||
logger = logging.getLogger("shagua.compare")
|
||||
|
||||
router = APIRouter(prefix="/api/v1", tags=["compare"])
|
||||
|
||||
|
||||
async def _passthrough(request: Request, upstream_path: str) -> dict[str, Any]:
|
||||
"""把请求体原样转发给 pricebot-backend 的 {upstream_path}。
|
||||
# ============================================================
|
||||
# harvest 落库(阻塞 SQLAlchemy → run_in_threadpool,独立 SessionLocal,不阻塞事件循环;
|
||||
# 任何写库异常都吞掉、绝不连累比价透传返回 —— 同 coupon.py 现有 best-effort 写法)。
|
||||
# ============================================================
|
||||
|
||||
跟 coupon_step 同款透传壳:不鉴权、不做 schema 校验,仅读 device_id/trace_id/step
|
||||
打日志。比价单帧是大上下文 / 逐帧 LLM,超时用 PRICEBOT_COMPARE_TIMEOUT_SEC(60s,
|
||||
比领券的 30s 长)。
|
||||
|
||||
def _harvest_running_blocking(
|
||||
trace_id: str, user_id: int | None, business_type: str,
|
||||
device_id: str | None, device_info: dict | None, trace_url: str | None,
|
||||
) -> None:
|
||||
with SessionLocal() as db:
|
||||
crud_compare.harvest_running(
|
||||
db, trace_id=trace_id, user_id=user_id, business_type=business_type,
|
||||
device_id=device_id, device_info=device_info, trace_url=trace_url,
|
||||
)
|
||||
logger.info(
|
||||
"harvest running row (user=%s)", user_id,
|
||||
extra={"phase": "harvest_running", "status": "running", "user_id": user_id},
|
||||
)
|
||||
|
||||
|
||||
def _harvest_done_blocking(
|
||||
trace_id: str, user_id: int | None, done_params: dict, business_type: str,
|
||||
device_id: str | None, device_info: dict | None, trace_url: str | None,
|
||||
) -> None:
|
||||
with SessionLocal() as db:
|
||||
rec, newly_success = crud_compare.harvest_done(
|
||||
db, trace_id=trace_id, user_id=user_id, done_params=done_params,
|
||||
business_type=business_type, device_id=device_id,
|
||||
device_info=device_info, trace_url=trace_url,
|
||||
)
|
||||
logger.info(
|
||||
"harvest done → %s saved=%s newly=%s", rec.status,
|
||||
rec.saved_amount_cents, newly_success,
|
||||
extra={"phase": "harvest_done", "status": rec.status,
|
||||
"saved_cents": rec.saved_amount_cents,
|
||||
"best_platform": rec.best_platform_id, "newly_success": newly_success,
|
||||
"user_id": user_id},
|
||||
)
|
||||
# 不在此处发邀请奖:#113 已把发奖口径从「比价」移到「实际下单」(order.py),harvest
|
||||
# 只记录比价、不发奖。否则比价先于下单 + try_reward 幂等闸会让奖落在「比价」这步,
|
||||
# 架空 #113 的「下单才发奖」防刷意图(newly_success 仅留作日志观测)。
|
||||
|
||||
|
||||
def _harvest_abort_blocking(
|
||||
trace_id: str, status_hint: str, reason: str | None, trace_url: str | None,
|
||||
) -> None:
|
||||
with SessionLocal() as db:
|
||||
rec = crud_compare.harvest_abort(
|
||||
db, trace_id=trace_id, status=status_hint, reason=reason, trace_url=trace_url,
|
||||
)
|
||||
logger.info(
|
||||
"harvest abort → %s", (rec.status if rec else "no-row"),
|
||||
extra={"phase": "harvest_abort",
|
||||
"status": (rec.status if rec else None), "reason": reason},
|
||||
)
|
||||
|
||||
|
||||
async def _forward(
|
||||
request: Request, upstream_path: str, user, *, harvest_first_frame: bool = True,
|
||||
) -> tuple[dict[str, Any], str, dict[str, Any]]:
|
||||
"""透传给 pricebot 的 {upstream_path},并 mint trace_id + 首帧建 running 行。
|
||||
返回 (resp_json, trace_id, meta)。
|
||||
|
||||
- 客户端首帧不带 trace_id → app-server 用 uuid 签发、写进转发 body、回填进响应顶层
|
||||
`trace_id`(客户端据此拿到后续帧都带上)。带了(老客户端 / 后续帧)→ 原样用、走原始 bytes 快路。
|
||||
- 仅**本次 mint(=首帧)**建 running 行(idempotent);后续帧不再写库。
|
||||
"""
|
||||
# 读原始字节,避免"反序列化→再序列化"的双重 JSON(省 ~一半透传 CPU,让 app-server
|
||||
# 单 worker 也扛得住高并发)。只 json.loads 一次拿 trace_id 做亲和 + 打日志,转发时
|
||||
# 直接发原始 bytes(content=raw),不重新 dumps。
|
||||
raw = await request.body()
|
||||
try:
|
||||
meta = json.loads(raw)
|
||||
@@ -52,26 +117,41 @@ async def _passthrough(request: Request, upstream_path: str) -> dict[str, Any]:
|
||||
if not isinstance(meta, dict):
|
||||
meta = {}
|
||||
|
||||
# 按 trace_id 一致性 hash 选 pricebot 实例(同一比价的所有帧落同一进程,内存维护 state)
|
||||
base = pick_pricebot(meta.get("trace_id"))
|
||||
trace_id = meta.get("trace_id")
|
||||
minted = False
|
||||
if not trace_id:
|
||||
trace_id = str(uuid.uuid4())
|
||||
meta["trace_id"] = trace_id
|
||||
raw = json.dumps(meta).encode() # 仅首帧重新序列化(注入 trace_id);后续帧走原始 bytes
|
||||
minted = True
|
||||
|
||||
# 请求级 trace_id 贯穿:此后本请求(含 run_in_threadpool 里的 harvest)每行日志自动带 trace_id
|
||||
trace_id_ctx.set(trace_id)
|
||||
|
||||
base = pick_pricebot(trace_id)
|
||||
url = f"{base.rstrip('/')}{upstream_path}"
|
||||
timeout = settings.PRICEBOT_COMPARE_TIMEOUT_SEC
|
||||
step = meta.get("step")
|
||||
|
||||
logger.info(
|
||||
"compare %s device_id=%s trace_id=%s step=%s",
|
||||
upstream_path,
|
||||
meta.get("device_id"),
|
||||
meta.get("trace_id"),
|
||||
meta.get("step"),
|
||||
"→ pricebot %s step=%s%s", upstream_path, step, " [mint]" if minted else "",
|
||||
extra={"phase": "forward", "endpoint": upstream_path, "step": step,
|
||||
"device_id": meta.get("device_id"), "minted": minted,
|
||||
"query": meta.get("query"), "user_id": (user.id if user else None)},
|
||||
)
|
||||
|
||||
t0 = time.monotonic()
|
||||
try:
|
||||
client = get_pricebot_client()
|
||||
resp = await client.post(
|
||||
url, content=raw, headers={"Content-Type": "application/json"}, timeout=timeout
|
||||
)
|
||||
except httpx.RequestError as e:
|
||||
logger.error("[pricebot] request failed: %s", e)
|
||||
logger.error(
|
||||
"← pricebot %s 不可达: %s", upstream_path, e,
|
||||
extra={"phase": "resp", "endpoint": upstream_path, "step": step,
|
||||
"error": "upstream_unreachable"},
|
||||
)
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_502_BAD_GATEWAY,
|
||||
detail=f"pricebot upstream unreachable: {e}",
|
||||
@@ -79,49 +159,112 @@ async def _passthrough(request: Request, upstream_path: str) -> dict[str, Any]:
|
||||
|
||||
if resp.status_code >= 500:
|
||||
logger.error(
|
||||
"[pricebot] 5xx status=%d body=%s",
|
||||
resp.status_code,
|
||||
resp.text[:500],
|
||||
"← pricebot %s 5xx=%d", upstream_path, resp.status_code,
|
||||
extra={"phase": "resp", "endpoint": upstream_path, "step": step,
|
||||
"http_status": resp.status_code, "error": "upstream_5xx"},
|
||||
)
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_502_BAD_GATEWAY,
|
||||
detail=f"pricebot upstream returned {resp.status_code}",
|
||||
)
|
||||
|
||||
return resp.json()
|
||||
resp_json = resp.json()
|
||||
cost_ms = int((time.monotonic() - t0) * 1000)
|
||||
if not isinstance(resp_json, dict):
|
||||
return {}, trace_id, meta
|
||||
# 回传 app-server 签发的 trace_id(新客户端首帧据此拿到,后续帧都带上它)
|
||||
resp_json.setdefault("trace_id", trace_id)
|
||||
_action = resp_json.get("action") or {}
|
||||
logger.info(
|
||||
"← pricebot %s cmd=%s cont=%s %dms", upstream_path,
|
||||
_action.get("command"), resp_json.get("continue"), cost_ms,
|
||||
extra={"phase": "resp", "endpoint": upstream_path, "step": step,
|
||||
"command": _action.get("command"), "continue": resp_json.get("continue"),
|
||||
"cost_ms": cost_ms},
|
||||
)
|
||||
|
||||
# 首帧建 running 行(仅本次 mint;老客户端自带 trace_id → 不 mint → 由 done / 其 POST 建行)
|
||||
if minted and harvest_first_frame:
|
||||
try:
|
||||
await run_in_threadpool(
|
||||
_harvest_running_blocking, trace_id,
|
||||
(user.id if user else None), "food",
|
||||
meta.get("device_id"), meta.get("device_info"),
|
||||
resp_json.get("trace_url"),
|
||||
)
|
||||
except Exception as e: # noqa: BLE001
|
||||
logger.warning("harvest_running failed trace=%s: %s", trace_id, e)
|
||||
|
||||
return resp_json, trace_id, meta
|
||||
|
||||
|
||||
@router.post("/intent/recognize", summary="外卖比价 Phase 1 意图识别 (透传到 pricebot)")
|
||||
async def intent_recognize(request: Request) -> dict[str, Any]:
|
||||
return await _passthrough(request, "/api/intent/recognize")
|
||||
@router.post("/intent/recognize", summary="外卖比价 Phase 1 意图识别 (透传 + 建 running 行)")
|
||||
async def intent_recognize(request: Request, user: OptionalUser) -> dict[str, Any]:
|
||||
resp, _, _ = await _forward(request, "/api/intent/recognize", user)
|
||||
return resp
|
||||
|
||||
|
||||
@router.post("/intent/step", summary="外卖比价 Phase 1 多帧意图识别 (透传到 pricebot, 仅淘宝源)")
|
||||
async def intent_step(request: Request) -> dict[str, Any]:
|
||||
# 多帧版意图识别(展开+滚动采集→提取): 循环调用直到 done(done 帧顶层带
|
||||
# result+calibration)。目前仅淘宝源走这条, 其它源走上面单次 /intent/recognize。
|
||||
return await _passthrough(request, "/api/intent/step")
|
||||
@router.post("/intent/step", summary="外卖比价 Phase 1 多帧意图识别 (透传, 仅淘宝源)")
|
||||
async def intent_step(request: Request, user: OptionalUser) -> dict[str, Any]:
|
||||
resp, _, _ = await _forward(request, "/api/intent/step", user)
|
||||
return resp
|
||||
|
||||
|
||||
@router.post(
|
||||
"/intent/precoupon/step",
|
||||
summary="外卖比价 Phase 0 意图识别前先用券 (透传到 pricebot, 仅美团源)",
|
||||
)
|
||||
async def intent_precoupon_step(request: Request) -> dict[str, Any]:
|
||||
# 美团源平台『意图识别前先用券』多帧循环: 客户端在调 /intent/recognize 之前先循环
|
||||
# 调本端点到 done(continue=false)。订单页底部有『点击使用X红包』就自动选最大免费券
|
||||
# 用上, 已用券/无券则首帧秒过。与 /intent/step 同属 intent 域, 复用同一透传壳。
|
||||
return await _passthrough(request, "/api/intent/precoupon/step")
|
||||
@router.post("/intent/precoupon/step", summary="外卖比价 Phase 0 识别前先用券 (透传, 仅美团源)")
|
||||
async def intent_precoupon_step(request: Request, user: OptionalUser) -> dict[str, Any]:
|
||||
resp, _, _ = await _forward(request, "/api/intent/precoupon/step", user)
|
||||
return resp
|
||||
|
||||
|
||||
@router.post("/price/step", summary="外卖比价 Phase 2 步进 (透传到 pricebot)")
|
||||
async def price_step(request: Request) -> dict[str, Any]:
|
||||
return await _passthrough(request, "/api/price/step")
|
||||
@router.post("/price/step", summary="外卖比价 Phase 2 步进 (透传 + done 落库)")
|
||||
async def price_step(request: Request, user: OptionalUser) -> dict[str, Any]:
|
||||
resp, trace_id, meta = await _forward(request, "/api/price/step", user)
|
||||
# 最终 done 帧(command=done 且 continue=false)→ harvest 更新成终态。
|
||||
# (单平台中途 done 被 pricebot 改写成 wait+continue=true,不会命中这里,同 coupon 语义。)
|
||||
action = resp.get("action") or {}
|
||||
if action.get("command") == "done" and not resp.get("continue", True):
|
||||
done_params = action.get("params") or {}
|
||||
try:
|
||||
await run_in_threadpool(
|
||||
_harvest_done_blocking, trace_id, (user.id if user else None),
|
||||
done_params, "food",
|
||||
meta.get("device_id"), meta.get("device_info"),
|
||||
resp.get("trace_url") or done_params.get("trace_url"),
|
||||
)
|
||||
except Exception as e: # noqa: BLE001
|
||||
logger.warning("harvest_done failed trace=%s: %s", trace_id, e)
|
||||
return resp
|
||||
|
||||
|
||||
@router.post("/trace/finalize", summary="比价 trace 收尾上云 (透传到 pricebot, 终止/未识别拿 trace_url)")
|
||||
async def trace_finalize(request: Request) -> dict[str, Any]:
|
||||
# 用户终止 / Phase1 未识别没走到 done 帧, pricebot 没上云也没回传 trace_url。客户端收尾时
|
||||
# 打这个, _passthrough 按 trace_id 一致性 hash 落到处理这条 trace 的同一 pricebot 进程
|
||||
# (dir_cache 在那, 才能算对 trace 目录), 由后者打包上云返回 {trace_url}。
|
||||
return await _passthrough(request, "/api/trace/finalize")
|
||||
@router.post("/trace/epilogue", summary="比价结果页尾声帧 (透传到 pricebot, 用户视角截图入 trace)")
|
||||
async def trace_epilogue(request: Request, user: OptionalUser) -> dict[str, Any]:
|
||||
# App 收到 done、渲染完结果页后,把自己页面的截图(base64)传给 pricebot 存进 trace
|
||||
# 目录并触发重传 —— trace 里补上"用户实际看到的汇总页"(步骤帧只有目标 App 画面)。
|
||||
# body ~几百 KB(截图 base64), 纯透传壳(harvest_first_frame=False:不建行/不落库,
|
||||
# 该 trace 的记录已由 done/finalize 落终态)。#112 原走 _passthrough,合并到 harvest
|
||||
# 分支后统一走 _forward(_passthrough 已并入它)。
|
||||
resp, _, _ = await _forward(
|
||||
request, "/api/trace/epilogue", user, harvest_first_frame=False,
|
||||
)
|
||||
return resp
|
||||
|
||||
|
||||
@router.post("/trace/finalize", summary="比价 trace 收尾上云 (透传 + 夭折落库)")
|
||||
async def trace_finalize(request: Request, user: OptionalUser) -> dict[str, Any]:
|
||||
# 用户终止 / Phase1 未识别没到 done 帧: pricebot 打包半截上云返回 {trace_url};
|
||||
# app-server 顺手把该 trace 的 running 行更新成夭折终态(**不降级 success**)。
|
||||
# 客户端 finalize body 带 status(cancelled/failed)+ reason;老客户端只带 trace_id →
|
||||
# 默认 cancelled,其后续 /compare/record 上报再补精确态。
|
||||
resp, trace_id, meta = await _forward(
|
||||
request, "/api/trace/finalize", user, harvest_first_frame=False,
|
||||
)
|
||||
try:
|
||||
await run_in_threadpool(
|
||||
_harvest_abort_blocking, trace_id,
|
||||
(meta.get("status") or "cancelled"),
|
||||
(meta.get("reason") or meta.get("information")),
|
||||
(resp.get("trace_url") if isinstance(resp, dict) else None),
|
||||
)
|
||||
except Exception as e: # noqa: BLE001
|
||||
logger.warning("harvest_abort failed trace=%s: %s", trace_id, e)
|
||||
return resp
|
||||
|
||||
@@ -1,15 +1,13 @@
|
||||
"""比价记录 endpoint(「我的比价记录」数据源)。
|
||||
"""比价记录 endpoint(「我的比价记录」+ admin 数据源)。
|
||||
|
||||
路由前缀 `/api/v1/compare`:
|
||||
POST /record 上报一次比价结果(幂等:同 user+trace_id 覆盖)
|
||||
POST /record (灰度期兼容)老客户端上报,按 **trace_id** 幂等 + 不降级 success
|
||||
GET /records 比价记录列表(游标分页)
|
||||
GET /records/{id} 单条详情(含 raw_payload 全量)
|
||||
|
||||
**均需鉴权**(CurrentUser)——与同文件无关的不鉴权透传 `compare.py` 分开:那个是
|
||||
转发壳(MVP 不鉴权),这里是按用户维度落库的业务接口,必须有 user_id。
|
||||
|
||||
注:本轮只做 server 端,客户端(android 仓)在 done 帧后调 POST /record 上报的改动
|
||||
另起一轮(见 app-server docs/待办与技术债.md P1)。
|
||||
**均需鉴权**(CurrentUser)。⚠️ 写路径现以 `compare.py` 透传壳的**后端 harvest** 为主
|
||||
(帧0 建 running 行 → done/finalize 落终态,新客户端不再 POST);本 POST /record 仅灰度期
|
||||
给老客户端用,与 harvest 按 trace_id reconcile。新版覆盖够高后可下线本 POST(阶段3)。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -21,16 +19,15 @@ from app.api.deps import CurrentUser, DbSession
|
||||
from app.db.session import SessionLocal
|
||||
from app.models.comparison import ComparisonRecord
|
||||
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 (
|
||||
CompareStatsOut,
|
||||
ComparisonRecordCreatedOut,
|
||||
ComparisonRecordDetailOut,
|
||||
ComparisonRecordIn,
|
||||
ComparisonRecordPage,
|
||||
ComparisonRecordOut,
|
||||
ComparisonRecordPage,
|
||||
)
|
||||
from app.services.pricebot_llm_calls import fetch_llm_calls
|
||||
|
||||
logger = logging.getLogger("shagua.compare_record")
|
||||
|
||||
@@ -53,18 +50,8 @@ def report_record(
|
||||
# 任务做,不阻塞上报响应(顺带给 pricebot 落盘留足余量)。upsert 已 commit,后台用
|
||||
# 独立 session 按 record id 回填 llm_calls + 派生 llm_call_count/retry_count。
|
||||
background_tasks.add_task(_backfill_llm_calls, rec.id, rec.trace_id)
|
||||
# 邀请 v2 发奖:被邀请人完成一次【成功】比价 → 给邀请人发邀请奖励金(幂等,只发一次)。
|
||||
# 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)
|
||||
# 注:邀请发奖已从"比价成功"挪到"实际下单"(见 api/v1/order.py report_order)——
|
||||
# 冰拍板:被邀请人完成比价并实际下单才算邀请成功,仅完成比价不再发奖。
|
||||
logger.info(
|
||||
"compare record user=%s trace=%s biz=%s status=%s saved=%s (llm_calls backfill queued)",
|
||||
user.id,
|
||||
|
||||
@@ -1,9 +1,14 @@
|
||||
import logging
|
||||
|
||||
from fastapi import APIRouter
|
||||
|
||||
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.schemas.order import OrderReportOut, OrderReportRequest
|
||||
|
||||
logger = logging.getLogger("shagua.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:
|
||||
# 记账唯一真相表是 savings_record(source='compare')。
|
||||
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(
|
||||
id=rec.id,
|
||||
platform=rec.platform or req.platform,
|
||||
|
||||
+19
-3
@@ -69,6 +69,18 @@ def complete_onboarding(
|
||||
return OkResponse()
|
||||
|
||||
|
||||
@router.post("/onboarding/reset", response_model=OkResponse, summary="重置新手引导(删该 设备+账号 完成标记,下次登录重走)")
|
||||
def reset_onboarding(
|
||||
req: OnboardingCompleteRequest, user: CurrentUser, db: DbSession
|
||||
) -> OkResponse:
|
||||
"""删该 (当前账号, device_id) 的引导完成标记 → 下次登录 onboarding_completed=false,客户端重走。
|
||||
与 /onboarding/complete 互逆。device_id 取客户端硬件级 ANDROID_ID(与登录/complete 一致)。
|
||||
幂等:无记录也返回 ok。仅删自己(当前 JWT 用户)这台设备的记录,不影响别的账号/设备。"""
|
||||
deleted = onboarding_repo.delete_completion(db, user_id=user.id, device_id=req.device_id)
|
||||
logger.info("onboarding reset user_id=%d device_len=%d deleted=%d", user.id, len(req.device_id), deleted)
|
||||
return OkResponse()
|
||||
|
||||
|
||||
@router.get(
|
||||
"/onboarding/status",
|
||||
response_model=OnboardingStatusResponse,
|
||||
@@ -86,15 +98,19 @@ def onboarding_status(
|
||||
return OnboardingStatusResponse(completed=completed)
|
||||
|
||||
|
||||
@router.delete("", response_model=OkResponse, summary="注销账号(软删除)")
|
||||
@router.delete("", response_model=OkResponse, summary="注销账号(清理个人数据 + 匿名化)")
|
||||
def delete_account(user: CurrentUser, db: DbSession) -> OkResponse:
|
||||
# 资金前置校验:注销是软删 + 匿名化,余额一旦留在 deleted 行就锁死(既退不出、新号也
|
||||
# 拿不回)。有可提现现金 / 在审提现单 → 拒绝注销,引导用户先把钱处理掉。
|
||||
# 资金前置校验:注销会物理清理个人数据 + 清零余额 + 匿名化,余额一旦清零就再也拿不回。
|
||||
# 有可提现现金 / 邀请奖励金 / 在审提现单 → 拒绝注销,引导用户先把钱处理掉。
|
||||
# (pending 提现转账已在途、对账 worker 不依赖 user.status 照常到账,故不拦 pending。)
|
||||
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="有提现正在审核中,请等审核完成后再注销"
|
||||
|
||||
@@ -81,7 +81,6 @@ class Settings(BaseSettings):
|
||||
SMS_SIGN_ID: int = 31729 # 极光短信签名 ID(非机密,可被 .env 覆盖)
|
||||
SMS_TEMPLATE_ID: int = 1 # 极光短信模板 ID(变量名 code,有效期 5 分钟)
|
||||
SMS_CODE_LENGTH: int = 6 # 验证码位数(本服务生成;前端 code 字段 4-8 位兼容)
|
||||
SMS_DAILY_LIMIT_PER_PHONE: int = 10 # 单手机号每日发送上限(防刷 + 控费)
|
||||
SMS_MAX_VERIFY_ATTEMPTS: int = 5 # 单个验证码最多校验失败次数,超过即作废(防爆破)
|
||||
|
||||
# ===== 测试账号(release 包全流程联调用)=====
|
||||
|
||||
+57
-8
@@ -2,9 +2,15 @@
|
||||
|
||||
业务代码用 `logger = logging.getLogger("shagua.xxx")` 即可, 本模块在 main.py 启动时调一次。
|
||||
|
||||
- 控制台(stdout): 人类可读文本, 给 systemd / 本地看。
|
||||
- 文件 `logs/app-server.log`: 单行 JSON, 供阿里云 SLS/Logtail 采集(JSON 模式零正则);
|
||||
异常栈作为字段内嵌不换行 → 每条日志一行。
|
||||
- 控制台(stdout): 人类可读文本, 给 systemd / 本地看; 有 trace 时行尾附 `trace=xxx`。
|
||||
- 文件 `logs/app-server.log`: 单行 JSON, 供**阿里云 SLS / Logtail** 采集(JSON 模式零正则);
|
||||
异常栈内嵌为字段不换行 → 每条日志一行。
|
||||
- **结构化字段(SLS 可直接查/聚合)**:
|
||||
- `trace_id`: 请求级贯穿——在入口 `trace_id_ctx.set(...)` 后, 本请求内**每一行日志**(含
|
||||
run_in_threadpool 里的 harvest, contextvars 自动拷进线程)都自动带上, 无需手写。
|
||||
SLS 里 `trace_id: "xxx"` 一查即得整条比价链路, 按 time 升序即请求顺序。
|
||||
- 任意 `logger.info(msg, extra={"phase": ..., "step": ..., "command": ...})` 的 extra
|
||||
键都会平铺进 JSON 顶层 → SLS 可按 phase/step/command/cost_ms 等过滤聚合。
|
||||
- 环境变量:
|
||||
- LOG_JSON_CONSOLE=1 控制台也输出 JSON
|
||||
- LOG_DIR / LOG_FILE 改落盘路径(默认 logs/app-server.log)
|
||||
@@ -17,13 +23,36 @@ import json
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
from contextvars import ContextVar
|
||||
from datetime import datetime
|
||||
from logging.handlers import RotatingFileHandler
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
# 请求级 trace_id:入口(如 compare.py 透传壳)set 之后, 本请求上下文(含 run_in_threadpool
|
||||
# 拷贝出去的线程)内所有日志自动带上。默认空串 = 非请求上下文(启动/后台 worker)。
|
||||
trace_id_ctx: ContextVar[str] = ContextVar("trace_id", default="")
|
||||
|
||||
|
||||
# 标准 LogRecord 属性 + 格式化期附加项:凡不在此集合的 record 属性都视为业务 extra, 平铺进 JSON。
|
||||
_RESERVED = set(
|
||||
logging.LogRecord("", 0, "", 0, "", (), None).__dict__
|
||||
) | {"message", "asctime", "trace_id", "taskName"}
|
||||
|
||||
|
||||
class _ContextFilter(logging.Filter):
|
||||
"""把 trace_id_ctx 注入每条 record(供两个 formatter 取用)。挂在 handler 上,
|
||||
命中每条(含 propagate 上来的)记录, 在 format 之前置好 record.trace_id。"""
|
||||
|
||||
def filter(self, record: logging.LogRecord) -> bool:
|
||||
if not hasattr(record, "trace_id"):
|
||||
record.trace_id = trace_id_ctx.get()
|
||||
return True
|
||||
|
||||
|
||||
class JsonFormatter(logging.Formatter):
|
||||
"""把 LogRecord 序列化成单行 JSON(SLS/Logtail 友好)。异常栈内嵌为字段, 整条仍是一行。"""
|
||||
"""LogRecord → 单行 JSON(SLS/Logtail 友好)。trace_id 提到顶层、extra 键平铺, 异常栈内嵌。"""
|
||||
|
||||
def __init__(self, service: str = "app-server"):
|
||||
super().__init__()
|
||||
self.service = service
|
||||
@@ -35,10 +64,17 @@ class JsonFormatter(logging.Formatter):
|
||||
"level": record.levelname,
|
||||
"service": self.service,
|
||||
"logger": record.name,
|
||||
"func": record.funcName,
|
||||
"line": record.lineno,
|
||||
"message": record.getMessage(),
|
||||
}
|
||||
tid = getattr(record, "trace_id", "") or trace_id_ctx.get()
|
||||
if tid:
|
||||
data["trace_id"] = tid
|
||||
# 业务 extra 字段(phase / step / endpoint / command / cost_ms / status ...)平铺进顶层
|
||||
for k, v in record.__dict__.items():
|
||||
if k not in _RESERVED and not k.startswith("_"):
|
||||
data[k] = v
|
||||
data["func"] = record.funcName
|
||||
data["line"] = record.lineno
|
||||
data["message"] = record.getMessage()
|
||||
if record.exc_info:
|
||||
data["exception"] = self.formatException(record.exc_info)
|
||||
if record.stack_info:
|
||||
@@ -46,6 +82,15 @@ class JsonFormatter(logging.Formatter):
|
||||
return json.dumps(data, ensure_ascii=False, default=str)
|
||||
|
||||
|
||||
class TextFormatter(logging.Formatter):
|
||||
"""控制台文本:标准行 + 有 trace_id 时行尾附 `trace=xxx`(无 trace 的启动/后台日志不加噪)。"""
|
||||
|
||||
def format(self, record: logging.LogRecord) -> str:
|
||||
base = super().format(record)
|
||||
tid = getattr(record, "trace_id", "") or trace_id_ctx.get()
|
||||
return f"{base} trace={tid}" if tid else base
|
||||
|
||||
|
||||
_CONFIGURED = False
|
||||
|
||||
|
||||
@@ -63,14 +108,17 @@ def setup_logging(debug: bool = False) -> None:
|
||||
for h in list(root.handlers):
|
||||
root.removeHandler(h)
|
||||
|
||||
ctx_filter = _ContextFilter()
|
||||
|
||||
# 控制台: 默认文本(systemd/本地看); LOG_JSON_CONSOLE=1 时输出 JSON
|
||||
console = logging.StreamHandler(sys.stdout)
|
||||
if os.getenv("LOG_JSON_CONSOLE") == "1":
|
||||
console.setFormatter(JsonFormatter(service))
|
||||
else:
|
||||
console.setFormatter(
|
||||
logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s")
|
||||
TextFormatter("%(asctime)s %(levelname)s %(name)s: %(message)s")
|
||||
)
|
||||
console.addFilter(ctx_filter)
|
||||
root.addHandler(console)
|
||||
|
||||
# 文件: 单行 JSON, 供 Logtail 采集(自动轮转, 单文件 10MB, 保留 5 个)
|
||||
@@ -82,6 +130,7 @@ def setup_logging(debug: bool = False) -> None:
|
||||
log_file, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8",
|
||||
)
|
||||
file_handler.setFormatter(JsonFormatter(service))
|
||||
file_handler.addFilter(ctx_filter)
|
||||
root.addHandler(file_handler)
|
||||
|
||||
# 第三方库降噪
|
||||
|
||||
+8
-25
@@ -8,15 +8,16 @@
|
||||
校验)→ 鉴权复用极光一键登录的 `JG_APP_KEY`/`JG_MASTER_SECRET`(同一极光应用)。
|
||||
|
||||
验证码存储:**进程内存**(单 worker uvicorn 够用)。重启丢失(用户重发即可)。多
|
||||
worker / 多机时内存不共享 → 冷却、每日上限、校验都会失效,届时迁移到 DB/Redis。
|
||||
worker / 多机时内存不共享 → 冷却、校验都会失效,届时迁移到 DB/Redis。
|
||||
见 docs/待办与技术债.md。
|
||||
|
||||
防刷三层(短信花钱 + `/sms/send` 在登录前无法 JWT 鉴权):
|
||||
防刷两层(短信花钱 + `/sms/send` 在登录前无法 JWT 鉴权):
|
||||
1. 单号 `SMS_SEND_INTERVAL_SEC` 冷却(本文件)
|
||||
2. 单号每日 `SMS_DAILY_LIMIT_PER_PHONE` 条上限(本文件)
|
||||
3. 单设备(device_id)每小时频控(api 层 auth.sms_send 内 enforce_rate_limit)+ 极光控制台 IP 白名单/防轰炸(运维侧)。
|
||||
2. 单设备(device_id)每小时频控(api 层 auth.sms_send 内 enforce_rate_limit)+ 极光控制台 IP 白名单/防轰炸(运维侧)。
|
||||
⚠️ 原「单 IP 频控(rate_limit 依赖)」2026-06-26 按产品要求删除、改设备维度;但 device_id 客户端可伪造/轮换,
|
||||
脚本轮换 id 能绕过本层 → 挡脚本狂发主要靠极光控制台侧(+ 可选 nginx 限流)。
|
||||
⚠️ 原「单号每日上限」2026-07-03 按精简要求删除(mentor 定:登录风控只留单号冷却 + 单设备频控);
|
||||
单号维度现仅剩 60s 冷却,「换号轰炸」由单设备频控封顶。
|
||||
另:单码校验失败 `SMS_MAX_VERIFY_ATTEMPTS` 次即作废(防爆破),验过即作废(一次性)。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
@@ -26,7 +27,6 @@ import logging
|
||||
import secrets
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from threading import Lock
|
||||
|
||||
import httpx
|
||||
@@ -56,22 +56,17 @@ class _CodeRecord:
|
||||
# 进程内存(单 worker 有效;多 worker 不共享,见模块 docstring)
|
||||
_codes: dict[str, _CodeRecord] = {} # phone -> 当前有效验证码
|
||||
_last_sent: dict[str, float] = {} # phone -> 上次发送 epoch(冷却)
|
||||
_daily_count: dict[str, tuple[str, int]] = {} # phone -> (date_str, 当日发送数)
|
||||
_lock = Lock()
|
||||
_GC_THRESHOLD = 10000 # 任一内存 dict 超此阈值,send 时顺手清过期项(防无限增长,仿 ratelimit)
|
||||
|
||||
|
||||
def _today() -> str:
|
||||
return datetime.now().strftime("%Y-%m-%d")
|
||||
|
||||
|
||||
def _gen_code() -> str:
|
||||
"""生成 N 位数字验证码(用 secrets 而非 random;允许前导 0)。"""
|
||||
return "".join(secrets.choice("0123456789") for _ in range(settings.SMS_CODE_LENGTH))
|
||||
|
||||
|
||||
def _gc(now: float) -> None:
|
||||
"""顺手清理过期内存项,防三个 dict 无限增长。仅在持锁时调用,且某 dict 超
|
||||
"""顺手清理过期内存项,防两个 dict 无限增长。仅在持锁时调用,且某 dict 超
|
||||
_GC_THRESHOLD 才扫它(低频,开销可忽略)。"""
|
||||
if len(_codes) > _GC_THRESHOLD:
|
||||
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
|
||||
for p in [p for p, ts in _last_sent.items() if ts < cutoff]:
|
||||
_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:
|
||||
@@ -102,17 +93,9 @@ def send_code(phone: str) -> int:
|
||||
remain = int(settings.SMS_SEND_INTERVAL_SEC - elapsed)
|
||||
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()
|
||||
# 预占:先记冷却/计数/存码,释放锁后再发网络(发失败保留冷却+计数,见下)
|
||||
# 预占:先记冷却/存码,释放锁后再发网络(发失败保留冷却,见下)
|
||||
_last_sent[phone] = now
|
||||
_daily_count[phone] = (today, cnt + 1)
|
||||
_codes[phone] = _CodeRecord(code=code, expires_at=now + settings.SMS_CODE_TTL_SEC)
|
||||
|
||||
# --- lock 外:真正发送(网络 IO 不持锁)---
|
||||
@@ -123,7 +106,7 @@ def send_code(phone: str) -> int:
|
||||
_send_via_jiguang(phone, code)
|
||||
logger.info("[SMS] sent to %s****", phone[:3])
|
||||
except Exception as e:
|
||||
# 发送失败:**保留冷却 + 每日计数**(失败也限速,挡住余额不足/签名失效时
|
||||
# 发送失败:**保留冷却**(失败也限速,挡住余额不足/签名失效时
|
||||
# 前端重试狂打极光),只清掉没发出去的码(用户收不到,留着无意义且占内存)。
|
||||
with _lock:
|
||||
_codes.pop(phone, None)
|
||||
|
||||
@@ -39,16 +39,19 @@ _JSON = JSON().with_variant(JSONB(), "postgresql")
|
||||
class ComparisonRecord(Base):
|
||||
__tablename__ = "comparison_record"
|
||||
__table_args__ = (
|
||||
# 同一用户同一次比价(trace_id)只存一条:客户端重试/误点重复上报时幂等覆盖。
|
||||
UniqueConstraint("user_id", "trace_id", name="uq_comparison_user_trace"),
|
||||
# trace_id 由 app-server 签发、全局唯一 → 一次比价一行,后端 harvest 按它 upsert。
|
||||
# (原 (user_id,trace_id) 复合唯一改为 trace_id 单列:harvest 帧0 建行时 user_id 可能暂缺。)
|
||||
UniqueConstraint("trace_id", name="uq_comparison_trace"),
|
||||
# 首页轮播 / 省钱战绩聚合都按 status='success' 过滤 + created_at 近期排序;
|
||||
# 复合索引避免随数据量增大退化成全表扫(单列 created_at 索引不含 status)。
|
||||
Index("ix_comparison_status_created", "status", "created_at"),
|
||||
)
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
user_id: Mapped[int] = mapped_column(
|
||||
Integer, ForeignKey("user.id"), index=True, nullable=False
|
||||
# 后端 harvest 在帧0(pricebot 出 trace_id)即建行,软鉴权下 user_id 可能暂缺(老客户端/匿名)→ 可空。
|
||||
# C 端「我的比价记录」按 user_id 过滤天然排除 null-user 行;admin 全看(含孤儿行)。
|
||||
user_id: Mapped[int | None] = mapped_column(
|
||||
Integer, ForeignKey("user.id"), index=True, nullable=True
|
||||
)
|
||||
# 仍记录设备号(同一用户多设备的行为区分 / 与不鉴权期 device_id 数据对账)
|
||||
device_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
|
||||
|
||||
@@ -74,7 +74,12 @@ def _derive(payload: ComparisonRecordIn) -> dict:
|
||||
def upsert_record(
|
||||
db: Session, *, user_id: int, payload: ComparisonRecordIn
|
||||
) -> ComparisonRecord:
|
||||
"""按 (user_id, trace_id) 幂等写入:已存在则覆盖(更完整的重试上报胜出),否则新建。"""
|
||||
"""按 **trace_id** 幂等写入(唯一键已从 user_id+trace_id 改为 trace_id):已存在则合并
|
||||
(回填 null user_id + 覆盖字段,但**不降级 success**),否则新建。
|
||||
|
||||
灰度期老客户端 POST /compare/record 走这条,与后端 harvest 按 trace_id reconcile;
|
||||
新客户端不再 POST(改由 compare.py 透传壳 harvest 落库)。
|
||||
"""
|
||||
derived = _derive(payload)
|
||||
fields = dict(
|
||||
device_id=payload.device_id,
|
||||
@@ -110,14 +115,29 @@ def upsert_record(
|
||||
**derived,
|
||||
)
|
||||
|
||||
# 按 trace_id 定位(唯一键已改 trace_id):后端 harvest 可能已建行,这里的客户端上报
|
||||
# (灰度期老客户端 / 新客户端不再走这条)与之 reconcile。
|
||||
existing = db.execute(
|
||||
select(ComparisonRecord).where(
|
||||
ComparisonRecord.user_id == user_id,
|
||||
ComparisonRecord.trace_id == payload.trace_id,
|
||||
)
|
||||
).scalar_one_or_none()
|
||||
|
||||
if existing is not None:
|
||||
# 不降级:harvest 或更早上报已落成 success,收尾期 fromFailure 的 cancelled/failed
|
||||
# 不许把它盖回去(老 comparisonReported bug 的服务端兜底)——只补 user_id / trace_url。
|
||||
if existing.status == "success" and fields.get("status") != "success":
|
||||
if existing.user_id is None and user_id is not None:
|
||||
existing.user_id = user_id
|
||||
if fields.get("trace_url"):
|
||||
existing.trace_url = fields["trace_url"]
|
||||
db.commit()
|
||||
db.refresh(existing)
|
||||
return existing
|
||||
# 只**填**空缺 user_id、不 reassign:trace_id 全局唯一,一条 trace 只属一个用户;
|
||||
# 绝不把已有归属的记录改判给另一个上报者(仅 harvest 建的 null-user 行在此绑上)。
|
||||
if existing.user_id is None:
|
||||
existing.user_id = user_id
|
||||
for k, v in fields.items():
|
||||
setattr(existing, k, v)
|
||||
db.commit()
|
||||
@@ -139,6 +159,201 @@ def upsert_record(
|
||||
return rec
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 后端 harvest:app-server 从 pricebot 透传响应里直接落库(不靠客户端上报)。
|
||||
# 帧0 建行(running) → 最终 done 更新(success/failed) → finalize 更新(aborted)。
|
||||
# 全按 trace_id upsert;success 行永不被后到的 failed/aborted 降级。
|
||||
# ============================================================
|
||||
|
||||
|
||||
def _derive_from_results(results: list[dict]) -> dict:
|
||||
"""从 done 帧 comparison_results(pricebot 原始 dict 列表)派生结构化列。
|
||||
等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。"""
|
||||
priced = [r for r in results if r.get("price") is not None]
|
||||
best = None
|
||||
if priced:
|
||||
best = min(
|
||||
priced,
|
||||
key=lambda r: (r.get("rank") if r.get("rank") is not None else 10**9, r["price"]),
|
||||
)
|
||||
src_row = next((r for r in results if r.get("is_source")), None)
|
||||
|
||||
source_price_cents = None
|
||||
if src_row is not None and src_row.get("price") is not None:
|
||||
source_price_cents = _yuan_to_cents(src_row["price"])
|
||||
best_price_cents = _yuan_to_cents(best["price"]) if best else None
|
||||
saved_amount_cents = None
|
||||
if source_price_cents is not None and best_price_cents is not None:
|
||||
saved_amount_cents = source_price_cents - best_price_cents
|
||||
|
||||
has_valid_target = any(
|
||||
(not r.get("is_source")) and r.get("price") is not None for r in results
|
||||
)
|
||||
return {
|
||||
"source_platform_id": (src_row or {}).get("platform_id"),
|
||||
"source_platform_name": (src_row or {}).get("platform_name"),
|
||||
"source_package": (src_row or {}).get("package"),
|
||||
"source_price_cents": source_price_cents,
|
||||
"best_platform_id": best.get("platform_id") if best else None,
|
||||
"best_platform_name": best.get("platform_name") if best else None,
|
||||
"best_price_cents": best_price_cents,
|
||||
"saved_amount_cents": saved_amount_cents,
|
||||
"is_source_best": best.get("is_source") if best else None,
|
||||
"store_name": (src_row or {}).get("store_name") or None,
|
||||
"status": "success" if has_valid_target else "failed",
|
||||
}
|
||||
|
||||
|
||||
def _device_cols_from_info(device_info: dict | None) -> dict:
|
||||
"""step 帧 device_info({locale,brand,model,android_version,rom_version})→ 表列。
|
||||
step 帧只带这几项;rom_name / android_sdk / app_version / 耗时步数 harvest 拿不到 → 留 None
|
||||
(灰度期老客户端 fromComparison 会补齐;要新客户端也全带需扩 collectDeviceInfo,后续)。"""
|
||||
d = device_info or {}
|
||||
rv = str(d.get("rom_version") or "")
|
||||
return {
|
||||
"device_model": d.get("model") or None,
|
||||
"device_manufacturer": d.get("brand") or None,
|
||||
"android_version": d.get("android_version") or None,
|
||||
"rom_version": int(rv) if rv.isdigit() else None,
|
||||
}
|
||||
|
||||
|
||||
def _get_by_trace(db: Session, trace_id: str) -> ComparisonRecord | None:
|
||||
return db.execute(
|
||||
select(ComparisonRecord).where(ComparisonRecord.trace_id == trace_id)
|
||||
).scalar_one_or_none()
|
||||
|
||||
|
||||
def harvest_running(
|
||||
db: Session,
|
||||
*,
|
||||
trace_id: str,
|
||||
user_id: int | None,
|
||||
business_type: str = "food",
|
||||
device_id: str | None = None,
|
||||
device_info: dict | None = None,
|
||||
trace_url: str | None = None,
|
||||
) -> ComparisonRecord:
|
||||
"""帧0(或任一尚未建行的帧)建/补 running 行。幂等:已存在只补空缺(user_id/trace_url/
|
||||
device_id/机型),绝不动已落定的 status / 结果。"""
|
||||
rec = _get_by_trace(db, trace_id)
|
||||
if rec is None:
|
||||
rec = ComparisonRecord(
|
||||
trace_id=trace_id,
|
||||
user_id=user_id,
|
||||
business_type=business_type or "food",
|
||||
device_id=device_id,
|
||||
status="running",
|
||||
trace_url=trace_url,
|
||||
created_at=datetime.now(CN_TZ).replace(tzinfo=None),
|
||||
**_device_cols_from_info(device_info),
|
||||
)
|
||||
db.add(rec)
|
||||
db.commit()
|
||||
db.refresh(rec)
|
||||
return rec
|
||||
changed = False
|
||||
if rec.user_id is None and user_id is not None:
|
||||
rec.user_id = user_id
|
||||
changed = True
|
||||
if not rec.trace_url and trace_url:
|
||||
rec.trace_url = trace_url
|
||||
changed = True
|
||||
if rec.device_id is None and device_id:
|
||||
rec.device_id = device_id
|
||||
changed = True
|
||||
for k, v in _device_cols_from_info(device_info).items():
|
||||
if getattr(rec, k) is None and v is not None:
|
||||
setattr(rec, k, v)
|
||||
changed = True
|
||||
if changed:
|
||||
db.commit()
|
||||
db.refresh(rec)
|
||||
return rec
|
||||
|
||||
|
||||
def harvest_done(
|
||||
db: Session,
|
||||
*,
|
||||
trace_id: str,
|
||||
user_id: int | None,
|
||||
done_params: dict,
|
||||
business_type: str = "food",
|
||||
device_id: str | None = None,
|
||||
device_info: dict | None = None,
|
||||
trace_url: str | None = None,
|
||||
) -> tuple[ComparisonRecord, bool]:
|
||||
"""最终 done 帧:running 行 → 终态(success/failed)+结果+trace_url。
|
||||
返回 (记录, 是否本次**新**落成 success)——供调用方据此幂等发一次邀请奖。
|
||||
行不存在(理论上帧0已建;防御)则新建。"""
|
||||
results = done_params.get("comparison_results") or []
|
||||
derived = _derive_from_results(results)
|
||||
fields = dict(
|
||||
business_type=business_type or "food",
|
||||
information=done_params.get("information") or None,
|
||||
# best_deeplink 来自客户端剪贴板采集,harvest 拿不到 → 留空(灰度期 fromComparison 会补;
|
||||
# 纯 harvest 行「再次比价」退化为按 package 拉起 App。要精确深链需客户端另传,后续)。
|
||||
trace_url=trace_url or done_params.get("trace_url"),
|
||||
total_dish_count=done_params.get("total_dish_count"),
|
||||
skipped_dish_count=done_params.get("skipped_dish_count"),
|
||||
skipped_dish_names=list(done_params.get("skipped_dish_names") or []),
|
||||
comparison_results=results,
|
||||
# 菜品:pricebot 已把源单菜品塞进 comparison_results[源行].items
|
||||
items=next((r.get("items") or [] for r in results if r.get("is_source")), []),
|
||||
raw_payload=done_params,
|
||||
**derived,
|
||||
**_device_cols_from_info(device_info),
|
||||
)
|
||||
|
||||
rec = _get_by_trace(db, trace_id)
|
||||
was_success = rec is not None and rec.status == "success"
|
||||
if rec is None:
|
||||
rec = ComparisonRecord(
|
||||
trace_id=trace_id,
|
||||
user_id=user_id,
|
||||
created_at=datetime.now(CN_TZ).replace(tzinfo=None),
|
||||
device_id=device_id,
|
||||
**fields,
|
||||
)
|
||||
db.add(rec)
|
||||
else:
|
||||
if user_id is not None and rec.user_id is None:
|
||||
rec.user_id = user_id
|
||||
if device_id and rec.device_id is None:
|
||||
rec.device_id = device_id
|
||||
for k, v in fields.items():
|
||||
setattr(rec, k, v)
|
||||
db.commit()
|
||||
db.refresh(rec)
|
||||
newly_success = (rec.status == "success") and not was_success
|
||||
return rec, newly_success
|
||||
|
||||
|
||||
def harvest_abort(
|
||||
db: Session,
|
||||
*,
|
||||
trace_id: str,
|
||||
status: str,
|
||||
reason: str | None,
|
||||
trace_url: str | None = None,
|
||||
) -> ComparisonRecord | None:
|
||||
"""finalize(用户终止/超时/异常,无 done 帧):running 行 → aborted/failed + trace_url。
|
||||
**不降级 success**:行已 success(收尾取消那种 finalize 后到)只 refresh trace_url。
|
||||
行不存在(极少:帧0没建成)→ 返回 None,不凭空造。"""
|
||||
rec = _get_by_trace(db, trace_id)
|
||||
if rec is None:
|
||||
return None
|
||||
if trace_url and not rec.trace_url:
|
||||
rec.trace_url = trace_url
|
||||
if rec.status != "success":
|
||||
rec.status = status or "cancelled"
|
||||
if reason:
|
||||
rec.information = reason
|
||||
db.commit()
|
||||
db.refresh(rec)
|
||||
return rec
|
||||
|
||||
|
||||
def _ordered_shop_names(db: Session, user_id: int) -> set[str]:
|
||||
"""该用户「真实下单」(source='compare')覆盖到的店名集合,用来给比价记录打「已下单」。
|
||||
|
||||
|
||||
@@ -304,6 +304,9 @@ def get_invitees(
|
||||
"avatar_url": u.avatar_url or u.wechat_avatar_url or None,
|
||||
"coins": rel.inviter_coin, # v3 起恒 0(邀请人收益改走邀请奖励金)
|
||||
"invited_at": rel.created_at,
|
||||
# 是否已完成过一次比价(= 已发过邀请奖励金)。客户端据此:好友列表分"去提醒/邀请成功"、
|
||||
# 在途列表只取未比价(is_compared=False)、算在途好友数与在途收益(未比价数×2元)。
|
||||
"is_compared": bool(rel.compare_reward_granted),
|
||||
})
|
||||
has_more = offset + len(rows) < int(total)
|
||||
return items, int(total), has_more
|
||||
|
||||
@@ -5,7 +5,7 @@ device_id 为空(老客户端 / 取不到 ANDROID_ID)一律按"未完成"处理,
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy import delete, select
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
@@ -33,3 +33,18 @@ def mark_completed(db: Session, *, user_id: int, device_id: str) -> None:
|
||||
except IntegrityError:
|
||||
# 并发 / 重复提交撞唯一约束:已有行即视为成功。
|
||||
db.rollback()
|
||||
|
||||
|
||||
def delete_completion(db: Session, *, user_id: int, device_id: str) -> int:
|
||||
"""删该 (账号, 设备) 的引导完成标记 → 下次登录 is_completed=False → 客户端重走。
|
||||
与 [mark_completed] 互逆。device_id 为空忽略(返 0);无记录也幂等(返 0)。返回删除行数。"""
|
||||
if not device_id:
|
||||
return 0
|
||||
result = db.execute(
|
||||
delete(OnboardingCompletion).where(
|
||||
OnboardingCompletion.user_id == user_id,
|
||||
OnboardingCompletion.device_id == device_id,
|
||||
)
|
||||
)
|
||||
db.commit()
|
||||
return result.rowcount
|
||||
|
||||
@@ -8,10 +8,29 @@ import secrets
|
||||
import string
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy import delete, select
|
||||
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.wallet import CoinAccount
|
||||
|
||||
|
||||
# ===== 创建时分配的标识:用户名(对外展示账号 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:
|
||||
"""注销账号:软删除 + 匿名化 + 释放唯一约束/外部绑定。
|
||||
"""注销账号:物理删除个人数据 + 清零余额 + 匿名化保留财务/关系 + 释放唯一约束。
|
||||
|
||||
把 phone 改成 `deleted_<id>` 释放手机号唯一约束,允许同号码重新注册成全新账号;
|
||||
清空 PII(昵称/头像),保留行用于审计。status=deleted 后将无法再登录该行
|
||||
(登录接口校验 status==active)。
|
||||
删除策略(按"删了是否破坏对账 / 影响别人 / 是否 PII"分类):
|
||||
物理删除(个人内容/成绩/行为/PII,只涉本人、删了不碰账目也不影响别人):比价/
|
||||
省钱/签到/领券/任务/里程碑领取/埋点/引导标记/设备活跃/价格上报/反馈/看广告时长日志
|
||||
(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: 不清 → 这个微信再也无法被任何账号绑定(提现通道彻底锁死);
|
||||
- invite_code: 不清 → 邀请码槽位被占 + 裸查仍命中死号(发奖已被 bind 的
|
||||
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.phone = f"deleted_{user.id}"
|
||||
user.nickname = None
|
||||
|
||||
@@ -296,6 +296,9 @@ def daily_auto_exchange(db: Session) -> dict:
|
||||
- **逐用户独立事务**:单个用户异常 rollback 不影响其他人;exchange_coins_to_cash 内部按用户 commit。
|
||||
返回统计 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()
|
||||
stats = {
|
||||
"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
|
||||
|
||||
|
||||
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:
|
||||
"""用户是否有「待审核(reviewing)」的提现单。
|
||||
|
||||
|
||||
@@ -69,6 +69,7 @@ class InviteeItem(BaseModel):
|
||||
avatar_url: str | None = None # 头像 URL;null = 前端画默认色块
|
||||
coins: int # 这次邀请给我(邀请人)发的金币
|
||||
invited_at: datetime # 邀请绑定时间(前端转"今天/3天前")
|
||||
is_compared: bool = False # 该好友是否已完成过一次比价(好友列表分"去提醒/邀请成功";在途列表只取 False)
|
||||
|
||||
|
||||
class InviteeListOut(BaseModel):
|
||||
|
||||
@@ -19,6 +19,9 @@ Type=oneshot
|
||||
User=root
|
||||
WorkingDirectory=/opt/shaguabijia-app-server
|
||||
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
|
||||
ExecStart=/opt/shaguabijia-app-server/.venv/bin/python -m scripts.daily_auto_exchange --once
|
||||
SyslogIdentifier=daily-exchange
|
||||
|
||||
@@ -4,8 +4,10 @@
|
||||
Description=Run daily auto-exchange coins->cash at midnight
|
||||
|
||||
[Timer]
|
||||
# 每天 0 点跑。客户端文案已注明「可能存在延迟」,可按需改 00:05 错开整点扎堆。
|
||||
OnCalendar=*-*-* 00:00:00
|
||||
# 每天【北京时】0 点跑,写死时区(用户 2026-07-01 硬约束:兑换一律北京 0 点,不随服务器本地时区漂)。
|
||||
# systemd OnCalendar 支持尾缀时区;不写时区会按服务器 OS 本地时区触发 → 服务器非 CST 时会在错误时刻兑。
|
||||
# 客户端文案已注明「可能存在延迟」,可按需改 00:05 错开整点扎堆。
|
||||
OnCalendar=*-*-* 00:00:00 Asia/Shanghai
|
||||
# 服务器宕机/重启后,补跑错过的那一轮(而不是干等次日)。
|
||||
Persistent=true
|
||||
AccuracySec=1min
|
||||
|
||||
@@ -2,10 +2,10 @@
|
||||
|
||||
> 所属:比价记录组(前缀 `/api/v1/compare`) | 鉴权:Bearer | [← 返回 API 索引](../README.md)
|
||||
|
||||
比价 `done` 帧后,客户端用**带 JWT 的通道**上报一条比价结果,落 `comparison_record` 表,作为「我的比价记录」数据源 + 用户级行为画像。
|
||||
比价 `done` 帧后上报一条比价结果,落 `comparison_record` 表,作为「我的比价记录」数据源 + 用户级行为画像。
|
||||
|
||||
> ⚠️ 与不鉴权的透传端点 [`/api/v1/price/step`](./compare-price-step.md) 不同:那是转发壳,本接口按用户维度落库,**必须鉴权**。
|
||||
> 本轮只做 server 端;客户端在 done 帧后调本接口的改动另起一轮(见 [待办与技术债.md](../guides/待办与技术债.md) P1)。
|
||||
> ⚠️ **灰度定位(2026-07)**:写 `comparison_record` 现以透传壳 `compare.py` 的**后端 harvest** 为主(帧0 建 `running` 行 → done/finalize 落终态,新客户端不再 POST)。本接口降级为**老客户端兼容**通道,与 harvest 按 `trace_id` reconcile(**success 不被降级**);新版覆盖够高后可下线本 POST。
|
||||
> 与软鉴权透传端点 [`/api/v1/price/step`](./compare-price-step.md) 不同:那是转发壳(顺带 harvest 落库),本接口按用户维度显式上报,**必须鉴权**。
|
||||
|
||||
## 入参(JSON body)
|
||||
|
||||
@@ -13,7 +13,7 @@
|
||||
|
||||
| 字段 | 类型 | 必填 | 默认 | 说明 |
|
||||
|---|---|---|---|---|
|
||||
| `trace_id` | string | ✅ | — | pricebot 侧 trace_id。**幂等键**:同用户同 trace_id 重复上报覆盖、返回同一 id |
|
||||
| `trace_id` | string | ✅ | — | 一次比价的唯一标识(app-server 帧0 签发)。**幂等键(trace_id 单列唯一)**:同 trace_id 重复上报覆盖、返回同一 id;与 harvest 行按它 reconcile |
|
||||
| `business_type` | string | ❌ | `food` | `food`(外卖,当前唯一接通) / `ecom`(电商) / `coupon`(领券) |
|
||||
| `device_id` | string \| null | ❌ | null | 设备号 |
|
||||
| `store_name` | string \| null | ❌ | null | 店铺名(外卖,来自 calibration.result) |
|
||||
|
||||
@@ -2,13 +2,13 @@
|
||||
|
||||
> 模型 `app/models/comparison.py` · 仓库 `app/repositories/comparison.py` · 接口 [compare-record-report](../api/compare-record-report.md) / [compare-records](../api/compare-records.md) / [compare-record-detail](../api/compare-record-detail.md) · [← 索引](./README.md) · [总览](./OVERVIEW.md)
|
||||
|
||||
每完成一次比价(外卖/电商/领券),客户端在 done 帧后用**带 JWT** 的通道上报一条。App「我的比价记录」列表/详情的数据源,也是比价战绩里程碑解锁进度的计数源(`status='success'` 条数),还被「上报更低价」反查原最低价。
|
||||
每完成一次比价(外卖/电商/领券)记一行。**写入以 app-server 后端 harvest 为主**(2026-07 起):比价透传壳 `compare.py` 在帧0(pricebot 出 trace_id)即建 `running` 行,随 done / `trace/finalize` 逐步补全成终态,客户端不再主动 POST 记录;老客户端仍可走**带 JWT** 的 `POST /compare/record` 兜底(灰度期两条写路径按 `trace_id` reconcile)。App「我的比价记录」列表/详情的数据源,也是比价战绩里程碑解锁进度的计数源(`status='success'` 条数),还被「上报更低价」反查原最低价。
|
||||
|
||||
> 与 `savings_record` 的区别:本表是「每一次**比价行为**的完整明细」(不省钱、甚至失败也记);`savings_record` 是「真正**下单成交**省了多少」。两表独立、互不喂数据。
|
||||
> 与 [`price_observation`](./price_observation.md) / `store_mapping` 的区别:本表是**用户视角**(登录后按 `user_id` 存「我的比价记录」);后两张是 server 侧无条件沉淀的**平台/门店视角客观事实**(价格事实 / 跨平台店铺身份映射),与本表 `trace_id` 同源但不互相 join,各存各的视角。
|
||||
|
||||
## 用在哪 / 增删改查
|
||||
- **C / U(upsert,幂等)**:`POST /compare/record`(`upsert_record`)。按 `(user_id, trace_id)` 查:不存在→新建;已存在→整行覆盖(客户端重试/重复上报时,**更完整的那次胜出**)。`best_*`/`saved_amount_cents`/`is_source_best`/`status` 由 `_derive` 从 `comparison_results` 算出(协议已按 price 升序、rank=1 最便宜),不信客户端自算。
|
||||
- **C / U(harvest 为主,按 `trace_id` 幂等)**:透传壳 `compare.py` 三段式落库(`app/repositories/comparison.py`)——`harvest_running`(帧0 建 `running` 行)→ `harvest_done`(done 帧转 `success`/`failed` + 派生 `best_*`/`saved_amount_cents` + 返 `newly_success` 供幂等发邀请奖)→ `harvest_abort`(`trace/finalize` 转 `cancelled`/`failed`,**不降级已 success**)。老客户端仍可 `POST /compare/record`(`upsert_record`,按 `trace_id` 查、整行覆盖)兜底。`best_*`/`saved_amount_cents`/`is_source_best`/`status` 一律由 `_derive` 从 `comparison_results` 算出(协议已按 price 升序、rank=1 最便宜),不信客户端自算。
|
||||
- **D**:无(关联的 `price_report` 也只把 `comparison_record_id` 置空,不删本表)。
|
||||
- **R**:`GET /compare/records`(列表,`created_at` 倒序游标 + 「已下单」标记)、`GET /compare/records/{id}`(详情,限本人);`count_success` 给里程碑;`get_stats`(`status='success'` 计数 + `saved_amount_cents` 求和)给 [`GET /compare/stats`](../api/compare-stats.md) 喂「我的」页省钱战绩卡(完成比价 + 累计发现可省,**比价口径**);`report.py` 反查 `best_*`;admin 大盘/明细。
|
||||
|
||||
@@ -16,10 +16,10 @@
|
||||
| 列 | 类型 | 约束 / 默认 | 说明(取值 / join) |
|
||||
|---|---|---|---|
|
||||
| `id` | Integer | PK, autoincrement | 被 `price_report.comparison_record_id` 引用 |
|
||||
| `user_id` | Integer | FK→user.id, index, NOT NULL | 归属用户 |
|
||||
| `user_id` | Integer | FK→user.id, index, **nullable**(2026-07 从 NOT NULL 放开) | 归属用户。后端 harvest 帧0 建行时(软鉴权 / 老客户端匿名)可能暂缺 → 可空;C 端「我的比价记录」按 `user_id` 过滤天然排除 null 行,admin 全看(含孤儿行) |
|
||||
| `device_id` | String(64) | nullable | 设备号(多设备区分 / 与不鉴权期对账) |
|
||||
| `business_type` | String(16) | NOT NULL, default `food`, index | 取值:`food`(当前唯一接通)/ `ecom` / `coupon` |
|
||||
| `trace_id` | String(64) | NOT NULL | pricebot 侧 trace_id(关联调试落盘 + 幂等键) |
|
||||
| `trace_id` | String(64) | NOT NULL, **UNIQUE** | 一次比价的唯一标识。**由 app-server 帧0 用 uuid 签发**(注入转发 body + 回填响应顶层给客户端;老客户端自带),全局唯一 = harvest upsert 键 + 关联 pricebot 调试落盘 |
|
||||
| `trace_url` | String(512) | nullable | 本次比价公网调试链接(`price.shaguabijia.com/traces/{dir}/`);dir 名含 pricebot 落盘时分秒,前端/server 拼不出必须存。查看接口按 `user.debug_trace_enabled`(或本机 agent 调试 `include_trace`)决定返不返回。旧记录 / 未开上云为 null |
|
||||
| `source_platform_id` / `_name` | String(32) | nullable | 源平台代号 / 中文名 |
|
||||
| `source_package` | String(128) | nullable | 源平台 Android 包名 |
|
||||
@@ -31,7 +31,7 @@
|
||||
| `is_source_best` | Boolean | nullable | 源平台就是最便宜(= 这次没省到) |
|
||||
| `store_name` | String(128) | nullable | 店铺名。**与 `savings_record.shop_name` 按字符串相等关联**,给本记录打「已下单」 |
|
||||
| `total_dish_count` / `skipped_dish_count` | Integer | nullable | 菜品总数 / 目标平台没找到被跳过数 |
|
||||
| `status` | String(16) | NOT NULL, default `success` | 取值:`success`(有非源且有价的目标结果)/ `failed`(出错/没采到目标价)。**里程碑只数 success** |
|
||||
| `status` | String(16) | NOT NULL, default `success` | 取值:`running`(harvest 帧0 建行、比价进行中)/ `success`(有非源且有价的目标结果)/ `failed`(出错/没采到目标价)/ `cancelled`(用户终止 / Phase1 未识别,`harvest_abort` 写,**不降级已 success**)。**里程碑只数 success** |
|
||||
| `information` | String(256) | nullable | done 帧文案;成功=摘要,失败=具体原因(前端失败时当原因展示) |
|
||||
| `items` | JSON(PG: JSONB) | NOT NULL, default [] | 下单菜品 `[{name, qty, specs?}]` |
|
||||
| `comparison_results` | JSON(PG: JSONB) | NOT NULL, default [] | 逐平台对比 `[{platform_id,platform_name,package,price(元),is_source,rank,coupon_saved(元),coupon_name,applied_coupons}]`;`coupon_saved`=该平台主优惠额(美团红包/淘宝平台红包/京东百亿补贴,只取一笔),`coupon_name`=优惠来源名(展示用),`applied_coupons`=`[{name,amount}]` 多券明细 |
|
||||
@@ -50,7 +50,8 @@
|
||||
- 被 `comparison_milestone_claim` 间接依赖:解锁进度 = 本表 `status='success'` 计数。
|
||||
|
||||
## 索引与约束
|
||||
- PK `id`;index `user_id`、`business_type`、`created_at`;复合 index `ix_comparison_status_created`(`status`, `created_at`)(按 `status='success'` 过滤 + 近期排序的聚合/轮播,避免随数据量退化为全表扫);UNIQUE(`user_id`, `trace_id`) = `uq_comparison_user_trace`(幂等覆盖)。
|
||||
- PK `id`;index `user_id`、`business_type`、`created_at`;复合 index `ix_comparison_status_created`(`status`, `created_at`)(按 `status='success'` 过滤 + 近期排序的聚合/轮播,避免随数据量退化为全表扫);UNIQUE(`trace_id`) = `uq_comparison_trace`(harvest 按它 upsert,一次比价一行)。
|
||||
> 2026-07 迁移 `comparison_record_trace_unique`:唯一键从复合 `uq_comparison_user_trace`(`user_id`,`trace_id`)改为 `trace_id` 单列——harvest 帧0 建行时 user_id 可能暂缺,不能再用复合键去重。上线前须确认历史无重复 `trace_id`(`SELECT trace_id,COUNT(*) c FROM comparison_record GROUP BY trace_id HAVING c>1`),否则建单列唯一会失败。
|
||||
|
||||
## 注意
|
||||
- 4 个 JSON 列用 `JSON().with_variant(JSONB(),"postgresql")`(SQLite 退化 JSON)。结构化金额列存「分」,`comparison_results.price`/`coupon_saved` 原样存「元」。
|
||||
|
||||
@@ -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 比对存码,匹配即作废,失败累计到上限作废 |
|
||||
| `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_TEMPLATE_ID` | 1 | 极光模板 ID(变量名 `code`) |
|
||||
| `SMS_CODE_LENGTH` | 6 | 验证码位数(本服务生成;前端 code 4-8 位兼容) |
|
||||
| `SMS_DAILY_LIMIT_PER_PHONE` | 10 | 单号每日发送上限(防刷 + 控费) |
|
||||
| `SMS_MAX_VERIFY_ATTEMPTS` | 5 | 单码最多校验失败次数,超过作废(防爆破) |
|
||||
|
||||
> **鉴权复用极光一键登录**:`/v1/messages` 用 `base64(JG_APP_KEY:JG_MASTER_SECRET)` 做 HTTP Basic Auth——短信与一键登录是**同一个极光应用**(同 AppKey)。**上线不需要额外凭证,只需 `SMS_MOCK=false`**(`JG_*` 一键登录已配)。
|
||||
|
||||
## 防刷(短信花钱 + `/sms/send` 在登录前无法 JWT 鉴权)
|
||||
1. 单号 `SMS_SEND_INTERVAL_SEC` 冷却
|
||||
2. 单号每日 `SMS_DAILY_LIMIT_PER_PHONE` 条上限
|
||||
3. 单 IP 频控:`/sms/send` 挂 `rate_limit(10,60)`、`/sms/login` 挂 `rate_limit(20,60)`
|
||||
4. 单码校验失败 `SMS_MAX_VERIFY_ATTEMPTS` 次即作废
|
||||
5. 运维侧建议在极光控制台叠加:**IP 白名单**(只许服务器 IP)+ **防轰炸设置**
|
||||
> 2026-07-03 精简:登录风控只留「单号冷却 + 单设备频控」两道主策略(删单号每日上限、删登录纯 IP `rate_limit`);单码失败上限属验证码安全底线,保留。
|
||||
1. 单号 `SMS_SEND_INTERVAL_SEC` 冷却(单号维度)
|
||||
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`
|
||||
3. 单码校验失败 `SMS_MAX_VERIFY_ATTEMPTS` 次即作废 + 验过即作废(一次性)
|
||||
4. 运维侧建议在极光控制台叠加:**IP 白名单**(只许服务器 IP)+ **防轰炸设置**
|
||||
|
||||
## 极光错误码(节选,映射在 `_send_via_jiguang`)
|
||||
| code | 含义 | 处理 |
|
||||
@@ -56,4 +55,4 @@
|
||||
3. 真机发一条验证:收到短信 + 能登录
|
||||
|
||||
## 已知局限
|
||||
**验证码存进程内存**:单 worker uvicorn 够用;重启丢码(用户重发即可);**多 worker / 多机不共享 → 冷却 / 每日上限 / 校验失效**,扩 worker 前迁移到 DB/Redis。见 [待办与技术债](../guides/待办与技术债.md)。
|
||||
**验证码存进程内存**:单 worker uvicorn 够用;重启丢码(用户重发即可);**多 worker / 多机不共享 → 冷却 / 校验失效**,扩 worker 前迁移到 DB/Redis。见 [待办与技术债](../guides/待办与技术债.md)。
|
||||
|
||||
+1
-1
@@ -169,7 +169,7 @@ POST /api/v1/auth/sms/login { phone, code } → 任意 6 位通过 → upsert
|
||||
|
||||
短信冷却表存进程内存(`--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 与刷新
|
||||
|
||||
|
||||
@@ -476,3 +476,100 @@ def test_withdraw_reconcile(admin_client: TestClient, finance_token: str, monkey
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
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
@@ -17,7 +17,6 @@ def _reset(phone: str) -> None:
|
||||
"""清该号的进程内存状态,隔离 real 模式用例。"""
|
||||
sms._codes.pop(phone, None)
|
||||
sms._last_sent.pop(phone, None)
|
||||
sms._daily_count.pop(phone, None)
|
||||
|
||||
|
||||
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:
|
||||
"""发码防刷:同一设备(device_id) + 同一 IP 每小时最多 N 次发码,超出 429。
|
||||
用不同手机号(绕开单号 60s 冷却)证明限流按设备封顶 —— 堵「换号绕开单号冷却/每日上限」的洞。
|
||||
用不同手机号(绕开单号 60s 冷却)证明限流按设备封顶 —— 堵「换号绕开单号冷却」的洞。
|
||||
conftest 默认关限流;本用例临时打开 + 调小阈值便于测 + 清计数隔离。"""
|
||||
from app.api.v1 import auth
|
||||
from app.core import ratelimit
|
||||
@@ -155,21 +154,6 @@ def test_phone_format_validation(client) -> None:
|
||||
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 路径(不真发)============================
|
||||
|
||||
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:
|
||||
"""GC 清过期码 / 旧冷却 / 隔日计数,但不动今天有效的(阈值设 0 强制每次扫)。"""
|
||||
"""GC 清过期码 / 旧冷却,但不动今天有效的(阈值设 0 强制每次扫)。"""
|
||||
monkeypatch.setattr(sms, "_GC_THRESHOLD", 0)
|
||||
sms._codes.clear()
|
||||
sms._last_sent.clear()
|
||||
sms._daily_count.clear()
|
||||
now = time.time()
|
||||
sms._codes["stale"] = sms._CodeRecord(code="111111", expires_at=now - 1)
|
||||
sms._codes["fresh"] = sms._CodeRecord(code="222222", expires_at=now + 999)
|
||||
sms._last_sent["old"] = now - 99999
|
||||
sms._last_sent["recent"] = now
|
||||
sms._daily_count["yesterday"] = ("2000-01-01", 3)
|
||||
sms._daily_count["today"] = (sms._today(), 1)
|
||||
|
||||
sms._gc(now)
|
||||
|
||||
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 "yesterday" not in sms._daily_count and "today" in sms._daily_count
|
||||
|
||||
|
||||
# ============================ 用户名 / 默认昵称 ============================
|
||||
|
||||
@@ -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 # 绑上了登录用户
|
||||
@@ -20,8 +20,10 @@ def _auth(token: str) -> dict[str, str]:
|
||||
|
||||
def _report_success(client, token: str, trace_id: str) -> None:
|
||||
"""上报一条成功比价(有非源有效价 → status 派生 success)。"""
|
||||
# trace_id 现全局唯一(prod 由 app-server 签发 UUID);测试共用 session DB,故按 token
|
||||
# (每用户不同)加前缀防跨用户/跨文件复用 "s-1" 撞车(见 comparison_record_trace_unique 迁移)。
|
||||
payload = {
|
||||
"trace_id": trace_id,
|
||||
"trace_id": f"{token[-16:]}:{trace_id}",
|
||||
"business_type": "food",
|
||||
"store_name": "测试店",
|
||||
"source_platform_id": "taobao_flash",
|
||||
@@ -40,7 +42,7 @@ def _report_success(client, token: str, trace_id: str) -> None:
|
||||
def _report_failed(client, token: str, trace_id: str) -> None:
|
||||
"""上报一条失败比价(只有源 → status 派生 failed,不计入解锁)。"""
|
||||
payload = {
|
||||
"trace_id": trace_id,
|
||||
"trace_id": f"{token[-16:]}:{trace_id}",
|
||||
"business_type": "food",
|
||||
"comparison_results": [
|
||||
{"platform_id": "taobao_flash", "platform_name": "淘宝闪购", "price": 30.0,
|
||||
|
||||
@@ -9,6 +9,7 @@ mock 掉对 pricebot 的 httpx 调用,验证(两个端点参数化同跑):
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import httpx
|
||||
@@ -41,7 +42,8 @@ def _stub_body() -> dict:
|
||||
|
||||
@pytest.mark.parametrize("path,upstream", ENDPOINTS)
|
||||
def test_passes_body_through(client, path, upstream) -> None:
|
||||
"""无 token + pricebot 200 → 响应原样透传,body 原样转发,URL 去掉 /v1。"""
|
||||
"""无 token(软鉴权放行)+ pricebot 200 → 响应透传(+ app-server 注入 trace_id),
|
||||
body 原样转发(自带 trace_id 时不重签),URL 去掉 /v1。"""
|
||||
fake_resp = {
|
||||
"success": True,
|
||||
"action": {"command": "wait", "params": {"duration_ms": 500}},
|
||||
@@ -49,20 +51,25 @@ def test_passes_body_through(client, path, upstream) -> None:
|
||||
}
|
||||
captured: dict = {}
|
||||
|
||||
async def fake_post(self, url, json=None, **kw):
|
||||
async def fake_post(self, url, content=None, **kw):
|
||||
# 透传壳用 content=raw(bytes)转发,不是 json=;这里捕获 content 解回来核对
|
||||
captured["url"] = url
|
||||
captured["json"] = json
|
||||
captured["content"] = content
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.status_code = 200
|
||||
mock_resp.json = lambda: fake_resp
|
||||
mock_resp.json = lambda: dict(fake_resp) # 每次新副本(端点会 setdefault trace_id 进去)
|
||||
return mock_resp
|
||||
|
||||
with patch.object(httpx.AsyncClient, "post", fake_post):
|
||||
r = client.post(path, json=_stub_body())
|
||||
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json() == fake_resp
|
||||
assert captured["json"] == _stub_body() # body 原样转发(不鉴权,无需 token)
|
||||
body = r.json()
|
||||
assert body["success"] is True
|
||||
assert body["action"] == fake_resp["action"]
|
||||
assert body["trace_id"] == "test-trace-1" # 自带 trace_id → 原样回传, 不 mint
|
||||
# body 原样转发(content=raw bytes;自带 trace_id 时不重新序列化)
|
||||
assert json.loads(captured["content"]) == _stub_body()
|
||||
assert captured["url"].endswith(upstream) # /api/v1/xxx → /api/xxx
|
||||
|
||||
|
||||
|
||||
@@ -1,14 +1,18 @@
|
||||
"""注销账号(DELETE /api/v1/user):软删 + 释放唯一约束 + 资金前置校验。
|
||||
"""注销账号(DELETE /api/v1/user):数据级联清理 + 软删 + 释放唯一约束 + 资金前置校验。
|
||||
|
||||
覆盖:
|
||||
- 干净注销:status=deleted + phone 匿名化 + wechat_openid/invite_code 等唯一槽全部释放
|
||||
- 回归原始 bug:注销释放 openid 后,别的账号能绑同一个微信
|
||||
- 资金前置闸:有可提现现金 / 在审提现单 → 409 拒绝注销,账号不被删
|
||||
- 注销后旧 token 即时失效(status==active 闸)
|
||||
- 级联清理:个人内容/成绩/行为/PII 表被物理删空、余额清零;财务流水/邀请关系/广告幂等
|
||||
+对账表按策略保留(soft_delete_account 的分类落地断言)
|
||||
"""
|
||||
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.models.user import User
|
||||
@@ -139,3 +143,130 @@ def test_deleted_user_old_token_is_rejected(client) -> None:
|
||||
# 注销后旧 token 立即失效(get_current_user 校验 status==active)
|
||||
r = client.get("/api/v1/user/onboarding/status", params={"device_id": "d"}, headers=_auth(token))
|
||||
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
|
||||
|
||||
@@ -65,3 +65,28 @@ def test_onboarding_missing_device_id_treated_as_incomplete(client) -> None:
|
||||
phone = "13800138003"
|
||||
data = _login(client, phone, device_id=None)
|
||||
assert data["onboarding_completed"] is False
|
||||
|
||||
|
||||
def _reset(client, access: str, device_id: str):
|
||||
return client.post(
|
||||
"/api/v1/user/onboarding/reset",
|
||||
json={"device_id": device_id},
|
||||
headers={"Authorization": f"Bearer {access}"},
|
||||
)
|
||||
|
||||
|
||||
def test_onboarding_reset_makes_relogin_reonboard(client) -> None:
|
||||
"""重置(删完成标记)后,同 (账号, 设备) 再登录 → onboarding_completed=False,重走引导。
|
||||
对应客户端「开发设置 → 重置新手引导」:先删后端行,再退登录重开。"""
|
||||
phone = "13800138004"
|
||||
dev = "android-id-reset"
|
||||
|
||||
# ① 走完引导 → 标记完成 → 再登录已完成
|
||||
access = _login(client, phone, dev)["access_token"]
|
||||
assert _mark_complete(client, access, dev).status_code == 200
|
||||
assert _login(client, phone, dev)["onboarding_completed"] is True
|
||||
|
||||
# ② 重置(幂等:重复重置也 200)→ 再登录变未完成 → 重走引导
|
||||
assert _reset(client, access, dev).status_code == 200
|
||||
assert _reset(client, access, dev).json()["ok"] is True
|
||||
assert _login(client, phone, dev)["onboarding_completed"] is False
|
||||
|
||||
Reference in New Issue
Block a user