功能：支持动态配置每日比价次数上限

alembic/versions/6d2309208549_merge_comparison_platforms_fail_reason_.py

@@ -1,26 +0,0 @@
-"""merge comparison platforms + fail_reason heads
-
-Revision ID: 6d2309208549
-Revises: comparison_platforms_col, comparison_record_fail_reason
-Create Date: 2026-07-29 01:48:41.868083
-
-"""
-from typing import Sequence, Union
-
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision: str = '6d2309208549'
-down_revision: Union[str, Sequence[str], None] = ('comparison_platforms_col', 'comparison_record_fail_reason')
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    pass
-
-
-def downgrade() -> None:
-    pass

alembic/versions/comparison_platforms_col.py

@@ -1,47 +0,0 @@
-"""add platforms unified array column to comparison_record
-
-展示模型统一数组(pricebot done.params.platforms 原样存): 每平台一行、自带
-status/is_best/display, 记录页据此直接渲染, 不再靠 comparison_results + 客户端合并 + 前端派生。
-纯新增列, 老记录为空 → 前端回退老 comparison_results。
-
-Revision ID: comparison_platforms_col
-Revises: user_manual_risk_fields
-"""
-from __future__ import annotations
-
-from collections.abc import Sequence
-
-import sqlalchemy as sa
-from sqlalchemy.dialects import postgresql
-
-from alembic import op
-
-revision: str = "comparison_platforms_col"
-down_revision: str | None = "user_manual_risk_fields"
-branch_labels: str | Sequence[str] | None = None
-depends_on: str | Sequence[str] | None = None
-
-_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
-
-
-def upgrade() -> None:
-    # 幂等: 线上为了提前给历史数据补 platforms(2026-07-29), 已手动
-    # `ALTER TABLE comparison_record ADD COLUMN IF NOT EXISTS platforms jsonb
-    #  NOT NULL DEFAULT '[]'::jsonb`(与本 migration 定义一致)。列已存在时跳过,
-    # 否则上线 alembic upgrade head 会撞 DuplicateColumn 直接部署失败。
-    bind = op.get_bind()
-    cols = {c["name"] for c in sa.inspect(bind).get_columns("comparison_record")}
-    if "platforms" in cols:
-        return
-    with op.batch_alter_table("comparison_record") as batch_op:
-        batch_op.add_column(
-            sa.Column(
-                "platforms", _JSON, nullable=False,
-                server_default=sa.text("'[]'"),
-            )
-        )
-
-
-def downgrade() -> None:
-    with op.batch_alter_table("comparison_record") as batch_op:
-        batch_op.drop_column("platforms")

app/admin/schemas/risk_monitor.py

@@ -23,7 +23,7 @@ class RiskMonitorSummary(BaseModel):
 class RiskRuleConfig(BaseModel):
     sms_hourly_threshold: int = Field(ge=1, le=5)
     oneclick_daily_threshold: int = Field(ge=1, le=100_000)
-    compare_daily_threshold: int = Field(ge=1, le=100)
+    compare_daily_threshold: int = Field(ge=1, le=100_000)
 
 
 class RiskIncidentItem(BaseModel):

app/api/v1/compare_record.py

@@ -38,13 +38,14 @@ router = APIRouter(prefix="/api/v1/compare", tags=["compare-record"])
 @router.post(
     "/start",
     response_model=CompareStartReserveOut,
-    summary="预占一次当日比价发起次数（每人每天最多100次）",
+    summary="预占一次当日比价发起次数（上限由风控监控配置）",
 )
 def reserve_compare_start(
     payload: CompareStartReserveIn,
     user: CurrentUser,
     db: DbSession,
 ) -> CompareStartReserveOut:
