fix(push): 完善厂商推送排障日志

2026-07-29 19:52:06 +08:00
37 changed files with 503 additions and 3610 deletions
@@ -1,128 +0,0 @@
-"""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")
@@ -31,7 +31,6 @@ 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
@@ -104,7 +103,6 @@ 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)
@@ -40,7 +40,6 @@ 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": "权限管理"},
@@ -59,14 +58,13 @@ 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", "limit-whitelist",
+        "event-logs", "audit-logs",
    ]},
 ]

@@ -1,455 +0,0 @@
-"""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
@@ -12,11 +12,7 @@ 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 (
-    AD_FEED_DAILY_LIMIT_KEY,
-    AD_REWARD_VIDEO_DAILY_LIMIT_KEY,
-    CONFIG_DEFS,
-)
+from app.core.config_schema import CONFIG_DEFS
 from app.core.rewards import SIGNIN_CYCLE_LEN
 from app.models.admin import AdminUser
 from app.repositories import app_config
@@ -92,23 +88,6 @@ 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,
@@ -26,6 +26,7 @@ router = APIRouter(
    dependencies=[Depends(get_current_admin)],
 )

+
 GuideScene = Literal["coupon", "comparison"]


@@ -62,8 +63,7 @@ def update_config(
    )
    write_audit(
        db, admin, action="guide_video.update", target_type="guide_video", target_id=None,
-        detail={"scene": scene, "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, scene)
@@ -87,12 +87,7 @@ async def upload_video(
    )
    write_audit(
        db, admin, action="guide_video.set_video", target_type="guide_video", target_id=None,
-        detail={
-            "scene": scene,
-            "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()
@@ -114,8 +109,7 @@ def delete_video(
    )
    write_audit(
        db, admin, action="guide_video.delete_video", target_type="guide_video", target_id=None,
-        detail={"scene": scene, "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"))
@@ -1,361 +0,0 @@
-"""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()
@@ -1,127 +0,0 @@
-"""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
@@ -21,9 +21,9 @@ class RiskMonitorSummary(BaseModel):


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


 class RiskIncidentItem(BaseModel):
@@ -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 limit_policy, rewards
+from app.core import rewards
 from app.core.config import settings
 from app.core.ratelimit import rate_limit
 from app.integrations import pangle
@@ -413,23 +413,12 @@ 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=(
-            feed_limit
-            if feed_limit is not None
-            else limit_policy.get_rule("ad.feed.daily").max_value
-        ),
+        daily_limit=rewards.get_ad_daily_limit(db),
    )


@@ -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 limit_policy, test_account
+from app.core import test_account
 from app.core.ratelimit import (
    RateLimitRule,
    check_rate_limits,
@@ -69,7 +69,6 @@ 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:
@@ -199,53 +198,16 @@ 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", hourly_limit, 3600,
-                      "操作过于频繁,请稍后再试", hourly_policy.bucket_version),
-        RateLimitRule("sms-send-device-daily", daily_limit, 86400,
-                      "今日验证码发送次数过多,请明天再试", daily_policy.bucket_version),
+        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,
+                      "今日验证码发送次数过多,请明天再试"),
    ]
-    check_rate_limits(request, subject=subject_id, rules=send_rules)
+    check_rate_limits(request, subject=req.device_id, rules=send_rules)

    try:
-        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)
-            ),
-        )
+        cooldown = send_code(req.phone)
    except SmsError as e:
        risk_repo.record_behavior_event(
            db,
@@ -262,7 +224,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=subject_id, rules=send_rules)
+    record_rate_limits(request, subject=req.device_id, rules=send_rules)

    from app.core.config import settings  # 局部 import 避免循环

@@ -321,48 +283,17 @@ 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=subject_id,
-        limit=login_limit,
+        subject=req.device_id,
+        limit=SMS_LOGIN_MAX_PER_HOUR,
        window_sec=3600,
        detail="登录尝试过于频繁,请稍后再试",
-        bucket_suffix=login_policy.bucket_version,
    )

    try:
-        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
-                )
-            ),
-        )
+        ok = verify_code(req.phone, req.code)
    except SmsError as e:  # provider 校验降级(如阿里云接口异常)→ 原样透出其状态码(503),别误报「验证码错误」
        raise HTTPException(status_code=e.status_code, detail=str(e)) from e
    if not ok:
@@ -463,25 +394,15 @@ 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,
        )
-        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,
-        )
+        blocked = rebind_repo.rebound_within_days(db, phone, settings.PHONE_REBIND_LIMIT_DAYS)
        logger.info(
            "wechat bind phone occupied phone=%s by user_id=%d has_wechat=%s",
            mask_phone(phone), existing.id, bool(existing.wechat_openid),
@@ -497,12 +418,7 @@ def _finish_wechat_bind(
            conflict_ticket=ticket,
            rebind_available=not blocked,
            rebind_blocked_days=(
-                rebind_repo.remaining_block_days(
-                    db,
-                    phone,
-                    rebind_days,
-                    reset_at=rebind_policy.reset_at,
-                )
+                rebind_repo.remaining_block_days(db, phone, settings.PHONE_REBIND_LIMIT_DAYS)
                if blocked else 0
            ),
        )
@@ -535,50 +451,18 @@ 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=subject_id,
-        limit=bind_limit,
+        subject=req.device_id,
+        limit=SMS_LOGIN_MAX_PER_HOUR,
        window_sec=3600,
        detail="登录尝试过于频繁,请稍后再试",
-        bucket_suffix=bind_policy.bucket_version,
    )

    try:
-        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
-                )
-            ),
-        )
+        ok = verify_code(req.phone, req.code)
    except SmsError as e:  # provider 校验降级(如阿里云接口异常)→ 原样透出其状态码(503),别误报「验证码错误」
        raise HTTPException(status_code=e.status_code, detail=str(e)) from e
    if not ok:
@@ -638,23 +522,9 @@ 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=subject_id,
-        limit=conflict_limit, window_sec=3600, detail="操作过于频繁,请稍后再试",
-        bucket_suffix=conflict_policy.bucket_version,
+        request, scope="wechat-conflict-device", subject=req.device_id,
+        limit=SMS_LOGIN_MAX_PER_HOUR, window_sec=3600, detail="操作过于频繁,请稍后再试",
    )

    user = user_repo.get_user_by_phone(db, claims["phone"])
@@ -692,50 +562,21 @@ 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=subject_id,
-        limit=conflict_limit, window_sec=3600, detail="操作过于频繁,请稍后再试",
-        bucket_suffix=conflict_policy.bucket_version,
+        request, scope="wechat-conflict-device", subject=req.device_id,
+        limit=SMS_LOGIN_MAX_PER_HOUR, window_sec=3600, detail="操作过于频繁,请稍后再试",
    )

    phone = claims["phone"]
