功能：统一限制策略与白名单管理

新增统一限制规则、临时不限和风控免告警白名单，接入比价、短信登录、广告、引导视频与账号冷却等业务链路。补充设备选择、权限、审计、迁移及回归测试。

alembic/versions/6d2309208549_merge_comparison_platforms_fail_reason_.py

@@ -0,0 +1,26 @@
+"""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

@@ -0,0 +1,47 @@
+"""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")

alembic/versions/comparison_record_fail_reason.py

@@ -0,0 +1,49 @@
+"""comparison_record.fail_reason (失败卡展示原因)
+
+Revision ID: comparison_record_fail_reason
+Revises: user_manual_risk_fields
+Create Date: 2026-07-28 12:00:00.000000
+
+失败记录的展示原因:information 具体则=它;笼统则由写路径从 platform_results 捞出的
+业务原因;纯系统失败为 None(端侧品牌兜底)。见 repositories.comparison._derive_fail_display。
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'comparison_record_fail_reason'
+down_revision: Union[str, Sequence[str], None] = 'user_manual_risk_fields'
+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:
+        batch_op.add_column(sa.Column('fail_reason', sa.String(length=256), nullable=True))
+    # 回填老失败记录:information 具体的直接搬过来(笼统/系统失败留 None → 端侧品牌兜底)。
+    # 新记录由写路径 _derive_fail_display 落库(含 platform_results 救援/补判),不走这条。
+    # platform_results 只在 raw_payload 里,SQL 里不易解析,故老记录不做救援/补判(可接受:
+    # 老 mixed/打烊记录回退品牌兜底);具体 information 的老记录本次即可显示真实原因。
+    op.execute(
+        """
+        UPDATE comparison_record
+        SET fail_reason = information
+        WHERE status = 'failed'
+          AND information IS NOT NULL
+          AND information <> ''
+          AND information NOT IN (
+              '比价过程出错，请稍后重试',
+              '比价出错',
+              '比价未完成',
+              'done 参数缺少可验证的目标平台结果'
+          )
+        """
+    )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table('comparison_record', schema=None) as batch_op:
+        batch_op.drop_column('fail_reason')

alembic/versions/guide_video_scene_unique.py

@@ -0,0 +1,31 @@
+"""guide video play count is independent for coupon and comparison
+
+Revision ID: guide_video_scene_unique
+Revises: 6d2309208549
+"""
+from alembic import op
+
+revision = "guide_video_scene_unique"
+down_revision = "6d2309208549"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.drop_index("uq_guide_video_play_user_seq", table_name="guide_video_play")
+    op.create_index(
+        "uq_guide_video_play_user_scene_seq",
+        "guide_video_play",
+        ["user_id", "scene", "seq"],
+        unique=True,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("uq_guide_video_play_user_scene_seq", table_name="guide_video_play")
+    op.create_index(
+        "uq_guide_video_play_user_seq",
+        "guide_video_play",
+        ["user_id", "seq"],
+        unique=True,
+    )

alembic/versions/limit_policy_whitelist.py

@@ -0,0 +1,128 @@
+"""add per-subject limit policy whitelist
+
+Revision ID: limit_policy_whitelist
+Revises: guide_video_scene_unique
+"""
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+from alembic import op
+
+revision: str = "limit_policy_whitelist"
+down_revision: str | Sequence[str] | None = "guide_video_scene_unique"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
+_PAGE = "limit-whitelist"
+_DEFAULT_ROLES = ("operator", "tech")
+
+
+def _role_table() -> sa.TableClause:
+    return sa.table(
+        "admin_role",
+        sa.column("name", sa.String),
+        sa.column("pages", _JSON),
+    )
+
+
+def _add_default_role_permissions() -> None:
+    """Grant the page without replacing existing role customisations."""
+    role = _role_table()
+    conn = op.get_bind()
+    rows = conn.execute(
+        sa.select(role.c.name, role.c.pages).where(
+            role.c.name.in_(_DEFAULT_ROLES)
+        )
+    ).all()
+    for name, pages in rows:
+        current_pages = list(pages or [])
+        if _PAGE not in current_pages:
+            conn.execute(
+                role.update()
+                .where(role.c.name == name)
+                .values(pages=[*current_pages, _PAGE])
+            )
+
+
+def _remove_default_role_permissions() -> None:
+    role = _role_table()
+    conn = op.get_bind()
+    rows = conn.execute(
+        sa.select(role.c.name, role.c.pages).where(
+            role.c.name.in_(_DEFAULT_ROLES)
+        )
+    ).all()
+    for name, pages in rows:
+        current_pages = list(pages or [])
+        if _PAGE in current_pages:
+            conn.execute(
+                role.update()
+                .where(role.c.name == name)
+                .values(
+                    pages=[page for page in current_pages if page != _PAGE]
+                )
+            )
+
+
+def upgrade() -> None:
+    op.create_table(
+        "limit_policy_override",
+        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column("subject_type", sa.String(length=16), nullable=False),
+        sa.Column("subject_value", sa.String(length=128), nullable=False),
+        sa.Column("rule_code", sa.String(length=64), nullable=False),
+        sa.Column("mode", sa.String(length=24), nullable=False),
+        sa.Column("limit_value", sa.Integer(), nullable=True),
+        sa.Column(
+            "enabled", sa.Boolean(), server_default=sa.true(), nullable=False
+        ),
+        sa.Column("starts_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("reset_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("reason", sa.String(length=256), nullable=True),
+        sa.Column("created_by_admin_id", sa.Integer(), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.func.now(),
+            nullable=False,
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.func.now(),
+            nullable=False,
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "subject_type",
+            "subject_value",
+            "rule_code",
+            name="uq_limit_policy_subject_rule",
+        ),
+    )
+    op.create_index(
+        "ix_limit_policy_lookup",
+        "limit_policy_override",
+        ["subject_type", "subject_value", "rule_code", "enabled"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_limit_policy_expires",
+        "limit_policy_override",
+        ["expires_at"],
+        unique=False,
+    )
+    _add_default_role_permissions()
+
+
+def downgrade() -> None:
+    _remove_default_role_permissions()
+    op.drop_index("ix_limit_policy_expires", table_name="limit_policy_override")
+    op.drop_index("ix_limit_policy_lookup", table_name="limit_policy_override")
+    op.drop_table("limit_policy_override")

app/admin/main.py

@@ -31,6 +31,7 @@ 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.guide_video import router as guide_video_router
 from app.admin.routers.huawei_review import router as huawei_review_router
+from app.admin.routers.limit_whitelist import router as limit_whitelist_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
@@ -103,6 +104,7 @@ admin_app.include_router(wallet_router)
 admin_app.include_router(withdraw_router)
 admin_app.include_router(price_report_router)
 admin_app.include_router(risk_monitor_router)
+admin_app.include_router(limit_whitelist_router)
 admin_app.include_router(feedback_router)
 admin_app.include_router(event_logs_router)
 admin_app.include_router(analytics_health_router)

app/admin/permissions.py

@@ -40,6 +40,7 @@ PERMISSION_CATALOG: list[dict] = [
         {"key": "analytics-health", "label": "埋点成功率"},
         {"key": "event-logs", "label": "埋点日志"},
         {"key": "audit-logs", "label": "审计日志"},
+        {"key": "limit-whitelist", "label": "白名单"},
     ]},
     {"group": "其他", "pages": [
         {"key": "admins", "label": "权限管理"},
@@ -58,13 +59,14 @@ BUILTIN_ROLES: list[dict] = [
     {"name": "operator", "label": "运营", "pages": [
         "dashboard", "coupon-data", "ad-revenue-report", "comparison-records",
         "cps", "risk-monitor", "device-liveness", "price-reports", "feedbacks", "huawei-review",
+        "limit-whitelist",
     ]},
     {"name": "finance", "label": "财务", "pages": [
         "dashboard", "ad-revenue-report", "cps", "invite-withdraws", "withdraws",
     ]},
     {"name": "tech", "label": "技术", "pages": [
         "dashboard", "risk-monitor", "device-liveness", "analytics-health", "config", "ad-revenue", "huawei-review",
-        "event-logs", "audit-logs",
+        "event-logs", "audit-logs", "limit-whitelist",
     ]},
 ]
 

app/admin/repositories/ad_revenue.py

@@ -43,7 +43,7 @@ _KNOWN_PROD_BUSINESS_CODE_IDS = frozenset({"104098712", "104099389"})
 _TEST_BUSINESS_CODE_IDS = frozenset({"104127529", "104127626", "104137445"})
 
 
-def _business_code_ids(db: Session, app_env: str | None) -> set[str]:
+def business_code_ids(db: Session, app_env: str | None) -> set[str]:
     """返回指定应用环境下可用于业务收益对账的 GroMore 聚合代码位。"""
     prod_config = app_config.get_ad_config(db)
     prod_ids = set(_KNOWN_PROD_BUSINESS_CODE_IDS) | {
@@ -320,10 +320,10 @@ def ad_revenue_report(
 
     # 业务口径仅保留正式配置/测试业务链路实际使用的代码位。穿山甲“全量”还包含广告测试
     # demo、插屏等没有客户端收益上报的曝光，两边直接比较会天然产生假差额。
-    business_code_ids: set[str] | None = None
+    business_ids: set[str] | None = None
     if revenue_scope == "business":
-        business_code_ids = _business_code_ids(db, app_env)
-        events = [e for e in events if e.get("our_code_id") in business_code_ids]
+        business_ids = business_code_ids(db, app_env)
+        events = [e for e in events if e.get("our_code_id") in business_ids]
 
     # 排序:time=按时间倒序(新→旧);ecpm=按 eCPM 数值倒序(eCPM 原值是字符串「分」,转数值排;
     # 纯发奖行用其发奖采用的 eCPM,缺失/非法计 0 排末尾)。
@@ -381,7 +381,7 @@ def ad_revenue_report(
             date_from=date_from,
             date_to=date_to,
             app_env=app_env,
-            our_code_ids=business_code_ids,
+            our_code_ids=business_ids,
         )
         if pangle_aggs:
             by_date = {a["date"]: a for a in pangle_aggs}

app/admin/repositories/limit_whitelist.py

@@ -0,0 +1,455 @@
+"""CRUD and presentation helpers for limit policy overrides."""
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from sqlalchemy import String, cast, func, or_, select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from app.core import limit_policy
+from app.models.app_config import AppConfig
+from app.models.comparison import ComparisonRecord
+from app.models.limit_policy import LimitPolicyOverride
+from app.models.risk import BehaviorEvent
+from app.models.user import User
+from app.repositories import app_config
+from app.repositories import risk as risk_repo
+
+
+class DuplicateOverrideError(Exception):
+    pass
+
+
+_EVENT_SOURCE_LABELS = {
+    risk_repo.EVENT_SMS_SEND: "短信验证码",
+    risk_repo.EVENT_SMS_LOGIN: "短信登录",
+    risk_repo.EVENT_ONECLICK_LOGIN: "一键登录",
+}
+
+
+def _matching_users(keyword: str):
+    pattern = f"%{keyword}%"
+    return select(User.id).where(
+        or_(
+            cast(User.id, String).ilike(pattern),
+            User.username.ilike(pattern),
+            User.phone.ilike(pattern),
+            User.nickname.ilike(pattern),
+        )
+    )
+
+
+def _user_maps(
+    db: Session, *, user_ids: set[int], phones: set[str]
+) -> tuple[dict[int, User], dict[str, User]]:
+    conditions = []
+    if user_ids:
+        conditions.append(User.id.in_(user_ids))
+    if phones:
+        conditions.append(User.phone.in_(phones))
+    if not conditions:
+        return {}, {}
+    users = list(db.scalars(select(User).where(or_(*conditions))).all())
+    return {user.id: user for user in users}, {user.phone: user for user in users}
+
+
+def _behavior_device_candidates(
+    db: Session, *, keyword: str | None, limit: int
+) -> list[dict]:
+    conditions = [
+        BehaviorEvent.subject_type == limit_policy.SUBJECT_DEVICE,
+        BehaviorEvent.subject_id != "",
+        BehaviorEvent.subject_id.not_like(
+            f"{limit_policy.LEGACY_IP_DEVICE_PREFIX}%"
+        ),
+    ]
+    if keyword:
+        value = keyword.strip()
+        pattern = f"%{value}%"
+        matched_users = _matching_users(value)
+        conditions.append(
+            or_(
+                BehaviorEvent.subject_id.ilike(pattern),
+                BehaviorEvent.device_id.ilike(pattern),
+                BehaviorEvent.device_model.ilike(pattern),
+                BehaviorEvent.phone.ilike(pattern),
+                cast(BehaviorEvent.user_id, String).ilike(pattern),
+                BehaviorEvent.user_id.in_(matched_users),
+                BehaviorEvent.phone.in_(
+                    select(User.phone).where(User.id.in_(_matching_users(value)))
+                ),
+            )
+        )
+
+    ranked = (
+        select(
+            BehaviorEvent.subject_id.label("device_id"),
+            BehaviorEvent.event_type,
+            BehaviorEvent.user_id,
+            BehaviorEvent.phone,
+            BehaviorEvent.device_model,
+            BehaviorEvent.occurred_at.label("last_active_at"),
+            func.row_number()
+            .over(
+                partition_by=BehaviorEvent.subject_id,
+                order_by=(
+                    BehaviorEvent.occurred_at.desc(),
+                    BehaviorEvent.id.desc(),
+                ),
+            )
+            .label("rank"),
+        )
+        .where(*conditions)
+        .subquery()
+    )
+    rows = db.execute(
+        select(ranked)
+        .where(ranked.c.rank == 1)
+        .order_by(ranked.c.last_active_at.desc(), ranked.c.device_id.asc())
+        .limit(limit)
+    ).all()
+    by_id, by_phone = _user_maps(
+        db,
+        user_ids={row.user_id for row in rows if row.user_id is not None},
+        phones={row.phone for row in rows if row.phone},
+    )
+    result = []
+    for row in rows:
+        user = by_id.get(row.user_id) if row.user_id is not None else None
+        user = user or (by_phone.get(row.phone) if row.phone else None)
+        result.append(
+            {
+                "device_id": row.device_id,
+                "source": row.event_type,
+                "source_label": _EVENT_SOURCE_LABELS.get(
+                    row.event_type, "登录/鉴权流水"
+                ),
+                "user_id": user.id if user else row.user_id,
+                "username": user.username if user else None,
+                "phone": user.phone if user else row.phone,
+                "nickname": user.nickname if user else None,
+                "device_model": row.device_model,
+                "last_active_at": row.last_active_at,
+            }
+        )
+    return result
+
+
+def _comparison_device_candidates(
+    db: Session, *, keyword: str | None, limit: int
+) -> list[dict]:
+    conditions = [
+        ComparisonRecord.device_id.is_not(None),
+        ComparisonRecord.device_id != "",
+    ]
+    if keyword:
+        value = keyword.strip()
+        pattern = f"%{value}%"
+        conditions.append(
+            or_(
+                ComparisonRecord.device_id.ilike(pattern),
+                ComparisonRecord.device_model.ilike(pattern),
+                cast(ComparisonRecord.user_id, String).ilike(pattern),
+                ComparisonRecord.user_id.in_(_matching_users(value)),
+            )
+        )
+    ranked = (
+        select(
+            ComparisonRecord.device_id,
+            ComparisonRecord.user_id,
+            ComparisonRecord.device_model,
+            ComparisonRecord.created_at.label("last_active_at"),
+            func.row_number()
+            .over(
+                partition_by=ComparisonRecord.device_id,
+                order_by=(
+                    ComparisonRecord.created_at.desc(),
+                    ComparisonRecord.id.desc(),
+                ),
+            )
+            .label("rank"),
+        )
+        .where(*conditions)
+        .subquery()
+    )
+    rows = db.execute(
+        select(ranked)
+        .where(ranked.c.rank == 1)
+        .order_by(ranked.c.last_active_at.desc(), ranked.c.device_id.asc())
+        .limit(limit)
+    ).all()
+    by_id, _ = _user_maps(
+        db,
+        user_ids={row.user_id for row in rows if row.user_id is not None},
+        phones=set(),
+    )
+    return [
+        {
+            "device_id": row.device_id,
+            "source": "comparison_record",
+            "source_label": "比价记录",
+            "user_id": user.id if user else row.user_id,
+            "username": user.username if user else None,
+            "phone": user.phone if user else None,
+            "nickname": user.nickname if user else None,
+            "device_model": row.device_model,
+            "last_active_at": row.last_active_at,
+        }
+        for row in rows
+        for user in [by_id.get(row.user_id) if row.user_id is not None else None]
+    ]
+
+
+def list_device_candidates(
+    db: Session,
+    *,
+    rule_code: str,
+    keyword: str | None = None,
+    limit: int = 30,
+) -> list[dict]:
+    rule = limit_policy.get_rule(rule_code)
+    if limit_policy.SUBJECT_DEVICE not in rule.subject_types:
+        raise ValueError("当前限制项不支持设备白名单")
+    if limit_policy.device_source_scope(rule_code) == "comparison":
+        return _comparison_device_candidates(db, keyword=keyword, limit=limit)
+    return _behavior_device_candidates(db, keyword=keyword, limit=limit)
+
+
+def _aware(value: datetime | None) -> datetime | None:
+    if value is None:
+        return None
+    return value.replace(tzinfo=UTC) if value.tzinfo is None else value
+
+
+def status_of(row: LimitPolicyOverride, *, now: datetime | None = None) -> str:
+    now = _aware(now) or datetime.now(UTC)
+    if not row.enabled:
+        return "disabled"
+    if _aware(row.starts_at) and _aware(row.starts_at) > now:
+        return "scheduled"
+    if _aware(row.expires_at) and _aware(row.expires_at) <= now:
+        return "expired"
+    return "active"
+
+
+def to_dict(db: Session, row: LimitPolicyOverride) -> dict:
+    rule = limit_policy.get_rule(row.rule_code)
+    effective = limit_policy.resolve(
+        db,
+        row.rule_code,
+        phone=row.subject_value if row.subject_type == "phone" else None,
+        device=row.subject_value if row.subject_type == "device" else None,
+    )
+    return {
+        "id": row.id,
+        "subject_type": row.subject_type,
+        "subject_value": row.subject_value,
+        "rule_code": row.rule_code,
+        "rule_label": rule.label,
+        "rule_group": rule.group,
+        "mode": row.mode,
+        "limit_value": row.limit_value,
+        "global_limit": effective.global_limit,
+        "effective_limit": effective.limit,
+        "enabled": row.enabled,
+        "starts_at": row.starts_at,
+        "expires_at": row.expires_at,
+        "reset_at": row.reset_at,
+        "reason": row.reason,
+        "status": status_of(row),
+        "created_by_admin_id": row.created_by_admin_id,
+        "created_at": row.created_at,
+        "updated_at": row.updated_at,
+    }
+
+
+def list_rows(
+    db: Session,
+    *,
+    subject_type: str | None = None,
+    keyword: str | None = None,
+    rule_code: str | None = None,
+    offset: int = 0,
+    limit: int = 100,
+) -> tuple[list[LimitPolicyOverride], int]:
+    stmt = select(LimitPolicyOverride)
+    count_stmt = select(func.count(LimitPolicyOverride.id))
+    conditions = []
+    if subject_type:
+        conditions.append(LimitPolicyOverride.subject_type == subject_type)
+    if keyword:
+        conditions.append(
+            LimitPolicyOverride.subject_value.ilike(f"%{keyword.strip()}%")
+        )
+    if rule_code:
+        conditions.append(LimitPolicyOverride.rule_code == rule_code)
+    if conditions:
+        stmt = stmt.where(*conditions)
+        count_stmt = count_stmt.where(*conditions)
+    total = int(db.scalar(count_stmt) or 0)
+    rows = list(
+        db.execute(
+            stmt.order_by(
+                LimitPolicyOverride.created_at.desc(),
+                LimitPolicyOverride.id.desc(),
+            )
+            .offset(offset)
+            .limit(limit)
+        ).scalars()
+    )
+    return rows, total
+
+
+def create(
+    db: Session,
+    *,
+    subject_type: str,
+    subject_value: str,
+    rule_code: str,
+    mode: str,
+    limit_value: int | None,
+    enabled: bool,
+    starts_at: datetime | None,
+    expires_at: datetime | None,
+    reason: str,
+    admin_id: int,
+    reactivate_inactive: bool = False,
+) -> LimitPolicyOverride:
+    rule = limit_policy.get_rule(rule_code)
+    subject_value = limit_policy.validate_whitelist_subject(
+        subject_type, subject_value
+    )
+    limit_policy.validate_override(
+        rule,
+        subject_type=subject_type,
+        mode=mode,
+        limit_value=limit_value,
+        starts_at=starts_at,
+        expires_at=expires_at,
+    )
+    existing = db.execute(
+        select(LimitPolicyOverride).where(
+            LimitPolicyOverride.subject_type == subject_type,
+            LimitPolicyOverride.subject_value == subject_value,
+            LimitPolicyOverride.rule_code == rule_code,
+        )
+    ).scalar_one_or_none()
+    if existing is not None:
+        if (
+            not reactivate_inactive
+            or status_of(existing) not in {"disabled", "expired"}
+        ):
+            raise DuplicateOverrideError
+        # “恢复全局”与自然过期都会保留原行供审计。再次加入白名单时
+        # 复用该行，避免唯一键让批量新增永久失败；完整变更仍写 admin 审计日志。
+        existing.mode = mode
+        existing.limit_value = (
+            limit_value if mode == limit_policy.MODE_OVERRIDE else None
+        )
+        existing.enabled = enabled
+        existing.starts_at = starts_at
+        existing.expires_at = expires_at
+        existing.reset_at = None
+        existing.reason = reason.strip() or None
+        db.flush()
+        return existing
+
+    row = LimitPolicyOverride(
+        subject_type=subject_type,
+        subject_value=subject_value,
+        rule_code=rule_code,
+        mode=mode,
+        limit_value=limit_value if mode == limit_policy.MODE_OVERRIDE else None,
+        enabled=enabled,
+        starts_at=starts_at,
+        expires_at=expires_at,
+        reason=reason.strip() or None,
+        created_by_admin_id=admin_id,
+    )
+    db.add(row)
+    try:
+        db.flush()
+    except IntegrityError as exc:
+        db.rollback()
+        raise DuplicateOverrideError from exc
+    return row
+
+
+def update(
+    db: Session,
+    row: LimitPolicyOverride,
+    *,
+    enabled: bool | None,
+    starts_at: datetime | None,
+    expires_at: datetime | None,
+    reason: str | None,
+    fields_set: set[str],
+) -> LimitPolicyOverride:
+    new_starts = starts_at if "starts_at" in fields_set else row.starts_at
+    new_expires = expires_at if "expires_at" in fields_set else row.expires_at
+    new_enabled = enabled if "enabled" in fields_set and enabled is not None else row.enabled
+    rule = limit_policy.get_rule(row.rule_code)
+    if row.mode == limit_policy.MODE_OVERRIDE and new_enabled:
+        raise ValueError("历史覆盖配置只能停用、恢复全局或删除")
+    if new_enabled:
+        limit_policy.validate_override(
+            rule,
+            subject_type=row.subject_type,
+            mode=row.mode,
+            limit_value=None,
+            starts_at=new_starts,
+            expires_at=new_expires,
+        )
+    if "enabled" in fields_set and enabled is not None:
+        row.enabled = enabled
+    if "starts_at" in fields_set:
+        row.starts_at = starts_at
+    if "expires_at" in fields_set:
+        row.expires_at = expires_at
+    if "reason" in fields_set:
+        row.reason = reason.strip() if reason else None
+    db.flush()
+    return row
+
+
+def update_global_limit(
+    db: Session, rule_code: str, value: int, *, admin_id: int
+) -> tuple[int, int]:
+    rule = limit_policy.get_rule(rule_code)
+    if not rule.min_value <= value <= rule.max_value:
+        raise ValueError(
+            f"限制值必须在 {rule.min_value} 到 {rule.max_value} 之间"
+        )
+    before = limit_policy.resolve(db, rule_code).global_limit
+    app_config.set_value(db, rule.config_key, value, admin_id=admin_id, commit=False)
+    # 引导视频仍有一个旧的专用配置页。同步其结构化配置，保证两个入口读取
+    # 同一数值；业务判定仍统一走 limit_policy。
+    if rule.legacy_config_key and rule.legacy_json_field:
+        legacy = db.get(AppConfig, rule.legacy_config_key)
+        legacy_value = (
+            dict(legacy.value)
+            if legacy is not None and isinstance(legacy.value, dict)
+            else {}
+        )
+        legacy_value[rule.legacy_json_field] = value
+        if legacy is None:
+            legacy = AppConfig(
+                key=rule.legacy_config_key,
+                value=legacy_value,
+                updated_by_admin_id=admin_id,
+            )
+            db.add(legacy)
+        else:
+            legacy.value = legacy_value
+            legacy.updated_by_admin_id = admin_id
+        db.flush()
+    return before, value
+
+
+def restore_global(row: LimitPolicyOverride) -> LimitPolicyOverride:
+    """Stop applying the exception while retaining its history for audit."""
+
+    row.enabled = False
+    row.reset_at = None
+    return row

app/admin/repositories/queries.py

@@ -14,6 +14,7 @@ from sqlalchemy.orm import Session
 
 from app.core import rewards
 from app.core.config import settings
+from app.models.ad_ecpm import AdEcpmRecord
 from app.models.ad_feed_reward import AdFeedRewardRecord
 from app.models.ad_reward import AdRewardRecord
 from app.models.admin import AdminAuditLog
@@ -1257,11 +1258,15 @@ def user_reward_stats(
     date_from: datetime | None = None,
     date_to: datetime | None = None,
     withdraw_source: str | None = None,
+    app_env: str | None = None,
+    revenue_scope: str = "all",
+    feed_scene: str | None = None,
 ) -> dict:
     """提现详情「用户统计区」10 项。窗口作用于除「现金余额」外的所有项(余额是当前快照)。
 