+    daily_limit = risk_repo.get_rule_threshold(db, risk_repo.RULE_COMPARE_DAILY)
     if risk_repo.is_restricted(
         db,
         subject_type="user",
@@ -57,25 +58,26 @@ def reserve_compare_start(
             db,
             user_id=user.id,
             trace_id=payload.trace_id,
+            daily_limit=daily_limit,
             business_type=payload.business_type,
             device_id=payload.device_id,
         )
     except crud_compare.DailyCompareStartLimitExceeded:
         raise HTTPException(
             status_code=status.HTTP_429_TOO_MANY_REQUESTS,
-            detail="今日已比价超过100次，请明天再试",
+            detail=f"今日已比价超过{daily_limit}次，请明天再试",
         ) from None
     except crud_compare.ComparisonTraceOwnershipError:
         raise HTTPException(
             status_code=status.HTTP_409_CONFLICT,
             detail="比价任务标识冲突，请重新发起",
         ) from None
-    # 风控阈值由后台动态配置，不能再只在固定的 100 次业务上限处同步。
+    # 告警和业务限流共用同一动态阈值，避免后台已修改但用户侧仍固定为 100 次。
     risk_repo.sync_compare_incident(db, user_id=user.id, at=rec.created_at)
     return CompareStartReserveOut(
-        limit=crud_compare.DAILY_COMPARE_START_LIMIT,
+        limit=daily_limit,
         used=used,
-        remaining=max(crud_compare.DAILY_COMPARE_START_LIMIT - used, 0),
+        remaining=max(daily_limit - used, 0),
     )
 
 

app/core/config_schema.py

@@ -134,12 +134,12 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
     },
     RISK_COMPARE_DAILY_THRESHOLD_KEY: {
         "default": 100,
-        "label": "比价账户每日告警阈值",
+        "label": "比价账户每日上限与告警阈值",
         "group": "风控",
         "type": "int",
         "min": 1,
-        "max": 100,
+        "max": 100_000,
         "hidden": True,
-        "help": "同一账户在北京时间同一自然日内发起比价达到该次数时告警。",
+        "help": "同一账户按北京时间自然日最多可发起的比价次数；达到该次数时同步告警。",
     },
 }

app/models/comparison.py

@@ -113,10 +113,6 @@ class ComparisonRecord(Base):
     items: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
     # 逐平台对比 [{platform_id, platform_name, package, price, is_source, rank, coupon_saved, coupon_name, applied_coupons}](price/coupon_saved 单位:元,原样存;coupon_name=优惠来源名;applied_coupons=[{name,amount}] 多券明细)
     comparison_results: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
-    # 展示模型统一数组(pricebot done.params.platforms 原样存): 每平台一行、自带
-    # status/is_best/display/display_order, 记录页据此直接渲染, 不再靠 comparison_results
-    # + 客户端合并 + 前端派生。老记录/旧客户端为空 → 前端回退老 comparison_results 渲染。
-    platforms: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
     # 目标平台未找到、跳过的菜名
     skipped_dish_names: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
     # 客户端上报的原始 payload(calibration + done.params 全量),未来取数兜底

app/repositories/comparison.py

@@ -17,8 +17,6 @@ from app.models.savings import SavingsRecord
 from app.models.user import User
 from app.schemas.compare_record import ComparisonRecordIn
 
-DAILY_COMPARE_START_LIMIT = 100
-
 
 class DailyCompareStartLimitExceeded(Exception):
     """The authenticated user has consumed today's comparison-start quota."""
@@ -164,9 +162,8 @@ def _derive(payload: ComparisonRecordIn) -> dict:
 
     is_source_best = best.is_source if best is not None else None
 