-    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,
-        )
+    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)
        raise HTTPException(status_code=409, detail=f"该手机号 {days} 天内已换绑过,暂不能再次换绑")

    user = user_repo.rebind_account(
@@ -16,7 +16,6 @@ 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 (
@@ -54,29 +53,17 @@ 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=(
-                f"今日已比价超过{policy.limit}次，请明天再试"
-                if policy.limit is not None
-                else "今日比价次数已达上限，请明天再试"
-            ),
+            detail="今日已比价超过100次，请明天再试",
        ) from None
    except crud_compare.ComparisonTraceOwnershipError:
        raise HTTPException(
@@ -84,16 +71,11 @@ def reserve_compare_start(
            detail="比价任务标识冲突，请重新发起",
        ) from None
    # 风控阈值由后台动态配置，不能再只在固定的 100 次业务上限处同步。
-    risk_repo.sync_compare_incident(
-        db,
-        user_id=user.id,
-        at=rec.created_at,
-        device_id=payload.device_id,
-    )
+    risk_repo.sync_compare_incident(db, user_id=user.id, at=rec.created_at)
    return CompareStartReserveOut(
-        limit=policy.limit,
+        limit=crud_compare.DAILY_COMPARE_START_LIMIT,
        used=used,
-        remaining=max(policy.limit - used, 0) if policy.limit is not None else None,
+        remaining=max(crud_compare.DAILY_COMPARE_START_LIMIT - used, 0),
    )


@@ -14,137 +14,12 @@ 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",
@@ -181,8 +56,7 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
    },
    "ad_daily_limit": {
        "default": r.DAILY_AD_REWARD_LIMIT, "label": "看广告每日上限(次)",
-        "group": "看广告", "type": "int", "hidden": True,
-        "help": "历史共享上限；新配置由白名单页分别管理激励视频与 Draw 信息流。",
+        "group": "看广告", "type": "int", "help": "福利页激励视频每日可发奖次数上限,默认 500。",
    },
    "ad_max_coin": {
        "default": r.MAX_AD_REWARD_COIN, "label": "看广告单次金币上限",
@@ -244,9 +118,9 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
        "group": "风控",
        "type": "int",
        "min": 1,
-        "max": 100_000,
+        "max": 5,
        "hidden": True,
-        "help": "同一设备在北京时间同一自然小时内成功下发短信达到该次数时告警。",
+        "help": "同一设备在北京时间同一自然小时内成功下发短信达到该次数时告警；不得高于现有每小时 5 次的发送上限。",
    },
    RISK_ONECLICK_DAILY_THRESHOLD_KEY: {
        "default": 20,
@@ -264,7 +138,7 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
        "group": "风控",
        "type": "int",
        "min": 1,
-        "max": 100_000,
+        "max": 100,
        "hidden": True,
        "help": "同一账户在北京时间同一自然日内发起比价达到该次数时告警。",
    },
@@ -1,525 +0,0 @@
-"""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()
@@ -75,11 +75,10 @@ def enforce_rate_limit(
    request: Request,
    scope: str,
    subject: str,
-    limit: int | None,
+    limit: int,
    window_sec: float,
    *,
    detail: str = "操作过于频繁,请稍后再试",
-    bucket_suffix: str = "",
 ) -> None:
    """在路由内部手动限流,按 (subject, 客户端 IP) 计数。

@@ -88,9 +87,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 or limit is None:
+    if not settings.RATE_LIMIT_ENABLED:
        return
-    key = f"{scope}:{subject}:{_client_ip(request)}:{bucket_suffix}"
+    key = f"{scope}:{subject}:{_client_ip(request)}"
    if not _hit(key, limit, window_sec):
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
@@ -113,10 +112,9 @@ class RateLimitRule(NamedTuple):
    """

    scope: str
-    limit: int | None
+    limit: int
    window_sec: float
    detail: str = "操作过于频繁,请稍后再试"
-    bucket_suffix: str = ""


 def _peek(key: str, limit: int, window_sec: float) -> bool:
@@ -152,13 +150,7 @@ def check_rate_limits(request: Request, subject: str, rules: list[RateLimitRule]
        return
    ip = _client_ip(request)
    for rule in rules:
-        if rule.limit is None:
-            continue
-        if not _peek(
-            f"{rule.scope}:{subject}:{ip}:{rule.bucket_suffix}",
-            rule.limit,
-            rule.window_sec,
-        ):
+        if not _peek(f"{rule.scope}:{subject}:{ip}", rule.limit, rule.window_sec):
            raise HTTPException(
                status_code=status.HTTP_429_TOO_MANY_REQUESTS,
                detail=rule.detail,
@@ -175,9 +167,4 @@ def record_rate_limits(request: Request, subject: str, rules: list[RateLimitRule
        return
    ip = _client_ip(request)
    for rule in rules:
-        if rule.limit is None:
-            continue
-        _commit(
-            f"{rule.scope}:{subject}:{ip}:{rule.bucket_suffix}",
-            rule.window_sec,
-        )
+        _commit(f"{rule.scope}:{subject}:{ip}", rule.window_sec)
@@ -27,22 +27,11 @@ def _provider():
    return _PROVIDERS.get(settings.SMS_PROVIDER, jiguang)


-def send_code(phone: str, *, cooldown_sec: int | None = None) -> int:
+def send_code(phone: str) -> int:
    """发送验证码,返回距下次可发的冷却秒数;失败抛 SmsError。委托给当前 provider。"""
-    provider = _provider()
-    if cooldown_sec is None:
-        return provider.send_code(phone)
-    return provider.send_code(phone, cooldown_sec=cooldown_sec)
+    return _provider().send_code(phone)


-def verify_code(
-    phone: str,
-    code: str,
-    *,
-    max_failed_attempts: int | None = None,
-) -> bool:
+def verify_code(phone: str, code: str) -> bool:
    """校验验证码,返回是否通过;provider 异常降级抛 SmsError。委托给当前 provider。"""
-    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)
+    return _provider().verify_code(phone, code)
@@ -47,26 +47,19 @@ _client = None  # 惰性构建的 SDK client(模块级缓存)

 # ============================ 对外:发码 / 校验 ============================

-def send_code(phone: str, *, cooldown_sec: int | None = None) -> int:
+def send_code(phone: str) -> 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 effective_cooldown
+        return settings.ALIYUN_SMS_INTERVAL_SEC
    if not settings.aliyun_sms_configured:
        raise SmsError("短信服务未配置(缺阿里云凭证)", status_code=503)

-    result = (
-        _call_send(phone)
-        if cooldown_sec is None
-        else _call_send(phone, cooldown_sec=effective_cooldown)
-    )
+    result = _call_send(phone)  # 传输/SDK 异常在内部抛 SmsError(503)

    if result["success"] and result["code"] == "OK":
        now = time.time()
@@ -75,7 +68,7 @@ def send_code(phone: str, *, cooldown_sec: int | None = None) -> int:
            _verify_attempts.pop(phone, None)  # 新码 = 新失败预算
            _verify_seen.pop(phone, None)
        logger.info("[SMS-aliyun] sent to %s****", phone[:3])
-        return effective_cooldown
+        return settings.ALIYUN_SMS_INTERVAL_SEC

    code = result["code"]
    logger.error("[SMS-aliyun] send failed code=%s msg=%s", code, result["message"])
@@ -85,12 +78,7 @@ def send_code(phone: str, *, cooldown_sec: int | None = None) -> int:
    raise SmsError(msg, status_code=status)


-def verify_code(
-    phone: str,
-    code: str,
-    *,
-    max_failed_attempts: int | None = None,
-) -> bool:
+def verify_code(phone: str, code: str) -> bool:
    """校验验证码(阿里云裁决)。

    - **mock**:放行任意 N 位数字(provider 无关,同极光)。
@@ -104,13 +92,8 @@ def verify_code(

    # 失败计数是 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) >= effective_max_attempts:
+        if _verify_attempts.get(phone, 0) >= settings.SMS_MAX_VERIFY_ATTEMPTS:
            return False  # 已作废:保持计数(直到 send_code 重置),与极光「达上限即作废」一致

    result = _call_check(phone, code)  # 传输/SDK 异常在内部抛 SmsError(503)
@@ -163,7 +146,7 @@ def _get_client():
    return _client


-def _call_send(phone: str, *, cooldown_sec: int | None = None) -> dict:
+def _call_send(phone: str) -> 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)
@@ -177,11 +160,7 @@ def _call_send(phone: str, *, cooldown_sec: int | None = None) -> 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
-                if cooldown_sec is None
-                else cooldown_sec
-            ),
+            interval=settings.ALIYUN_SMS_INTERVAL_SEC,
            scheme_name=settings.ALIYUN_SMS_SCHEME_NAME or None,
        )
        body = _get_client().send_sms_verify_code(req).body