-    口径:激励视频/信息流只统计 granted;数量——视频按条数、信息流按份数(unit_count 累加);
-    平均 eCPM 用原始分值(分/千次)按记录取算术平均;各「提现」= 该来源累计金币折现。
+    口径:激励视频/信息流奖励数量只统计 granted;数量——视频按条数、信息流按份数(unit_count 累加)。
+    平均 Draw eCPM 与广告收益报表一致:基于 ad_ecpm_record 的全部 draw/feed 展示记录计算，
+    不以是否发奖为筛选条件。各「提现」= 该来源累计金币折现。
     传统任务 = 窗口内正向金币中,排除广告(reward_video/feed_ad_reward)与人工调整后的折现。
     """
     withdraw_source_conds = (
@@ -1296,30 +1301,68 @@ def user_reward_stats(
 
     # 只投影本统计实际使用的列。避免滚动发布或旧本地库尚未补齐无关新列时，
     # SQLAlchemy 因 select(ORM) 自动展开整表字段而让提现详情整体 500。
+    business_ids: set[str] | None = None
+    if revenue_scope == "business":
+        # 与广告收益报表共用正式/测试业务代码位集合，避免两个页面随配置切换后再次漂移。
+        from app.admin.repositories.ad_revenue import business_code_ids
+
+        business_ids = business_code_ids(db, app_env)
+
+    rv_conds = [
+        AdRewardRecord.user_id == user_id,
+        AdRewardRecord.reward_scene == "reward_video",
+        AdRewardRecord.status == "granted",
+        *_window_conds(AdRewardRecord.created_at, date_from, date_to),
+    ]
+    if app_env is not None:
+        rv_conds.append(AdRewardRecord.app_env == app_env)
+    if business_ids is not None:
+        rv_conds.append(AdRewardRecord.our_code_id.in_(business_ids))
     rv = db.execute(
         select(AdRewardRecord.ecpm_raw, AdRewardRecord.coin).where(
-            AdRewardRecord.user_id == user_id,
-            AdRewardRecord.reward_scene == "reward_video",
-            AdRewardRecord.status == "granted",
-            *_window_conds(AdRewardRecord.created_at, date_from, date_to),
+            *rv_conds,
         )
     ).all()
     rv_ecpms = [rewards.parse_ecpm_fen(r.ecpm_raw) for r in rv if r.ecpm_raw]
     rv_coins = sum(r.coin for r in rv)
 
-    feed = db.execute(
+    feed_reward_conds = [
+        AdFeedRewardRecord.user_id == user_id,
+        AdFeedRewardRecord.status == "granted",
+        *_window_conds(AdFeedRewardRecord.created_at, date_from, date_to),
+    ]
+    if app_env is not None:
+        feed_reward_conds.append(AdFeedRewardRecord.app_env == app_env)
+    if feed_scene is not None:
+        feed_reward_conds.append(AdFeedRewardRecord.feed_scene == feed_scene)
+    if business_ids is not None:
+        feed_reward_conds.append(AdFeedRewardRecord.our_code_id.in_(business_ids))
+    feed_rewards = db.execute(
         select(
             AdFeedRewardRecord.unit_count,
-            AdFeedRewardRecord.ecpm_raw,
             AdFeedRewardRecord.coin,
         ).where(
-            AdFeedRewardRecord.user_id == user_id,
-            AdFeedRewardRecord.status == "granted",
-            *_window_conds(AdFeedRewardRecord.created_at, date_from, date_to),
+            *feed_reward_conds,
         )
     ).all()
-    feed_ecpms = [rewards.parse_ecpm_fen(f.ecpm_raw) for f in feed if f.ecpm_raw]
-    feed_coins = sum(f.coin for f in feed)
+    feed_coins = sum(f.coin for f in feed_rewards)
+
+    feed_impression_conds = [
+        AdEcpmRecord.user_id == user_id,
+        AdEcpmRecord.ad_type.in_(("draw", "feed")),
+        *_window_conds(AdEcpmRecord.created_at, date_from, date_to),
+    ]
+    if app_env is not None:
+        feed_impression_conds.append(AdEcpmRecord.app_env == app_env)
+    if feed_scene is not None:
+        feed_impression_conds.append(AdEcpmRecord.feed_scene == feed_scene)
+    if business_ids is not None:
+        feed_impression_conds.append(AdEcpmRecord.our_code_id.in_(business_ids))
+    feed_impressions = db.execute(
+        select(AdEcpmRecord.ecpm_raw).where(*feed_impression_conds)
+    ).all()
+    # 与 ad_revenue.category_stats 相同:每次展示权重相同，非法原值按 parse_ecpm_fen 记 0。
+    feed_ecpms = [rewards.parse_ecpm_fen(row.ecpm_raw) for row in feed_impressions]
 
     trad_coins = db.execute(
         select(func.coalesce(func.sum(CoinTransaction.amount), 0)).where(
@@ -1338,7 +1381,7 @@ def user_reward_stats(
         "reward_video_count": len(rv),
         "reward_video_avg_ecpm": round(sum(rv_ecpms) / len(rv_ecpms), 2) if rv_ecpms else 0.0,
         "reward_video_cash_cents": _coins_to_cents(rv_coins),
-        "feed_count": int(sum(f.unit_count for f in feed)),
+        "feed_count": int(sum(f.unit_count for f in feed_rewards)),
         "feed_avg_ecpm": round(sum(feed_ecpms) / len(feed_ecpms), 2) if feed_ecpms else 0.0,
         "feed_cash_cents": _coins_to_cents(feed_coins),
     }

app/admin/routers/config.py

@@ -12,7 +12,11 @@ from fastapi import APIRouter, Depends, HTTPException, Request
 from app.admin.audit import write_audit
 from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
 from app.admin.schemas.config import ConfigItemOut, ConfigUpdateRequest
-from app.core.config_schema import CONFIG_DEFS
+from app.core.config_schema import (
+    AD_FEED_DAILY_LIMIT_KEY,
+    AD_REWARD_VIDEO_DAILY_LIMIT_KEY,
+    CONFIG_DEFS,
+)
 from app.core.rewards import SIGNIN_CYCLE_LEN
 from app.models.admin import AdminUser
 from app.repositories import app_config
@@ -88,6 +92,23 @@ def update_config(
 
     before = app_config.get_value(db, key)
     app_config.set_value(db, key, body.value, admin_id=admin.id, commit=False)
+    if key == "ad_daily_limit":
+        # 旧系统配置接口过去只有一个广告日上限。仍有人直接调用时，同时同步
+        # 新的激励视频/Draw 两项，避免旧入口写入后业务实际值不变。
+        app_config.set_value(
+            db,
+            AD_REWARD_VIDEO_DAILY_LIMIT_KEY,
+            body.value,
+            admin_id=admin.id,
+            commit=False,
+        )
+        app_config.set_value(
+            db,
+            AD_FEED_DAILY_LIMIT_KEY,
+            body.value,
+            admin_id=admin.id,
+            commit=False,
+        )
     write_audit(
         db, admin, action="config.set", target_type="config", target_id=key,
         detail={"before": before, "after": body.value}, ip=get_client_ip(request), commit=False,

app/admin/routers/guide_video.py

@@ -9,7 +9,7 @@ client_max_body_size,见 shaguabijia-admin-web/deploy/nginx/admin.shaguabijia.co
 """
 from __future__ import annotations
 
-from typing import Annotated
+from typing import Annotated, Literal
 
 from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile
 
@@ -26,15 +26,21 @@ router = APIRouter(
     dependencies=[Depends(get_current_admin)],
 )
 
+GuideScene = Literal["coupon", "comparison"]
 
-def _out(db: AdminDb) -> GuideVideoConfigOut:
+
+def _out(db: AdminDb, scene: GuideScene) -> GuideVideoConfigOut:
     """配置 + 播放统计合成响应(四个写接口都以最新状态返回,前端一次同步到位)。"""
-    return GuideVideoConfigOut(**guide_video.get_config(db), **guide_video.play_stats(db))
+    return GuideVideoConfigOut(
+        scene=scene,
+        **guide_video.get_config(db, scene),
+        **guide_video.play_stats(db, scene),
+    )
 
 
 @router.get("", response_model=GuideVideoConfigOut, summary="新手引导视频配置(领券浮层)")
-def get_config(db: AdminDb) -> GuideVideoConfigOut:
-    return _out(db)
+def get_config(db: AdminDb, scene: GuideScene = "coupon") -> GuideVideoConfigOut:
+    return _out(db, scene)
 
 
 @router.patch("", response_model=GuideVideoConfigOut, summary="改开关/次数/金币(带审计)")
@@ -43,21 +49,24 @@ def update_config(
     request: Request,
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
+    scene: GuideScene = "coupon",
 ) -> GuideVideoConfigOut:
     before, after = guide_video.update_config(
         db,
         enabled=body.enabled,
         max_plays=body.max_plays,
         reward_coin=body.reward_coin,
+        scene=scene,
         admin_id=admin.id,
         commit=False,
     )
     write_audit(
         db, admin, action="guide_video.update", target_type="guide_video", target_id=None,
-        detail={"before": before, "after": after}, ip=get_client_ip(request), commit=False,
+        detail={"scene": scene, "before": before, "after": after},
+        ip=get_client_ip(request), commit=False,
     )
     db.commit()
-    return _out(db)
+    return _out(db, scene)
 
 
 @router.post("/video", response_model=GuideVideoConfigOut, summary="上传新手引导视频(MP4,带审计)")
@@ -65,23 +74,31 @@ async def upload_video(
     request: Request,
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
-    file: UploadFile = File(...),
+    file: Annotated[UploadFile, File()],
+    scene: GuideScene = "coupon",
 ) -> GuideVideoConfigOut:
     data = await file.read()
     try:
         url = media.save_guide_video(data)
     except media.MediaError as e:
         raise HTTPException(status_code=400, detail=str(e)) from e
-    before, after = guide_video.set_video(db, url, admin_id=admin.id, commit=False)
+    before, after = guide_video.set_video(
+        db, url, scene=scene, admin_id=admin.id, commit=False
+    )
     write_audit(
         db, admin, action="guide_video.set_video", target_type="guide_video", target_id=None,
-        detail={"before": before.get("video_url"), "after": url, "bytes": len(data)},
+        detail={
+            "scene": scene,
+            "before": before.get("video_url"),
+            "after": url,
+            "bytes": len(data),
+        },
         ip=get_client_ip(request), commit=False,
     )
     db.commit()
     # 提交成功后再删旧片,避免新片没落库就把旧片丢了
     media.delete_guide_video(before.get("video_url"))
-    return _out(db)
+    return _out(db, scene)
 
 
 @router.delete("/video", response_model=GuideVideoConfigOut, summary="移除新手引导视频(带审计)")
@@ -89,13 +106,17 @@ def delete_video(
     request: Request,
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
+    scene: GuideScene = "coupon",
 ) -> GuideVideoConfigOut:
     """移除后 /guide-video/start 一律返回 should_play=false,领券浮层回到「只放广告」。"""
-    before, after = guide_video.set_video(db, None, admin_id=admin.id, commit=False)
+    before, after = guide_video.set_video(
+        db, None, scene=scene, admin_id=admin.id, commit=False
+    )
     write_audit(
         db, admin, action="guide_video.delete_video", target_type="guide_video", target_id=None,
-        detail={"before": before.get("video_url")}, ip=get_client_ip(request), commit=False,
+        detail={"scene": scene, "before": before.get("video_url")},
+        ip=get_client_ip(request), commit=False,
     )
     db.commit()
     media.delete_guide_video(before.get("video_url"))
-    return _out(db)
+    return _out(db, scene)

app/admin/routers/limit_whitelist.py

@@ -0,0 +1,361 @@
+"""Unified limit rules and per-phone/device whitelist overrides."""
+from __future__ import annotations
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
+
+from app.admin.audit import write_audit
+from app.admin.deps import AdminDb, CurrentAdmin, get_client_ip, require_page
+from app.admin.repositories import limit_whitelist as repo
+from app.admin.schemas.limit_whitelist import (
+    DeviceCandidateOut,
+    GlobalLimitUpdate,
+    LimitOverrideBulkWrite,
+    LimitOverrideList,
+    LimitOverrideOut,
+    LimitOverridePatch,
+    LimitOverrideWrite,
+    LimitRuleOut,
+)
+from app.core import limit_policy
+from app.models.limit_policy import LimitPolicyOverride
+from app.repositories import risk as risk_repo
+
+router = APIRouter(
+    prefix="/admin/api/limit-whitelist",
+    tags=["admin-limit-whitelist"],
+    dependencies=[Depends(require_page("limit-whitelist"))],
+)
+
+def _reconcile_risk_rule(db, rule_code: str) -> None:
+    now = risk_repo.utcnow()
+    if rule_code == "risk.sms.hourly":
+        risk_repo.reconcile_behavior_rule(
+            db,
+            rule_code=risk_repo.RULE_SMS_HOURLY,
+            at=now,
+            commit=False,
+        )
+    elif rule_code == "risk.oneclick.daily":
+        risk_repo.reconcile_behavior_rule(
+            db,
+            rule_code=risk_repo.RULE_ONECLICK_DAILY,
+            at=now,
+            commit=False,
+        )
+    elif rule_code == "risk.compare.daily":
+        risk_repo.reconcile_compare_rule(db, at=now, commit=False)
+
+
+def _row_or_404(db, override_id: int) -> LimitPolicyOverride:
+    row = db.get(LimitPolicyOverride, override_id)
+    if row is None:
+        raise HTTPException(status_code=404, detail="白名单配置不存在")
+    return row
+
+
+def _out(db, row: LimitPolicyOverride) -> LimitOverrideOut:
+    return LimitOverrideOut(**repo.to_dict(db, row))
+
+
+def _audit_payload(value: LimitOverrideOut) -> dict:
+    return value.model_dump(mode="json")
+
+
+@router.get("/rules", response_model=list[LimitRuleOut], summary="读取所有限制规则")
+def list_rules(db: AdminDb) -> list[LimitRuleOut]:
+    return [LimitRuleOut(**item) for item in limit_policy.rule_catalog(db)]
+
+
+@router.patch(
+    "/rules/{rule_code}",
+    response_model=LimitRuleOut,
+    summary="修改规则的全局限制值",
+)
+def update_rule(
+    rule_code: str,
+    body: GlobalLimitUpdate,
+    request: Request,
+    admin: CurrentAdmin,
+    db: AdminDb,
+) -> LimitRuleOut:
+    try:
+        before, after = repo.update_global_limit(
+            db, rule_code, body.value, admin_id=admin.id
+        )
+    except KeyError as exc:
+        raise HTTPException(status_code=404, detail="未知限制规则") from exc
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    _reconcile_risk_rule(db, rule_code)
+    write_audit(
+        db,
+        admin,
+        action="limit.rule.update",
+        target_type="limit_rule",
+        target_id=rule_code,
+        detail={"before": before, "after": after},
+        ip=get_client_ip(request),
+        commit=False,
+    )
+    db.commit()
+    item = next(
+        item for item in limit_policy.rule_catalog(db) if item["code"] == rule_code
+    )
+    return LimitRuleOut(**item)
+
+
+@router.get(
+    "/device-candidates",
+    response_model=list[DeviceCandidateOut],
+    summary="按限制项搜索可加入白名单的设备",
+)
+def list_device_candidates(
+    db: AdminDb,
+    rule_code: str = Query(..., min_length=1, max_length=64),
+    keyword: str | None = Query(None, max_length=128),
+    limit: int = Query(30, ge=1, le=50),
+) -> list[DeviceCandidateOut]:
+    try:
+        rows = repo.list_device_candidates(
+            db,
+            rule_code=rule_code,
+            keyword=keyword,
+            limit=limit,
+        )
+    except KeyError as exc:
+        raise HTTPException(status_code=404, detail="未知限制规则") from exc
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    return [DeviceCandidateOut(**row) for row in rows]
+
+
+@router.get("", response_model=LimitOverrideList, summary="读取白名单配置")
+def list_overrides(
+    db: AdminDb,
+    subject_type: str | None = Query(None, pattern="^(phone|device)$"),
+    keyword: str | None = Query(None, max_length=128),
+    rule_code: str | None = Query(None, max_length=64),
+    offset: int = Query(0, ge=0),
+    limit: int = Query(100, ge=1, le=500),
+) -> LimitOverrideList:
+    rows, total = repo.list_rows(
+        db,
+        subject_type=subject_type,
+        keyword=keyword,
+        rule_code=rule_code,
+        offset=offset,
+        limit=limit,
+    )
+    return LimitOverrideList(items=[_out(db, row) for row in rows], total=total)
+
+
+@router.post(
+    "",
+    response_model=LimitOverrideOut,
+    status_code=status.HTTP_201_CREATED,
+    summary="新增白名单配置",
+)
+def create_override(
+    body: LimitOverrideWrite,
+    request: Request,
+    admin: CurrentAdmin,
+    db: AdminDb,
+) -> LimitOverrideOut:
+    try:
+        row = repo.create(
+            db,
+            **body.model_dump(),
+            limit_value=None,
+            admin_id=admin.id,
+        )
+    except KeyError as exc:
+        raise HTTPException(status_code=404, detail="未知限制规则") from exc
+    except repo.DuplicateOverrideError as exc:
+        raise HTTPException(
+            status_code=409,
+            detail="该手机号或设备已配置此规则，请编辑现有配置",
+        ) from exc
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    _reconcile_risk_rule(db, row.rule_code)
+    result = _out(db, row)
+    write_audit(
+        db,
+        admin,
+        action="limit.override.create",
+        target_type="limit_override",
+        target_id=str(row.id),
+        detail={"after": _audit_payload(result)},
+        ip=get_client_ip(request),
+        commit=False,
+    )
+    db.commit()
+    return _out(db, row)
+
+
+@router.post(
+    "/bulk",
+    response_model=list[LimitOverrideOut],
+    status_code=status.HTTP_201_CREATED,
+    summary="批量新增临时不限或免告警白名单",
+)
+def create_overrides_bulk(
+    body: LimitOverrideBulkWrite,
+    request: Request,
+    admin: CurrentAdmin,
+    db: AdminDb,
+) -> list[LimitOverrideOut]:
+    rows: list[LimitPolicyOverride] = []
+    current_rule_code = ""
+    try:
+        if body.subject_type == limit_policy.SUBJECT_DEVICE:
+            limit_policy.validate_device_rule_scope(body.rule_codes)
+        for current_rule_code in body.rule_codes:
+            rule = limit_policy.get_rule(current_rule_code)
+            mode = (
+                limit_policy.MODE_SUPPRESS_ALERT
+                if rule.alert_only
+                else limit_policy.MODE_UNLIMITED
+            )
+            rows.append(
+                repo.create(
+                    db,
+                    subject_type=body.subject_type,
+                    subject_value=body.subject_value,
+                    rule_code=current_rule_code,
+                    mode=mode,
+                    limit_value=None,
+                    enabled=body.enabled,
+                    starts_at=body.starts_at,
+                    expires_at=body.expires_at,
+                    reason=body.reason,
+                    admin_id=admin.id,
+                    reactivate_inactive=True,
+                )
+            )
+    except KeyError as exc:
+        raise HTTPException(status_code=404, detail="未知限制规则") from exc
+    except repo.DuplicateOverrideError as exc:
+        rule = limit_policy.get_rule(current_rule_code)
+        raise HTTPException(
+            status_code=409,
+            detail=f"“{rule.label}”已有白名单配置，请编辑或重置现有配置",
+        ) from exc
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+    for row in rows:
+        _reconcile_risk_rule(db, row.rule_code)
+    results = [_out(db, row) for row in rows]
+    write_audit(
+        db,
+        admin,
+        action="limit.override.bulk_create",
+        target_type="limit_override",
+        target_id=",".join(str(row.id) for row in rows),
+        detail={"after": [_audit_payload(item) for item in results]},
+        ip=get_client_ip(request),
+        commit=False,
+    )
+    db.commit()
+    return [_out(db, row) for row in rows]
+
+
+@router.patch(
+    "/{override_id}",
+    response_model=LimitOverrideOut,
+    summary="编辑白名单配置",
+)
+def update_override(
+    override_id: int,
+    body: LimitOverridePatch,
+    request: Request,
+    admin: CurrentAdmin,
+    db: AdminDb,
+) -> LimitOverrideOut:
+    row = _row_or_404(db, override_id)
+    before = _audit_payload(_out(db, row))
+    try:
+        repo.update(
+            db,
+            row,
+            **body.model_dump(),
+            fields_set=set(body.model_fields_set),
+        )
+    except (KeyError, ValueError) as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    _reconcile_risk_rule(db, row.rule_code)
+    after = _out(db, row)
+    write_audit(
+        db,
+        admin,
+        action="limit.override.update",
+        target_type="limit_override",
+        target_id=str(row.id),
+        detail={"before": before, "after": _audit_payload(after)},
+        ip=get_client_ip(request),
+        commit=False,
+    )
+    db.commit()
+    return _out(db, row)
+
+
+@router.post(
+    "/{override_id}/reset",
+    response_model=LimitOverrideOut,
+    summary="停用例外配置并恢复全局策略",
+)
+def restore_global_policy(
+    override_id: int,
+    request: Request,
+    admin: CurrentAdmin,
+    db: AdminDb,
+) -> LimitOverrideOut:
+    row = _row_or_404(db, override_id)
+    before = _audit_payload(_out(db, row))
+    repo.restore_global(row)
+    _reconcile_risk_rule(db, row.rule_code)
+    after = _out(db, row)
+    write_audit(
+        db,
+        admin,
+        action="limit.override.restore_global",
+        target_type="limit_override",
+        target_id=str(row.id),
+        detail={"before": before, "after": _audit_payload(after)},
+        ip=get_client_ip(request),
+        commit=False,
+    )
+    db.commit()
+    return _out(db, row)
+
+
+@router.delete(
+    "/{override_id}",
+    status_code=status.HTTP_204_NO_CONTENT,
+    summary="删除白名单配置",
+)
+def delete_override(
+    override_id: int,
+    request: Request,
+    admin: CurrentAdmin,
+    db: AdminDb,
+) -> None:
+    row = _row_or_404(db, override_id)
+    before = _audit_payload(_out(db, row))
+    target_id = str(row.id)
+    rule_code = row.rule_code
+    db.delete(row)
+    db.flush()
+    _reconcile_risk_rule(db, rule_code)
+    write_audit(
+        db,
+        admin,
+        action="limit.override.delete",
+        target_type="limit_override",
+        target_id=target_id,
+        detail={"before": before},
+        ip=get_client_ip(request),
+        commit=False,
+    )
+    db.commit()

app/admin/routers/users.py

@@ -88,6 +88,11 @@ def get_user_reward_stats(
     withdraw_source: Annotated[
         str | None, Query(pattern="^(coin_cash|invite_cash)$")
     ] = None,
+    app_env: Annotated[str | None, Query(pattern="^(prod|test)$")] = None,
+    revenue_scope: Annotated[str, Query(pattern="^(business|all)$")] = "all",
+    feed_scene: Annotated[
+        str | None, Query(pattern="^(comparison|coupon|welfare)$")
+    ] = None,
 ) -> UserRewardStats:
     """提现详情抽屉「用户统计区」。date_from/date_to 都不传 = 注册至今(全量)。"""
     if not user_repo.user_exists(db, user_id):
@@ -99,6 +104,9 @@ def get_user_reward_stats(
             date_from=date_from,
             date_to=date_to,
             withdraw_source=withdraw_source,
+            app_env=app_env,
+            revenue_scope=revenue_scope,
+            feed_scene=feed_scene,
         )
     )
 

app/admin/schemas/guide_video.py

@@ -7,6 +7,7 @@ from app.repositories.guide_video import MAX_PLAYS_LIMIT, REWARD_COIN_LIMIT
 
 
 class GuideVideoConfigOut(BaseModel):
+    scene: str
     enabled: bool
     video_url: str | None = None   # 相对地址 /media/guide_video/xxx.mp4;未配片 = None
     max_plays: int

app/admin/schemas/limit_whitelist.py