-    # status:优先 pricebot record_status(区分 below_minimum/store_closed) → 客户端显式 status
-    # → 兜底"非源且有价"=success/否则 failed。record_status 让"未满起送"不再塌缩成 failed。
-    status = payload.record_status or payload.status
+    # status:客户端显式给了就用;否则有"非源且有价"的结果=success,否则 failed
+    status = payload.status
     if status is None:
         has_valid_target = any(
             (not r.is_source) and r.price is not None for r in results
@@ -198,35 +195,16 @@ def upsert_record(
     灰度期老客户端 POST /compare/record 走这条,与后端 harvest 按 trace_id reconcile;
     新客户端不再 POST(改由 compare.py 透传壳 harvest 落库)。
     """
-    # 单源派生: 与 harvest_done 一致, payload 带 platforms 时从它派生(唯一真相源
-    # _derive_from_platforms), 老客户端不带 platforms 时回退 _derive(从 comparison_results)。
-    if payload.platforms:
-        derived = _derive_from_platforms(payload.platforms, payload.record_status)
-        # 对齐 _derive 返回键(#189 fail_reason): 两路径 fields 键集一致, 覆盖已有行时不残留旧值
-        derived["fail_reason"] = (
-            _derive_fail_display(payload.information, payload.platform_results or {})
-            if derived["status"] == "failed"
-            else None
-        )
-        # 单源派生取自 platforms 源行(常无源平台元数据/店名)→ 空则用 payload 兜底不丢字段。
-        # 下面 fields 不再显式写这四个键, 统一由 derived 提供(否则 dict(store_name=..., **derived)
-        # 与 _derive_from_platforms 同名键撞键 TypeError)。
-        for _k in ("store_name", "source_platform_id", "source_platform_name", "source_package"):
-            if not derived.get(_k):
-                derived[_k] = getattr(payload, _k)
-    else:
-        derived = _derive(payload)
-        # _derive 只从 comparison_results 派生, 不含源平台四件套 / store_name → 从 payload 补,
-        # 与上面 platforms 分支键集对齐(fields 统一靠 **derived 提供这些列)。
-        for _k in ("store_name", "source_platform_id", "source_platform_name", "source_package"):
-            derived[_k] = getattr(payload, _k)
+    derived = _derive(payload)
     items = [it.model_dump(exclude_none=True) for it in payload.items]
     fields = dict(
         device_id=payload.device_id,
         business_type=payload.business_type,
+        store_name=payload.store_name,
         product_names=_product_names_from_items(items),
-        # store_name / source_platform_id / source_platform_name / source_package 统一由
-        # derived 提供(见上方两分支补齐), 不在此显式写 —— 否则与 _derive_from_platforms 撞键。
+        source_platform_id=payload.source_platform_id,
+        source_platform_name=payload.source_platform_name,
+        source_package=payload.source_package,
         information=payload.information,
         best_deeplink=payload.best_deeplink,
         trace_url=payload.trace_url,
@@ -234,7 +212,6 @@ def upsert_record(
         skipped_dish_count=payload.skipped_dish_count,
         items=items,
         comparison_results=[r.model_dump() for r in payload.comparison_results],
-        platforms=list(payload.platforms or []),
         skipped_dish_names=list(payload.skipped_dish_names),
         # 客户端环境 / 性能(debug,客户端上报;旧客户端为 None)
         device_model=payload.device_model,
@@ -307,8 +284,7 @@ def upsert_record(
 
 
 def _derive_from_results(
-    results: list[dict], platform_results: dict | None = None,
-    record_status: str | None = None,
+    results: list[dict], platform_results: dict | None = None
 ) -> dict:
     """从 done 帧 comparison_results(pricebot 原始 dict 列表)派生结构化列。
     等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。
@@ -356,53 +332,7 @@ def _derive_from_results(
         "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,
-        # 记录级结局: 优先用 pricebot 下发的 record_status(区分 below_minimum/store_closed,
-        # 不再把"未满起送"塌缩成 failed → 记录页不再误报"网络开小差"); 旧 pricebot 未下发时
-        # 回退老的 success/failed 二态派生, 向后兼容。
-        "status": record_status or ("success" if has_valid_target else "failed"),
-    }
-
-
-def _derive_from_platforms(
-    platforms: list, record_status: str | None = None,
-) -> dict:
-    """从 done 帧 platforms(每平台一行、渲染就绪)派生结构化列——**单一真相源**。
-
-    best_* 直接取 platforms 里 is_best 的那一行、source_* 取 role=source 行,与前端读的
-    platforms 天然一致(不再像 _derive_from_results 那样从 comparison_results 二次评最优,
-    消除"标量列 vs platforms"双源不一致)。platforms 非空时优先走这里;老 pricebot 无
-    platforms 时调用方回退 _derive_from_results(向后兼容)。"""
-    rows = [p for p in (platforms or []) if isinstance(p, dict)]
-    src = next((p for p in rows if p.get("role") == "source"), None)
-    best = next((p for p in rows if p.get("is_best")), None)
-    source_price_cents = _yuan_to_cents(src.get("price")) if src else None
-    best_price_cents = _yuan_to_cents(best.get("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(
-        p.get("role") != "source" and p.get("price") is not None for p in rows
-    )
-    # store_name: 优先源行; recompare 场景源平台自己当目标、源行被目标覆盖(pricebot
-    # _build_platform_rows 有意去重, platforms 无 role=source 行)→ 回退 best 行 → 首个有店名
-    # 的行(显示现场实际比到的店), 免得记录页店名空掉兜底显示成"比价"。正常比价有源行不走回退。
-    store_name = (
-        (src or {}).get("store_name")
-        or (best or {}).get("store_name")
-        or next((p.get("store_name") for p in rows if p.get("store_name")), None)
-    )
-    return {
-        "source_platform_id": (src or {}).get("platform_id"),
-        "source_platform_name": (src or {}).get("platform_name"),
-        "source_package": (src or {}).get("package"),
-        "source_price_cents": source_price_cents,
-        "best_platform_id": (best or {}).get("platform_id"),
-        "best_platform_name": (best or {}).get("platform_name"),
-        "best_price_cents": best_price_cents,
-        "saved_amount_cents": saved_amount_cents,
-        "is_source_best": (best.get("role") == "source") if best else None,
-        "store_name": store_name or None,
-        "status": record_status or ("success" if has_valid_target else "failed"),
+        "status": "success" if has_valid_target else "failed",
     }
 
 
@@ -431,11 +361,12 @@ def reserve_daily_start(
     *,
     user_id: int,
     trace_id: str,
+    daily_limit: int,
     business_type: str = "food",
     device_id: str | None = None,
     now: datetime | None = None,
 ) -> tuple[ComparisonRecord, int]:
-    """Atomically reserve one of a user's 100 Beijing-day comparison starts.
+    """Atomically reserve one of a user's configured Beijing-day comparison starts.
 
     ``trace_id`` makes client retries idempotent. Locking the user row serializes
     concurrent starts for one account, so parallel requests cannot both consume
@@ -481,7 +412,7 @@ def reserve_daily_start(
             ComparisonRecord.created_at < day_end,
         )
     ) or 0
-    if used >= DAILY_COMPARE_START_LIMIT:
+    if used >= daily_limit:
         raise DailyCompareStartLimitExceeded
 
     rec = ComparisonRecord(
@@ -561,29 +492,7 @@ def harvest_done(
     返回 (记录, 是否本次**新**落成 success)——供调用方据此幂等发一次邀请奖。
     行不存在(理论上帧0已建;防御)则新建。"""
     results = done_params.get("comparison_results") or []
-    # 展示模型统一数组(pricebot 新增, 每平台一行自带 status/is_best): 原样存, 记录页据此直渲染。
-    # record_status: 记录级结局(success/below_minimum/store_closed/failed), 覆盖老二态派生。
-    platforms = done_params.get("platforms") or []
-    record_status = done_params.get("record_status")
-    # 单源派生: platforms(含 pricebot 权威 is_best)是唯一真相源, best_*/source_*/saved/status
-    # 全从它取 → 与前端读的 platforms 天然一致; 菜品也取 platforms 源行。老 pricebot 无
-    # platforms 时回退从 comparison_results 派生(向后兼容)。
-    if platforms:
-        derived = _derive_from_platforms(platforms, record_status)
-        # 菜品优先源行; recompare 无源行 → 回退 best 行 → 首个有菜品的行(同 store_name 回退)
-        _item_row = (
-            next((p for p in platforms if isinstance(p, dict) and p.get("role") == "source"), None)
-            or next((p for p in platforms if isinstance(p, dict) and p.get("is_best")), None)
-            or next((p for p in platforms if isinstance(p, dict) and p.get("items")), None)
-        )
-        items = (_item_row or {}).get("items") or []
-    else:
-        derived = _derive_from_results(
-            results, done_params.get("platform_results"), record_status
-        )
-        # pricebot 已把源单菜品塞进 comparison_results[源行].items
-        items = next((r.get("items") or [] for r in results if r.get("is_source")), [])
-    # 失败展示原因(#189): platforms / results 两个派生分支的 status 都可能 failed, 统一在此算
+    derived = _derive_from_results(results, done_params.get("platform_results"))
     fail_reason = (
         _derive_fail_display(
             done_params.get("information"), done_params.get("platform_results")
@@ -591,6 +500,8 @@ def harvest_done(
         if derived["status"] == "failed"
         else None
     )
+    # 菜品:pricebot 已把源单菜品塞进 comparison_results[源行].items
+    items = next((r.get("items") or [] for r in results if r.get("is_source")), [])
     fields = dict(
         business_type=business_type or "food",
         information=done_params.get("information") or None,
@@ -602,7 +513,6 @@ def harvest_done(
         skipped_dish_count=done_params.get("skipped_dish_count"),
         skipped_dish_names=list(done_params.get("skipped_dish_names") or []),
         comparison_results=results,
-        platforms=platforms,
         items=items,
         product_names=_product_names_from_items(items),
         raw_payload=done_params,

app/schemas/compare_record.py

@@ -107,13 +107,6 @@ class ComparisonRecordIn(BaseModel):
     # 明细
     items: list[ComparisonItemIn] = Field(default_factory=list)
     comparison_results: list[ComparisonResultIn] = Field(default_factory=list)
-    # 展示模型统一数组(pricebot done.params.platforms 原样透传): 每平台一行、自带
-    # status/is_best/display/display_order,记录页据此直渲染。宽松 list[dict] 存(结构由
-    # pricebot 定,server 只原样落库),前端读它、老记录空时回退 comparison_results。
-    platforms: list[dict] = Field(default_factory=list)
-    # 记录级结局(pricebot 下发): success/below_minimum/store_closed/failed。让"未满起送"不再
-    # 被塌缩成 failed。_derive 优先用它、其次客户端 status、再兜底二态派生。
-    record_status: str | None = None
     # 逐平台结局摘要(含失败平台的细分原因 status: store_not_found/items_not_found/below_minimum/
     # unsupported/...)。来自 done.params.platform_results,客户端透传;落 raw_payload(不单列),
     # admin「卡在哪一步」从这里读。dict{platform_id: {...}} 宽松存(结构由 pricebot 定——是
@@ -184,9 +177,6 @@ class ComparisonRecordOut(BaseModel):
     fail_reason: str | None = None
     items: list = []
     comparison_results: list = []
-    # 展示模型统一数组(每平台一行、自带 status/is_best/display/display_order): 记录页据此
-    # 直渲染, 不再靠 comparison_results + 前端派生。老记录为空 → 前端回退 comparison_results。
-    platforms: list = []
     skipped_dish_names: list = []
     total_ms: int | None = None
     # 「已下单」(店级):该店名在该用户真实下单(source='compare')里出现过即 True。

tests/test_compare_daily_limit.py

@@ -3,14 +3,32 @@ from __future__ import annotations
 import time
 from datetime import datetime, timedelta
 
+import pytest
 from sqlalchemy import func, select
 
+from app.core.config_schema import RISK_COMPARE_DAILY_THRESHOLD_KEY
 from app.core.rewards import CN_TZ
 from app.core.security import decode_token
 from app.db.session import SessionLocal
+from app.models.app_config import AppConfig
 from app.models.comparison import ComparisonRecord
 
 
+@pytest.fixture(autouse=True)
+def _reset_compare_daily_limit():
+    with SessionLocal() as db:
+        db.query(AppConfig).filter(
+            AppConfig.key == RISK_COMPARE_DAILY_THRESHOLD_KEY
+        ).delete()
+        db.commit()
+    yield
+    with SessionLocal() as db:
+        db.query(AppConfig).filter(
+            AppConfig.key == RISK_COMPARE_DAILY_THRESHOLD_KEY
+        ).delete()
+        db.commit()
+
+
 def _login(client) -> tuple[str, int]:
     phone = f"137{int(time.time() * 1000) % 100000000:08d}"
     sent = client.post("/api/v1/auth/sms/send", json={"phone": phone})
@@ -118,3 +136,33 @@ def test_compare_start_rejects_101st_beijing_day_attempt(client) -> None:
                 ComparisonRecord.trace_id == rejected_trace
             )
         ) == 0
+
+
+def test_compare_start_uses_dynamic_risk_monitor_limit(client) -> None:
+    with SessionLocal() as db:
+        db.add(AppConfig(key=RISK_COMPARE_DAILY_THRESHOLD_KEY, value=2))
+        db.commit()
+
+    token, user_id = _login(client)
+    first = client.post(
+        "/api/v1/compare/start",
+        json={"trace_id": f"quota-dynamic-{user_id}-1"},
+        headers=_headers(token),
+    )
+    second = client.post(
+        "/api/v1/compare/start",
+        json={"trace_id": f"quota-dynamic-{user_id}-2"},
+        headers=_headers(token),
+    )
+    rejected = client.post(
+        "/api/v1/compare/start",
+        json={"trace_id": f"quota-dynamic-{user_id}-3"},
+        headers=_headers(token),
+    )
+
+    assert first.status_code == 200
+    assert first.json() == {"limit": 2, "used": 1, "remaining": 1}
+    assert second.status_code == 200
+    assert second.json() == {"limit": 2, "used": 2, "remaining": 0}
+    assert rejected.status_code == 429
+    assert rejected.json()["detail"] == "今日已比价超过2次，请明天再试"

tests/test_risk_monitor.py

@@ -529,6 +529,18 @@ def test_admin_can_edit_rules_and_current_window_is_reconciled() -> None:
                 now + timedelta(seconds=2)
             ).replace(tzinfo=None)
 
+        compare_limit = client.patch(
+            "/admin/api/risk-monitor/rules",
+            headers=headers,
+            json={
+                "sms_hourly_threshold": 3,
+                "oneclick_daily_threshold": 20,
+                "compare_daily_threshold": 120,
+            },
+        )
+        assert compare_limit.status_code == 200
+        assert compare_limit.json()["compare_daily_threshold"] == 120
+
         invalid = client.patch(
             "/admin/api/risk-monitor/rules",
             headers=headers,