@@ -84,23 +84,20 @@ def _gc(now: float) -> None:
            _last_sent.pop(p, None)


-def send_code(phone: str, *, cooldown_sec: int | None = None) -> int:
+def send_code(phone: str) -> 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 < effective_cooldown:
-            remain = int(effective_cooldown - elapsed)
+        if elapsed < settings.SMS_SEND_INTERVAL_SEC:
+            remain = int(settings.SMS_SEND_INTERVAL_SEC - elapsed)
            raise SmsError(f"发送过于频繁,请 {remain}s 后再试")

        code = _gen_code()
@@ -125,15 +122,10 @@ def send_code(phone: str, *, cooldown_sec: int | None = None) -> int:
        logger.exception("[SMS-chuanglan] send failed phone=%s****", phone[:3])
        raise SmsError("验证码发送失败,请稍后重试", status_code=503) from e

-    return effective_cooldown
+    return settings.SMS_SEND_INTERVAL_SEC


-def verify_code(
-    phone: str,
-    code: str,
-    *,
-    max_failed_attempts: int | None = None,
-) -> bool:
+def verify_code(phone: str, code: str) -> bool:
    """校验验证码。

    - **mock 模式**:放行任意 N 位数字(测试/开发便利,不真校验)。
@@ -144,11 +136,6 @@ def verify_code(
        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:
@@ -156,7 +143,7 @@ def verify_code(
        if time.time() > rec.expires_at:
            _codes.pop(phone, None)
            return False
-        if rec.attempts >= effective_max_attempts:
+        if rec.attempts >= settings.SMS_MAX_VERIFY_ATTEMPTS:
            _codes.pop(phone, None)  # 试错过多,作废
            return False
        if secrets.compare_digest(code.encode("utf-8"), rec.code.encode("utf-8")):
@@ -70,23 +70,20 @@ def _gc(now: float) -> None:
            _last_sent.pop(p, None)


-def send_code(phone: str, *, cooldown_sec: int | None = None) -> int:
+def send_code(phone: str) -> 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 < effective_cooldown:
-            remain = int(effective_cooldown - elapsed)
+        if elapsed < settings.SMS_SEND_INTERVAL_SEC:
+            remain = int(settings.SMS_SEND_INTERVAL_SEC - elapsed)
            raise SmsError(f"发送过于频繁,请 {remain}s 后再试")

        code = _gen_code()
@@ -111,15 +108,10 @@ def send_code(phone: str, *, cooldown_sec: int | None = None) -> int:
        logger.exception("[SMS] send failed phone=%s****", phone[:3])
        raise SmsError("验证码发送失败,请稍后重试", status_code=503) from e

-    return effective_cooldown
+    return settings.SMS_SEND_INTERVAL_SEC


-def verify_code(
-    phone: str,
-    code: str,
-    *,
-    max_failed_attempts: int | None = None,
-) -> bool:
+def verify_code(phone: str, code: str) -> bool:
    """校验验证码。

    - **mock 模式**:放行任意 N 位数字(测试/开发便利,不真校验)。