@@ -0,0 +1,127 @@
+"""Admin contracts for global limit rules and per-subject overrides."""
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+SubjectType = Literal["phone", "device"]
+PolicyMode = Literal["unlimited", "suppress_alert"]
+
+
+class LimitRuleOut(BaseModel):
+    code: str
+    label: str
+    group: str
+    global_limit: int
+    default_limit: int
+    window_label: str
+    subject_types: list[str]
+    allowed_modes: list[str]
+    min_value: int
+    max_value: int
+    supports_reset: bool
+    alert_only: bool
+
+
+class GlobalLimitUpdate(BaseModel):
+    value: int = Field(ge=0, le=1_000_000)
+
+
+class DeviceCandidateOut(BaseModel):
+    device_id: str
+    source: str
+    source_label: str
+    user_id: int | None = None
+    username: str | None = None
+    phone: str | None = None
+    nickname: str | None = None
+    device_model: str | None = None
+    last_active_at: datetime
+
+
+class LimitOverrideWrite(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    subject_type: SubjectType
+    subject_value: str = Field(min_length=1, max_length=128)
+    rule_code: str = Field(min_length=1, max_length=64)
+    mode: PolicyMode
+    enabled: bool = True
+    starts_at: datetime | None = None
+    expires_at: datetime | None = None
+    reason: str = Field("", max_length=256)
+
+    @model_validator(mode="after")
+    def validate_time_range(self):
+        if self.starts_at and self.expires_at and self.expires_at <= self.starts_at:
+            raise ValueError("失效时间必须晚于生效时间")
+        return self
+
+
+class LimitOverrideBulkWrite(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    subject_type: SubjectType
+    subject_value: str = Field(min_length=1, max_length=128)
+    rule_codes: list[str] = Field(min_length=1, max_length=32)
+    enabled: bool = True
+    starts_at: datetime | None = None
+    expires_at: datetime
+    reason: str = Field("", max_length=256)
+
+    @model_validator(mode="after")
+    def validate_bulk_request(self):
+        self.rule_codes = list(dict.fromkeys(self.rule_codes))
+        starts_at = (
+            self.starts_at.replace(tzinfo=UTC)
+            if self.starts_at and self.starts_at.tzinfo is None
+            else self.starts_at
+        )
+        expires_at = (
+            self.expires_at.replace(tzinfo=UTC)
+            if self.expires_at.tzinfo is None
+            else self.expires_at
+        )
+        if starts_at and expires_at <= starts_at:
+            raise ValueError("失效时间必须晚于生效时间")
+        if expires_at <= datetime.now(UTC):
+            raise ValueError("失效时间必须晚于当前时间")
+        return self
+
+
+class LimitOverridePatch(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    enabled: bool | None = None
+    starts_at: datetime | None = None
+    expires_at: datetime | None = None
+    reason: str | None = Field(None, max_length=256)
+
+
+class LimitOverrideOut(BaseModel):
+    id: int
+    subject_type: str
+    subject_value: str
+    rule_code: str
+    rule_label: str
+    rule_group: str
+    mode: str
+    limit_value: int | None
+    global_limit: int
+    effective_limit: int | None
+    enabled: bool
+    starts_at: datetime | None
+    expires_at: datetime | None
+    reset_at: datetime | None
+    reason: str | None
+    status: str
+    created_by_admin_id: int | None
+    created_at: datetime
+    updated_at: datetime
+
+
+class LimitOverrideList(BaseModel):
+    items: list[LimitOverrideOut]
+    total: int

app/admin/schemas/risk_monitor.py

@@ -21,9 +21,9 @@ class RiskMonitorSummary(BaseModel):
 
 
 class RiskRuleConfig(BaseModel):
-    sms_hourly_threshold: int = Field(ge=1, le=5)
+    sms_hourly_threshold: int = Field(ge=1, le=100_000)
     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/admin/schemas/user.py

@@ -60,7 +60,7 @@ class UserRewardStats(BaseModel):
     reward_video_avg_ecpm: float  # 平均激励视频 eCPM(分/千次)
     reward_video_cash_cents: int  # 激励视频提现(金币折现)
     feed_count: int  # 累计信息流广告数(granted 份数,unit_count 累加)
-    feed_avg_ecpm: float  # 平均信息流广告 eCPM(分/千次)
+    feed_avg_ecpm: float  # 全部 Draw/feed 实际展示的平均 eCPM(分/千次,含未发奖展示)
     feed_cash_cents: int  # 信息流广告提现(金币折现)
 
 

app/api/v1/ad.py

@@ -18,7 +18,7 @@ import uuid
 from fastapi import APIRouter, Depends, HTTPException, Path, Request, status
 
 from app.api.deps import CurrentUser, DbSession
-from app.core import rewards
+from app.core import limit_policy, rewards
 from app.core.config import settings
 from app.core.ratelimit import rate_limit
 from app.integrations import pangle
@@ -413,12 +413,23 @@ def feed_reward(payload: FeedRewardIn, user: CurrentUser, db: DbSession) -> Feed
         "feed ad reward user_id=%d event=%s status=%s units=%d coin=%d",
         user.id, rec.client_event_id, rec.status, rec.unit_count, rec.coin,
     )
+    feed_policy = limit_policy.resolve_for_user(db, "ad.feed.daily", user.id)
+    feed_limit = (
+        rewards.get_ad_daily_limit(db)
+        if feed_policy.override_id is None
+        and feed_policy.bucket_version == "default"
+        else feed_policy.limit
+    )
     return FeedRewardOut(
         granted=(rec.status == "granted"),
         status=rec.status,
         coin=rec.coin,
         unit_count=rec.unit_count,
-        daily_limit=rewards.get_ad_daily_limit(db),
+        daily_limit=(
+            feed_limit
+            if feed_limit is not None
+            else limit_policy.get_rule("ad.feed.daily").max_value
+        ),
     )
 
 

app/api/v1/auth.py

@@ -16,7 +16,7 @@ from fastapi import APIRouter, HTTPException, Request
 from sqlalchemy.exc import IntegrityError
 
 from app.api.deps import CurrentUser, DbSession
-from app.core import test_account
+from app.core import limit_policy, test_account
 from app.core.ratelimit import (
     RateLimitRule,
     check_rate_limits,
@@ -69,6 +69,7 @@ SMS_LOGIN_MAX_PER_HOUR = 5
 # 堵「换手机号绕开单号 60s 冷却」的洞 —— 冷却是单号维度,一机换号能绕开。
 SMS_SEND_MAX_PER_HOUR_PER_DEVICE = 5     # 每小时上限
 SMS_SEND_MAX_PER_DAY_PER_DEVICE = 20     # 每天上限(再叠一层日封顶,挡低频长时间轰炸)
+UNLIMITED_VERIFY_ATTEMPTS = 2_147_483_647
 
 
 def _client_ip(request: Request) -> str:
@@ -198,16 +199,53 @@ def sms_send(req: SmsSendRequest, request: Request, db: DbSession) -> SmsSendRes
     # 补「换手机号绕开单号 60s 冷却」的洞(冷却是单号维度,一机换号能绕);设备维度按机器封顶,挡短信轰炸/烧钱。
     # 关键:被单号 60s 冷却挡下的重发是「没真发、没烧钱」→ 不该占额度。故 check(先判)放在真发之前
     # (超限直接 429、不真发),record(计数)只在 send_code 成功后调 —— 冷却/供应商失败抛 429 时直接返回、不计数。
+    hourly_policy = limit_policy.resolve(
+        db,
+        "sms.send.hourly",
+        phone=req.phone,
+        device=subject_id,
+    )
+    daily_policy = limit_policy.resolve(
+        db,
+        "sms.send.daily",
+        phone=req.phone,
+        device=subject_id,
+    )
+    cooldown_policy = limit_policy.resolve(
+        db,
+        "sms.phone.cooldown",
+        phone=req.phone,
+    )
+    hourly_limit = (
+        SMS_SEND_MAX_PER_HOUR_PER_DEVICE
+        if hourly_policy.override_id is None
+        and hourly_policy.bucket_version == "default"
+        else hourly_policy.limit
+    )
+    daily_limit = (
+        SMS_SEND_MAX_PER_DAY_PER_DEVICE
+        if daily_policy.override_id is None
+        and daily_policy.bucket_version == "default"
+        else daily_policy.limit
+    )
     send_rules = [
-        RateLimitRule("sms-send-device", SMS_SEND_MAX_PER_HOUR_PER_DEVICE, 3600,
-                      "操作过于频繁,请稍后再试"),
-        RateLimitRule("sms-send-device-daily", SMS_SEND_MAX_PER_DAY_PER_DEVICE, 86400,
-                      "今日验证码发送次数过多,请明天再试"),
+        RateLimitRule("sms-send-device", hourly_limit, 3600,
+                      "操作过于频繁,请稍后再试", hourly_policy.bucket_version),
+        RateLimitRule("sms-send-device-daily", daily_limit, 86400,
+                      "今日验证码发送次数过多,请明天再试", daily_policy.bucket_version),
     ]
-    check_rate_limits(request, subject=req.device_id, rules=send_rules)
+    check_rate_limits(request, subject=subject_id, rules=send_rules)
 
     try:
-        cooldown = send_code(req.phone)
+        cooldown = send_code(
+            req.phone,
+            cooldown_sec=(
+                None
+                if cooldown_policy.override_id is None
+                and cooldown_policy.bucket_version == "default"
+                else (0 if cooldown_policy.limit is None else cooldown_policy.limit)
+            ),
+        )
     except SmsError as e:
         risk_repo.record_behavior_event(
             db,
@@ -224,7 +262,7 @@ def sms_send(req: SmsSendRequest, request: Request, db: DbSession) -> SmsSendRes
         raise HTTPException(status_code=e.status_code, detail=str(e)) from e
 
     # 发码成功 → 两道闸各 +1(被单号冷却挡下的重发走不到这里,故不占额度)
-    record_rate_limits(request, subject=req.device_id, rules=send_rules)
+    record_rate_limits(request, subject=subject_id, rules=send_rules)
 
     from app.core.config import settings  # 局部 import 避免循环
 
@@ -283,17 +321,48 @@ def sms_login(req: SmsLoginRequest, request: Request, db: DbSession) -> TokenWit
     # **之前** → 输错验证码的失败尝试也计数,才挡得住撞库/爆破(另有单码失败 SMS_MAX_VERIFY_ATTEMPTS 次即作废兜底)。
     # ⚠️ 按设备而非手机号 → 一台机器换不同手机号刷登录也受限(防一机狂登多号);device_id 空(老客户端)时
     #    退化为该 IP 下所有空设备聚一桶,仍受限。
+    login_policy = limit_policy.resolve(
+        db,
+        "sms.login.hourly",
+        phone=req.phone,
+        device=subject_id,
+    )
+    verify_policy = limit_policy.resolve(
+        db,
+        "sms.code.failed_attempts",
+        phone=req.phone,
+    )
+    login_limit = (
+        SMS_LOGIN_MAX_PER_HOUR
+        if login_policy.override_id is None
+        and login_policy.bucket_version == "default"
+        else login_policy.limit
+    )
     enforce_rate_limit(
         request,
         scope="sms-login-device",
-        subject=req.device_id,
-        limit=SMS_LOGIN_MAX_PER_HOUR,
+        subject=subject_id,
+        limit=login_limit,
         window_sec=3600,
         detail="登录尝试过于频繁,请稍后再试",
+        bucket_suffix=login_policy.bucket_version,
     )
 
     try:
-        ok = verify_code(req.phone, req.code)
+        ok = verify_code(
+            req.phone,
+            req.code,
+            max_failed_attempts=(
+                None
+                if verify_policy.override_id is None
+                and verify_policy.bucket_version == "default"
+                else (
+                    UNLIMITED_VERIFY_ATTEMPTS
+                    if verify_policy.limit is None
+                    else verify_policy.limit
+                )
+            ),
+        )
     except SmsError as e:  # provider 校验降级(如阿里云接口异常)→ 原样透出其状态码(503),别误报「验证码错误」
         raise HTTPException(status_code=e.status_code, detail=str(e)) from e
     if not ok:
@@ -394,15 +463,25 @@ def _finish_wechat_bind(
     未占用 → 新建微信账号(channel=wechat,昵称头像取微信)→ 签 token 登入。"""
     existing = user_repo.get_user_by_phone(db, phone)
     if existing is not None:
-        from app.core.config import settings  # 局部 import,避免循环
-
         ticket = create_conflict_ticket(
             openid=openid,
             wechat_nickname=wechat_nickname,
             wechat_avatar_url=wechat_avatar_url,
             phone=phone,
         )
-        blocked = rebind_repo.rebound_within_days(db, phone, settings.PHONE_REBIND_LIMIT_DAYS)
+        rebind_policy = limit_policy.resolve(
+            db,
+            "phone.rebind.days",
+            phone=phone,
+            device=device_id,
+        )
+        rebind_days = rebind_policy.limit or 0
+        blocked = rebind_repo.rebound_within_days(
+            db,
+            phone,
+            rebind_days,
+            reset_at=rebind_policy.reset_at,
+        )
         logger.info(
             "wechat bind phone occupied phone=%s by user_id=%d has_wechat=%s",
             mask_phone(phone), existing.id, bool(existing.wechat_openid),
@@ -418,7 +497,12 @@ def _finish_wechat_bind(
             conflict_ticket=ticket,
             rebind_available=not blocked,
             rebind_blocked_days=(
-                rebind_repo.remaining_block_days(db, phone, settings.PHONE_REBIND_LIMIT_DAYS)
+                rebind_repo.remaining_block_days(
+                    db,
+                    phone,
+                    rebind_days,
+                    reset_at=rebind_policy.reset_at,
+                )
                 if blocked else 0
             ),
         )
@@ -451,18 +535,50 @@ def wechat_bind_phone_sms(
     except TokenError as e:
         raise HTTPException(status_code=401, detail="授权已过期,请重新用微信登录") from e
 
+    subject_id = _device_subject(req.device_id, request)
     # 防刷:同 sms/login,按 设备+IP 每小时限流(放在验证码校验之前,失败也计数)
+    bind_policy = limit_policy.resolve(
+        db,
+        "wechat.bind.hourly",
+        phone=req.phone,
+        device=subject_id,
+    )
+    verify_policy = limit_policy.resolve(
+        db,
+        "sms.code.failed_attempts",
+        phone=req.phone,
+    )
+    bind_limit = (
+        SMS_LOGIN_MAX_PER_HOUR
+        if bind_policy.override_id is None
+        and bind_policy.bucket_version == "default"
+        else bind_policy.limit
+    )
     enforce_rate_limit(
         request,
         scope="wechat-bind-sms-device",
-        subject=req.device_id,
-        limit=SMS_LOGIN_MAX_PER_HOUR,
+        subject=subject_id,
+        limit=bind_limit,
         window_sec=3600,
         detail="登录尝试过于频繁,请稍后再试",
+        bucket_suffix=bind_policy.bucket_version,
     )
 
     try:
-        ok = verify_code(req.phone, req.code)
+        ok = verify_code(
+            req.phone,
+            req.code,
+            max_failed_attempts=(
+                None
+                if verify_policy.override_id is None
+                and verify_policy.bucket_version == "default"
+                else (
+                    UNLIMITED_VERIFY_ATTEMPTS
+                    if verify_policy.limit is None
+                    else verify_policy.limit
+                )
+            ),
+        )
     except SmsError as e:  # provider 校验降级(如阿里云接口异常)→ 原样透出其状态码(503),别误报「验证码错误」
         raise HTTPException(status_code=e.status_code, detail=str(e)) from e
     if not ok:
@@ -522,9 +638,23 @@ def wechat_conflict_continue(
     except TokenError as e:
         raise HTTPException(status_code=401, detail="操作超时,请重新用微信登录") from e
 
+    subject_id = _device_subject(req.device_id, request)
+    conflict_policy = limit_policy.resolve(
+        db,
+        "wechat.conflict.hourly",
+        phone=claims["phone"],
+        device=subject_id,
+    )
+    conflict_limit = (
+        SMS_LOGIN_MAX_PER_HOUR
+        if conflict_policy.override_id is None
+        and conflict_policy.bucket_version == "default"
+        else conflict_policy.limit
+    )
     enforce_rate_limit(
-        request, scope="wechat-conflict-device", subject=req.device_id,
-        limit=SMS_LOGIN_MAX_PER_HOUR, window_sec=3600, detail="操作过于频繁,请稍后再试",
+        request, scope="wechat-conflict-device", subject=subject_id,
+        limit=conflict_limit, window_sec=3600, detail="操作过于频繁,请稍后再试",
+        bucket_suffix=conflict_policy.bucket_version,
     )
 
     user = user_repo.get_user_by_phone(db, claims["phone"])
@@ -562,21 +692,50 @@ def wechat_conflict_continue(
 def wechat_conflict_rebind(
     req: WechatConflictRebindRequest, request: Request, db: DbSession
 ) -> WechatBindResultResponse:
-    from app.core.config import settings  # 局部 import,避免循环
-
     try:
         claims = decode_conflict_ticket(req.conflict_ticket)
     except TokenError as e:
         raise HTTPException(status_code=401, detail="操作超时,请重新用微信登录") from e
 
+    subject_id = _device_subject(req.device_id, request)
+    conflict_policy = limit_policy.resolve(
+        db,
+        "wechat.conflict.hourly",
+        phone=claims["phone"],
+        device=subject_id,
+    )
+    conflict_limit = (
+        SMS_LOGIN_MAX_PER_HOUR
+        if conflict_policy.override_id is None
+        and conflict_policy.bucket_version == "default"
+        else conflict_policy.limit
+    )
     enforce_rate_limit(
-        request, scope="wechat-conflict-device", subject=req.device_id,
-        limit=SMS_LOGIN_MAX_PER_HOUR, window_sec=3600, detail="操作过于频繁,请稍后再试",
+        request, scope="wechat-conflict-device", subject=subject_id,
+        limit=conflict_limit, window_sec=3600, detail="操作过于频繁,请稍后再试",
+        bucket_suffix=conflict_policy.bucket_version,
     )
 
     phone = claims["phone"]
-    if rebind_repo.rebound_within_days(db, phone, settings.PHONE_REBIND_LIMIT_DAYS):
-        days = rebind_repo.remaining_block_days(db, phone, settings.PHONE_REBIND_LIMIT_DAYS)
+    rebind_policy = limit_policy.resolve(
+        db,
+        "phone.rebind.days",
+        phone=phone,
+        device=req.device_id,
+    )
+    rebind_days = rebind_policy.limit or 0
+    if rebind_repo.rebound_within_days(
+        db,
+        phone,
+        rebind_days,
+        reset_at=rebind_policy.reset_at,
+    ):
+        days = rebind_repo.remaining_block_days(
+            db,
+            phone,
+            rebind_days,
+            reset_at=rebind_policy.reset_at,
+        )
         raise HTTPException(status_code=409, detail=f"该手机号 {days} 天内已换绑过,暂不能再次换绑")
 
     user = user_repo.rebind_account(

app/api/v1/compare.py

@@ -108,7 +108,7 @@ def _harvest_done_blocking(
 
 def _harvest_abort_blocking(
     trace_id: str, status_hint: str, reason: str | None, trace_url: str | None,
-) -> None:
+) -> int | None:
     with SessionLocal() as db:
         rec = crud_compare.harvest_abort(
             db, trace_id=trace_id, status=status_hint, reason=reason, trace_url=trace_url,
@@ -118,6 +118,7 @@ def _harvest_abort_blocking(
         extra={"phase": "harvest_abort",
                "status": (rec.status if rec else None), "reason": reason},
     )
+    return rec.id if rec is not None else None
 
 
 async def _forward(
@@ -291,7 +292,10 @@ async def trace_epilogue(
 
 @router.post("/trace/finalize", summary="比价 trace 收尾上云 (透传 + 夭折落库)")
 async def trace_finalize(
-    request: Request, user: OptionalUser, db: DbSession
+    request: Request,
+    background_tasks: BackgroundTasks,
+    user: OptionalUser,
+    db: DbSession,
 ) -> dict[str, Any]:
     _ensure_compare_allowed(user, db)
     # 用户终止 / Phase1 未识别没到 done 帧: pricebot 打包半截上云返回 {trace_url};
@@ -302,12 +306,16 @@ async def trace_finalize(
         request, "/api/trace/finalize", user, harvest_first_frame=False,
     )
     try:
-        await run_in_threadpool(
+        record_id = 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),
         )
+        if record_id is not None:
+            background_tasks.add_task(
+                backfill_comparison_llm_cost, record_id, trace_id
+            )
     except Exception as e:  # noqa: BLE001
         logger.warning("harvest_abort failed trace=%s: %s", trace_id, e)
     return resp

app/api/v1/compare_record.py

@@ -16,6 +16,7 @@ import logging
 from fastapi import APIRouter, BackgroundTasks, HTTPException, Query, status
 
 from app.api.deps import CurrentUser, DbSession
+from app.core import limit_policy
 from app.repositories import comparison as crud_compare
 from app.repositories import risk as risk_repo
 from app.schemas.compare_record import (
@@ -53,17 +54,29 @@ def reserve_compare_start(
     ):
         raise HTTPException(status_code=403, detail="账号存在异常，该功能暂不可用")
     try:
+        policy = limit_policy.resolve(
+            db,
+            "compare.start.daily",
+            phone=user.phone,
+            device=payload.device_id,
+        )
         rec, used = crud_compare.reserve_daily_start(
             db,
             user_id=user.id,
             trace_id=payload.trace_id,
             business_type=payload.business_type,
             device_id=payload.device_id,
+            limit=policy.limit,
+            reset_at=policy.reset_at,
         )
     except crud_compare.DailyCompareStartLimitExceeded:
         raise HTTPException(
             status_code=status.HTTP_429_TOO_MANY_REQUESTS,
-            detail="今日已比价超过100次，请明天再试",
+            detail=(
+                f"今日已比价超过{policy.limit}次，请明天再试"
+                if policy.limit is not None
+                else "今日比价次数已达上限，请明天再试"
+            ),
         ) from None
     except crud_compare.ComparisonTraceOwnershipError:
         raise HTTPException(
@@ -71,11 +84,16 @@ def reserve_compare_start(
             detail="比价任务标识冲突，请重新发起",
         ) from None
     # 风控阈值由后台动态配置，不能再只在固定的 100 次业务上限处同步。
-    risk_repo.sync_compare_incident(db, user_id=user.id, at=rec.created_at)
+    risk_repo.sync_compare_incident(
+        db,
+        user_id=user.id,
+        at=rec.created_at,
+        device_id=payload.device_id,
+    )
     return CompareStartReserveOut(
-        limit=crud_compare.DAILY_COMPARE_START_LIMIT,
+        limit=policy.limit,
         used=used,
-        remaining=max(crud_compare.DAILY_COMPARE_START_LIMIT - used, 0),
+        remaining=max(policy.limit - used, 0) if policy.limit is not None else None,
     )
 
 

app/core/config_schema.py

@@ -14,12 +14,137 @@ from app.core import rewards as r
 RISK_SMS_HOURLY_THRESHOLD_KEY = "risk_sms_hourly_threshold"
 RISK_ONECLICK_DAILY_THRESHOLD_KEY = "risk_oneclick_daily_threshold"
 RISK_COMPARE_DAILY_THRESHOLD_KEY = "risk_compare_daily_threshold"
+COMPARE_DAILY_LIMIT_KEY = "compare_daily_limit"
+SMS_SEND_HOURLY_LIMIT_KEY = "sms_send_hourly_limit"
+SMS_SEND_DAILY_LIMIT_KEY = "sms_send_daily_limit"
+SMS_LOGIN_HOURLY_LIMIT_KEY = "sms_login_hourly_limit"
+WECHAT_BIND_SMS_HOURLY_LIMIT_KEY = "wechat_bind_sms_hourly_limit"
+WECHAT_CONFLICT_HOURLY_LIMIT_KEY = "wechat_conflict_hourly_limit"
+SMS_PHONE_COOLDOWN_SECONDS_KEY = "sms_phone_cooldown_seconds"
+SMS_CODE_MAX_FAILED_ATTEMPTS_KEY = "sms_code_max_failed_attempts"
+AD_REWARD_VIDEO_DAILY_LIMIT_KEY = "ad_reward_video_daily_limit"
+AD_FEED_DAILY_LIMIT_KEY = "ad_feed_daily_limit"
+PHONE_REBIND_DAYS_KEY = "phone_rebind_days"
+GUIDE_VIDEO_MAX_PLAYS_KEY = "guide_video_max_plays"
 
 # type 约定(给前端渲染编辑控件用):int / int_list / dict_str_int / bool / enum
 # hidden=True:仍是合法可配项(业务照常 get_value / admin 可经专用端点读写),但**不在通用
 #   「系统配置」页渲染**(admin/routers/config.py:list_config 按此过滤)。用于把已下线/已改由
 #   专用页管理的项从福利页 Tab 收起,同时保留后端默认值与写入能力。
 CONFIG_DEFS: dict[str, dict[str, Any]] = {
+    COMPARE_DAILY_LIMIT_KEY: {
+        "default": 100,
+        "label": "账号每日比价次数上限",
+        "group": "限制策略",
+        "type": "int",
+        "min": 1,
+        "max": 100_000,
+        "hidden": True,
+        "help": "北京时间自然日内每个账号可发起的比价次数。",
+    },
+    SMS_SEND_HOURLY_LIMIT_KEY: {
+        "default": 5,
+        "label": "短信每小时发送上限",
+        "group": "限制策略",
+        "type": "int",
+        "min": 1,
+        "max": 10_000,
+        "hidden": True,
+        "help": "同一设备和 IP 在固定 1 小时窗口内成功发送短信的次数。",
+    },
+    SMS_SEND_DAILY_LIMIT_KEY: {
+        "default": 20,
+        "label": "短信 24 小时发送上限",
+        "group": "限制策略",
+        "type": "int",
+        "min": 1,
+        "max": 100_000,
+        "hidden": True,
+        "help": "同一设备和 IP 在固定 24 小时窗口内成功发送短信的次数。",
+    },
+    SMS_LOGIN_HOURLY_LIMIT_KEY: {
+        "default": 5,
+        "label": "短信登录每小时尝试上限",
+        "group": "限制策略",
+        "type": "int",
+        "min": 1,
+        "max": 10_000,
+        "hidden": True,
+        "help": "成功与失败均计入。",
+    },
+    WECHAT_BIND_SMS_HOURLY_LIMIT_KEY: {
+        "default": 5,
+        "label": "微信短信绑定每小时尝试上限",
+        "group": "限制策略",
+        "type": "int",
+        "min": 1,
+        "max": 10_000,
+        "hidden": True,
+    },
+    WECHAT_CONFLICT_HOURLY_LIMIT_KEY: {
+        "default": 5,
+        "label": "微信冲突处理每小时尝试上限",
+        "group": "限制策略",
+        "type": "int",
+        "min": 1,
+        "max": 10_000,
+        "hidden": True,
+        "help": "继续登录与重新绑定共享此额度。",
+    },
+    SMS_PHONE_COOLDOWN_SECONDS_KEY: {
+        "default": 60,
+        "label": "同手机号短信发送冷却秒数",
+        "group": "限制策略",
+        "type": "int",
+        "min": 0,
+        "max": 86_400,
+        "hidden": True,
+    },
+    SMS_CODE_MAX_FAILED_ATTEMPTS_KEY: {
+        "default": 5,
+        "label": "单验证码最大失败次数",
+        "group": "限制策略",
+        "type": "int",
+        "min": 1,
+        "max": 100,
+        "hidden": True,
+    },
+    AD_REWARD_VIDEO_DAILY_LIMIT_KEY: {
+        "default": r.DAILY_AD_REWARD_LIMIT,
+        "label": "激励视频每日发奖次数",
+        "group": "限制策略",
+        "type": "int",
+        "min": 1,
+        "max": 100_000,
+        "hidden": True,
+    },
+    AD_FEED_DAILY_LIMIT_KEY: {
+        "default": r.DAILY_AD_REWARD_LIMIT,
+        "label": "Draw 信息流每日发奖次数",
+        "group": "限制策略",
+        "type": "int",
+        "min": 1,
+        "max": 100_000,
+        "hidden": True,
+    },
+    PHONE_REBIND_DAYS_KEY: {
+        "default": 30,
+        "label": "手机/微信换绑冷却天数",
+        "group": "限制策略",
+        "type": "int",
+        "min": 0,
+        "max": 3650,
+        "hidden": True,
+    },
+    GUIDE_VIDEO_MAX_PLAYS_KEY: {
+        "default": 3,
+        "label": "领券引导视频最大播放次数",
+        "group": "限制策略",
+        "type": "int",
+        "min": 0,
+        "max": 50,
+        "hidden": True,
+    },
     "signin_rewards": {
         "default": list(r.SIGNIN_REWARDS), "label": "签到 7 天金币档位",
         "group": "签到", "type": "int_list",
@@ -56,7 +181,8 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
     },
     "ad_daily_limit": {
         "default": r.DAILY_AD_REWARD_LIMIT, "label": "看广告每日上限(次)",
-        "group": "看广告", "type": "int", "help": "福利页激励视频每日可发奖次数上限,默认 500。",
+        "group": "看广告", "type": "int", "hidden": True,
+        "help": "历史共享上限；新配置由白名单页分别管理激励视频与 Draw 信息流。",
     },
     "ad_max_coin": {
         "default": r.MAX_AD_REWARD_COIN, "label": "看广告单次金币上限",
@@ -118,9 +244,9 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
         "group": "风控",
         "type": "int",
         "min": 1,
-        "max": 5,
+        "max": 100_000,
         "hidden": True,
-        "help": "同一设备在北京时间同一自然小时内成功下发短信达到该次数时告警；不得高于现有每小时 5 次的发送上限。",
+        "help": "同一设备在北京时间同一自然小时内成功下发短信达到该次数时告警。",
     },
     RISK_ONECLICK_DAILY_THRESHOLD_KEY: {
         "default": 20,
@@ -138,7 +264,7 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
         "group": "风控",
         "type": "int",
         "min": 1,
-        "max": 100,
+        "max": 100_000,
         "hidden": True,
         "help": "同一账户在北京时间同一自然日内发起比价达到该次数时告警。",
     },

app/core/limit_policy.py

@@ -0,0 +1,525 @@
+"""Unified global limits and per-phone/device policy overrides.
+
+The registry is the single source of truth for the whitelist page.  Existing
+constants remain as backwards-compatible defaults, while business call sites
+resolve an effective value here.
+"""
+from __future__ import annotations
+
+from collections.abc import Iterable
+from dataclasses import dataclass
+from datetime import UTC, datetime
+
+from sqlalchemy import or_, select
+from sqlalchemy.orm import Session
+
+from app.core.config_schema import (
+    AD_FEED_DAILY_LIMIT_KEY,
+    AD_REWARD_VIDEO_DAILY_LIMIT_KEY,
+    COMPARE_DAILY_LIMIT_KEY,
+    GUIDE_VIDEO_MAX_PLAYS_KEY,
+    PHONE_REBIND_DAYS_KEY,
+    RISK_COMPARE_DAILY_THRESHOLD_KEY,
+    RISK_ONECLICK_DAILY_THRESHOLD_KEY,
+    RISK_SMS_HOURLY_THRESHOLD_KEY,
+    SMS_CODE_MAX_FAILED_ATTEMPTS_KEY,
+    SMS_LOGIN_HOURLY_LIMIT_KEY,
+    SMS_PHONE_COOLDOWN_SECONDS_KEY,
+    SMS_SEND_DAILY_LIMIT_KEY,
+    SMS_SEND_HOURLY_LIMIT_KEY,
+    WECHAT_BIND_SMS_HOURLY_LIMIT_KEY,
+    WECHAT_CONFLICT_HOURLY_LIMIT_KEY,
+)
+from app.models.app_config import AppConfig
+from app.models.limit_policy import LimitPolicyOverride
+from app.models.user import User
+from app.repositories import app_config
+
+MODE_INHERIT = "inherit"
+MODE_OVERRIDE = "override"
+MODE_UNLIMITED = "unlimited"
+MODE_SUPPRESS_ALERT = "suppress_alert"
+
+SUBJECT_PHONE = "phone"
+SUBJECT_DEVICE = "device"
+SUBJECT_TYPES = (SUBJECT_PHONE, SUBJECT_DEVICE)
+SUBJECT_PRECEDENCE = {SUBJECT_DEVICE: 0, SUBJECT_PHONE: 1}
+LEGACY_IP_DEVICE_PREFIX = "legacy-ip:"
+
+
+@dataclass(frozen=True)
+class RuleDefinition:
+    code: str
+    label: str
+    group: str
+    config_key: str
+    default_limit: int
+    window_label: str
+    subject_types: tuple[str, ...]
+    min_value: int = 1
+    max_value: int = 100_000
+    allow_unlimited: bool = True
+    supports_reset: bool = True
+    alert_only: bool = False
+    legacy_config_key: str | None = None
+    legacy_json_field: str | None = None
+
+    @property
+    def allowed_modes(self) -> tuple[str, ...]:
+        if self.alert_only:
+            return (MODE_SUPPRESS_ALERT,)
+        return (MODE_UNLIMITED,) if self.allow_unlimited else ()
+
+
+RULES: tuple[RuleDefinition, ...] = (
+    RuleDefinition(
+        "compare.start.daily",
+        "每日发起比价次数",
+        "比价",
+        COMPARE_DAILY_LIMIT_KEY,
+        100,
+        "北京时间自然日",
+        SUBJECT_TYPES,
+    ),
+    RuleDefinition(
+        "sms.send.hourly",
+        "短信每小时成功发送次数",
+        "短信与登录",
+        SMS_SEND_HOURLY_LIMIT_KEY,
+        5,
+        "固定 1 小时窗口",
+        SUBJECT_TYPES,
+        max_value=10_000,
+    ),
+    RuleDefinition(
+        "sms.send.daily",
+        "短信 24 小时成功发送次数",
+        "短信与登录",
+        SMS_SEND_DAILY_LIMIT_KEY,
+        20,
+        "固定 24 小时窗口",
+        SUBJECT_TYPES,
+    ),
+    RuleDefinition(
+        "sms.phone.cooldown",
+        "同手机号短信发送冷却",
+        "短信与登录",
+        SMS_PHONE_COOLDOWN_SECONDS_KEY,
+        60,
+        "秒",
+        (SUBJECT_PHONE,),
+        min_value=0,
+        max_value=86_400,
+    ),
+    RuleDefinition(
+        "sms.code.failed_attempts",
+        "单验证码最大失败次数",
+        "短信与登录",
+        SMS_CODE_MAX_FAILED_ATTEMPTS_KEY,
+        5,
+        "单个验证码",
+        (SUBJECT_PHONE,),
+        max_value=100,
+    ),
+    RuleDefinition(
+        "sms.login.hourly",
+        "短信登录每小时尝试次数",
+        "短信与登录",
+        SMS_LOGIN_HOURLY_LIMIT_KEY,
+        5,
+        "固定 1 小时窗口",
+        SUBJECT_TYPES,
+        max_value=10_000,
+    ),
+    RuleDefinition(
+        "wechat.bind.hourly",
+        "微信短信绑定每小时尝试次数",
+        "短信与登录",
+        WECHAT_BIND_SMS_HOURLY_LIMIT_KEY,
+        5,
+        "固定 1 小时窗口",
+        SUBJECT_TYPES,
+        max_value=10_000,
+    ),
+    RuleDefinition(
+        "wechat.conflict.hourly",
+        "微信冲突处理每小时尝试次数",
+        "短信与登录",
+        WECHAT_CONFLICT_HOURLY_LIMIT_KEY,
+        5,
+        "固定 1 小时窗口",
+        SUBJECT_TYPES,
+        max_value=10_000,
+    ),
+    RuleDefinition(
+        "ad.reward_video.daily",
+        "激励视频每日发奖次数",
+        "广告",
+        AD_REWARD_VIDEO_DAILY_LIMIT_KEY,
+        500,
+        "北京时间自然日",
+        (SUBJECT_PHONE,),
+        legacy_config_key="ad_daily_limit",
+    ),
+    RuleDefinition(
+        "ad.feed.daily",
+        "Draw 信息流每日发奖次数",
+        "广告",
+        AD_FEED_DAILY_LIMIT_KEY,
+        500,
+        "北京时间自然日",
+        (SUBJECT_PHONE,),
+        legacy_config_key="ad_daily_limit",
+    ),
+    RuleDefinition(
+        "ad.reward_video.cooldown",
+        "激励视频发奖后冷却秒数",
+        "广告",
+        "ad_cooldown_sec",
+        3,
+        "秒",
+        (SUBJECT_PHONE,),
+        min_value=0,
+        max_value=86_400,
+    ),
+    RuleDefinition(
+        "guide.video.lifetime",
+        "领券引导视频最大播放次数",
+        "引导与账号",
+        GUIDE_VIDEO_MAX_PLAYS_KEY,
+        3,
+        "账号生命周期",
+        (SUBJECT_PHONE,),
+        min_value=0,
+        max_value=50,
+        legacy_config_key="coupon_guide_video",
+        legacy_json_field="max_plays",
+    ),
+    RuleDefinition(
+        "phone.rebind.days",
+        "手机/微信换绑冷却天数",
+        "引导与账号",
+        PHONE_REBIND_DAYS_KEY,
+        30,
+        "自然日",
+        (SUBJECT_PHONE,),
+        min_value=0,
+        max_value=3650,
+    ),
+    RuleDefinition(
+        "risk.sms.hourly",
+        "短信设备每小时告警",
+        "风控免告警",
+        RISK_SMS_HOURLY_THRESHOLD_KEY,
+        5,
+        "北京时间自然小时",
+        (SUBJECT_DEVICE,),
+        max_value=100_000,
+        allow_unlimited=False,
+        alert_only=True,
+    ),
+    RuleDefinition(
+        "risk.oneclick.daily",
+        "一键登录设备每日告警",
+        "风控免告警",
+        RISK_ONECLICK_DAILY_THRESHOLD_KEY,
+        20,
+        "北京时间自然日",
+        (SUBJECT_DEVICE,),
+        max_value=100_000,
+        allow_unlimited=False,
+        alert_only=True,
+    ),
+    RuleDefinition(
+        "risk.compare.daily",
+        "比价账号每日告警",
+        "风控免告警",
+        RISK_COMPARE_DAILY_THRESHOLD_KEY,
+        100,
+        "北京时间自然日",
+        (SUBJECT_PHONE,),
+        max_value=100_000,
+        allow_unlimited=False,
+        alert_only=True,
+    ),
+)
+RULE_MAP = {rule.code: rule for rule in RULES}
+
+
+@dataclass(frozen=True)
+class EffectiveLimit:
+    rule_code: str
+    global_limit: int
+    limit: int | None
+    mode: str
+    suppressed: bool
+    override_id: int | None
+    matched_subject_type: str | None
+    matched_subject_value: str | None
+    reset_at: datetime | None
+    bucket_version: str
+
+    @property
+    def unlimited(self) -> bool:
+        return self.limit is None
+
+
+def normalize_subject(subject_type: str, value: str) -> str:
+    value = (value or "").strip()
+    if subject_type == SUBJECT_PHONE:
+        value = "".join(ch for ch in value if ch.isdigit())
+    if subject_type not in SUBJECT_TYPES:
+        raise ValueError(f"unsupported subject type: {subject_type}")
+    if not value:
+        raise ValueError("subject value is empty")
+    return value[:128]
+
+
+def validate_whitelist_subject(subject_type: str, value: str) -> str:
+    """Normalize a whitelist subject and reject unsafe pseudo-devices.
+
+    Old clients without a device ID are grouped by public IP for rate limiting.
+    That fallback remains valid at runtime, but it is not a stable, unique
+    device identity and must never be persisted as a device whitelist target.
+    """
+
+    normalized = normalize_subject(subject_type, value)
+    if (
+        subject_type == SUBJECT_DEVICE
+        and normalized.startswith(LEGACY_IP_DEVICE_PREFIX)
+    ):
+        raise ValueError(
+            "旧客户端未上报真实设备 ID，不能加入设备白名单，请升级客户端后重试"
+        )
+    return normalized
+
+
+def get_rule(rule_code: str) -> RuleDefinition:
+    try:
+        return RULE_MAP[rule_code]
+    except KeyError as exc:
+        raise ValueError(f"unknown rule: {rule_code}") from exc
+
+
+def device_source_scope(rule_code: str) -> str:
+    """返回设备候选数据所属命名空间，防止一个设备 ID 跨来源误套规则。"""
+    rule = get_rule(rule_code)
+    if SUBJECT_DEVICE not in rule.subject_types:
+        raise ValueError("当前限制项不支持设备白名单")
+    return "comparison" if rule_code == "compare.start.daily" else "auth"
+
+
+def validate_device_rule_scope(rule_codes: Iterable[str]) -> None:
+    scopes = {device_source_scope(rule_code) for rule_code in rule_codes}
+    if len(scopes) > 1:
+        raise ValueError("比价设备限制与短信/登录设备限制不能在同一白名单中混选")
+
+
+def _global_limit(db: Session, rule: RuleDefinition) -> tuple[int, str]:
+    row = db.get(AppConfig, rule.config_key)
+    if row is not None:
+        return int(row.value), "configured"
+
+    if rule.legacy_config_key:
+        legacy = db.get(AppConfig, rule.legacy_config_key)
+        if legacy is not None:
+            value = legacy.value
+            if rule.legacy_json_field:
+                value = value.get(rule.legacy_json_field) if isinstance(value, dict) else None
+            if value is not None:
+                return int(value), "legacy"
+
+    try:
+        return int(app_config.get_value(db, rule.config_key)), "default"
+    except KeyError:
+        return rule.default_limit, "default"
+
+
+def _aware(value: datetime | None) -> datetime | None:
+    if value is None:
+        return None
+    return value.replace(tzinfo=UTC) if value.tzinfo is None else value
+
+
+def _matching_overrides(
+    db: Session,
+    rule: RuleDefinition,
+    subjects: dict[str, str | None],
+    now: datetime,
+) -> list[LimitPolicyOverride]:
+    pairs: list[tuple[str, str]] = []
+    for subject_type in rule.subject_types:
+        raw = subjects.get(subject_type)
+        if raw:
+            pairs.append((subject_type, normalize_subject(subject_type, raw)))
+    if not pairs:
+        return []
+    clauses = [
+        (
+            (LimitPolicyOverride.subject_type == subject_type)
+            & (LimitPolicyOverride.subject_value == subject_value)
+        )
+        for subject_type, subject_value in pairs
+    ]
+    rows = list(
+        db.execute(
+            select(LimitPolicyOverride).where(
+                LimitPolicyOverride.rule_code == rule.code,
+                LimitPolicyOverride.enabled.is_(True),
+                or_(*clauses),
+            )
+        ).scalars()
+    )
+    active = [
+        row
+        for row in rows
+        if (_aware(row.starts_at) is None or _aware(row.starts_at) <= now)
+        and (_aware(row.expires_at) is None or _aware(row.expires_at) > now)
+    ]
+    return sorted(active, key=lambda row: SUBJECT_PRECEDENCE[row.subject_type])
+
+
+def resolve(
+    db: Session,
+    rule_code: str,
+    *,
+    phone: str | None = None,
+    device: str | None = None,
+    now: datetime | None = None,
+) -> EffectiveLimit:
+    """Resolve global config plus the most specific active override.
+
+    Device overrides win over phone overrides when both match.
+    """
+
+    rule = get_rule(rule_code)
+    now = _aware(now) or datetime.now(UTC)
+    global_limit, global_version = _global_limit(db, rule)
+    matches = _matching_overrides(
+        db, rule, {SUBJECT_PHONE: phone, SUBJECT_DEVICE: device}, now
+    )
+    row = matches[0] if matches else None
+    if row is None:
+        return EffectiveLimit(
+            rule.code,
+            global_limit,
+            global_limit,
+            MODE_INHERIT,
+            False,
+            None,
+            None,
+            None,
+            None,
+            global_version,
+        )
+
+    limit: int | None = global_limit
+    suppressed = False
+    if row.mode == MODE_OVERRIDE:
+        limit = int(row.limit_value) if row.limit_value is not None else global_limit
+    elif row.mode == MODE_UNLIMITED:
+        limit = None
+    elif row.mode == MODE_SUPPRESS_ALERT:
+        suppressed = True
+    # 编辑限制值/备注不应隐式清空计数；只有显式“重置状态”才切换桶。
+    reset_version = _aware(row.reset_at)
+    version = (
+        f"{global_version}:o:{row.id}:"
+        f"{reset_version.isoformat() if reset_version else '0'}"
+    )
+    return EffectiveLimit(
+        rule.code,
+        global_limit,
+        limit,
+        row.mode,
+        suppressed,
+        row.id,
+        row.subject_type,
+        row.subject_value,
+        _aware(row.reset_at),
+        version,
+    )
+
+
+def resolve_for_user(
+    db: Session,
+    rule_code: str,
+    user_id: int,
+    *,
+    device: str | None = None,
+) -> EffectiveLimit:
+    user = db.get(User, user_id)
+    return resolve(
+        db,
+        rule_code,
+        phone=user.phone if user is not None else None,
+        device=device,
+    )
+
+
+def rule_catalog(db: Session) -> list[dict]:
+    out: list[dict] = []
+    for rule in RULES:
+        value, _ = _global_limit(db, rule)
+        out.append(
+            {
+                "code": rule.code,
+                "label": rule.label,
+                "group": rule.group,
+                "global_limit": value,
+                "default_limit": rule.default_limit,
+                "window_label": rule.window_label,
+                "subject_types": list(rule.subject_types),
+                "allowed_modes": list(rule.allowed_modes),
+                "min_value": rule.min_value,
+                "max_value": rule.max_value,
+                "supports_reset": rule.supports_reset,
+                "alert_only": rule.alert_only,
+            }
+        )
+    return out
+
+
+def validate_override(
+    rule: RuleDefinition,
+    *,
+    subject_type: str,
+    mode: str,
+    limit_value: int | None,
+    starts_at: datetime | None,
+    expires_at: datetime | None,
+) -> None:
+    if subject_type not in rule.subject_types:
+        raise ValueError("该规则不支持此主体类型")
+    if mode not in rule.allowed_modes:
+        raise ValueError("该规则不支持此策略模式")
+    if limit_value is not None:
+        raise ValueError("白名单不支持覆盖指定值")
+    if mode in {MODE_UNLIMITED, MODE_SUPPRESS_ALERT} and expires_at is None:
+        raise ValueError("临时白名单必须设置失效时间")
+    if starts_at and expires_at and _aware(expires_at) <= _aware(starts_at):
+        raise ValueError("失效时间必须晚于生效时间")
+    if (
+        mode in {MODE_UNLIMITED, MODE_SUPPRESS_ALERT}
+        and expires_at is not None
+        and _aware(expires_at) <= datetime.now(UTC)
+    ):
+        raise ValueError("临时白名单的失效时间必须晚于当前时间")
+    if mode == MODE_SUPPRESS_ALERT and not rule.alert_only:
+        raise ValueError("免告警只支持风控监控的三项规则")
+
+
+def active_overrides(
+    db: Session,
+    *,
+    subject_type: str | None = None,
+    keyword: str | None = None,
+) -> Iterable[LimitPolicyOverride]:
+    stmt = select(LimitPolicyOverride)
+    if subject_type:
+        stmt = stmt.where(LimitPolicyOverride.subject_type == subject_type)
+    if keyword:
+        stmt = stmt.where(LimitPolicyOverride.subject_value.ilike(f"%{keyword.strip()}%"))
+    return db.execute(
+        stmt.order_by(LimitPolicyOverride.updated_at.desc(), LimitPolicyOverride.id.desc())
+    ).scalars()

app/core/ratelimit.py

@@ -75,10 +75,11 @@ def enforce_rate_limit(
     request: Request,
     scope: str,
     subject: str,
-    limit: int,
+    limit: int | None,
     window_sec: float,
     *,
     detail: str = "操作过于频繁,请稍后再试",
+    bucket_suffix: str = "",
 ) -> None:
     """在路由内部手动限流,按 (subject, 客户端 IP) 计数。
 
@@ -87,9 +88,9 @@ def enforce_rate_limit(
     key = `scope:subject:client_ip`;同一 (subject, IP) 在 window_sec 内超过 limit 次 → 抛 429。
     受 [settings.RATE_LIMIT_ENABLED] 总开关控制(与 [rate_limit] 一致)。
     """
-    if not settings.RATE_LIMIT_ENABLED:
+    if not settings.RATE_LIMIT_ENABLED or limit is None:
         return
-    key = f"{scope}:{subject}:{_client_ip(request)}"
+    key = f"{scope}:{subject}:{_client_ip(request)}:{bucket_suffix}"
     if not _hit(key, limit, window_sec):
         raise HTTPException(
             status_code=status.HTTP_429_TOO_MANY_REQUESTS,
@@ -112,9 +113,10 @@ class RateLimitRule(NamedTuple):
     """
 
     scope: str
-    limit: int
+    limit: int | None
     window_sec: float
     detail: str = "操作过于频繁,请稍后再试"
+    bucket_suffix: str = ""
 
 
 def _peek(key: str, limit: int, window_sec: float) -> bool:
@@ -150,7 +152,13 @@ def check_rate_limits(request: Request, subject: str, rules: list[RateLimitRule]
         return
     ip = _client_ip(request)
     for rule in rules:
-        if not _peek(f"{rule.scope}:{subject}:{ip}", rule.limit, rule.window_sec):
+        if rule.limit is None:
+            continue
+        if not _peek(
+            f"{rule.scope}:{subject}:{ip}:{rule.bucket_suffix}",
+            rule.limit,
+            rule.window_sec,
+        ):
             raise HTTPException(
                 status_code=status.HTTP_429_TOO_MANY_REQUESTS,
                 detail=rule.detail,
@@ -167,4 +175,9 @@ def record_rate_limits(request: Request, subject: str, rules: list[RateLimitRule
         return
     ip = _client_ip(request)
     for rule in rules:
-        _commit(f"{rule.scope}:{subject}:{ip}", rule.window_sec)
+        if rule.limit is None:
+            continue
+        _commit(
+            f"{rule.scope}:{subject}:{ip}:{rule.bucket_suffix}",
+            rule.window_sec,
+        )

app/integrations/sms/__init__.py

@@ -27,11 +27,22 @@ def _provider():
     return _PROVIDERS.get(settings.SMS_PROVIDER, jiguang)
 
 
-def send_code(phone: str) -> int:
+def send_code(phone: str, *, cooldown_sec: int | None = None) -> int:
     """发送验证码,返回距下次可发的冷却秒数;失败抛 SmsError。委托给当前 provider。"""
-    return _provider().send_code(phone)
+    provider = _provider()
+    if cooldown_sec is None:
+        return provider.send_code(phone)
+    return provider.send_code(phone, cooldown_sec=cooldown_sec)
 
 
-def verify_code(phone: str, code: str) -> bool:
+def verify_code(
+    phone: str,
+    code: str,
+    *,
+    max_failed_attempts: int | None = None,
+) -> bool:
     """校验验证码,返回是否通过;provider 异常降级抛 SmsError。委托给当前 provider。"""
-    return _provider().verify_code(phone, code)
+    provider = _provider()
+    if max_failed_attempts is None:
+        return provider.verify_code(phone, code)
+    return provider.verify_code(phone, code, max_failed_attempts=max_failed_attempts)

app/integrations/sms/aliyun.py

@@ -47,19 +47,26 @@ _client = None  # 惰性构建的 SDK client(模块级缓存)
 
 # ============================ 对外:发码 / 校验 ============================
 
-def send_code(phone: str) -> int:
+def send_code(phone: str, *, cooldown_sec: int | None = None) -> int:
     """发送验证码(阿里云生成+下发)。
 
     Returns: 距下次可发的秒数(= ALIYUN_SMS_INTERVAL_SEC,冷却由阿里云 Interval 侧执行)。
     Raises: SmsError(手机号无效 400 / 过频·天级流控 429 / 未配置·未开通·其他 503)。
     """
+    effective_cooldown = (
+        settings.ALIYUN_SMS_INTERVAL_SEC if cooldown_sec is None else cooldown_sec
+    )
     if settings.SMS_MOCK:
         logger.info("[SMS-aliyun-MOCK] to %s**** (不真发)", phone[:3])
-        return settings.ALIYUN_SMS_INTERVAL_SEC
+        return effective_cooldown
     if not settings.aliyun_sms_configured:
         raise SmsError("短信服务未配置(缺阿里云凭证)", status_code=503)
 
-    result = _call_send(phone)  # 传输/SDK 异常在内部抛 SmsError(503)
+    result = (
+        _call_send(phone)
+        if cooldown_sec is None
+        else _call_send(phone, cooldown_sec=effective_cooldown)
+    )
 
     if result["success"] and result["code"] == "OK":
         now = time.time()
@@ -68,7 +75,7 @@ def send_code(phone: str) -> int:
             _verify_attempts.pop(phone, None)  # 新码 = 新失败预算
             _verify_seen.pop(phone, None)
         logger.info("[SMS-aliyun] sent to %s****", phone[:3])
-        return settings.ALIYUN_SMS_INTERVAL_SEC
+        return effective_cooldown
 
     code = result["code"]
     logger.error("[SMS-aliyun] send failed code=%s msg=%s", code, result["message"])
@@ -78,7 +85,12 @@ def send_code(phone: str) -> int:
     raise SmsError(msg, status_code=status)
 
 
-def verify_code(phone: str, code: str) -> bool:
+def verify_code(
+    phone: str,
+    code: str,
+    *,
+    max_failed_attempts: int | None = None,
+) -> bool:
     """校验验证码(阿里云裁决)。
 
     - **mock**:放行任意 N 位数字(provider 无关,同极光)。
@@ -92,8 +104,13 @@ def verify_code(phone: str, code: str) -> bool:
 
     # 失败计数是 best-effort:网络调用不持锁(不能锁跨 IO),故并发下同号可能多放行个位数次。
     # 无碍——API 层登录频控(设备+IP 5/时)是硬上限,阿里云码有效期 + DuplicatePolicy 亦兜底。
+    effective_max_attempts = (
+        settings.SMS_MAX_VERIFY_ATTEMPTS
+        if max_failed_attempts is None
+        else max_failed_attempts
+    )
     with _lock:
-        if _verify_attempts.get(phone, 0) >= settings.SMS_MAX_VERIFY_ATTEMPTS:
+        if _verify_attempts.get(phone, 0) >= effective_max_attempts:
             return False  # 已作废:保持计数(直到 send_code 重置),与极光「达上限即作废」一致
 
     result = _call_check(phone, code)  # 传输/SDK 异常在内部抛 SmsError(503)
@@ -146,7 +163,7 @@ def _get_client():
     return _client
 
 
-def _call_send(phone: str) -> dict:
+def _call_send(phone: str, *, cooldown_sec: int | None = None) -> dict:
     """调 SendSmsVerifyCode。返回归一化 {success, code, message};import/建 client/调用 任一失败抛 SmsError(503)。"""
     valid_min = max(1, settings.ALIYUN_SMS_VALID_TIME_SEC // 60)
     template_param = json.dumps({"code": "##code##", "min": str(valid_min)}, ensure_ascii=False)
@@ -160,7 +177,11 @@ def _call_send(phone: str) -> dict:
             template_param=template_param,
             code_length=settings.ALIYUN_SMS_CODE_LENGTH,
             valid_time=settings.ALIYUN_SMS_VALID_TIME_SEC,
-            interval=settings.ALIYUN_SMS_INTERVAL_SEC,
+            interval=(
+                settings.ALIYUN_SMS_INTERVAL_SEC
+                if cooldown_sec is None
+                else cooldown_sec
+            ),
             scheme_name=settings.ALIYUN_SMS_SCHEME_NAME or None,
         )
         body = _get_client().send_sms_verify_code(req).body

app/integrations/sms/chuanglan.py

@@ -84,20 +84,23 @@ def _gc(now: float) -> None:
             _last_sent.pop(p, None)
 
 
-def send_code(phone: str) -> int:
+def send_code(phone: str, *, cooldown_sec: int | None = None) -> int:
     """发送验证码。
 
     Returns: 距下次可发的秒数(= SMS_SEND_INTERVAL_SEC)
     Raises: SmsError(过频 429 / 手机号无效 400 / 供应商失败 503)
     """
     now = time.time()
+    effective_cooldown = (
+        settings.SMS_SEND_INTERVAL_SEC if cooldown_sec is None else cooldown_sec
+    )
 
     # --- lock 内:防刷检查 + 预占(防并发重复发烧钱)---
     with _lock:
         _gc(now)  # 顺手清过期内存(超阈值才扫)
         elapsed = now - _last_sent.get(phone, 0.0)
-        if elapsed < settings.SMS_SEND_INTERVAL_SEC:
-            remain = int(settings.SMS_SEND_INTERVAL_SEC - elapsed)
+        if elapsed < effective_cooldown:
+            remain = int(effective_cooldown - elapsed)
             raise SmsError(f"发送过于频繁,请 {remain}s 后再试")
 
         code = _gen_code()
@@ -122,10 +125,15 @@ def send_code(phone: str) -> int:
         logger.exception("[SMS-chuanglan] send failed phone=%s****", phone[:3])
         raise SmsError("验证码发送失败,请稍后重试", status_code=503) from e
 
-    return settings.SMS_SEND_INTERVAL_SEC
+    return effective_cooldown
 
 
-def verify_code(phone: str, code: str) -> bool:
+def verify_code(
+    phone: str,
+    code: str,
+    *,
+    max_failed_attempts: int | None = None,
+) -> bool:
     """校验验证码。
 
     - **mock 模式**:放行任意 N 位数字(测试/开发便利,不真校验)。
@@ -136,6 +144,11 @@ def verify_code(phone: str, code: str) -> bool:
         logger.info("[SMS-chuanglan-MOCK] verify %s for %s****", "ok" if ok else "fail", phone[:3])
         return ok
 
+    effective_max_attempts = (
+        settings.SMS_MAX_VERIFY_ATTEMPTS
+        if max_failed_attempts is None
+        else max_failed_attempts
+    )
     with _lock:
         rec = _codes.get(phone)
         if rec is None:
@@ -143,7 +156,7 @@ def verify_code(phone: str, code: str) -> bool:
         if time.time() > rec.expires_at:
             _codes.pop(phone, None)
             return False
-        if rec.attempts >= settings.SMS_MAX_VERIFY_ATTEMPTS:
+        if rec.attempts >= effective_max_attempts:
             _codes.pop(phone, None)  # 试错过多,作废
             return False
         if secrets.compare_digest(code.encode("utf-8"), rec.code.encode("utf-8")):

app/integrations/sms/jiguang.py

@@ -70,20 +70,23 @@ def _gc(now: float) -> None:
             _last_sent.pop(p, None)
 
 
-def send_code(phone: str) -> int:
+def send_code(phone: str, *, cooldown_sec: int | None = None) -> int:
     """发送验证码。
 
     Returns: 距下次可发的秒数(= SMS_SEND_INTERVAL_SEC)
     Raises: SmsError(过频 429 / 当日超限 429 / 供应商失败 503 / 手机号无效 400)
     """
     now = time.time()
+    effective_cooldown = (
+        settings.SMS_SEND_INTERVAL_SEC if cooldown_sec is None else cooldown_sec
+    )
 
     # --- lock 内:防刷检查 + 预占(防并发重复发烧钱)---
     with _lock:
         _gc(now)  # 顺手清过期内存(超阈值才扫)
         elapsed = now - _last_sent.get(phone, 0.0)
-        if elapsed < settings.SMS_SEND_INTERVAL_SEC:
-            remain = int(settings.SMS_SEND_INTERVAL_SEC - elapsed)
+        if elapsed < effective_cooldown:
+            remain = int(effective_cooldown - elapsed)
             raise SmsError(f"发送过于频繁,请 {remain}s 后再试")
 
         code = _gen_code()
@@ -108,10 +111,15 @@ def send_code(phone: str) -> int:
         logger.exception("[SMS] send failed phone=%s****", phone[:3])
         raise SmsError("验证码发送失败,请稍后重试", status_code=503) from e
 
-    return settings.SMS_SEND_INTERVAL_SEC
+    return effective_cooldown
 
 
-def verify_code(phone: str, code: str) -> bool:
+def verify_code(
+    phone: str,
+    code: str,
+    *,
+    max_failed_attempts: int | None = None,
+) -> bool:
     """校验验证码。
 
     - **mock 模式**:放行任意 N 位数字(测试/开发便利,不真校验)。
@@ -122,6 +130,11 @@ def verify_code(phone: str, code: str) -> bool:
         logger.info("[SMS-MOCK] verify %s for %s****", "ok" if ok else "fail", phone[:3])
         return ok
 
+    effective_max_attempts = (
+        settings.SMS_MAX_VERIFY_ATTEMPTS
+        if max_failed_attempts is None
+        else max_failed_attempts
+    )
     with _lock:
         rec = _codes.get(phone)
         if rec is None:
@@ -129,7 +142,7 @@ def verify_code(phone: str, code: str) -> bool:
         if time.time() > rec.expires_at:
             _codes.pop(phone, None)
             return False
-        if rec.attempts >= settings.SMS_MAX_VERIFY_ATTEMPTS:
+        if rec.attempts >= effective_max_attempts:
             _codes.pop(phone, None)  # 试错过多,作废
             return False
         if secrets.compare_digest(code.encode("utf-8"), rec.code.encode("utf-8")):

app/models/__init__.py

@@ -36,6 +36,7 @@ from app.models.inactivity import (  # noqa: F401
 from app.models.invite import InviteRelation  # noqa: F401
 from app.models.invite_fingerprint import InviteFingerprint  # noqa: F401
 from app.models.launch_confirm_sample import LaunchConfirmSample  # noqa: F401
+from app.models.limit_policy import LimitPolicyOverride  # noqa: F401
 from app.models.meituan_coupon import MeituanCoupon  # noqa: F401
 from app.models.notification import Notification  # noqa: F401
 from app.models.onboarding import OnboardingCompletion  # noqa: F401

app/models/comparison.py

@@ -102,12 +102,21 @@ class ComparisonRecord(Base):
     # done 帧 information 文案。成功:"在美团找到同店,到手价 ¥X…";
     # 失败:具体原因(如"美团、京东外卖均未找到该商品")。前端在比价失败时当原因展示。
     information: Mapped[str | None] = mapped_column(String(256), nullable=True)
+    # 失败卡「原因」行的展示文案(仅 status=failed 时非空):information 具体则=它;笼统则从
+    # platform_results 捞出的业务原因(打烊/未起送/找不到店或菜/单点不配送);纯系统失败为 None
+    # → 端侧显示品牌兜底「网络开小差…」。写路径(harvest_done / upsert_record)落库时派生。
+    # 见 repositories.comparison._derive_fail_display。
+    fail_reason: Mapped[str | None] = mapped_column(String(256), nullable=True)
 
     # ===== 明细(JSON,越详细越好)=====
     # 下单菜品 [{name, qty, specs?}]
     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/models/guide_video.py

@@ -1,7 +1,7 @@
 """新手引导视频播放记录(领券浮层前 N 次用它替代广告)。
 
 产品规则(2026-07 拍板):新用户点「一键自动领取」后的等候浮层,**前 3 次**不放广告,
-改放运营后台上传的引导视频;每次固定 120 金币,中途关闭也算看完照发。
+改放运营后台上传的引导视频;默认每次固定 100 金币,中途关闭也算看完照发。
 
 口径:
 - **计次按账号**(user_id),与设备无关 —— 换设备不重新送 3 次。
@@ -38,7 +38,13 @@ class GuideVideoPlay(Base):
         # start_play 捕获 IntegrityError 降级成"这次不放视频"。
         # 用 unique Index 而非 UniqueConstraint:与迁移里的 create_index 对齐(SQLite 加约束
         # 要整表重建),autogenerate 才不会每次报一条假 diff。
-        Index("uq_guide_video_play_user_seq", "user_id", "seq", unique=True),
+        Index(
+            "uq_guide_video_play_user_scene_seq",
+            "user_id",
+            "scene",
+            "seq",
+            unique=True,
+        ),
     )
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)

app/models/limit_policy.py

@@ -0,0 +1,77 @@
+"""Per-subject limit policy overrides used by the admin whitelist page."""
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import (
+    Boolean,
+    DateTime,
+    Index,
+    Integer,
+    String,
+    UniqueConstraint,
+    func,
+    true,
+)
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base
+
+
+class LimitPolicyOverride(Base):
+    """One rule override for one phone or device.
+
+    ``reset_at`` is a non-destructive usage baseline.  Business records and
+    security events remain intact; quota readers only count rows at or after
+    this timestamp.
+    """
+
+    __tablename__ = "limit_policy_override"
+    __table_args__ = (
+        UniqueConstraint(
+            "subject_type",
+            "subject_value",
+            "rule_code",
+            name="uq_limit_policy_subject_rule",
+        ),
+        Index(
+            "ix_limit_policy_lookup",
+            "subject_type",
+            "subject_value",
+            "rule_code",
+            "enabled",
+        ),
+        Index("ix_limit_policy_expires", "expires_at"),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    subject_type: Mapped[str] = mapped_column(String(16), nullable=False)
+    subject_value: Mapped[str] = mapped_column(String(128), nullable=False)
+    rule_code: Mapped[str] = mapped_column(String(64), nullable=False)
+    # 产品只保留“临时不限/免告警”。即使有内部脚本绕过 API 直接建 ORM
+    # 对象，也不能再悄悄落成已经下线的 override 模式。
+    mode: Mapped[str] = mapped_column(String(24), nullable=False, default="unlimited")
+    limit_value: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    enabled: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=True, server_default=true()
+    )
+    starts_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+    expires_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+    reset_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+    reason: Mapped[str | None] = mapped_column(String(256), nullable=True)
+    created_by_admin_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        server_default=func.now(),
+        onupdate=func.now(),
+        nullable=False,
+    )

app/repositories/ad_feed_reward.py

@@ -5,11 +5,13 @@
 """
 from __future__ import annotations
 
+from datetime import datetime
+
 from sqlalchemy import func, select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
-from app.core import rewards
+from app.core import limit_policy, rewards
 from app.core.rewards import cn_today
 from app.models.ad_feed_reward import AdFeedRewardRecord
 from app.repositories import wallet as crud_wallet
@@ -29,8 +31,14 @@ def _find_by_event(db: Session, client_event_id: str) -> AdFeedRewardRecord | No
     ).scalar_one_or_none()
 
 
-def _granted_today(db: Session, user_id: int, reward_date: str) -> int:
-    return db.execute(
+def _granted_today(
+    db: Session,
+    user_id: int,
+    reward_date: str,
+    *,
+    reset_at: datetime | None = None,
+) -> int:
+    stmt = (
         select(func.count())
         .select_from(AdFeedRewardRecord)
         .where(
@@ -38,7 +46,10 @@ def _granted_today(db: Session, user_id: int, reward_date: str) -> int:
             AdFeedRewardRecord.reward_date == reward_date,
             AdFeedRewardRecord.status == "granted",
         )
-    ).scalar_one()
+    )
+    if reset_at is not None:
+        stmt = stmt.where(AdFeedRewardRecord.created_at >= reset_at)
+    return db.execute(stmt).scalar_one()
 
 
 def granted_unit_total(db: Session, user_id: int) -> int:
@@ -121,7 +132,27 @@ def grant_feed_reward(
         )
         return _commit_record(db, rec, client_event_id)
 
-    if _granted_today(db, user_id, today) >= rewards.get_ad_daily_limit(db):
+    daily_policy = limit_policy.resolve_for_user(
+        db,
+        "ad.feed.daily",
+        user_id,
+    )
+    daily_limit = (
+        rewards.get_ad_daily_limit(db)
+        if daily_policy.override_id is None
+        and daily_policy.bucket_version == "default"
+        else daily_policy.limit
+    )
+    if (
+        daily_limit is not None
+        and _granted_today(
+            db,
+            user_id,
+            today,
+            reset_at=daily_policy.reset_at,
+        )
+        >= daily_limit
+    ):
         rec = AdFeedRewardRecord(
             client_event_id=client_event_id,
             user_id=user_id,

app/repositories/ad_reward.py

@@ -10,13 +10,13 @@
 """
 from __future__ import annotations
 
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 
 from sqlalchemy import func, select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
-from app.core import rewards
+from app.core import limit_policy, rewards
 from app.core.ad_cooldown import compute_cooldown
 from app.core.rewards import DAILY_AD_WATCH_SECONDS_LIMIT, cn_today
 from app.models.ad_reward import AdRewardRecord
@@ -96,8 +96,14 @@ def round_coin_total(db: Session, user_id: int, boost_round_id: str) -> int:
     )
 
 
-def _granted_today(db: Session, user_id: int, reward_date: str) -> int:
-    return db.execute(
+def _granted_today(
+    db: Session,
+    user_id: int,
+    reward_date: str,
+    *,
+    reset_at: datetime | None = None,
+) -> int:
+    stmt = (
         select(func.count())
         .select_from(AdRewardRecord)
         .where(
@@ -106,7 +112,10 @@ def _granted_today(db: Session, user_id: int, reward_date: str) -> int:
             AdRewardRecord.status == "granted",
             AdRewardRecord.reward_scene == "reward_video",
         )
-    ).scalar_one()
+    )
+    if reset_at is not None:
+        stmt = stmt.where(AdRewardRecord.created_at >= reset_at)
+    return db.execute(stmt).scalar_one()
 
 
 def _granted_cumulative(db: Session, user_id: int) -> int:
@@ -165,7 +174,27 @@ def grant_ad_reward(
         DAILY_AD_WATCH_SECONDS_LIMIT > 0
         and watched_seconds_today(db, user_id, today=today) >= DAILY_AD_WATCH_SECONDS_LIMIT
     )
-    over_count = _granted_today(db, user_id, today) >= rewards.get_ad_daily_limit(db)
+    daily_policy = limit_policy.resolve_for_user(
+        db,
+        "ad.reward_video.daily",
+        user_id,
+    )
+    daily_limit = (
+        rewards.get_ad_daily_limit(db)
+        if daily_policy.override_id is None
+        and daily_policy.bucket_version == "default"
+        else daily_policy.limit
+    )
+    over_count = (
+        daily_limit is not None
+        and _granted_today(
+            db,
+            user_id,
+            today,
+            reset_at=daily_policy.reset_at,
+        )
+        >= daily_limit
+    )
     if over_time or over_count:
         rec = AdRewardRecord(
             trans_id=trans_id, user_id=user_id, coin=0, status="capped",
@@ -321,19 +350,24 @@ def _commit_record(db: Session, rec: AdRewardRecord, trans_id: str) -> AdRewardR
     return rec
 
 
-def _granted_times_today_desc(db: Session, user_id: int, reward_date: str) -> list[datetime]:
+def _granted_times_today_desc(
+    db: Session,
+    user_id: int,
+    reward_date: str,
+    *,
+    reset_at: datetime | None = None,
+) -> list[datetime]:
     """当日 status=granted 记录的 created_at,按时间倒序(最新在前)——冷却策略的输入数据。"""
-    return list(
-        db.execute(
-            select(AdRewardRecord.created_at)
-            .where(
+    stmt = select(AdRewardRecord.created_at).where(
                 AdRewardRecord.user_id == user_id,
                 AdRewardRecord.reward_date == reward_date,
                 AdRewardRecord.status == "granted",
                 AdRewardRecord.reward_scene == "reward_video",
             )
-            .order_by(AdRewardRecord.created_at.desc())
-        ).scalars()
+    if reset_at is not None:
+        stmt = stmt.where(AdRewardRecord.created_at >= reset_at)
+    return list(
+        db.execute(stmt.order_by(AdRewardRecord.created_at.desc())).scalars()
     )
 
 
@@ -349,16 +383,47 @@ def today_status(
     旧客户端兼容,当前 DAILY_AD_WATCH_SECONDS_LIMIT=0 表示不启用时长闸。
     """
     today = cn_today().isoformat()
-    granted_desc = _granted_times_today_desc(db, user_id, today)
+    daily_policy = limit_policy.resolve_for_user(
+        db,
+        "ad.reward_video.daily",
+        user_id,
+    )
+    cooldown_policy = limit_policy.resolve_for_user(
+        db,
+        "ad.reward_video.cooldown",
+        user_id,
+    )
+    daily_limit = (
+        rewards.get_ad_daily_limit(db)
+        if daily_policy.override_id is None
+        and daily_policy.bucket_version == "default"
+        else daily_policy.limit
+    )
+    cooldown_seconds = (
+        rewards.get_ad_cooldown_sec(db)
+        if cooldown_policy.override_id is None
+        and cooldown_policy.bucket_version == "default"
+        else (cooldown_policy.limit or 0)
+    )
+    granted_desc = _granted_times_today_desc(
+        db,
+        user_id,
+        today,
+        reset_at=daily_policy.reset_at,
+    )
     state = compute_cooldown(
         granted_desc,
-        datetime.now(timezone.utc),
+        datetime.now(UTC),
         round_size=rewards.get_ad_round_count(db),
-        cooldown_seconds=rewards.get_ad_cooldown_sec(db),
+        cooldown_seconds=cooldown_seconds,
     )
     return (
         len(granted_desc),
-        rewards.get_ad_daily_limit(db),
+        (
+            daily_limit
+            if daily_limit is not None
+            else limit_policy.get_rule("ad.reward_video.daily").max_value
+        ),
         0,
         state.round_count,
         state.cooldown_until,

app/repositories/comparison.py

@@ -52,6 +52,81 @@ def _product_names_from_items(items: list | None) -> str | None:
     return joined[:500] or None
 
 
+# ---- 失败记录的展示文案(记录页失败卡「原因」行)------------------------------
+# information 具体就直出;笼统(_GENERIC_INFO)则从 platform_results 捞一条用户可读的业务
+# 原因;捞不到 → None(端侧显示品牌兜底「网络开小差…」)。pricebot 把 store_closed /
+# no_delivery 漏成了 status=failed,这里按 reason 关键字补判;打烊类 reason 常带一坨脏店名
+# (店名+月售+起送+配送…),统一成简短模板。自动化黑话(搜索失败/读价失败/购物车残留/裸
+# FAILED…)不给用户看 → 归入品牌兜底。
+
+# pricebot 组不出具体原因时的笼统 information(线上统计的大头),一律走品牌兜底。
+_GENERIC_INFO = {
+    "比价过程出错，请稍后重试",
+    "比价出错",
+    "比价未完成",
+    "done 参数缺少可验证的目标平台结果",
+}
+# 干净业务结局 status(直接可信),按展示优先级(越靠前越先选)。
+_BIZ_STATUS_PRIORITY = (
+    "below_minimum",
+    "no_delivery",
+    "store_closed",
+    "items_not_found",
+    "store_not_found",
+)
+
+
+def _store_closed_text(reason: str | None) -> str:
+    """打烊/暂停营业/休息类 reason 常带脏店名元数据 → 只留结论,套简短模板。"""
+    r = reason or ""
+    if "暂停营业" in r:
+        state = "暂停营业"
+    elif "休息" in r:
+        state = "休息中"
+    else:
+        state = "已打烊"
+    return f"门店{state}，无法比价"
+
+
+def _target_display_reason(platform_results: dict | None) -> str | None:
+    """从逐平台结果里挑一条"可展示给用户"的失败原因;挑不到返回 None。
+    ① status 命中干净业务结局集 → 直接采信(打烊套模板,其余用 reason);
+    ② 补判 pricebot 漏成 status=failed 的两类:打烊(套模板)、单点不配送(reason 本身干净);
+    自动化黑话(搜索失败/读价失败/购物车残留/裸 FAILED…)一律不展示 → None。"""
+    pr = platform_results or {}
+    targets = [
+        v for v in pr.values() if isinstance(v, dict) and not v.get("is_source")
+    ]
+    for want in _BIZ_STATUS_PRIORITY:  # ① 干净 status 优先
+        for v in targets:
+            if v.get("status") == want:
+                if want == "store_closed":
+                    return _store_closed_text(v.get("reason"))
+                if v.get("reason"):
+                    return v["reason"]
+    for v in targets:  # ② 漏成 failed 的业务结局补判
+        if v.get("status") != "failed":
+            continue
+        reason = (v.get("reason") or "").strip()
+        if any(k in reason for k in ("打烊", "暂停营业", "休息")):
+            return _store_closed_text(reason)
+        if "单点不配送" in reason:
+            return reason
+    return None
+
+
+def _derive_fail_display(
+    information: str | None, platform_results: dict | None
+) -> str | None:
+    """失败记录展示文案:information 具体则直出;笼统则从 platform_results 捞/补判;
+    都拿不到 → None(端侧品牌兜底)。仅在 status=failed 时调用。"""
+    info = (information or "").strip()
+    text = info if (info and info not in _GENERIC_INFO) else _target_display_reason(
+        platform_results
+    )
+    return text[:256] if text else None
+
+
 def _derive(payload: ComparisonRecordIn) -> dict:
     """从上报 payload 派生结构化列(best/saved/is_source_best/status)。"""
     results = payload.comparison_results
@@ -89,8 +164,9 @@ def _derive(payload: ComparisonRecordIn) -> dict:
 
     is_source_best = best.is_source if best is not None else None
 
-    # status:客户端显式给了就用;否则有"非源且有价"的结果=success,否则 failed
-    status = payload.status
+    # status:优先 pricebot record_status(区分 below_minimum/store_closed) → 客户端显式 status
+    # → 兜底"非源且有价"=success/否则 failed。record_status 让"未满起送"不再塌缩成 failed。
+    status = payload.record_status or payload.status
     if status is None:
         has_valid_target = any(
             (not r.is_source) and r.price is not None for r in results
@@ -105,6 +181,11 @@ def _derive(payload: ComparisonRecordIn) -> dict:
         "saved_amount_cents": saved_amount_cents,
         "is_source_best": is_source_best,
         "status": status,
+        "fail_reason": (
+            _derive_fail_display(payload.information, _pr)
+            if status == "failed"
+            else None
+        ),
     }
 
 
@@ -117,16 +198,35 @@ def upsert_record(
     灰度期老客户端 POST /compare/record 走这条,与后端 harvest 按 trace_id reconcile;
     新客户端不再 POST(改由 compare.py 透传壳 harvest 落库)。
     """
-    derived = _derive(payload)
+    # 单源派生: 与 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)
     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),
-        source_platform_id=payload.source_platform_id,
-        source_platform_name=payload.source_platform_name,
-        source_package=payload.source_package,
+        # store_name / source_platform_id / source_platform_name / source_package 统一由
+        # derived 提供(见上方两分支补齐), 不在此显式写 —— 否则与 _derive_from_platforms 撞键。
         information=payload.information,
         best_deeplink=payload.best_deeplink,
         trace_url=payload.trace_url,
@@ -134,6 +234,7 @@ 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,
@@ -206,7 +307,8 @@ def upsert_record(
 
 
 def _derive_from_results(
-    results: list[dict], platform_results: dict | None = None
+    results: list[dict], platform_results: dict | None = None,
+    record_status: str | None = None,
 ) -> dict:
     """从 done 帧 comparison_results(pricebot 原始 dict 列表)派生结构化列。
     等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。
@@ -254,7 +356,53 @@ 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,
-        "status": "success" if has_valid_target else "failed",
+        # 记录级结局: 优先用 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"),
     }
 
 
@@ -286,6 +434,8 @@ def reserve_daily_start(
     business_type: str = "food",
     device_id: str | None = None,
     now: datetime | None = None,
+    limit: int | None = DAILY_COMPARE_START_LIMIT,
+    reset_at: datetime | None = None,
 ) -> tuple[ComparisonRecord, int]:
     """Atomically reserve one of a user's 100 Beijing-day comparison starts.
 
@@ -311,6 +461,11 @@ def reserve_daily_start(
         if existing_at.tzinfo is not None:
             existing_at = existing_at.astimezone(CN_TZ).replace(tzinfo=None)
         day_start = existing_at.replace(hour=0, minute=0, second=0, microsecond=0)
+        if reset_at is not None:
+            reset_start = reset_at
+            if reset_start.tzinfo is not None:
+                reset_start = reset_start.astimezone(CN_TZ).replace(tzinfo=None)
+            day_start = max(day_start, reset_start)
         day_end = day_start + timedelta(days=1)
         used = db.scalar(
             select(func.count(ComparisonRecord.id)).where(
@@ -325,6 +480,11 @@ def reserve_daily_start(
     if current.tzinfo is not None:
         current = current.astimezone(CN_TZ).replace(tzinfo=None)
     day_start = current.replace(hour=0, minute=0, second=0, microsecond=0)
+    if reset_at is not None:
+        reset_start = reset_at
+        if reset_start.tzinfo is not None:
+            reset_start = reset_start.astimezone(CN_TZ).replace(tzinfo=None)
+        day_start = max(day_start, reset_start)
     day_end = day_start + timedelta(days=1)
     used = db.scalar(
         select(func.count(ComparisonRecord.id)).where(
@@ -333,7 +493,7 @@ def reserve_daily_start(
             ComparisonRecord.created_at < day_end,
         )
     ) or 0
-    if used >= DAILY_COMPARE_START_LIMIT:
+    if limit is not None and used >= limit:
         raise DailyCompareStartLimitExceeded
 
     rec = ComparisonRecord(
@@ -413,12 +573,40 @@ def harvest_done(
     返回 (记录, 是否本次**新**落成 success)——供调用方据此幂等发一次邀请奖。
     行不存在(理论上帧0已建;防御)则新建。"""
     results = done_params.get("comparison_results") or []
-    derived = _derive_from_results(results, done_params.get("platform_results"))
-    # 菜品:pricebot 已把源单菜品塞进 comparison_results[源行].items
-    items = next((r.get("items") or [] for r in results if r.get("is_source")), [])
+    # 展示模型统一数组(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, 统一在此算
+    fail_reason = (
+        _derive_fail_display(
+            done_params.get("information"), done_params.get("platform_results")
+        )
+        if derived["status"] == "failed"
+        else None
+    )
     fields = dict(
         business_type=business_type or "food",
         information=done_params.get("information") or None,
+        fail_reason=fail_reason,
         # best_deeplink 来自客户端剪贴板采集,harvest 拿不到 → 留空(灰度期 fromComparison 会补;
         # 纯 harvest 行「再次比价」退化为按 package 拉起 App。要精确深链需客户端另传,后续)。
         trace_url=trace_url or done_params.get("trace_url"),
@@ -426,6 +614,7 @@ 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/repositories/guide_video.py

@@ -1,8 +1,8 @@
 """新手引导视频:运营配置读写 + 播放计次 + 发币。
 
-**配置**(开关 / 视频地址 / 前几次 / 每次金币)整体作为一个 JSON 存进通用 app_config 表
-(key=coupon_guide_video),写法完全对齐 feedback_qr —— 不进 CONFIG_DEFS,所以不会污染
-系统配置页的通用列表,由本模块独占维护。
+**配置**(开关 / 视频地址 / 前几次 / 每次金币)按场景分别作为一个 JSON 存进通用
+app_config 表。领券使用 coupon_guide_video，比价使用 comparison_guide_video；
+写法完全对齐 feedback_qr，不进入系统配置页的通用列表。
 
 **计次**按账号(user_id)、**开播即计数**:客户端每次要展示领券等候浮层时调
 `/api/v1/guide-video/start`,命中则当场写一行 guide_video_play(status='playing')。
@@ -25,12 +25,16 @@ from sqlalchemy import func, select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
-from app.core import rewards
+from app.core import limit_policy, rewards
 from app.models.app_config import AppConfig
 from app.models.guide_video import GuideVideoPlay
 from app.repositories import wallet as crud_wallet
 
-_KEY = "coupon_guide_video"
+SCENES = ("coupon", "comparison")
+_KEY_BY_SCENE = {
+    "coupon": "coupon_guide_video",
+    "comparison": "comparison_guide_video",
+}
 
 #: 金币流水 biz_type。客户端收益明细按它显示「新手引导视频奖励」。
 BIZ_TYPE = "guide_video"
@@ -41,7 +45,7 @@ _DEFAULTS: dict[str, Any] = {
     "enabled": True,
     "video_url": None,   # None/空 = 未配片 → 不下发,浮层照旧放广告
     "max_plays": 3,      # 每个账号前 N 次浮层放引导视频
-    "reward_coin": 120,  # 每次固定金币
+    "reward_coin": 100,  # 每次固定金币
 }
 
 _FIELDS = tuple(_DEFAULTS.keys())
@@ -65,19 +69,38 @@ def _merge(raw: Any) -> dict[str, Any]:
     return out
 
 
-def get_config(db: Session) -> dict[str, Any]:
+def _config_key(scene: str) -> str:
+    if scene not in _KEY_BY_SCENE:
+        raise ValueError(f"unsupported guide video scene: {scene}")
+    return _KEY_BY_SCENE[scene]
+
+
+def get_config(db: Session, scene: str = "coupon") -> dict[str, Any]:
     """完整配置 + updated_at(admin 读 / 业务读共用)。"""
-    row = db.get(AppConfig, _KEY)
+    row = db.get(AppConfig, _config_key(scene))
     cfg = _merge(row.value if row is not None else None)
+    if scene == "coupon":
+        # 白名单里的引导视频规则只对应领券场景；比价仍使用自己的独立配置。
+        cfg["max_plays"] = limit_policy.resolve(
+            db, "guide.video.lifetime"
+        ).global_limit
     cfg["updated_at"] = row.updated_at.isoformat() if row is not None and row.updated_at else None
     return cfg
 
 
-def _write(db: Session, value: dict[str, Any], *, admin_id: int, commit: bool) -> dict[str, Any]:
+def _write(
+    db: Session,
+    value: dict[str, Any],
+    *,
+    scene: str,
+    admin_id: int,
+    commit: bool,
+) -> dict[str, Any]:
     """整体覆写该行(value 须为完整字段 dict),返回合并后的完整配置(含 updated_at)。"""
-    row = db.get(AppConfig, _KEY)
+    key = _config_key(scene)
+    row = db.get(AppConfig, key)
     if row is None:
-        row = AppConfig(key=_KEY, value=value, updated_by_admin_id=admin_id)
+        row = AppConfig(key=key, value=value, updated_by_admin_id=admin_id)
         db.add(row)
     else:
         row.value = value  # 整体重新赋值,SQLAlchemy 才侦测得到变更
@@ -98,11 +121,12 @@ def update_config(
     enabled: bool | None = None,
     max_plays: int | None = None,
     reward_coin: int | None = None,
+    scene: str = "coupon",
     admin_id: int,
     commit: bool = True,
 ) -> tuple[dict[str, Any], dict[str, Any]]:
     """改开关 / 次数 / 金币(只改传了的字段;视频走 set_video)。返回 (before, after) 供审计。"""
-    row = db.get(AppConfig, _KEY)
+    row = db.get(AppConfig, _config_key(scene))
     before = _merge(row.value if row is not None else None)
     new_value = {k: before[k] for k in _FIELDS}
     if enabled is not None:
@@ -111,45 +135,91 @@ def update_config(
         new_value["max_plays"] = max(0, min(int(max_plays), MAX_PLAYS_LIMIT))
     if reward_coin is not None:
         new_value["reward_coin"] = max(0, min(int(reward_coin), REWARD_COIN_LIMIT))
-    after = _write(db, new_value, admin_id=admin_id, commit=commit)
+    after = _write(
+        db,
+        new_value,
+        scene=scene,
+        admin_id=admin_id,
+        commit=False,
+    )
+    if scene == "coupon" and max_plays is not None:
+        # 兼容仍调用旧专用接口的客户端/脚本，并把统一策略全局值一并更新。
+        from app.core.config_schema import GUIDE_VIDEO_MAX_PLAYS_KEY
+        from app.repositories import app_config
+
+        app_config.set_value(
+            db,
+            GUIDE_VIDEO_MAX_PLAYS_KEY,
+            new_value["max_plays"],
+            admin_id=admin_id,
+            commit=False,
+        )
+    if commit:
+        db.commit()
+        after = get_config(db, scene)
     return before, after
 
 
 def set_video(
-    db: Session, video_url: str | None, *, admin_id: int, commit: bool = True
+    db: Session,
+    video_url: str | None,
+    *,
+    scene: str = "coupon",
+    admin_id: int,
+    commit: bool = True,
 ) -> tuple[dict[str, Any], dict[str, Any]]:
     """设置/清空引导视频地址。返回 (before, after);before['video_url'] 供调用方删旧文件。"""
-    row = db.get(AppConfig, _KEY)
+    row = db.get(AppConfig, _config_key(scene))
     before = _merge(row.value if row is not None else None)
     new_value = {k: before[k] for k in _FIELDS}
     new_value["video_url"] = video_url
-    after = _write(db, new_value, admin_id=admin_id, commit=commit)
+    after = _write(
+        db,
+        new_value,
+        scene=scene,
+        admin_id=admin_id,
+        commit=commit,
+    )
     return before, after
 
 
 # ===== 播放计次 =====
 
 
-def used_plays(db: Session, user_id: int) -> int:
+def used_plays(
+    db: Session,
+    user_id: int,
+    *,
+    scene: str = "coupon",
+    reset_at: datetime | None = None,
+) -> int:
     """该账号已用掉的引导视频次数(开播即算,含未发币的)。"""
-    return int(
-        db.execute(
-            select(func.count()).select_from(GuideVideoPlay).where(
-                GuideVideoPlay.user_id == user_id
-            )
-        ).scalar_one()
+    stmt = select(func.count()).select_from(GuideVideoPlay).where(
+        GuideVideoPlay.user_id == user_id,
+        GuideVideoPlay.scene == scene,
     )
+    if reset_at is not None:
+        # 该表的既有写入口使用北京时间 naive 墙钟，重置基线必须转换成
+        # 同一存储口径再比较；单独改成 UTC 会让存量/增量记录偏移 8 小时。
+        reset_value = reset_at.astimezone(rewards.CN_TZ).replace(tzinfo=None)
+        stmt = stmt.where(GuideVideoPlay.started_at >= reset_value)
+    return int(db.execute(stmt).scalar_one())
 
 
-def play_stats(db: Session) -> dict[str, int]:
+def play_stats(db: Session, scene: str = "coupon") -> dict[str, int]:
     """全站播放统计(admin 页展示):总播放次数 / 其中已发币次数。"""
     total = int(
-        db.execute(select(func.count()).select_from(GuideVideoPlay)).scalar_one()
+        db.execute(
+            select(func.count()).select_from(GuideVideoPlay).where(
+                GuideVideoPlay.scene == scene
+            )
+        ).scalar_one()
     )
     granted = int(
         db.execute(
             select(func.count()).select_from(GuideVideoPlay).where(
-                GuideVideoPlay.status == "granted"
+                GuideVideoPlay.status == "granted",
+                GuideVideoPlay.scene == scene,
             )
         ).scalar_one()
     )
@@ -168,11 +238,22 @@ def start_play(
       reward_coin  播完/中途关闭都发的固定金币
       seq / remaining  第几次 / 发完这次还剩几次(仅展示与排查用)
     """
-    cfg = get_config(db)
+    _config_key(scene)
+    cfg = get_config(db, scene)
     video_url = (cfg.get("video_url") or "").strip()
-    max_plays = int(cfg.get("max_plays") or 0)
+    policy = (
+        limit_policy.resolve_for_user(db, "guide.video.lifetime", user_id)
+        if scene == "coupon"
+        else None
+    )
+    max_plays = policy.limit if policy is not None else int(cfg.get("max_plays") or 0)
+    exposed_max_plays = max_plays if max_plays is not None else 1_000_000
     reward_coin = int(cfg.get("reward_coin") or 0)
-    used = used_plays(db, user_id)
+    used = (
+        used_plays(db, user_id, scene=scene, reset_at=policy.reset_at)
+        if policy is not None and policy.reset_at is not None
+        else used_plays(db, user_id, scene=scene)
+    )
 
     def _miss(used_now: int) -> dict[str, Any]:
         return {
@@ -181,20 +262,29 @@ def start_play(
             "play_token": "",
             "reward_coin": reward_coin,
             "seq": used_now,
-            "remaining": max(0, max_plays - used_now),
+            "remaining": max(0, exposed_max_plays - used_now),
         }
 
-    if not cfg.get("enabled") or not video_url or max_plays <= 0 or used >= max_plays:
+    if (
+        not cfg.get("enabled")
+        or not video_url
+        or (max_plays is not None and (max_plays <= 0 or used >= max_plays))
+    ):
         return _miss(used)
 
-    seq = used + 1
+    # ``seq`` remains scene 内 lifetime-monotonic because
+    # (user_id, scene, seq) is unique.
+    # A whitelist reset only changes the quota baseline; reusing seq=1 would
+    # collide with historical rows and make every post-reset start fail.
+    seq = used_plays(db, user_id, scene=scene) + 1
     play = GuideVideoPlay(
         user_id=user_id,
         play_token=uuid.uuid4().hex,
         scene=scene,
         seq=seq,
         video_url=video_url,
-        coin=0,
+        # 固化本次承诺发放的金币，避免运营改价后已开播记录按新价结算。
+        coin=reward_coin,
         status="playing",
         completed=0,
         started_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
@@ -210,14 +300,19 @@ def start_play(
             db.flush()
     except IntegrityError:
         db.rollback()
-        return _miss(used_plays(db, user_id))
+        used_now = (
+            used_plays(db, user_id, scene=scene, reset_at=policy.reset_at)
+            if policy is not None and policy.reset_at is not None
+            else used_plays(db, user_id, scene=scene)
+        )
+        return _miss(used_now)
     return {
         "should_play": True,
         "video_url": video_url,
         "play_token": play.play_token,
         "reward_coin": reward_coin,
         "seq": seq,
-        "remaining": max(0, max_plays - seq),
+        "remaining": max(0, exposed_max_plays - (used + 1)),
     }
 
 
@@ -240,8 +335,9 @@ def grant_play(
     重复上报返回 granted=False + 已发金币(客户端据此不重复累加 toast 金额)。
     """
     token = (play_token or "").strip()
-    # 金币额度以**服务端配置**为准,不信客户端(客户端只上报"播完/关闭")。
-    coin = int(get_config(db).get("reward_coin") or 0)
+    # 金币额度取开播时由服务端固化的值，不信客户端，也不受后续配置变更影响。
+    play = _find_play(db, user_id, token)
+    coin = int(play.coin if play is not None else 0)
 
     # 幂等核心:把 status 放进 WHERE 做条件更新(compare-and-set),而不是"先读再判再写"。
     # 「播完」与「✕ 关闭」抢跑、或客户端超时重试时,两个请求会都读到 status='playing',

app/repositories/phone_rebind.py

@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 import math
-from datetime import datetime, timedelta, timezone
+from datetime import UTC, datetime, timedelta
 
 from sqlalchemy import func, select
 from sqlalchemy.orm import Session
@@ -10,9 +10,22 @@ from sqlalchemy.orm import Session
 from app.models.phone_rebind_log import PhoneRebindLog
 
 
-def rebound_within_days(db: Session, phone: str, days: int) -> bool:
+def rebound_within_days(
+    db: Session,
+    phone: str,
+    days: int,
+    *,
+    reset_at: datetime | None = None,
+) -> bool:
     """该手机号在最近 days 天内是否换绑过(命中 → 禁止再次换绑)。"""
-    since = datetime.now(timezone.utc) - timedelta(days=days)
+    since = datetime.now(UTC) - timedelta(days=days)
+    if reset_at is not None:
+        reset_value = (
+            reset_at.replace(tzinfo=UTC)
+            if reset_at.tzinfo is None
+            else reset_at.astimezone(UTC)
+        )
+        since = max(since, reset_value)
     stmt = (
         select(PhoneRebindLog.id)
         .where(PhoneRebindLog.phone == phone, PhoneRebindLog.rebound_at >= since)
@@ -21,16 +34,25 @@ def rebound_within_days(db: Session, phone: str, days: int) -> bool:
     return db.execute(stmt).first() is not None
 
 
-def remaining_block_days(db: Session, phone: str, days: int) -> int:
+def remaining_block_days(
+    db: Session,
+    phone: str,
+    days: int,
+    *,
+    reset_at: datetime | None = None,
+) -> int:
     """距离该手机号可再次换绑还剩几天(向上取整;无记录返回 0)。"""
+    conditions = [PhoneRebindLog.phone == phone]
+    if reset_at is not None:
+        conditions.append(PhoneRebindLog.rebound_at >= reset_at)
     last = db.execute(
-        select(func.max(PhoneRebindLog.rebound_at)).where(PhoneRebindLog.phone == phone)
+        select(func.max(PhoneRebindLog.rebound_at)).where(*conditions)
     ).scalar_one_or_none()
     if last is None:
         return 0
     if last.tzinfo is None:  # SQLite 取回 naive datetime,按 UTC 归一
-        last = last.replace(tzinfo=timezone.utc)
-    remaining = (last + timedelta(days=days) - datetime.now(timezone.utc)).total_seconds()
+        last = last.replace(tzinfo=UTC)
+    remaining = (last + timedelta(days=days) - datetime.now(UTC)).total_seconds()
     return max(0, math.ceil(remaining / 86400))
 
 

app/repositories/risk.py

@@ -9,6 +9,7 @@ from sqlalchemy import func, select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
+from app.core import limit_policy
 from app.core.config_schema import (
     RISK_COMPARE_DAILY_THRESHOLD_KEY,
     RISK_ONECLICK_DAILY_THRESHOLD_KEY,
@@ -257,22 +258,34 @@ def _upsert_incident(
 
 
 def evaluate_behavior_rule(
-    db: Session, *, rule_code: str, subject_id: str, at: datetime
+    db: Session,
+    *,
+    rule_code: str,
+    subject_id: str,
+    at: datetime,
+    threshold: int | None = None,
+    subject_reset_at: datetime | None = None,
 ) -> RiskIncident | None:
     spec = RULES[rule_code]
-    threshold = get_rule_threshold(db, rule_code)
+    effective_threshold = threshold or get_rule_threshold(db, rule_code)
     window_key, window_start, end = _window_bounds(at, spec.window)
     reset_at = get_rule_reset_at(db, rule_code)
-    start = max(window_start, reset_at) if reset_at else window_start
+    baselines = [value for value in (reset_at, subject_reset_at) if value is not None]
+    start = max(window_start, *baselines) if baselines else window_start
     count, first_at, last_at, triggered_at = _event_stats(
         db,
         spec,
         subject_id=subject_id,
         start=start,
         end=end,
-        threshold=threshold,
+        threshold=effective_threshold,
     )
-    if count < threshold or first_at is None or last_at is None or triggered_at is None:
+    if (
+        count < effective_threshold
+        or first_at is None
+        or last_at is None
+        or triggered_at is None
+    ):
         return None
     return _upsert_incident(
         db,
@@ -300,7 +313,6 @@ def reconcile_behavior_rule(
     """按当前阈值重算短信/一键登录当前窗口，并收起已不再命中的待处理告警。"""
     spec = RULES[rule_code]
     current = at or utcnow()
-    threshold = get_rule_threshold(db, rule_code)
     window_key, window_start, end = _window_bounds(current, spec.window)
     reset_at = get_rule_reset_at(db, rule_code)
     start = max(window_start, reset_at) if reset_at else window_start
@@ -311,23 +323,39 @@ def reconcile_behavior_rule(
         BehaviorEvent.occurred_at >= start,
         BehaviorEvent.occurred_at < end,
     )
-    qualifying_rows = db.execute(
+    subject_rows = db.execute(
         select(
             BehaviorEvent.subject_id,
             func.max(BehaviorEvent.occurred_at),
+            func.max(BehaviorEvent.phone),
         )
         .where(*filters)
         .group_by(BehaviorEvent.subject_id)
-        .having(func.count(BehaviorEvent.id) >= threshold)
     ).all()
-    qualifying = {str(subject_id) for subject_id, _ in qualifying_rows}
-    for subject_id, last_at in qualifying_rows:
-        evaluate_behavior_rule(
+    qualifying: set[str] = set()
+    policy_code = {
+        RULE_SMS_HOURLY: "risk.sms.hourly",
+        RULE_ONECLICK_DAILY: "risk.oneclick.daily",
+    }[rule_code]
+    for subject_id, last_at, phone in subject_rows:
+        policy = limit_policy.resolve(
+            db,
+            policy_code,
+            phone=phone,
+            device=str(subject_id),
+        )
+        if policy.suppressed or policy.limit is None:
+            continue
+        incident = evaluate_behavior_rule(
             db,
             rule_code=rule_code,
             subject_id=str(subject_id),
             at=last_at or current,
+            threshold=policy.limit,
+            subject_reset_at=policy.reset_at,
         )
+        if incident is not None:
+            qualifying.add(str(subject_id))
 
     open_incidents = db.scalars(
         select(RiskIncident).where(
@@ -383,7 +411,29 @@ def record_behavior_event(
     db.add(event)
     db.flush()
     if evaluate_rule:
-        evaluate_behavior_rule(db, rule_code=evaluate_rule, subject_id=subject_id, at=at)
+        policy_code = {
+            RULE_SMS_HOURLY: "risk.sms.hourly",
+            RULE_ONECLICK_DAILY: "risk.oneclick.daily",
+        }.get(evaluate_rule)
+        policy = (
+            limit_policy.resolve(
+                db,
+                policy_code,
+                phone=phone,
+                device=device_id or subject_id,
+            )
+            if policy_code
+            else None
+        )
+        if policy is None or not policy.suppressed:
+            evaluate_behavior_rule(
+                db,
+                rule_code=evaluate_rule,
+                subject_id=subject_id,
+                at=at,
+                threshold=policy.limit if policy else None,
+                subject_reset_at=policy.reset_at if policy else None,
+            )
     if commit:
         db.commit()
         db.refresh(event)
@@ -396,20 +446,33 @@ def sync_compare_incident(
     user_id: int,
     at: datetime,
     threshold: int | None = None,
+    device_id: str | None = None,
     commit: bool = True,
 ) -> RiskIncident | None:
-    effective_threshold = threshold or get_rule_threshold(db, RULE_COMPARE_DAILY)
+    policy = limit_policy.resolve_for_user(
+        db,
+        "risk.compare.daily",
+        user_id,
+        device=device_id,
+    )
+    if policy.suppressed:
+        return None
+    effective_threshold = threshold or policy.limit or get_rule_threshold(
+        db, RULE_COMPARE_DAILY
+    )
     # comparison_record 的既有写入口统一落“北京时间 naive”时间；这里必须沿用同一
     # 口径，否则 SQLite/PG session timezone 不同时会把凌晨记录算到前一天。
     local = at.astimezone(CN_TZ).replace(tzinfo=None) if at.tzinfo else at
     window_start = local.replace(hour=0, minute=0, second=0, microsecond=0)
     end = window_start + timedelta(days=1)
     window_key = window_start.strftime("%Y-%m-%d")
-    reset_at = get_rule_reset_at(db, RULE_COMPARE_DAILY)
-    reset_local = (
-        reset_at.astimezone(CN_TZ).replace(tzinfo=None) if reset_at else None
-    )
-    start = max(window_start, reset_local) if reset_local else window_start
+    global_reset_at = get_rule_reset_at(db, RULE_COMPARE_DAILY)
+    baselines = [
+        value.astimezone(CN_TZ).replace(tzinfo=None)
+        for value in (global_reset_at, policy.reset_at)
+        if value is not None
+    ]
+    start = max(window_start, *baselines) if baselines else window_start
     filters = (
         ComparisonRecord.user_id == user_id,
         ComparisonRecord.created_at >= start,
@@ -464,7 +527,6 @@ def reconcile_compare_rule(
         reset_at.astimezone(CN_TZ).replace(tzinfo=None) if reset_at else None
     )
     start = max(window_start, reset_local) if reset_local else window_start
-    threshold = get_rule_threshold(db, RULE_COMPARE_DAILY)
     rows = db.execute(
         select(
             ComparisonRecord.user_id,
@@ -476,17 +538,17 @@ def reconcile_compare_rule(
             ComparisonRecord.created_at < end,
         )
         .group_by(ComparisonRecord.user_id)
-        .having(func.count(ComparisonRecord.id) >= threshold)
     ).all()
-    qualifying = {str(user_id) for user_id, _ in rows}
+    qualifying: set[str] = set()
     for user_id, last_at in rows:
-        sync_compare_incident(
+        incident = sync_compare_incident(
             db,
             user_id=int(user_id),
             at=last_at or current,
-            threshold=threshold,
             commit=False,
         )
+        if incident is not None:
+            qualifying.add(str(user_id))
 
     open_incidents = db.scalars(
         select(RiskIncident).where(

app/schemas/compare_record.py

@@ -107,6 +107,13 @@ 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 定——是
@@ -173,8 +180,13 @@ class ComparisonRecordOut(BaseModel):
     skipped_dish_count: int | None = None
     status: str
     information: str | None = None
+    # 失败卡「原因」文案:具体失败给具体原因,纯系统失败为 None(端侧品牌兜底)。见模型 fail_reason。
+    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。
@@ -210,9 +222,9 @@ class CompareStartReserveIn(BaseModel):
 
 
 class CompareStartReserveOut(BaseModel):
-    limit: int
+    limit: int | None
     used: int
-    remaining: int
+    remaining: int | None
 
 
 class CompareStatsOut(BaseModel):

app/schemas/guide_video.py

@@ -1,13 +1,15 @@
 """新手引导视频(领券等候浮层前 N 次替代广告)的客户端请求/响应契约。"""
 from __future__ import annotations
 
+from typing import Literal
+
 from pydantic import BaseModel, Field
 
 
 class GuideVideoStartIn(BaseModel):
-    """开播询问。scene 目前只有 coupon(领券浮层);预留给日后比价等场景。"""
+    """开播询问。领券与比价分别使用独立配置和独立次数。"""
 
-    scene: str = Field(default="coupon", max_length=16)
+    scene: Literal["coupon", "comparison"] = "coupon"
 
 
 class GuideVideoStartOut(BaseModel):

app/services/comparison_llm_backfill.py

@@ -135,7 +135,7 @@ def repair_missing_comparison_llm_costs(
                 select(ComparisonRecord.id, ComparisonRecord.trace_id)
                 .where(
                     *date_conditions,
-                    ComparisonRecord.status.in_(("success", "failed")),
+                    ComparisonRecord.status.in_(("success", "failed", "cancelled")),
                     ComparisonRecord.llm_cost_yuan.is_(None),
                 )
                 .order_by(ComparisonRecord.created_at.desc(), ComparisonRecord.id.desc())

tests/test_admin_config.py

@@ -68,12 +68,12 @@ def test_list_config(admin_client: TestClient, token: str) -> None:
     r = admin_client.get("/admin/api/config", headers=_auth(token))
     assert r.status_code == 200, r.text
     items = {i["key"]: i for i in r.json()}
-    # 非 hidden 项照常返回;看广告组保留可见的:每日上限 / 单次金币上限 / 关闭后冷却。
-    assert "signin_rewards" in items and "ad_daily_limit" in items and "ad_cooldown_sec" in items
-    # hidden 项(任务/里程碑、首页轮播数据源、看广告组的单次金币/每轮次数/信息流广告开关)不在配置页返回。
+    # 非 hidden 项照常返回；广告次数上限迁到「白名单」统一配置。
+    assert "signin_rewards" in items and "ad_cooldown_sec" in items
+    # hidden 项(任务/里程碑、首页轮播数据源、广告次数/单次金币/每轮次数/信息流广告开关)不在配置页返回。
     for hidden_key in (
         "task_rewards", "record_milestones", "marquee_feed_mode",
-        "ad_reward_coin", "ad_round_count", "comparing_ad_enabled",
+        "ad_daily_limit", "ad_reward_coin", "ad_round_count", "comparing_ad_enabled",
     ):
         assert hidden_key not in items, f"{hidden_key} 应被 hidden 过滤"
     assert items["signin_rewards"]["value"] == [

tests/test_admin_read.py

@@ -219,6 +219,115 @@ def test_user_reward_stats_can_scope_withdrawals_by_account(
     assert invite.json()["cash_balance_cents"] == 456
 
 
+def test_user_reward_stats_draw_ecpm_uses_all_filtered_impressions(
+    admin_client: TestClient, admin_token: str
+) -> None:
+    """Draw 平均 eCPM 应与收益报表一致，不能只平均成功发奖记录。"""
+    from app.models.ad_ecpm import AdEcpmRecord
+    from app.models.ad_feed_reward import AdFeedRewardRecord
+
+    uid = _seed_user_with_data("13800000024")
+    created_at = datetime(2038, 1, 15, 4, tzinfo=UTC)
+    db = SessionLocal()
+    try:
+        db.add_all(
+            [
+                AdFeedRewardRecord(
+                    client_event_id="reward-stats-granted-high",
+                    ad_session_id="reward-stats-granted-high",
+                    user_id=uid,
+                    reward_date="2038-01-15",
+                    duration_seconds=10,
+                    unit_count=1,
+                    ecpm_raw="9000",
+                    ad_type="draw",
+                    feed_scene="coupon",
+                    app_env="prod",
+                    our_code_id="104098712",
+                    coin=9,
+                    status="granted",
+                    created_at=created_at,
+                ),
+                AdEcpmRecord(
+                    user_id=uid,
+                    ad_type="draw",
+                    feed_scene="coupon",
+                    ad_session_id="reward-stats-impression-low",
+                    app_env="prod",
+                    our_code_id="104098712",
+                    ecpm_raw="1000",
+                    report_date="2038-01-15",
+                    created_at=created_at,
+                ),
+                AdEcpmRecord(
+                    user_id=uid,
+                    ad_type="feed",
+                    feed_scene="coupon",
+                    ad_session_id="reward-stats-impression-mid",
+                    app_env="prod",
+                    our_code_id="104098712",
+                    ecpm_raw="3000",
+                    report_date="2038-01-15",
+                    created_at=created_at,
+                ),
+                # 同用户但不同场景/环境/非业务代码位，均不应进入本次详情筛选。
+                AdEcpmRecord(
+                    user_id=uid,
+                    ad_type="draw",
+                    feed_scene="comparison",
+                    ad_session_id="reward-stats-other-scene",
+                    app_env="prod",
+                    our_code_id="104098712",
+                    ecpm_raw="7000",
+                    report_date="2038-01-15",
+                    created_at=created_at,
+                ),
+                AdEcpmRecord(
+                    user_id=uid,
+                    ad_type="draw",
+                    feed_scene="coupon",
+                    ad_session_id="reward-stats-test-env",
+                    app_env="test",
+                    our_code_id="104127529",
+                    ecpm_raw="8000",
+                    report_date="2038-01-15",
+                    created_at=created_at,
+                ),
+                AdEcpmRecord(
+                    user_id=uid,
+                    ad_type="draw",
+                    feed_scene="coupon",
+                    ad_session_id="reward-stats-non-business",
+                    app_env="prod",
+                    our_code_id="demo-slot",
+                    ecpm_raw="9000",
+                    report_date="2038-01-15",
+                    created_at=created_at,
+                ),
+            ]
+        )
+        db.commit()
+    finally:
+        db.close()
+
+    response = admin_client.get(
+        f"/admin/api/users/{uid}/reward-stats",
+        params={
+            "date_from": "2038-01-15T00:00:00Z",
+            "date_to": "2038-01-15T23:59:59Z",
+            "app_env": "prod",
+            "revenue_scope": "business",
+            "feed_scene": "coupon",
+        },
+        headers=_auth(admin_token),
+    )
+    assert response.status_code == 200, response.text
+    data = response.json()
+    assert data["feed_count"] == 1
+    # 全部真实展示 (1000 + 3000) / 2；不能返回成功发奖记录的 9000。
+    assert data["feed_avg_ecpm"] == 2000.0
+
+
 def test_user_coin_record_sort_accepts_mixed_timezone_datetimes() -> None:
     """线上 PostgreSQL 返回 aware，SQLite/历史转换可能返回 naive，二者必须可混排。"""
     naive = datetime(2038, 1, 1, 8, 0)

tests/test_admin_roles.py

@@ -73,6 +73,7 @@ def test_monitoring_audit_catalog_and_api_permissions(
         "analytics-health",
         "event-logs",
         "audit-logs",
+        "limit-whitelist",
     ]
 
     # 运营默认可查风控和设备存活，不能绕过导航直调技术/审计接口。
@@ -82,6 +83,9 @@ def test_monitoring_audit_catalog_and_api_permissions(
     assert admin_client.get(
         "/admin/api/device-liveness/stats", headers=_auth(operator_token)
     ).status_code == 200
+    assert admin_client.get(
+        "/admin/api/limit-whitelist/rules", headers=_auth(operator_token)
+    ).status_code == 200
     for path in (
         "/admin/api/analytics-health/overview?date_from=2026-07-01T00:00:00Z&date_to=2026-07-02T00:00:00Z",
         "/admin/api/event-logs",
@@ -89,13 +93,14 @@ def test_monitoring_audit_catalog_and_api_permissions(
     ):
         assert admin_client.get(path, headers=_auth(operator_token)).status_code == 403
 
-    # 技术角色默认拥有监控审计组全部五项权限。
+    # 技术角色默认拥有监控审计组全部六项权限。
     for path in (
         "/admin/api/risk-monitor/summary",
         "/admin/api/device-liveness/stats",
         "/admin/api/analytics-health/overview?date_from=2026-07-01T00:00:00Z&date_to=2026-07-02T00:00:00Z",
         "/admin/api/event-logs",
         "/admin/api/audit-logs",
+        "/admin/api/limit-whitelist/rules",
     ):
         assert admin_client.get(path, headers=_auth(tech_token)).status_code == 200
 
@@ -199,7 +204,7 @@ def test_builtin_roles_labels_and_pages(admin_client, super_token) -> None:
     }
     assert set(roles["tech"]["pages"]) == {
         "dashboard", "risk-monitor", "device-liveness", "analytics-health", "config",
-        "ad-revenue", "huawei-review", "event-logs", "audit-logs",
+        "ad-revenue", "huawei-review", "event-logs", "audit-logs", "limit-whitelist",
     }
 
 

tests/test_compare_harvest.py

@@ -102,6 +102,7 @@ def test_harvest_done_derives_and_newly_success_once(client) -> None:
         assert rec.is_source_best is False
         assert rec.store_name == "测试店"
         assert rec.information == "美团更便宜"
+        assert rec.fail_reason is None               # 成功记录不派生失败原因
         assert rec.items == [{"name": "肥牛饭", "qty": 1}]
         assert rec.trace_url.endswith("/done/")
         # 再来一次(重试 done)→ 已 success,newly_success=False(发奖不重复触发)
@@ -110,6 +111,35 @@ def test_harvest_done_derives_and_newly_success_once(client) -> None:
         assert newly2 is False
 
 
+def test_harvest_done_failed_derives_fail_reason(client) -> None:
+    """failed 记录:记录级 information 笼统,但 fail_reason 从 platform_results 救出具体原因
+    (id 3030 型:美团系统失败 + 京东 items_not_found → 展示京东那条)。"""
+    tid = _tid()
+    done_failed = {
+        "comparison_results": [
+            {"platform_id": "taobao_flash", "platform_name": "淘宝闪购",
+             "package": "com.taobao.taobao", "price": 23.04, "is_source": True, "rank": 1,
+             "items": [{"name": "肥牛饭", "qty": 1}]},
+        ],
+        "platform_results": {
+            "taobao_flash": {"is_source": True, "status": "source", "price": 23.04},
+            "meituan_waimai": {"is_source": False, "status": "failed",
+                               "reason": "搜索店铺失败, 无法跳转到搜索页"},
+            "jd_waimai_standalone": {"is_source": False, "status": "items_not_found",
+                                     "reason": "京东外卖此店内未找到这些菜品"},
+        },
+        "information": "比价过程出错，请稍后重试",
+    }
+    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_failed)
+        assert newly is False                        # 没落成 success
+        assert rec.status == "failed"
+        assert rec.fail_reason == "京东外卖此店内未找到这些菜品"
+        assert rec.information == "比价过程出错，请稍后重试"  # 原文案仍留存
+
+
 def test_harvest_abort_cancels_running(client) -> None:
     tid = _tid()
     with SessionLocal() as db:
@@ -233,7 +263,9 @@ def test_trace_finalize_harvests_abort(client) -> None:
     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:
+    with p, patch(
+        "app.api.v1.compare.backfill_comparison_llm_cost"
+    ) as backfill:
         r = client.post("/api/v1/trace/finalize",
                         json={"trace_id": tid, "status": "cancelled", "reason": "用户终止"})
     assert r.status_code == 200
@@ -241,6 +273,7 @@ def test_trace_finalize_harvests_abort(client) -> None:
         rec = _get(db, tid)
         assert rec is not None and rec.status == "cancelled"
         assert rec.trace_url.endswith("/fin/")
+        backfill.assert_called_once_with(rec.id, tid)
 
 
 def test_price_step_binds_user_when_authed(client) -> None:

tests/test_comparison_llm_backfill.py

@@ -72,6 +72,7 @@ def test_backfill_retries_then_persists_cost(monkeypatch):
 
 def test_repair_batch_only_targets_terminal_missing_rows(monkeypatch):
     missing_id = _record("llm-repair-missing")
+    cancelled_id = _record("llm-repair-cancelled", status="cancelled")
     running_id = _record("llm-repair-running", status="running")
     calls = [
         {
@@ -96,12 +97,15 @@ def test_repair_batch_only_targets_terminal_missing_rows(monkeypatch):
         )
         assert result["repaired"] >= 1
         assert "llm-repair-missing" in seen
+        assert "llm-repair-cancelled" in seen
         assert "llm-repair-running" not in seen
         with SessionLocal() as db:
             assert db.get(ComparisonRecord, missing_id).llm_cost_yuan is not None
+            assert db.get(ComparisonRecord, cancelled_id).llm_cost_yuan is not None
             assert db.get(ComparisonRecord, running_id).llm_cost_yuan is None
     finally:
         _delete(missing_id)
+        _delete(cancelled_id)
         _delete(running_id)
 
 

tests/test_fail_reason.py

@@ -0,0 +1,122 @@
+"""失败卡展示原因派生(repositories.comparison._derive_fail_display)单元测试。
+
+用例取自线上真实 failed 记录(platform_results 形态),覆盖:
+  - information 具体 → 直出
+  - information 笼统 + platform_results 有干净业务结局 → 救援出该原因(id 3030/2964 型)
+  - information 笼统 + 仅系统失败(搜索失败等黑话)→ None(端侧品牌兜底,id 3027 型)
+  - store_closed / no_delivery 被 pricebot 漏成 status=failed → 按 reason 关键字补判
+  - 打烊类脏店名 blob → 统一简短模板
+  - platform_results 为空 / 非对象 → None
+"""
+from __future__ import annotations
+
+from app.repositories import comparison as crud
+
+
+def test_specific_information_passthrough() -> None:
+    # information 本身具体(未达起送/找不到菜等)→ 直出,不看 platform_results
+    assert (
+        crud._derive_fail_display("淘宝闪购未达起送门槛，可加菜凑单后下单", {})
+        == "淘宝闪购未达起送门槛，可加菜凑单后下单"
+    )
+    assert crud._derive_fail_display("未识别到商品", {}) == "未识别到商品"
+
+
+def test_generic_info_rescued_from_items_not_found() -> None:
+    # id 3030 型:美团系统失败 + 京东 items_not_found,记录级 information 笼统 → 救出京东那条
+    pr = {
+        "eleme": {"is_source": True, "status": "source", "price": 23.04},
+        "meituan_waimai": {
+            "is_source": False, "status": "failed",
+            "reason": "搜索店铺失败, 无法跳转到搜索页",
+        },
+        "jd_waimai_standalone": {
+            "is_source": False, "status": "items_not_found",
+            "reason": "京东外卖此店内未找到这些菜品",
+        },
+    }
+    assert (
+        crud._derive_fail_display("比价过程出错，请稍后重试", pr)
+        == "京东外卖此店内未找到这些菜品"
+    )
+
+
+def test_generic_info_rescued_from_store_not_found() -> None:
+    # id 2964 型:美团系统失败 + 京东 store_not_found → 救出京东相似店铺文案
+    pr = {
+        "taobao_flash": {"is_source": True, "status": "source", "price": 127.98},
+        "meituan": {
+            "is_source": False, "status": "failed",
+            "reason": "比价过程出错，请稍后重试",
+        },
+        "jd_waimai": {
+            "is_source": False, "status": "store_not_found",
+            "reason": "未在京东找到「黔珍味·贵州牛肉蘸水健康菜 (望京店)」相似店铺",
+        },
+    }
+    assert (
+        crud._derive_fail_display("比价过程出错，请稍后重试", pr)
+        == "未在京东找到「黔珍味·贵州牛肉蘸水健康菜 (望京店)」相似店铺"
+    )
+
+
+def test_generic_info_pure_system_failure_returns_none() -> None:
+    # id 3027 型:唯一目标平台是自动化黑话失败 → 不给用户看 → None(端侧品牌兜底)
+    pr = {
+        "eleme": {"is_source": True, "status": "source", "price": 18.83},
+        "meituan_waimai": {
+            "is_source": False, "status": "failed",
+            "reason": "搜索店铺失败, 无法跳转到搜索页",
+        },
+    }
+    assert crud._derive_fail_display("比价过程出错，请稍后重试", pr) is None
+
+
+def test_store_closed_leaked_to_failed_is_rescued_and_cleaned() -> None:
+    # 打烊被漏成 status=failed;reason 常带脏店名 blob → 统一简短模板
+    pr = {
+        "jd_waimai": {"is_source": True, "status": "source", "price": 25},
+        "taobao_flash": {
+            "is_source": False, "status": "failed",
+            "reason": "淘宝闪购「沙胆彪炭炉牛杂煲(...)，蜂鸟准时达，月售300+，起送¥20」本店已休息,无法比价",
+        },
+    }
+    assert crud._derive_fail_display("比价过程出错，请稍后重试", pr) == "门店休息中，无法比价"
+
+    pr2 = {
+        "taobao_flash": {"is_source": True, "status": "source", "price": 31.83},
+        "meituan": {
+            "is_source": False, "status": "failed",
+            "reason": "美团「奈雪的茶（北京王府井奥莱·香江」门店已打烊,无法比价",
+        },
+    }
+    assert crud._derive_fail_display("比价出错", pr2) == "门店已打烊，无法比价"
+
+
+def test_no_delivery_leaked_to_failed_is_rescued() -> None:
+    # 单点不配送被漏成 status=failed;reason 本身干净 → 直接用
+    reason = "京东外卖该商家所选商品单点不配送，无法进入结算比价"
+    pr = {
+        "taobao_flash": {"is_source": True, "status": "source", "price": 20.1},
+        "jd_waimai_standalone": {
+            "is_source": False, "status": "failed", "reason": reason,
+        },
+    }
+    assert crud._derive_fail_display("比价过程出错，请稍后重试", pr) == reason
+
+
+def test_empty_or_missing_platform_results_returns_none() -> None:
+    # 「比价出错」+ 空 {} / None / 非对象:引擎早夭,无可展示原因 → None
+    assert crud._derive_fail_display("比价出错", {}) is None
+    assert crud._derive_fail_display("比价过程出错，请稍后重试", None) is None
+    assert crud._derive_fail_display("比价出错", []) is None  # 老 array 形态,防御
+
+
+def test_clean_status_wins_over_priority_order() -> None:
+    # 多个业务结局同现时按 _BIZ_STATUS_PRIORITY 选(below_minimum 优先于 store_not_found)
+    pr = {
+        "src": {"is_source": True, "status": "source", "price": 30},
+        "a": {"is_source": False, "status": "store_not_found", "reason": "未找到店铺A"},
+        "b": {"is_source": False, "status": "below_minimum", "reason": "B未达起送门槛"},
+    }
+    assert crud._derive_fail_display("比价过程出错，请稍后重试", pr) == "B未达起送门槛"

tests/test_guide_video.py

@@ -75,7 +75,7 @@ def _assert_ledger_balanced(user_id: int) -> None:
 
 @pytest.fixture()
 def guide_configured():
-    """给全局配置塞一支片子(默认 max_plays=3 / reward_coin=120),用完还原成未配片。"""
+    """给全局配置塞一支片子(默认 max_plays=3 / reward_coin=100),用完还原成未配片。"""
     with SessionLocal() as db:
         crud_guide.set_video(db, VIDEO_URL, admin_id=1)
         cfg = crud_guide.get_config(db)
@@ -201,7 +201,11 @@ def test_start_loses_seq_race_degrades_to_ad(client, guide_configured, monkeypat
     assert first["should_play"] is True and first["seq"] == 1
 
     # 本次请求读到的是过期计数 → 仍会算出 seq=1
-    monkeypatch.setattr(crud_guide, "used_plays", lambda db, user_id: 0)
+    monkeypatch.setattr(
+        crud_guide,
+        "used_plays",
+        lambda db, user_id, *, scene="coupon", reset_at=None: 0,
+    )
     with SessionLocal() as db:
         result = crud_guide.start_play(db, uid)
 
@@ -261,3 +265,99 @@ def test_reward_loses_race_does_not_double_mint(client, guide_configured) -> Non
     assert _coin_balance(uid) == coin, "一次播放只能发一次币"
     assert len(_guide_txns(uid)) == 1, "一个 play_token 只能有一条金币流水"
     _assert_ledger_balanced(uid)
+
+
+def test_coupon_and_comparison_configs_and_counts_are_independent(client) -> None:
+    phone = "13920000008"
+    token = _login(client, phone)
+    uid = _user_id(phone)
+    comparison_url = "/media/guide_video/pytest_comparison.mp4"
+
+    with SessionLocal() as db:
+        coupon_before = crud_guide.get_config(db, "coupon")
+        comparison_before = crud_guide.get_config(db, "comparison")
+        crud_guide.set_video(db, VIDEO_URL, scene="coupon", admin_id=1)
+        crud_guide.update_config(
+            db,
+            scene="comparison",
+            enabled=True,
+            max_plays=1,
+            reward_coin=321,
+            admin_id=1,
+        )
+        crud_guide.set_video(
+            db,
+            comparison_url,
+            scene="comparison",
+            admin_id=1,
+        )
+
+    try:
+        coupon = client.post(
+            "/api/v1/guide-video/start",
+            json={"scene": "coupon"},
+            headers=_auth(token),
+        ).json()
+        comparison = client.post(
+            "/api/v1/guide-video/start",
+            json={"scene": "comparison"},
+            headers=_auth(token),
+        ).json()
+        comparison_blocked = client.post(
+            "/api/v1/guide-video/start",
+            json={"scene": "comparison"},
+            headers=_auth(token),
+        ).json()
+
+        assert coupon["should_play"] is True
+        assert coupon["video_url"] == VIDEO_URL
+        assert coupon["seq"] == 1
+        assert comparison["should_play"] is True
+        assert comparison["video_url"] == comparison_url
+        assert comparison["reward_coin"] == 321
+        assert comparison["seq"] == 1
+        assert comparison_blocked["should_play"] is False
+
+        with SessionLocal() as db:
+            assert crud_guide.play_stats(db, "coupon")["total_plays"] >= 1
+            assert crud_guide.play_stats(db, "comparison")["total_plays"] == 1
+            rows = list(
+                db.scalars(
+                    select(GuideVideoPlay)
+                    .where(GuideVideoPlay.user_id == uid)
+                    .order_by(GuideVideoPlay.scene)
+                )
+            )
+            assert [(row.scene, row.seq) for row in rows] == [
+                ("comparison", 1),
+                ("coupon", 1),
+            ]
+    finally:
+        with SessionLocal() as db:
+            crud_guide.update_config(
+                db,
+                scene="coupon",
+                enabled=coupon_before["enabled"],
+                reward_coin=coupon_before["reward_coin"],
+                admin_id=1,
+            )
+            crud_guide.set_video(
+                db,
+                coupon_before["video_url"],
+                scene="coupon",
+                admin_id=1,
+            )
+            crud_guide.update_config(
+                db,
+                scene="comparison",
+                enabled=comparison_before["enabled"],
+                max_plays=comparison_before["max_plays"],
+                reward_coin=comparison_before["reward_coin"],
+                admin_id=1,
+            )
+            crud_guide.set_video(
+                db,
+                comparison_before["video_url"],
+                scene="comparison",
+                admin_id=1,
+            )

tests/test_limit_whitelist.py

@@ -0,0 +1,943 @@
+from __future__ import annotations
+
+from copy import deepcopy
+from datetime import UTC, datetime, timedelta
+from uuid import uuid4
+
+import pytest
+from fastapi.testclient import TestClient
+
+from app.admin.main import admin_app
+from app.admin.repositories import admin_user as admin_repo
+from app.admin.security import create_admin_token
+from app.api.v1 import auth as auth_api
+from app.core import limit_policy
+from app.core.config_schema import (
+    AD_FEED_DAILY_LIMIT_KEY,
+    AD_REWARD_VIDEO_DAILY_LIMIT_KEY,
+    GUIDE_VIDEO_MAX_PLAYS_KEY,
+    PHONE_REBIND_DAYS_KEY,
+    SMS_PHONE_COOLDOWN_SECONDS_KEY,
+)
+from app.core.security import create_token
+from app.db.session import SessionLocal
+from app.main import app
+from app.models.app_config import AppConfig
+from app.models.comparison import ComparisonRecord
+from app.models.limit_policy import LimitPolicyOverride
+from app.models.risk import RiskIncident
+from app.repositories import risk as risk_repo
+from app.repositories import user as user_repo
+
+
+def _snapshot_configs(keys: list[str]) -> dict[str, tuple[bool, object, int | None]]:
+    with SessionLocal() as db:
+        snapshot = {}
+        for key in keys:
+            row = db.get(AppConfig, key)
+            snapshot[key] = (
+                row is not None,
+                deepcopy(row.value) if row is not None else None,
+                row.updated_by_admin_id if row is not None else None,
+            )
+        return snapshot
+
+
+def _restore_configs(snapshot: dict[str, tuple[bool, object, int | None]]) -> None:
+    with SessionLocal() as db:
+        for key, (existed, value, admin_id) in snapshot.items():
+            row = db.get(AppConfig, key)
+            if not existed:
+                if row is not None:
+                    db.delete(row)
+                continue
+            if row is None:
+                db.add(
+                    AppConfig(
+                        key=key,
+                        value=deepcopy(value),
+                        updated_by_admin_id=admin_id,
+                    )
+                )
+            else:
+                row.value = deepcopy(value)
+                row.updated_by_admin_id = admin_id
+        db.commit()
+
+
+@pytest.fixture()
+def admin_headers() -> dict[str, str]:
+    username = f"limit_admin_{uuid4().hex[:8]}"
+    with SessionLocal() as db:
+        admin = admin_repo.create_admin(
+            db,
+            username=username,
+            password="limit-admin-pass",
+            role="super_admin",
+        )
+        admin_id = admin.id
+    token, _ = create_admin_token(admin_id=admin_id, role="super_admin")
+    return {"Authorization": f"Bearer {token}"}
+
+
+@pytest.fixture()
+def custom_admin_headers() -> dict[str, str]:
+    username = f"limit_custom_{uuid4().hex[:8]}"
+    with SessionLocal() as db:
+        admin = admin_repo.create_admin(
+            db,
+            username=username,
+            password="limit-admin-pass",
+            role="custom",
+            pages_override=["limit-whitelist"],
+        )
+        admin_id = admin.id
+    token, _ = create_admin_token(admin_id=admin_id, role="custom")
+    return {"Authorization": f"Bearer {token}"}
+
+
+def test_custom_admin_with_page_permission_can_manage_whitelist(
+    custom_admin_headers,
+) -> None:
+    phone = f"139{int(uuid4().hex[:8], 16) % 100000000:08d}"
+    expires_at = (datetime.now(UTC) + timedelta(hours=2)).isoformat()
+
+    with TestClient(admin_app) as client:
+        created = client.post(
+            "/admin/api/limit-whitelist/bulk",
+            headers=custom_admin_headers,
+            json={
+                "subject_type": "phone",
+                "subject_value": phone,
+                "rule_codes": ["compare.start.daily"],
+                "expires_at": expires_at,
+                "reason": "custom role page permission",
+            },
+        )
+        assert created.status_code == 201, created.text
+        override_id = created.json()[0]["id"]
+        deleted = client.delete(
+            f"/admin/api/limit-whitelist/{override_id}",
+            headers=custom_admin_headers,
+        )
+        assert deleted.status_code == 204, deleted.text
+
+
+def test_bulk_create_reactivates_existing_rows(admin_headers) -> None:
+    phone = f"139{int(uuid4().hex[:8], 16) % 100000000:08d}"
+    first_expiry = (datetime.now(UTC) + timedelta(hours=1)).isoformat()
+    second_expiry = (datetime.now(UTC) + timedelta(hours=3)).isoformat()
+
+    with TestClient(admin_app) as client:
+        first = client.post(
+            "/admin/api/limit-whitelist/bulk",
+            headers=admin_headers,
+            json={
+                "subject_type": "phone",
+                "subject_value": phone,
+                "rule_codes": ["compare.start.daily"],
+                "expires_at": first_expiry,
+                "reason": "first period",
+            },
+        )
+        assert first.status_code == 201, first.text
+        first_id = first.json()[0]["id"]
+        reset = client.post(
+            f"/admin/api/limit-whitelist/{first_id}/reset",
+            headers=admin_headers,
+        )
+        assert reset.status_code == 200, reset.text
+
+        recreated = client.post(
+            "/admin/api/limit-whitelist/bulk",
+            headers=admin_headers,
+            json={
+                "subject_type": "phone",
+                "subject_value": phone,
+                "rule_codes": [
+                    "compare.start.daily",
+                    "sms.phone.cooldown",
+                ],
+                "expires_at": second_expiry,
+                "reason": "renewed period",
+            },
+        )
+        assert recreated.status_code == 201, recreated.text
+        rows = {item["rule_code"]: item for item in recreated.json()}
+        assert rows["compare.start.daily"]["id"] == first_id
+        assert rows["compare.start.daily"]["enabled"] is True
+        assert rows["compare.start.daily"]["status"] == "active"
+        assert rows["compare.start.daily"]["reason"] == "renewed period"
+        assert rows["sms.phone.cooldown"]["enabled"] is True
+
+
+def test_admin_whitelist_crud_and_policy_precedence(admin_headers) -> None:
+    suffix = uuid4().hex[:10]
+    phone = f"139{int(suffix[:8], 16) % 100000000:08d}"
+    device = f"limit-device-{suffix}"
+    expires_at = (datetime.now(UTC) + timedelta(hours=2)).isoformat()
+
+    with TestClient(admin_app) as client:
+        rules = client.get(
+            "/admin/api/limit-whitelist/rules",
+            headers=admin_headers,
+        )
+        assert rules.status_code == 200, rules.text
+        assert {item["code"] for item in rules.json()} >= {
+            "compare.start.daily",
+            "sms.send.hourly",
+            "ad.reward_video.daily",
+            "risk.compare.daily",
+        }
+        rules_by_code = {item["code"]: item for item in rules.json()}
+        assert all(
+            "inherit" not in item["allowed_modes"] for item in rules.json()
+        )
+        assert all(
+            "override" not in item["allowed_modes"] for item in rules.json()
+        )
+        assert "unlimited" in rules_by_code["sms.phone.cooldown"]["allowed_modes"]
+        assert "unlimited" in rules_by_code["sms.code.failed_attempts"]["allowed_modes"]
+        assert "suppress_alert" in rules_by_code["risk.compare.daily"]["allowed_modes"]
+
+        removed_inherit = client.post(
+            "/admin/api/limit-whitelist",
+            headers=admin_headers,
+            json={
+                "subject_type": "phone",
+                "subject_value": phone,
+                "rule_code": "compare.start.daily",
+                "mode": "inherit",
+                "enabled": True,
+            },
+        )
+        assert removed_inherit.status_code == 422
+
+        invalid_alert = client.post(
+            "/admin/api/limit-whitelist",
+            headers=admin_headers,
+            json={
+                "subject_type": "phone",
+                "subject_value": phone,
+                "rule_code": "compare.start.daily",
+                "mode": "suppress_alert",
+                "enabled": True,
+                "reason": "invalid",
+            },
+        )
+        assert invalid_alert.status_code == 400
+
+        missing_expiry = client.post(
+            "/admin/api/limit-whitelist",
+            headers=admin_headers,
+            json={
+                "subject_type": "phone",
+                "subject_value": phone,
+                "rule_code": "compare.start.daily",
+                "mode": "unlimited",
+                "enabled": True,
+                "reason": "temporary QA",
+            },
+        )
+        assert missing_expiry.status_code == 400
+
+        expired_unlimited = client.post(
+            "/admin/api/limit-whitelist",
+            headers=admin_headers,
+            json={
+                "subject_type": "phone",
+                "subject_value": phone,
+                "rule_code": "compare.start.daily",
+                "mode": "unlimited",
+                "enabled": True,
+                "expires_at": "2020-01-01T00:00:00Z",
+                "reason": "expired temporary policy",
+            },
+        )
+        assert expired_unlimited.status_code == 400
+        assert "晚于当前时间" in expired_unlimited.json()["detail"]
+
+        removed_override = client.post(
+            "/admin/api/limit-whitelist",
+            headers=admin_headers,
+            json={
+                "subject_type": "phone",
+                "subject_value": phone,
+                "rule_code": "compare.start.daily",
+                "mode": "override",
+                "limit_value": 2,
+                "enabled": True,
+                "expires_at": expires_at,
+            },
+        )
+        assert removed_override.status_code == 422
+
+        phone_created = client.post(
+            "/admin/api/limit-whitelist",
+            headers=admin_headers,
+            json={
+                "subject_type": "phone",
+                "subject_value": phone,
+                "rule_code": "compare.start.daily",
+                "mode": "unlimited",
+                "enabled": True,
+                "expires_at": expires_at,
+                "reason": "temporary QA",
+            },
+        )
+        assert phone_created.status_code == 201, phone_created.text
+        phone_id = phone_created.json()["id"]
+        assert phone_created.json()["effective_limit"] is None
+
+        phone_updated = client.patch(
+            f"/admin/api/limit-whitelist/{phone_id}",
+            headers=admin_headers,
+            json={
+                "enabled": True,
+                "reason": "updated from admin",
+            },
+        )
+        assert phone_updated.status_code == 200, phone_updated.text
+        assert phone_updated.json()["effective_limit"] is None
+        assert phone_updated.json()["reason"] == "updated from admin"
+        with SessionLocal() as db:
+            persisted = db.get(LimitPolicyOverride, phone_id)
+            assert persisted is not None
+            assert persisted.mode == "unlimited"
+            assert persisted.limit_value is None
+            assert persisted.reason == "updated from admin"
+
+        removed_patch_fields = client.patch(
+            f"/admin/api/limit-whitelist/{phone_id}",
+            headers=admin_headers,
+            json={"mode": "override", "limit_value": 4},
+        )
+        assert removed_patch_fields.status_code == 422
+
+        inverted_time = client.patch(
+            f"/admin/api/limit-whitelist/{phone_id}",
+            headers=admin_headers,
+            json={
+                "starts_at": "2030-01-02T00:00:00Z",
+                "expires_at": "2030-01-01T00:00:00Z",
+            },
+        )
+        assert inverted_time.status_code == 400
+        assert "晚于生效时间" in inverted_time.json()["detail"]
+
+        duplicate = client.post(
+            "/admin/api/limit-whitelist",
+            headers=admin_headers,
+            json={
+                "subject_type": "phone",
+                "subject_value": phone,
+                "rule_code": "compare.start.daily",
+                "mode": "unlimited",
+                "enabled": True,
+                "expires_at": expires_at,
+            },
+        )
+        assert duplicate.status_code == 409
+
+        device_created = client.post(
+            "/admin/api/limit-whitelist",
+            headers=admin_headers,
+            json={
+                "subject_type": "device",
+                "subject_value": device,
+                "rule_code": "compare.start.daily",
+                "mode": "unlimited",
+                "enabled": True,
+                "expires_at": expires_at,
+                "reason": "device QA",
+            },
+        )
+        assert device_created.status_code == 201, device_created.text
+        device_id = device_created.json()["id"]
+        assert device_created.json()["effective_limit"] is None
+
+        with SessionLocal() as db:
+            effective = limit_policy.resolve(
+                db,
+                "compare.start.daily",
+                phone=phone,
+                device=device,
+            )
+            assert effective.unlimited is True
+            assert effective.matched_subject_type == "device"
+
+        restored = client.post(
+            f"/admin/api/limit-whitelist/{phone_id}/reset",
+            headers=admin_headers,
+        )
+        assert restored.status_code == 200, restored.text
+        assert restored.json()["enabled"] is False
+        assert restored.json()["status"] == "disabled"
+        assert restored.json()["effective_limit"] == restored.json()["global_limit"]
+        assert restored.json()["reset_at"] is None
+        with SessionLocal() as db:
+            effective = limit_policy.resolve(
+                db,
+                "compare.start.daily",
+                phone=phone,
+                device=None,
+            )
+            assert effective.override_id is None
+            assert effective.limit == effective.global_limit
+
+        ordered = client.get(
+            "/admin/api/limit-whitelist",
+            headers=admin_headers,
+            params={"rule_code": "compare.start.daily", "limit": 500},
+        )
+        assert ordered.status_code == 200, ordered.text
+        ordered_ids = [item["id"] for item in ordered.json()["items"]]
+        assert ordered_ids.index(device_id) < ordered_ids.index(phone_id)
+
+        listed = client.get(
+            "/admin/api/limit-whitelist",
+            headers=admin_headers,
+            params={"keyword": suffix[:5]},
+        )
+        assert listed.status_code == 200
+        assert listed.json()["total"] >= 1
+
+        deleted = client.delete(
+            f"/admin/api/limit-whitelist/{device_id}",
+            headers=admin_headers,
+        )
+        assert deleted.status_code == 204
+        client.delete(
+            f"/admin/api/limit-whitelist/{phone_id}",
+            headers=admin_headers,
+        )
+
+
+def test_device_candidates_are_rule_aware_and_searchable(admin_headers) -> None:
+    suffix = uuid4().hex[:8]
+    phone = f"135{int(suffix, 16) % 100000000:08d}"
+    auth_device = f"android-id-{suffix}"
+    compare_device = f"device-PJZ110-{suffix}"
+    now = datetime.now(UTC).replace(microsecond=0)
+    with SessionLocal() as db:
+        user = user_repo.upsert_user_for_login(
+            db,
+            phone=phone,
+            register_channel="sms",
+        )
+        user.nickname = f"候选设备用户{suffix}"
+        risk_repo.record_behavior_event(
+            db,
+            event_type=risk_repo.EVENT_ONECLICK_LOGIN,
+            subject_type="device",
+            subject_id=auth_device,
+            user_id=user.id,
+            device_id=auth_device,
+            device_model="OPPO Find X8",
+            phone=phone,
+            outcome="success",
+            occurred_at=now,
+        )
+        db.add(
+            ComparisonRecord(
+                user_id=user.id,
+                device_id=compare_device,
+                trace_id=f"device-candidate-{suffix}",
+                device_model="PJZ110",
+                status="success",
+                items=[],
+                comparison_results=[],
+                skipped_dish_names=[],
+                created_at=now.replace(tzinfo=None),
+            )
+        )
+        db.commit()
+        user_id = user.id
+        username = user.username
+
+    with TestClient(admin_app) as client:
+        for keyword in (phone, str(user_id), username, "Find X8"):
+            response = client.get(
+                "/admin/api/limit-whitelist/device-candidates",
+                headers=admin_headers,
+                params={
+                    "rule_code": "risk.oneclick.daily",
+                    "keyword": keyword,
+                },
+            )
+            assert response.status_code == 200, response.text
+            item = next(
+                row for row in response.json() if row["device_id"] == auth_device
+            )
+            assert item["source_label"] == "一键登录"
+            assert item["user_id"] == user_id
+            assert item["phone"] == phone
+            assert item["device_model"] == "OPPO Find X8"
+
+        comparison = client.get(
+            "/admin/api/limit-whitelist/device-candidates",
+            headers=admin_headers,
+            params={
+                "rule_code": "compare.start.daily",
+                "keyword": "PJZ110",
+            },
+        )
+        assert comparison.status_code == 200, comparison.text
+        assert len(comparison.json()) == 1
+        comparison_item = comparison.json()[0]
+        assert comparison_item["device_id"] == compare_device
+        assert comparison_item["source"] == "comparison_record"
+        assert comparison_item["source_label"] == "比价记录"
+        assert comparison_item["user_id"] == user_id
+        assert comparison_item["username"] == username
+        assert comparison_item["phone"] == phone
+        assert comparison_item["nickname"] == f"候选设备用户{suffix}"
+        assert comparison_item["device_model"] == "PJZ110"
+        assert comparison_item["last_active_at"].startswith(
+            now.replace(tzinfo=None).isoformat()
+        )
+
+        phone_only = client.get(
+            "/admin/api/limit-whitelist/device-candidates",
+            headers=admin_headers,
+            params={"rule_code": "risk.compare.daily"},
+        )
+        assert phone_only.status_code == 400
+        assert "不支持设备白名单" in phone_only.json()["detail"]
+
+
+def test_legacy_ip_is_not_a_device_candidate_or_valid_whitelist_subject(
+    admin_headers,
+) -> None:
+    suffix = uuid4().hex[:8]
+    legacy_device = f"legacy-ip:203.0.113.{int(suffix[:2], 16) % 200 + 1}"
+    expires_at = (datetime.now(UTC) + timedelta(hours=2)).isoformat()
+    with SessionLocal() as db:
+        risk_repo.record_behavior_event(
+            db,
+            event_type=risk_repo.EVENT_SMS_SEND,
+            subject_type="device",
+            subject_id=legacy_device,
+            device_id=None,
+            phone=f"136{int(suffix, 16) % 100000000:08d}",
+            outcome="success",
+        )
+
+    with TestClient(admin_app) as client:
+        candidates = client.get(
+            "/admin/api/limit-whitelist/device-candidates",
+            headers=admin_headers,
+            params={
+                "rule_code": "sms.send.hourly",
+                "keyword": legacy_device,
+            },
+        )
+        assert candidates.status_code == 200, candidates.text
+        assert all(
+            row["device_id"] != legacy_device for row in candidates.json()
+        )
+
+        single = client.post(
+            "/admin/api/limit-whitelist",
+            headers=admin_headers,
+            json={
+                "subject_type": "device",
+                "subject_value": legacy_device,
+                "rule_code": "sms.send.hourly",
+                "mode": "unlimited",
+                "enabled": True,
+                "expires_at": expires_at,
+            },
+        )
+        assert single.status_code == 400, single.text
+        assert "未上报真实设备 ID" in single.json()["detail"]
+
+        bulk = client.post(
+            "/admin/api/limit-whitelist/bulk",
+            headers=admin_headers,
+            json={
+                "subject_type": "device",
+                "subject_value": legacy_device,
+                "rule_codes": ["sms.send.hourly", "sms.send.daily"],
+                "enabled": True,
+                "expires_at": expires_at,
+            },
+        )
+        assert bulk.status_code == 400, bulk.text
+        assert "未上报真实设备 ID" in bulk.json()["detail"]
+
+
+def test_bulk_create_automatically_selects_unlimited_and_suppress_modes(
+    admin_headers,
+) -> None:
+    suffix = uuid4().hex[:8]
+    phone = f"137{int(suffix, 16) % 100000000:08d}"
+    expires_at = (datetime.now(UTC) + timedelta(hours=2)).isoformat()
+
+    with TestClient(admin_app) as client:
+        created = client.post(
+            "/admin/api/limit-whitelist/bulk",
+            headers=admin_headers,
+            json={
+                "subject_type": "phone",
+                "subject_value": phone,
+                "rule_codes": [
+                    "compare.start.daily",
+                    "risk.compare.daily",
+                ],
+                "expires_at": expires_at,
+                "reason": "批量白名单测试",
+            },
+        )
+        assert created.status_code == 201, created.text
+        rows = {item["rule_code"]: item for item in created.json()}
+        assert rows["compare.start.daily"]["mode"] == "unlimited"
+        assert rows["compare.start.daily"]["effective_limit"] is None
+        assert rows["risk.compare.daily"]["mode"] == "suppress_alert"
+        assert rows["risk.compare.daily"]["effective_limit"] == rows[
+            "risk.compare.daily"
+        ]["global_limit"]
+
+        duplicate_batch = client.post(
+            "/admin/api/limit-whitelist/bulk",
+            headers=admin_headers,
+            json={
+                "subject_type": "phone",
+                "subject_value": phone,
+                "rule_codes": [
+                    "compare.start.daily",
+                    "sms.send.daily",
+                ],
+                "expires_at": expires_at,
+            },
+        )
+        assert duplicate_batch.status_code == 409
+        with SessionLocal() as db:
+            rolled_back = (
+                db.query(LimitPolicyOverride)
+                .filter(
+                    LimitPolicyOverride.subject_type == "phone",
+                    LimitPolicyOverride.subject_value == phone,
+                    LimitPolicyOverride.rule_code == "sms.send.daily",
+                )
+                .one_or_none()
+            )
+            assert rolled_back is None
+
+        for item in created.json():
+            deleted = client.delete(
+                f"/admin/api/limit-whitelist/{item['id']}",
+                headers=admin_headers,
+            )
+            assert deleted.status_code == 204
+
+
+def test_device_bulk_rejects_mixed_candidate_namespaces(admin_headers) -> None:
+    expires_at = (datetime.now(UTC) + timedelta(hours=2)).isoformat()
+    with TestClient(admin_app) as client:
+        response = client.post(
+            "/admin/api/limit-whitelist/bulk",
+            headers=admin_headers,
+            json={
+                "subject_type": "device",
+                "subject_value": f"mixed-device-{uuid4().hex[:8]}",
+                "rule_codes": [
+                    "compare.start.daily",
+                    "sms.send.hourly",
+                ],
+                "expires_at": expires_at,
+            },
+        )
+    assert response.status_code == 400
+    assert "不能在同一白名单中混选" in response.json()["detail"]
+
+
+def test_legacy_ip_device_whitelist_matches_sms_policy(monkeypatch) -> None:
+    suffix = uuid4().hex[:8]
+    client_ip = f"203.0.113.{int(suffix[:2], 16) % 200 + 1}"
+    legacy_device = f"legacy-ip:{client_ip}"
+    phone = f"134{int(suffix, 16) % 100000000:08d}"
+    expires_at = datetime.now(UTC) + timedelta(hours=2)
+    with SessionLocal() as db:
+        row = LimitPolicyOverride(
+            subject_type="device",
+            subject_value=legacy_device,
+            rule_code="sms.send.hourly",
+            mode="unlimited",
+            enabled=True,
+            expires_at=expires_at,
+        )
+        db.add(row)
+        db.commit()
+        override_id = row.id
+
+    original_resolve = auth_api.limit_policy.resolve
+    matched_override_ids: list[int | None] = []
+
+    def spy_resolve(db, rule_code, **subjects):
+        result = original_resolve(db, rule_code, **subjects)
+        if rule_code == "sms.send.hourly":
+            matched_override_ids.append(result.override_id)
+        return result
+
+    monkeypatch.setattr(auth_api.limit_policy, "resolve", spy_resolve)
+    try:
+        with TestClient(app) as client:
+            response = client.post(
+                "/api/v1/auth/sms/send",
+                headers={"x-forwarded-for": client_ip},
+                json={"phone": phone},
+            )
+        assert response.status_code == 200, response.text
+        assert matched_override_ids == [override_id]
+    finally:
+        with SessionLocal() as db:
+            row = db.get(LimitPolicyOverride, override_id)
+            if row is not None:
+                db.delete(row)
+                db.commit()
+
+
+def test_global_guide_limit_stays_in_sync_with_legacy_config(admin_headers) -> None:
+    snapshot = _snapshot_configs(
+        [
+            GUIDE_VIDEO_MAX_PLAYS_KEY,
+            "coupon_guide_video",
+            "comparison_guide_video",
+        ]
+    )
+    try:
+        with TestClient(admin_app) as client:
+            changed = client.patch(
+                "/admin/api/limit-whitelist/rules/guide.video.lifetime",
+                headers=admin_headers,
+                json={"value": 7},
+            )
+            assert changed.status_code == 200, changed.text
+            assert changed.json()["global_limit"] == 7
+
+            legacy_page = client.get(
+                "/admin/api/guide-video",
+                headers=admin_headers,
+            )
+            assert legacy_page.status_code == 200, legacy_page.text
+            assert legacy_page.json()["max_plays"] == 7
+            assert legacy_page.json()["scene"] == "coupon"
+
+            comparison_changed = client.patch(
+                "/admin/api/guide-video",
+                headers=admin_headers,
+                params={"scene": "comparison"},
+                json={"max_plays": 9},
+            )
+            assert comparison_changed.status_code == 200, comparison_changed.text
+            assert comparison_changed.json()["scene"] == "comparison"
+            assert comparison_changed.json()["max_plays"] == 9
+
+            coupon_again = client.get(
+                "/admin/api/guide-video",
+                headers=admin_headers,
+                params={"scene": "coupon"},
+            )
+            assert coupon_again.status_code == 200, coupon_again.text
+            assert coupon_again.json()["max_plays"] == 7
+    finally:
+        _restore_configs(snapshot)
+
+
+def test_legacy_ad_limit_update_syncs_split_global_rules(admin_headers) -> None:
+    snapshot = _snapshot_configs(
+        ["ad_daily_limit", AD_REWARD_VIDEO_DAILY_LIMIT_KEY, AD_FEED_DAILY_LIMIT_KEY]
+    )
+    try:
+        with TestClient(admin_app) as client:
+            changed = client.patch(
+                "/admin/api/config/ad_daily_limit",
+                headers=admin_headers,
+                json={"value": 321},
+            )
+            assert changed.status_code == 200, changed.text
+
+            rules = client.get(
+                "/admin/api/limit-whitelist/rules",
+                headers=admin_headers,
+            )
+            by_code = {item["code"]: item for item in rules.json()}
+            assert by_code["ad.reward_video.daily"]["global_limit"] == 321
+            assert by_code["ad.feed.daily"]["global_limit"] == 321
+    finally:
+        _restore_configs(snapshot)
+
+
+def test_zero_disables_cooldown_style_global_rules(admin_headers) -> None:
+    snapshot = _snapshot_configs(
+        [SMS_PHONE_COOLDOWN_SECONDS_KEY, PHONE_REBIND_DAYS_KEY]
+    )
+    try:
+        with TestClient(admin_app) as client:
+            for rule_code in ("sms.phone.cooldown", "phone.rebind.days"):
+                changed = client.patch(
+                    f"/admin/api/limit-whitelist/rules/{rule_code}",
+                    headers=admin_headers,
+                    json={"value": 0},
+                )
+                assert changed.status_code == 200, changed.text
+                assert changed.json()["global_limit"] == 0
+    finally:
+        _restore_configs(snapshot)
+
+
+def test_compare_start_uses_phone_override_and_reset_baseline() -> None:
+    suffix = uuid4().hex[:8]
+    phone = f"136{int(suffix, 16) % 100000000:08d}"
+    device = f"compare-policy-{suffix}"
+    with SessionLocal() as db:
+        user = user_repo.upsert_user_for_login(
+            db,
+            phone=phone,
+            register_channel="sms",
+        )
+        db.add(
+            LimitPolicyOverride(
+                subject_type="phone",
+                subject_value=phone,
+                rule_code="compare.start.daily",
+                mode="override",
+                limit_value=1,
+                enabled=True,
+            )
+        )
+        db.commit()
+        user_id = user.id
+
+    token, _ = create_token(user_id=user_id, token_type="access")
+    headers = {"Authorization": f"Bearer {token}"}
+    with TestClient(app) as client:
+        first = client.post(
+            "/api/v1/compare/start",
+            headers=headers,
+            json={
+                "trace_id": f"limit-policy-first-{suffix}",
+                "business_type": "food",
+                "device_id": device,
+            },
+        )
+        assert first.status_code == 200, first.text
+        assert first.json() == {"limit": 1, "used": 1, "remaining": 0}
+
+        blocked = client.post(
+            "/api/v1/compare/start",
+            headers=headers,
+            json={
+                "trace_id": f"limit-policy-blocked-{suffix}",
+                "business_type": "food",
+                "device_id": device,
+            },
+        )
+        assert blocked.status_code == 429
+
+        with SessionLocal() as db:
+            override = db.query(LimitPolicyOverride).filter(
+                LimitPolicyOverride.subject_type == "phone",
+                LimitPolicyOverride.subject_value == phone,
+                LimitPolicyOverride.rule_code == "compare.start.daily",
+            ).one()
+            override.reset_at = datetime.now(UTC)
+            db.commit()
+
+        after_reset = client.post(
+            "/api/v1/compare/start",
+            headers=headers,
+            json={
+                "trace_id": f"limit-policy-reset-{suffix}",
+                "business_type": "food",
+                "device_id": device,
+            },
+        )
+        assert after_reset.status_code == 200, after_reset.text
+        assert after_reset.json() == {"limit": 1, "used": 1, "remaining": 0}
+
+
+def test_suppress_alert_keeps_events_but_creates_no_incident() -> None:
+    suffix = uuid4().hex[:10]
+    device = f"suppress-device-{suffix}"
+    phone = f"138{int(suffix[:8], 16) % 100000000:08d}"
+    now = risk_repo.utcnow().replace(microsecond=0)
+    with SessionLocal() as db:
+        db.add(
+            LimitPolicyOverride(
+                subject_type="device",
+                subject_value=device,
+                rule_code="risk.sms.hourly",
+                mode="suppress_alert",
+                enabled=True,
+            )
+        )
+        db.commit()
+        for index in range(10):
+            risk_repo.record_behavior_event(
+                db,
+                event_type=risk_repo.EVENT_SMS_SEND,
+                subject_type="device",
+                subject_id=device,
+                device_id=device,
+                phone=phone,
+                outcome="success",
+                occurred_at=now + timedelta(seconds=index),
+                evaluate_rule=risk_repo.RULE_SMS_HOURLY,
+            )
+
+    with SessionLocal() as db:
+        incident = db.query(RiskIncident).filter(
+            RiskIncident.rule_code == risk_repo.RULE_SMS_HOURLY,
+            RiskIncident.subject_id == device,
+        ).one_or_none()
+        assert incident is None
+
+
+def test_adding_suppress_alert_resolves_existing_incident(admin_headers) -> None:
+    suffix = uuid4().hex[:10]
+    device = f"suppress-existing-{suffix}"
+    now = risk_repo.utcnow().replace(microsecond=0)
+    with SessionLocal() as db:
+        for index in range(5):
+            risk_repo.record_behavior_event(
+                db,
+                event_type=risk_repo.EVENT_SMS_SEND,
+                subject_type="device",
+                subject_id=device,
+                device_id=device,
+                phone=f"13700001{index:03d}",
+                outcome="success",
+                occurred_at=now + timedelta(seconds=index),
+                evaluate_rule=risk_repo.RULE_SMS_HOURLY,
+            )
+        incident = db.query(RiskIncident).filter(
+            RiskIncident.rule_code == risk_repo.RULE_SMS_HOURLY,
+            RiskIncident.subject_id == device,
+        ).one()
+        assert incident.status == "open"
+
+    with TestClient(admin_app) as client:
+        created = client.post(
+            "/admin/api/limit-whitelist",
+            headers=admin_headers,
+            json={
+                "subject_type": "device",
+                "subject_value": device,
+                "rule_code": "risk.sms.hourly",
+                "mode": "suppress_alert",
+                "enabled": True,
+                "expires_at": (datetime.now(UTC) + timedelta(hours=2)).isoformat(),
+                "reason": "QA device",
+            },
+        )
+        assert created.status_code == 201, created.text
+
+    with SessionLocal() as db:
+        incident = db.query(RiskIncident).filter(
+            RiskIncident.rule_code == risk_repo.RULE_SMS_HOURLY,
+            RiskIncident.subject_id == device,
+        ).one()
+        assert incident.status == "resolved"
+        assert incident.action_reason == risk_repo.AUTO_RESOLVED_REASON

tests/test_risk_monitor.py

@@ -533,7 +533,7 @@ def test_admin_can_edit_rules_and_current_window_is_reconciled() -> None:
             "/admin/api/risk-monitor/rules",
             headers=headers,
             json={
-                "sms_hourly_threshold": 21,
+                "sms_hourly_threshold": 100_001,
                 "oneclick_daily_threshold": 20,
                 "compare_daily_threshold": 100,
             },