@@ -130,11 +122,6 @@ def verify_code(
        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:
@@ -142,7 +129,7 @@ def verify_code(
        if time.time() > rec.expires_at:
            _codes.pop(phone, None)
            return False
-        if rec.attempts >= effective_max_attempts:
+        if rec.attempts >= settings.SMS_MAX_VERIFY_ATTEMPTS:
            _codes.pop(phone, None)  # 试错过多,作废
            return False
        if secrets.compare_digest(code.encode("utf-8"), rec.code.encode("utf-8")):
@@ -22,7 +22,7 @@ import uuid
 from collections.abc import Callable
 from dataclasses import dataclass
 from typing import Any
-from urllib.parse import quote
+from urllib.parse import quote, urlsplit

 import httpx

@@ -69,6 +69,18 @@ class _CachedToken:

 _token_cache: dict[str, _CachedToken] = {}

+_LOG_SUMMARY_MAX_CHARS = 1500
+_LOG_STRING_MAX_CHARS = 200
+_SENSITIVE_LOG_KEYS = {
+    "accesstoken",
+    "appkey",
+    "authtoken",
+    "authorization",
+    "clientsecret",
+    "mastersecret",
+    "sign",
+}
+

 def normalize_vendor(push_vendor: str | None) -> str | None:
    if not push_vendor:
@@ -115,8 +127,16 @@ def send_notification(

    if mock:
        logger.info(
-            "[mock push] vendor=%s token=%s... title=%s body=%s extras=%s",
-            vendor, token[:12], title, body, extras,
+            "vendor push mock vendor=%s request=%s",
+            vendor,
+            _log_summary(
+                {
+                    "push_token": token,
+                    "title": title,
+                    "body": body,
+                    "extras": extras,
+                }
+            ),
        )
        return {
            "mock": True,
@@ -133,7 +153,28 @@ def send_notification(
        "xiaomi": _send_xiaomi,
        "oppo": _send_oppo,
    }
-    return dispatch[vendor](token, title, body, extras)
+    started = time.perf_counter()
+    try:
+        result = dispatch[vendor](token, title, body, extras)
+    except Exception:
+        logger.exception(
+            "vendor push dispatch failed vendor=%s token=%s notification_type=%s "
+            "elapsed_ms=%.1f",
+            vendor,
+            token,
+            extras.get("type", ""),
+            _elapsed_ms(started),
+        )
+        raise
+    logger.info(
+        "vendor push dispatch succeeded vendor=%s token=%s notification_type=%s "
+        "elapsed_ms=%.1f",
+        vendor,
+        token,
+        extras.get("type", ""),
+        _elapsed_ms(started),
+    )
+    return result


 def send_accessibility_disabled(
@@ -159,13 +200,120 @@ def _require(value: str, name: str) -> str:
    return value


+def _elapsed_ms(started: float) -> float:
+    return (time.perf_counter() - started) * 1000
+
+
+def _redacted_value(value: Any) -> str:
+    raw = json.dumps(value, ensure_ascii=False, sort_keys=True, default=str)
+    digest = hashlib.sha256(raw.encode()).hexdigest()[:10]
+    return f"<redacted len={len(raw)} sha256={digest}>"
+
+
+def _sanitize_for_log(value: Any, *, key: str = "", depth: int = 0) -> Any:
+    """生成有排障价值、但不暴露服务端鉴权凭据的紧凑日志摘要。"""
+    normalized_key = "".join(char for char in key.lower() if char.isalnum())
+    if normalized_key in _SENSITIVE_LOG_KEYS or normalized_key.endswith("secret"):
+        return _redacted_value(value)
+    if depth >= 10:
+        return f"<{type(value).__name__}>"
+    if isinstance(value, dict):
+        items = list(value.items())
+        sanitized = {
+            str(item_key): _sanitize_for_log(item_value, key=str(item_key), depth=depth + 1)
+            for item_key, item_value in items[:30]
+        }
+        if len(items) > 30:
+            sanitized["<omitted_keys>"] = len(items) - 30
+        return sanitized
+    if isinstance(value, (list, tuple)):
+        items = list(value)
+        sanitized = [_sanitize_for_log(item, key=key, depth=depth + 1) for item in items[:20]]
+        if len(items) > 20:
+            sanitized.append(f"<omitted_items={len(items) - 20}>")
+        return sanitized
+    if isinstance(value, str):
+        if value.lstrip().startswith(("{", "[")):
+            try:
+                decoded = json.loads(value)
+            except ValueError:
+                pass
+            else:
+                return _sanitize_for_log(decoded, key=key, depth=depth + 1)
+        if len(value) > _LOG_STRING_MAX_CHARS:
+            return f"{value[:_LOG_STRING_MAX_CHARS]}…<len={len(value)}>"
+    return value
+
+
+def _log_summary(value: Any) -> str:
+    rendered = json.dumps(
+        _sanitize_for_log(value),
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+        default=str,
+    )
+    if len(rendered) > _LOG_SUMMARY_MAX_CHARS:
+        return f"{rendered[:_LOG_SUMMARY_MAX_CHARS]}…<len={len(rendered)}>"
+    return rendered
+
+
+def _raw_log_summary(value: Any) -> str:
+    """厂商响应摘要：保留原始字段和值，仅限制单条日志长度。"""
+    rendered = json.dumps(
+        value,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+        default=str,
+    )
+    if len(rendered) > _LOG_SUMMARY_MAX_CHARS:
+        return f"{rendered[:_LOG_SUMMARY_MAX_CHARS]}…<len={len(rendered)}>"
+    return rendered
+
+
+def _endpoint_for_log(url: str) -> str:
+    parsed = urlsplit(url)
+    return f"{parsed.scheme}://{parsed.netloc}{parsed.path}"
+
+
+def _request_summary(kwargs: dict[str, Any]) -> str:
+    summary: dict[str, Any] = {}
+    if "json" in kwargs:
+        summary["body_type"] = "json"
+        summary["request_payload"] = kwargs["json"]
+    elif "data" in kwargs:
+        summary["body_type"] = "form"
+        summary["request_payload"] = kwargs["data"]
+    if kwargs.get("params"):
+        summary["params"] = kwargs["params"]
+    if kwargs.get("headers"):
+        summary["headers"] = kwargs["headers"]
+    return _log_summary(summary)
+
+
+def _response_summary(resp: Any, parsed: Any | None = None) -> str:
+    if parsed is None:
+        try:
+            parsed = resp.json()
+        except ValueError:
+            pass
+    if parsed is not None:
+        return _raw_log_summary(parsed)
+    return _raw_log_summary(getattr(resp, "text", ""))
+
+
 def _request_json(
    method: str,
    url: str,
    *,
+    vendor: str,
+    operation: str,
    expected_status: tuple[int, ...] = (200,),
    **kwargs: Any,
 ) -> dict[str, Any]:
+    started = time.perf_counter()
+    request_summary = _request_summary(kwargs)
    try:
        resp = httpx.request(
            method,
@@ -174,41 +322,82 @@ def _request_json(
            **kwargs,
        )
    except httpx.HTTPError as e:
+        logger.error(
+            "vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
+            "outcome=network_error request=%s response=%s elapsed_ms=%.1f error=%s",
+            vendor,
+            operation,
+            method,
+            _endpoint_for_log(url),
+            request_summary,
+            "<no_response>",
+            _elapsed_ms(started),
+            type(e).__name__,
+        )
        raise VendorPushError(f"push http error: {e}") from e

    if resp.status_code not in expected_status:
-        logger.error("vendor push http failed url=%s http=%s body=%s", url, resp.status_code, resp.text[:500])
+        logger.error(
+            "vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
+            "outcome=http_error http_status=%s request=%s response=%s elapsed_ms=%.1f",
+            vendor,
+            operation,
+            method,
+            _endpoint_for_log(url),
+            resp.status_code,
+            request_summary,
+            _response_summary(resp),
+            _elapsed_ms(started),
+        )
        raise VendorPushError(f"push http {resp.status_code}")
    try:
-        return resp.json()
+        data = resp.json()
    except ValueError as e:
+        logger.error(
+            "vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
+            "outcome=invalid_json http_status=%s request=%s response=%s elapsed_ms=%.1f",
+            vendor,
+            operation,
+            method,
+            _endpoint_for_log(url),
+            resp.status_code,
+            request_summary,
+            _response_summary(resp),
+            _elapsed_ms(started),
+        )
        raise VendorPushError(f"push invalid json: {resp.text[:200]}") from e
+    logger.info(
+        "vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
+        "outcome=success http_status=%s request=%s response=%s elapsed_ms=%.1f",
+        vendor,
+        operation,
+        method,
+        _endpoint_for_log(url),
+        resp.status_code,
+        request_summary,
+        _response_summary(resp, data),
+        _elapsed_ms(started),
+    )
+    return data


 def _request_form(
    method: str,
    url: str,
    *,
+    vendor: str,
+    operation: str,
    expected_status: tuple[int, ...] = (200,),
    **kwargs: Any,
 ) -> dict[str, Any]:
-    try:
-        resp = httpx.request(
-            method,
-            url,
-            timeout=settings.PUSH_REQUEST_TIMEOUT_SEC,
-            **kwargs,
-        )
-    except httpx.HTTPError as e:
-        raise VendorPushError(f"push http error: {e}") from e
-
-    if resp.status_code not in expected_status:
-        logger.error("vendor push http failed url=%s http=%s body=%s", url, resp.status_code, resp.text[:500])
-        raise VendorPushError(f"push http {resp.status_code}")
-    try:
-        return resp.json()
-    except ValueError as e:
-        raise VendorPushError(f"push invalid json: {resp.text[:200]}") from e
+    return _request_json(
+        method,
+        url,
+        vendor=vendor,
+        operation=operation,
+        expected_status=expected_status,
+        **kwargs,
+    )


 def _cache_get(key: str) -> str | None:
@@ -234,6 +423,8 @@ def _honor_access_token() -> str:
    data = _request_form(
        "POST",
        settings.HONOR_PUSH_TOKEN_ENDPOINT,
+        vendor="honor",
+        operation="authenticate",
        data={
            "grant_type": "client_credentials",
            "client_id": client_id,
@@ -243,7 +434,7 @@ def _honor_access_token() -> str:
    )
    token = data.get("access_token")
    if not token:
-        raise VendorPushError(f"honor auth failed: {data}")
+        raise VendorPushError(f"honor auth failed: {_raw_log_summary(data)}")
    return _cache_put(cache_key, str(token), data.get("expires_in"))


@@ -270,6 +461,8 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
    data = _request_json(
        "POST",
        settings.HONOR_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
+        vendor="honor",
+        operation="send_notification",
        json=payload,
        headers={
            "Content-Type": "application/json; charset=UTF-8",
@@ -279,7 +472,7 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
    )
    code = data.get("code")
    if code is not None and int(code) != 200:
-        raise VendorPushError(f"honor push failed: {data}")
+        raise VendorPushError(f"honor push failed: {_raw_log_summary(data)}")
    return data


@@ -294,6 +487,8 @@ def _huawei_access_token() -> str:
    data = _request_form(
        "POST",
        settings.HUAWEI_PUSH_TOKEN_ENDPOINT,
+        vendor="huawei",
+        operation="authenticate",
        data={
            "grant_type": "client_credentials",
            "client_id": app_id,
@@ -303,7 +498,7 @@ def _huawei_access_token() -> str:
    )
    token = data.get("access_token")
    if not token:
-        raise VendorPushError(f"huawei auth failed: {data}")
+        raise VendorPushError(f"huawei auth failed: {_raw_log_summary(data)}")
    return _cache_put(cache_key, str(token), data.get("expires_in"))


@@ -333,6 +528,8 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d
    data = _request_json(
        "POST",
        settings.HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
+        vendor="huawei",
+        operation="send_notification",
        json=payload,
        headers={
            "Content-Type": "application/json; charset=UTF-8",
@@ -340,7 +537,7 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d
        },
    )
    if str(data.get("code", "")) != "80000000":
-        raise VendorPushError(f"huawei push failed: {data}")
+        raise VendorPushError(f"huawei push failed: {_raw_log_summary(data)}")
    return data


@@ -357,6 +554,8 @@ def _vivo_auth_token() -> str:
    data = _request_json(
        "POST",
        settings.VIVO_PUSH_AUTH_ENDPOINT,
+        vendor="vivo",
+        operation="authenticate",
        json={
            "appId": app_id,
            "appKey": app_key,
@@ -366,10 +565,10 @@ def _vivo_auth_token() -> str:
        headers={"Content-Type": "application/json"},
    )
    if int(data.get("result", -1)) != 0:
-        raise VendorPushError(f"vivo auth failed: {data}")
+        raise VendorPushError(f"vivo auth failed: {_raw_log_summary(data)}")
    token = data.get("authToken")
    if not token:
-        raise VendorPushError(f"vivo auth missing authToken: {data}")
+        raise VendorPushError(f"vivo auth missing authToken: {_raw_log_summary(data)}")
    return _cache_put(cache_key, str(token), 24 * 3600)


@@ -401,6 +600,8 @@ def _send_vivo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
    data = _request_json(
        "POST",
        settings.VIVO_PUSH_SEND_ENDPOINT,
+        vendor="vivo",
+        operation="send_notification",
        json=payload,
        headers={
            "Content-Type": "application/json",
@@ -408,7 +609,7 @@ def _send_vivo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
        },
    )
    if int(data.get("result", -1)) != 0:
-        raise VendorPushError(f"vivo push failed: {data}")
+        raise VendorPushError(f"vivo push failed: {_raw_log_summary(data)}")
    return data


@@ -446,14 +647,16 @@ def _send_xiaomi(token: str, title: str, body: str, extras: dict[str, str]) -> d
    data = _request_form(
        "POST",
        settings.XIAOMI_PUSH_SEND_ENDPOINT,
+        vendor="xiaomi",
+        operation="send_notification",
        data=form,
        headers={"Authorization": f"key={app_secret}"},
    )
    code = data.get("code")
    if code not in (0, "0", None):
-        raise VendorPushError(f"xiaomi push failed: {data}")
+        raise VendorPushError(f"xiaomi push failed: {_raw_log_summary(data)}")
    if str(data.get("result", "ok")).lower() not in ("ok", "success"):
-        raise VendorPushError(f"xiaomi push failed: {data}")
+        raise VendorPushError(f"xiaomi push failed: {_raw_log_summary(data)}")
    return data


@@ -522,6 +725,8 @@ def _oppo_auth_token() -> str:
    data = _request_form(
        "POST",
        settings.OPPO_PUSH_AUTH_ENDPOINT,
+        vendor="oppo",
+        operation="authenticate",
        data={
            "app_key": app_key,
            "timestamp": timestamp,
@@ -530,10 +735,10 @@ def _oppo_auth_token() -> str:
        headers={"Content-Type": "application/x-www-form-urlencoded"},
    )
    if int(data.get("code", -1)) != 0:
-        raise VendorPushError(f"oppo auth failed: {data}")
+        raise VendorPushError(f"oppo auth failed: {_raw_log_summary(data)}")
    token = (data.get("data") or {}).get("auth_token") or data.get("auth_token")
    if not token:
-        raise VendorPushError(f"oppo auth missing auth_token: {data}")
+        raise VendorPushError(f"oppo auth missing auth_token: {_raw_log_summary(data)}")
    return _cache_put(cache_key, str(token), 24 * 3600)


@@ -573,6 +778,8 @@ def _send_oppo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
    data = _request_form(
        "POST",
        settings.OPPO_PUSH_SEND_ENDPOINT,
+        vendor="oppo",
+        operation="send_notification",
        data={
            "auth_token": auth_token,
            "message": json.dumps(message, ensure_ascii=False),
@@ -580,5 +787,5 @@ def _send_oppo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
        headers={"Content-Type": "application/x-www-form-urlencoded"},
    )
    if int(data.get("code", -1)) != 0:
-        raise VendorPushError(f"oppo push failed: {data}")
+        raise VendorPushError(f"oppo push failed: {_raw_log_summary(data)}")
    return data
@@ -36,7 +36,6 @@ 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
@@ -1,7 +1,7 @@
 """新手引导视频播放记录(领券浮层前 N 次用它替代广告)。

 产品规则(2026-07 拍板):新用户点「一键自动领取」后的等候浮层,**前 3 次**不放广告,
-改放运营后台上传的引导视频;默认每次固定 100 金币,中途关闭也算看完照发。
+改放运营后台上传的引导视频;每次固定 120 金币,中途关闭也算看完照发。

 口径:
 - **计次按账号**(user_id),与设备无关 —— 换设备不重新送 3 次。
@@ -38,13 +38,7 @@ class GuideVideoPlay(Base):
        # start_play 捕获 IntegrityError 降级成"这次不放视频"。
        # 用 unique Index 而非 UniqueConstraint:与迁移里的 create_index 对齐(SQLite 加约束
        # 要整表重建),autogenerate 才不会每次报一条假 diff。
-        Index(
-            "uq_guide_video_play_user_scene_seq",
-            "user_id",
-            "scene",
-            "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)
@@ -1,77 +0,0 @@
-"""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,
-    )
@@ -5,13 +5,11 @@
 """
 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 limit_policy, rewards
+from app.core import rewards
 from app.core.rewards import cn_today
 from app.models.ad_feed_reward import AdFeedRewardRecord
 from app.repositories import wallet as crud_wallet
@@ -31,14 +29,8 @@ 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,
-    *,
-    reset_at: datetime | None = None,
-) -> int:
-    stmt = (
+def _granted_today(db: Session, user_id: int, reward_date: str) -> int:
+    return db.execute(
        select(func.count())
        .select_from(AdFeedRewardRecord)
        .where(
@@ -46,10 +38,7 @@ def _granted_today(
            AdFeedRewardRecord.reward_date == reward_date,
            AdFeedRewardRecord.status == "granted",
        )
-    )
-    if reset_at is not None:
-        stmt = stmt.where(AdFeedRewardRecord.created_at >= reset_at)
-    return db.execute(stmt).scalar_one()
+    ).scalar_one()


 def granted_unit_total(db: Session, user_id: int) -> int:
@@ -132,27 +121,7 @@ def grant_feed_reward(
        )
        return _commit_record(db, rec, client_event_id)

-    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
-    ):
+    if _granted_today(db, user_id, today) >= rewards.get_ad_daily_limit(db):
        rec = AdFeedRewardRecord(
            client_event_id=client_event_id,
            user_id=user_id,
@@ -10,13 +10,13 @@
 """
 from __future__ import annotations

-from datetime import UTC, datetime
+from datetime import datetime, timezone

 from sqlalchemy import func, select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session

-from app.core import limit_policy, rewards
+from app.core import 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,14 +96,8 @@ 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,
-    *,
-    reset_at: datetime | None = None,
-) -> int:
-    stmt = (
+def _granted_today(db: Session, user_id: int, reward_date: str) -> int:
+    return db.execute(
        select(func.count())
        .select_from(AdRewardRecord)
        .where(
@@ -112,10 +106,7 @@ def _granted_today(
            AdRewardRecord.status == "granted",
            AdRewardRecord.reward_scene == "reward_video",
        )
-    )
-    if reset_at is not None:
-        stmt = stmt.where(AdRewardRecord.created_at >= reset_at)
-    return db.execute(stmt).scalar_one()
+    ).scalar_one()


 def _granted_cumulative(db: Session, user_id: int) -> int:
@@ -174,27 +165,7 @@ def grant_ad_reward(
        DAILY_AD_WATCH_SECONDS_LIMIT > 0
        and watched_seconds_today(db, user_id, today=today) >= DAILY_AD_WATCH_SECONDS_LIMIT
    )
-    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
-    )
+    over_count = _granted_today(db, user_id, today) >= rewards.get_ad_daily_limit(db)
    if over_time or over_count:
        rec = AdRewardRecord(
            trans_id=trans_id, user_id=user_id, coin=0, status="capped",
@@ -350,24 +321,19 @@ 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,
-    *,
-    reset_at: datetime | None = None,
-) -> list[datetime]:
+def _granted_times_today_desc(db: Session, user_id: int, reward_date: str) -> list[datetime]:
    """当日 status=granted 记录的 created_at,按时间倒序(最新在前)——冷却策略的输入数据。"""
-    stmt = select(AdRewardRecord.created_at).where(
+    return list(
+        db.execute(
+            select(AdRewardRecord.created_at)
+            .where(
                AdRewardRecord.user_id == user_id,
                AdRewardRecord.reward_date == reward_date,
                AdRewardRecord.status == "granted",
                AdRewardRecord.reward_scene == "reward_video",
            )
-    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()
+            .order_by(AdRewardRecord.created_at.desc())
+        ).scalars()
    )


@@ -383,47 +349,16 @@ def today_status(
    旧客户端兼容,当前 DAILY_AD_WATCH_SECONDS_LIMIT=0 表示不启用时长闸。
    """
    today = cn_today().isoformat()
-    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,
-    )
+    granted_desc = _granted_times_today_desc(db, user_id, today)
    state = compute_cooldown(
        granted_desc,
-        datetime.now(UTC),
+        datetime.now(timezone.utc),
        round_size=rewards.get_ad_round_count(db),
-        cooldown_seconds=cooldown_seconds,
+        cooldown_seconds=rewards.get_ad_cooldown_sec(db),
    )
    return (
        len(granted_desc),
-        (
-            daily_limit
-            if daily_limit is not None
-            else limit_policy.get_rule("ad.reward_video.daily").max_value
-        ),
+        rewards.get_ad_daily_limit(db),
        0,
        state.round_count,
        state.cooldown_until,
@@ -434,8 +434,6 @@ 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.

@@ -461,11 +459,6 @@ 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(
@@ -480,11 +473,6 @@ 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(
@@ -493,7 +481,7 @@ def reserve_daily_start(
            ComparisonRecord.created_at < day_end,
        )
    ) or 0
-    if limit is not None and used >= limit:
+    if used >= DAILY_COMPARE_START_LIMIT:
        raise DailyCompareStartLimitExceeded

    rec = ComparisonRecord(
@@ -1,8 +1,8 @@
 """新手引导视频:运营配置读写 + 播放计次 + 发币。

-**配置**(开关 / 视频地址 / 前几次 / 每次金币)按场景分别作为一个 JSON 存进通用
-app_config 表。领券使用 coupon_guide_video，比价使用 comparison_guide_video；
-写法完全对齐 feedback_qr，不进入系统配置页的通用列表。
+**配置**(开关 / 视频地址 / 前几次 / 每次金币)整体作为一个 JSON 存进通用 app_config 表
+(key=coupon_guide_video),写法完全对齐 feedback_qr —— 不进 CONFIG_DEFS,所以不会污染
+系统配置页的通用列表,由本模块独占维护。

 **计次**按账号(user_id)、**开播即计数**:客户端每次要展示领券等候浮层时调
 `/api/v1/guide-video/start`,命中则当场写一行 guide_video_play(status='playing')。
@@ -25,7 +25,7 @@ from sqlalchemy import func, select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session

-from app.core import limit_policy, rewards
+from app.core import rewards
 from app.models.app_config import AppConfig
 from app.models.guide_video import GuideVideoPlay
 from app.repositories import wallet as crud_wallet
@@ -79,22 +79,12 @@ def get_config(db: Session, scene: str = "coupon") -> dict[str, Any]:
    """完整配置 + updated_at(admin 读 / 业务读共用)。"""
    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],
-    *,
-    scene: str,
-    admin_id: int,
-    commit: bool,
+    db: Session, value: dict[str, Any], *, scene: str, admin_id: int, commit: bool
 ) -> dict[str, Any]:
    """整体覆写该行(value 须为完整字段 dict),返回合并后的完整配置(含 updated_at)。"""
    key = _config_key(scene)
@@ -135,75 +125,36 @@ 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,
-        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)
+    after = _write(db, new_value, scene=scene, admin_id=admin_id, commit=commit)
    return before, after


 def set_video(
-    db: Session,
-    video_url: str | None,
-    *,
-    scene: str = "coupon",
-    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, _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,
-        scene=scene,
-        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,
-    *,
-    scene: str = "coupon",
-    reset_at: datetime | None = None,
-) -> int:
+def used_plays(db: Session, user_id: int, scene: str = "coupon") -> int:
    """该账号已用掉的引导视频次数(开播即算,含未发币的)。"""
-    stmt = select(func.count()).select_from(GuideVideoPlay).where(
-        GuideVideoPlay.user_id == user_id,
-        GuideVideoPlay.scene == scene,
+    return int(
+        db.execute(
+            select(func.count()).select_from(GuideVideoPlay).where(
+                GuideVideoPlay.user_id == user_id,
+                GuideVideoPlay.scene == scene,
+            )
+        ).scalar_one()
    )
-    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, scene: str = "coupon") -> dict[str, int]:
@@ -241,19 +192,9 @@ def start_play(
    _config_key(scene)
    cfg = get_config(db, scene)
    video_url = (cfg.get("video_url") or "").strip()
-    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
+    max_plays = int(cfg.get("max_plays") or 0)
    reward_coin = int(cfg.get("reward_coin") or 0)
-    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)
-    )
+    used = used_plays(db, user_id, scene)

    def _miss(used_now: int) -> dict[str, Any]:
        return {
@@ -262,28 +203,19 @@ def start_play(
            "play_token": "",
            "reward_coin": reward_coin,
            "seq": used_now,
-            "remaining": max(0, exposed_max_plays - used_now),
+            "remaining": max(0, max_plays - used_now),
        }

-    if (
-        not cfg.get("enabled")
-        or not video_url
-        or (max_plays is not None and (max_plays <= 0 or used >= max_plays))
-    ):
+    if not cfg.get("enabled") or not video_url or max_plays <= 0 or used >= max_plays:
        return _miss(used)

-    # ``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
+    seq = used + 1
    play = GuideVideoPlay(
        user_id=user_id,
        play_token=uuid.uuid4().hex,
        scene=scene,
        seq=seq,
        video_url=video_url,
-        # 固化本次承诺发放的金币，避免运营改价后已开播记录按新价结算。
        coin=reward_coin,
        status="playing",
        completed=0,
@@ -300,19 +232,14 @@ def start_play(
            db.flush()
    except IntegrityError:
        db.rollback()
-        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 _miss(used_plays(db, user_id, scene))
    return {
        "should_play": True,
        "video_url": video_url,
        "play_token": play.play_token,
        "reward_coin": reward_coin,
        "seq": seq,
-        "remaining": max(0, exposed_max_plays - (used + 1)),
+        "remaining": max(0, max_plays - seq),
    }


@@ -335,7 +262,7 @@ def grant_play(
    重复上报返回 granted=False + 已发金币(客户端据此不重复累加 toast 金额)。
    """
    token = (play_token or "").strip()
-    # 金币额度取开播时由服务端固化的值，不信客户端，也不受后续配置变更影响。
+    # 金币额度以**服务端配置**为准,不信客户端(客户端只上报"播完/关闭")。
    play = _find_play(db, user_id, token)
    coin = int(play.coin if play is not None else 0)

@@ -2,7 +2,7 @@
 from __future__ import annotations

 import math
-from datetime import UTC, datetime, timedelta
+from datetime import datetime, timedelta, timezone

 from sqlalchemy import func, select
 from sqlalchemy.orm import Session
@@ -10,22 +10,9 @@ from sqlalchemy.orm import Session
 from app.models.phone_rebind_log import PhoneRebindLog


-def rebound_within_days(
-    db: Session,
-    phone: str,
-    days: int,
-    *,
-    reset_at: datetime | None = None,
-) -> bool:
+def rebound_within_days(db: Session, phone: str, days: int) -> bool:
    """该手机号在最近 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)
+    since = datetime.now(timezone.utc) - timedelta(days=days)
    stmt = (
        select(PhoneRebindLog.id)
        .where(PhoneRebindLog.phone == phone, PhoneRebindLog.rebound_at >= since)
@@ -34,25 +21,16 @@ def rebound_within_days(
    return db.execute(stmt).first() is not None


-def remaining_block_days(
-    db: Session,
-    phone: str,
-    days: int,
-    *,
-    reset_at: datetime | None = None,
-) -> int:
+def remaining_block_days(db: Session, phone: str, days: int) -> 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(*conditions)
+        select(func.max(PhoneRebindLog.rebound_at)).where(PhoneRebindLog.phone == phone)
    ).scalar_one_or_none()
    if last is None:
        return 0
    if last.tzinfo is None:  # SQLite 取回 naive datetime,按 UTC 归一
-        last = last.replace(tzinfo=UTC)
-    remaining = (last + timedelta(days=days) - datetime.now(UTC)).total_seconds()
+        last = last.replace(tzinfo=timezone.utc)
+    remaining = (last + timedelta(days=days) - datetime.now(timezone.utc)).total_seconds()
    return max(0, math.ceil(remaining / 86400))


@@ -9,7 +9,6 @@ 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,
@@ -258,34 +257,22 @@ def _upsert_incident(


 def evaluate_behavior_rule(
-    db: Session,
-    *,
-    rule_code: str,
-    subject_id: str,
-    at: datetime,
-    threshold: int | None = None,
-    subject_reset_at: datetime | None = None,
+    db: Session, *, rule_code: str, subject_id: str, at: datetime
 ) -> RiskIncident | None:
    spec = RULES[rule_code]
-    effective_threshold = threshold or get_rule_threshold(db, rule_code)
+    threshold = 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)
-    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
+    start = max(window_start, reset_at) if reset_at else window_start
    count, first_at, last_at, triggered_at = _event_stats(
        db,
        spec,
        subject_id=subject_id,
        start=start,
        end=end,
-        threshold=effective_threshold,
+        threshold=threshold,
    )
-    if (
-        count < effective_threshold
-        or first_at is None
-        or last_at is None
-        or triggered_at is None
-    ):
+    if count < threshold or first_at is None or last_at is None or triggered_at is None:
        return None
    return _upsert_incident(
        db,
@@ -313,6 +300,7 @@ 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
@@ -323,39 +311,23 @@ def reconcile_behavior_rule(
        BehaviorEvent.occurred_at >= start,
        BehaviorEvent.occurred_at < end,
    )
-    subject_rows = db.execute(
+    qualifying_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: 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(
+    qualifying = {str(subject_id) for subject_id, _ in qualifying_rows}
+    for subject_id, last_at in qualifying_rows:
+        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(
@@ -411,29 +383,7 @@ def record_behavior_event(
    db.add(event)
    db.flush()
    if evaluate_rule:
-        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,
-            )
+        evaluate_behavior_rule(db, rule_code=evaluate_rule, subject_id=subject_id, at=at)
    if commit:
        db.commit()
        db.refresh(event)
@@ -446,33 +396,20 @@ def sync_compare_incident(
    user_id: int,
    at: datetime,
    threshold: int | None = None,
-    device_id: str | None = None,
    commit: bool = True,
 ) -> RiskIncident | None:
-    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
-    )
+    effective_threshold = threshold 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")
-    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
+    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
    filters = (
        ComparisonRecord.user_id == user_id,
        ComparisonRecord.created_at >= start,
@@ -527,6 +464,7 @@ 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,
@@ -538,17 +476,17 @@ def reconcile_compare_rule(
            ComparisonRecord.created_at < end,
        )
        .group_by(ComparisonRecord.user_id)
+        .having(func.count(ComparisonRecord.id) >= threshold)
    ).all()
-    qualifying: set[str] = set()
+    qualifying = {str(user_id) for user_id, _ in rows}
    for user_id, last_at in rows:
-        incident = sync_compare_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(
@@ -222,9 +222,9 @@ class CompareStartReserveIn(BaseModel):


 class CompareStartReserveOut(BaseModel):
-    limit: int | None
+    limit: int
    used: int
-    remaining: int | None
+    remaining: int


 class CompareStatsOut(BaseModel):
@@ -7,7 +7,7 @@ from pydantic import BaseModel, Field


 class GuideVideoStartIn(BaseModel):
-    """开播询问。领券与比价分别使用独立配置和独立次数。"""
+    """开播询问。scene 目前只有 coupon(领券浮层);预留给日后比价等场景。"""

    scene: Literal["coupon", "comparison"] = "coupon"

@@ -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_cooldown_sec" in items
-    # hidden 项(任务/里程碑、首页轮播数据源、广告次数/单次金币/每轮次数/信息流广告开关)不在配置页返回。
+    # 非 hidden 项照常返回;看广告组保留可见的:每日上限 / 单次金币上限 / 关闭后冷却。
+    assert "signin_rewards" in items and "ad_daily_limit" in items and "ad_cooldown_sec" in items
+    # hidden 项(任务/里程碑、首页轮播数据源、看广告组的单次金币/每轮次数/信息流广告开关)不在配置页返回。
    for hidden_key in (
        "task_rewards", "record_milestones", "marquee_feed_mode",
-        "ad_daily_limit", "ad_reward_coin", "ad_round_count", "comparing_ad_enabled",
+        "ad_reward_coin", "ad_round_count", "comparing_ad_enabled",
    ):
        assert hidden_key not in items, f"{hidden_key} 应被 hidden 过滤"
    assert items["signin_rewards"]["value"] == [
@@ -73,7 +73,6 @@ def test_monitoring_audit_catalog_and_api_permissions(
        "analytics-health",
        "event-logs",
        "audit-logs",
-        "limit-whitelist",
    ]

    # 运营默认可查风控和设备存活，不能绕过导航直调技术/审计接口。
@@ -83,9 +82,6 @@ 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",
@@ -93,14 +89,13 @@ 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

@@ -204,7 +199,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", "limit-whitelist",
+        "ad-revenue", "huawei-review", "event-logs", "audit-logs",
    }


@@ -75,7 +75,7 @@ def _assert_ledger_balanced(user_id: int) -> None:

@pytest.fixture()
 def guide_configured():
-    """给全局配置塞一支片子(默认 max_plays=3 / reward_coin=100),用完还原成未配片。"""
+    """给全局配置塞一支片子(默认 max_plays=3 / reward_coin=120),用完还原成未配片。"""
    with SessionLocal() as db:
        crud_guide.set_video(db, VIDEO_URL, admin_id=1)
        cfg = crud_guide.get_config(db)
@@ -201,11 +201,7 @@ 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, *, scene="coupon", reset_at=None: 0,
-    )
+    monkeypatch.setattr(crud_guide, "used_plays", lambda db, user_id, scene="coupon": 0)
    with SessionLocal() as db:
        result = crud_guide.start_play(db, uid)

@@ -265,99 +261,3 @@ 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,
-            )
@@ -1,943 +0,0 @@
-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
@@ -7,7 +7,9 @@ mock/真发/变量覆盖/站内联动/设备反查/参数校验。厂商 HTTP
 from __future__ import annotations

 import json
+import logging

+import httpx
 import pytest
 from fastapi.testclient import TestClient

@@ -110,6 +112,99 @@ def test_huawei_non_success_code_raises(monkeypatch) -> None:
        vendor_push.send_notification("huawei", "bad-token", title="t", body="b")


+def test_vendor_http_logs_device_token_raw_response_and_elapsed(monkeypatch, caplog) -> None:
+    vendor_push._token_cache.clear()
+
+    def _fake_request(method, url, **kwargs):  # noqa: ANN001
+        if url == vendor_push.settings.HUAWEI_PUSH_TOKEN_ENDPOINT:
+            return _Resp({"access_token": "raw-access-token", "expires_in": 3600})
+        return _Resp({"code": "80000000", "msg": "Success", "requestId": "vendor-request-1"})
+
+    monkeypatch.setattr(vendor_push.settings, "HUAWEI_PUSH_APP_ID", "10086001")
+    monkeypatch.setattr(vendor_push.settings, "HUAWEI_PUSH_APP_SECRET", "raw-app-secret")
+    monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
+
+    with caplog.at_level(logging.INFO, logger=vendor_push.logger.name):
+        vendor_push.send_notification(
+            "huawei",
+            "raw-device-token",
+            title="需要进入日志的标题",
+            body="需要进入日志的正文",
+            extras={"type": "withdraw_success", "notificationId": "90001"},
+        )
+
+    messages = "\n".join(record.getMessage() for record in caplog.records)
+    assert "operation=authenticate" in messages
+    assert "operation=send_notification" in messages
+    assert "request=" in messages
+    assert "response=" in messages
+    assert "elapsed_ms=" in messages
+    assert "vendor-request-1" in messages
+    assert "withdraw_success" in messages
+    assert "raw-access-token" in messages
+    assert "raw-device-token" in messages
+    assert "需要进入日志的标题" in messages
+    assert "需要进入日志的正文" in messages
+    assert "<redacted" in messages
+    assert "raw-app-secret" not in messages
+    send_http_log = next(
+        record.getMessage()
+        for record in caplog.records
+        if "operation=send_notification" in record.getMessage()
+        and "vendor push http completed" in record.getMessage()
+    )
+    assert "raw-device-token" in send_http_log
+    assert "raw-access-token" not in send_http_log  # 请求头中的厂商鉴权 token 仍脱敏
+
+
+def test_vendor_http_network_error_logs_request_context(monkeypatch, caplog) -> None:
+    def _fake_request(*args, **kwargs):  # noqa: ANN002, ANN003
+        raise httpx.ConnectError("connection refused")
+
+    monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_APP_SECRET", "raw-xiaomi-secret")
+    monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
+
+    with (
+        caplog.at_level(logging.ERROR, logger=vendor_push.logger.name),
+        pytest.raises(vendor_push.VendorPushError, match="push http error"),
+    ):
+        vendor_push.send_notification(
+            "xiaomi",
+            "raw-xiaomi-token",
+            title="title",
+            body="body",
+            extras={"type": "push_test"},
+        )
+
+    messages = "\n".join(record.getMessage() for record in caplog.records)
+    assert "outcome=network_error" in messages
+    assert "response=<no_response>" in messages
+    assert "operation=send_notification" in messages
+    assert "elapsed_ms=" in messages
+    assert "raw-xiaomi-secret" not in messages
+    assert "raw-xiaomi-token" in messages
+
+
+@pytest.mark.parametrize(
+    ("request_kwargs", "device_token"),
+    [
+        ({"json": {"token": ["honor-device-token"]}}, "honor-device-token"),
+        ({"json": {"message": {"token": ["huawei-device-token"]}}}, "huawei-device-token"),
+        ({"json": {"regId": "vivo-device-token"}}, "vivo-device-token"),
+        ({"data": {"registration_id": "xiaomi-device-token"}}, "xiaomi-device-token"),
+        (
+            {"data": {"message": '{"target_value":"oppo-device-token"}'}},
+            "oppo-device-token",
+        ),
+    ],
+)
+def test_all_vendor_device_token_fields_remain_visible_in_request_summary(
+    request_kwargs, device_token
+) -> None:
+    summary = vendor_push._request_summary(request_kwargs)
+    assert device_token in summary
+
+
 def test_vendor_aliases_normalize() -> None:
    assert vendor_push.normalize_vendor("华为") == "huawei"
    assert vendor_push.normalize_vendor("HMS") == "huawei"
@@ -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": 100_001,
+                "sms_hourly_threshold": 21,
                "oneclick_daily_threshold": 20,
                "compare_daily_threshold": 100,
            },