更新 run.sh

.env.example

@@ -175,9 +175,3 @@ PANGLE_REPORT_SITE_ID_TEST=5832303
 # APPLOG_MAX_BATCH=500                      # 单批最大条数(超 → 422;导入期常量,改需重启)
 # APPLOG_MAX_BODY_BYTES=2097152            # 请求体上限 2MB(超 → 413;运行期可调)
 # APPLOG_MAX_MSG_BYTES=8192                # 单条 msg 超此字节数截断
-
-# ===== 华为 Push =====
-HUAWEI_PUSH_APP_ID=
-HUAWEI_PUSH_APP_SECRET=
-# 0=正式消息(默认)；1=测试消息(仅开发联调，生产必须保持 0)
-HUAWEI_PUSH_TARGET_USER_TYPE=0

alembic/versions/limit_policy_global_bundle.py

@@ -1,193 +0,0 @@
-"""store all global limit values in one complete JSON document
-
-Revision ID: limit_policy_global_bundle
-Revises: limit_policy_whitelist
-"""
-from __future__ import annotations
-
-from collections.abc import Sequence
-from typing import Any
-
-import sqlalchemy as sa
-from sqlalchemy.dialects import postgresql
-
-from alembic import op
-
-revision: str = "limit_policy_global_bundle"
-down_revision: str | Sequence[str] | None = "limit_policy_whitelist"
-branch_labels: str | Sequence[str] | None = None
-depends_on: str | Sequence[str] | None = None
-
-_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
-_BUNDLE_KEY = "limit_policy_global"
-
-# rule_code, old sparse key, default, legacy structured key, legacy JSON field
-_RULES: tuple[tuple[str, str, int, str | None, str | None], ...] = (
-    ("compare.start.daily", "compare_daily_limit", 100, None, None),
-    ("sms.send.hourly", "sms_send_hourly_limit", 5, None, None),
-    ("sms.send.daily", "sms_send_daily_limit", 20, None, None),
-    ("sms.phone.cooldown", "sms_phone_cooldown_seconds", 60, None, None),
-    ("sms.code.failed_attempts", "sms_code_max_failed_attempts", 5, None, None),
-    ("sms.login.hourly", "sms_login_hourly_limit", 5, None, None),
-    ("wechat.bind.hourly", "wechat_bind_sms_hourly_limit", 5, None, None),
-    ("wechat.conflict.hourly", "wechat_conflict_hourly_limit", 5, None, None),
-    (
-        "ad.reward_video.daily",
-        "ad_reward_video_daily_limit",
-        500,
-        "ad_daily_limit",
-        None,
-    ),
-    (
-        "ad.feed.daily",
-        "ad_feed_daily_limit",
-        500,
-        "ad_daily_limit",
-        None,
-    ),
-    ("ad.reward_video.cooldown", "ad_cooldown_sec", 3, None, None),
-    (
-        "guide.video.lifetime",
-        "guide_video_max_plays",
-        3,
-        "coupon_guide_video",
-        "max_plays",
-    ),
-    ("phone.rebind.days", "phone_rebind_days", 30, None, None),
-    ("risk.sms.hourly", "risk_sms_hourly_threshold", 5, None, None),
-    (
-        "risk.oneclick.daily",
-        "risk_oneclick_daily_threshold",
-        20,
-        None,
-        None,
-    ),
-    (
-        "risk.compare.daily",
-        "risk_compare_daily_threshold",
-        100,
-        None,
-        None,
-    ),
-)
-
-
-def _table() -> sa.TableClause:
-    return sa.table(
-        "app_config",
-        sa.column("key", sa.String(64)),
-        sa.column("value", _JSON),
-        sa.column("updated_by_admin_id", sa.Integer),
-        sa.column("updated_at", sa.DateTime(timezone=True)),
-    )
-
-
-def _row(conn, table, key: str):
-    return conn.execute(
-        sa.select(
-            table.c.value,
-            table.c.updated_by_admin_id,
-        ).where(table.c.key == key)
-    ).mappings().first()
-
-
-def _int_or_none(value: Any) -> int | None:
-    if isinstance(value, bool):
-        return None
-    try:
-        return int(value)
-    except (TypeError, ValueError):
-        return None
-
-
-def upgrade() -> None:
-    conn = op.get_bind()
-    table = _table()
-    values = {rule_code: default for rule_code, _, default, _, _ in _RULES}
-
-    # Old shared/structured values are the lowest-precedence compatibility
-    # source.  Dedicated sparse keys override them.
-    for rule_code, _, _, legacy_key, legacy_field in _RULES:
-        if legacy_key is None:
-            continue
-        legacy = _row(conn, table, legacy_key)
-        if legacy is None:
-            continue
-        raw = legacy["value"]
-        if legacy_field is not None:
-            raw = raw.get(legacy_field) if isinstance(raw, dict) else None
-        parsed = _int_or_none(raw)
-        if parsed is not None:
-            values[rule_code] = parsed
-
-    for rule_code, sparse_key, _, _, _ in _RULES:
-        sparse = _row(conn, table, sparse_key)
-        parsed = _int_or_none(sparse["value"]) if sparse is not None else None
-        if parsed is not None:
-            values[rule_code] = parsed
-
-    # If a deployment already wrote the new key, preserve it over old keys.
-    bundle = _row(conn, table, _BUNDLE_KEY)
-    if bundle is not None and isinstance(bundle["value"], dict):
-        for rule_code, raw in bundle["value"].items():
-            if rule_code not in values:
-                continue
-            parsed = _int_or_none(raw)
-            if parsed is not None:
-                values[rule_code] = parsed
-
-    if bundle is None:
-        conn.execute(
-            table.insert().values(
-                key=_BUNDLE_KEY,
-                value=values,
-                updated_by_admin_id=None,
-            )
-        )
-    else:
-        conn.execute(
-            table.update()
-            .where(table.c.key == _BUNDLE_KEY)
-            .values(value=values, updated_at=sa.func.now())
-        )
-
-    sparse_keys = [sparse_key for _, sparse_key, _, _, _ in _RULES]
-    conn.execute(table.delete().where(table.c.key.in_(sparse_keys)))
-
-
-def downgrade() -> None:
-    conn = op.get_bind()
-    table = _table()
-    bundle = _row(conn, table, _BUNDLE_KEY)
-    values = (
-        bundle["value"]
-        if bundle is not None and isinstance(bundle["value"], dict)
-        else {}
-    )
-    admin_id = bundle["updated_by_admin_id"] if bundle is not None else None
-
-    for rule_code, sparse_key, default, _, _ in _RULES:
-        value = _int_or_none(values.get(rule_code))
-        if value is None:
-            value = default
-        existing = _row(conn, table, sparse_key)
-        if existing is None:
-            conn.execute(
-                table.insert().values(
-                    key=sparse_key,
-                    value=value,
-                    updated_by_admin_id=admin_id,
-                )
-            )
-        else:
-            conn.execute(
-                table.update()
-                .where(table.c.key == sparse_key)
-                .values(
-                    value=value,
-                    updated_by_admin_id=admin_id,
-                    updated_at=sa.func.now(),
-                )
-            )
-
-    conn.execute(table.delete().where(table.c.key == _BUNDLE_KEY))

alembic/versions/limit_policy_whitelist.py

@@ -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_ten_circle_v2"
-branch_labels: str | Sequence[str] | None = None
-depends_on: str | Sequence[str] | None = None
-
-_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
-_PAGE = "limit-whitelist"
-_DEFAULT_ROLES = ("operator", "tech")
-
-
-def _role_table() -> sa.TableClause:
-    return sa.table(
-        "admin_role",
-        sa.column("name", sa.String),
-        sa.column("pages", _JSON),
-    )
-
-
-def _add_default_role_permissions() -> None:
-    """Grant the page without replacing existing role customisations."""
-    role = _role_table()
-    conn = op.get_bind()
-    rows = conn.execute(
-        sa.select(role.c.name, role.c.pages).where(
-            role.c.name.in_(_DEFAULT_ROLES)
-        )
-    ).all()
-    for name, pages in rows:
-        current_pages = list(pages or [])
-        if _PAGE not in current_pages:
-            conn.execute(
-                role.update()
-                .where(role.c.name == name)
-                .values(pages=[*current_pages, _PAGE])
-            )
-
-
-def _remove_default_role_permissions() -> None:
-    role = _role_table()
-    conn = op.get_bind()
-    rows = conn.execute(
-        sa.select(role.c.name, role.c.pages).where(
-            role.c.name.in_(_DEFAULT_ROLES)
-        )
-    ).all()
-    for name, pages in rows:
-        current_pages = list(pages or [])
-        if _PAGE in current_pages:
-            conn.execute(
-                role.update()
-                .where(role.c.name == name)
-                .values(
-                    pages=[page for page in current_pages if page != _PAGE]
-                )
-            )
-
-
-def upgrade() -> None:
-    op.create_table(
-        "limit_policy_override",
-        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
-        sa.Column("subject_type", sa.String(length=16), nullable=False),
-        sa.Column("subject_value", sa.String(length=128), nullable=False),
-        sa.Column("rule_code", sa.String(length=64), nullable=False),
-        sa.Column("mode", sa.String(length=24), nullable=False),
-        sa.Column("limit_value", sa.Integer(), nullable=True),
-        sa.Column(
-            "enabled", sa.Boolean(), server_default=sa.true(), nullable=False
-        ),
-        sa.Column("starts_at", sa.DateTime(timezone=True), nullable=True),
-        sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True),
-        sa.Column("reset_at", sa.DateTime(timezone=True), nullable=True),
-        sa.Column("reason", sa.String(length=256), nullable=True),
-        sa.Column("created_by_admin_id", sa.Integer(), nullable=True),
-        sa.Column(
-            "created_at",
-            sa.DateTime(timezone=True),
-            server_default=sa.func.now(),
-            nullable=False,
-        ),
-        sa.Column(
-            "updated_at",
-            sa.DateTime(timezone=True),
-            server_default=sa.func.now(),
-            nullable=False,
-        ),
-        sa.PrimaryKeyConstraint("id"),
-        sa.UniqueConstraint(
-            "subject_type",
-            "subject_value",
-            "rule_code",
-            name="uq_limit_policy_subject_rule",
-        ),
-    )
-    op.create_index(
-        "ix_limit_policy_lookup",
-        "limit_policy_override",
-        ["subject_type", "subject_value", "rule_code", "enabled"],
-        unique=False,
-    )
-    op.create_index(
-        "ix_limit_policy_expires",
-        "limit_policy_override",
-        ["expires_at"],
-        unique=False,
-    )
-    _add_default_role_permissions()
-
-
-def downgrade() -> None:
-    _remove_default_role_permissions()
-    op.drop_index("ix_limit_policy_expires", table_name="limit_policy_override")
-    op.drop_index("ix_limit_policy_lookup", table_name="limit_policy_override")
-    op.drop_table("limit_policy_override")

app/admin/main.py

@@ -31,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
@@ -72,8 +71,8 @@ admin_app = FastAPI(
 
 # admin 前端独立部署。生产同域(nginx)无需 CORS;本地 next dev 跨域需放行开发源。
 _dev_origins = [
-    "http://localhost:3001",
-    "http://127.0.0.1:3001",
+    "http://localhost:3002",
+    "http://127.0.0.1:3002",
     "http://localhost:3000",
     "http://127.0.0.1:3000",
 ]
@@ -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)

app/admin/permissions.py

@@ -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",
     ]},
 ]
 

app/admin/repositories/limit_whitelist.py

@@ -1,583 +0,0 @@
-"""CRUD and presentation helpers for limit policy overrides."""
-
-from __future__ import annotations
-
-from datetime import UTC, datetime
-from hashlib import blake2b
-
-from sqlalchemy import String, cast, func, or_, select, text
-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 risk as risk_repo
-
-
-class DuplicateOverrideError(Exception):
-    pass
-
-
-def lock_subject(db: Session, *, subject_type: str, subject_value: str) -> None:
-    """Serialize writes for one logical whitelist subject on PostgreSQL.
-
-    Row locks cannot protect a brand-new subject because there is no row to
-    lock yet.  A transaction-scoped advisory lock closes that gap and also
-    serializes concurrent appends that touch different rule rows.
-    """
-
-    bind = db.get_bind()
-    if bind.dialect.name != "postgresql":
-        return
-    identity = f"limit-whitelist:{subject_type}:{subject_value}".encode()
-    lock_id = int.from_bytes(
-        blake2b(identity, digest_size=8).digest(),
-        byteorder="big",
-        signed=True,
-    )
-    db.execute(
-        text("SELECT pg_advisory_xact_lock(:lock_id)"),
-        {"lock_id": lock_id},
-    )
-
-
-_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 list_subject_rows(
-    db: Session,
-    *,
-    subject_type: str | None = None,
-    keyword: str | None = None,
-    rule_code: str | None = None,
-    offset: int = 0,
-    limit: int = 10,
-) -> tuple[list[tuple[str, str, list[LimitPolicyOverride], datetime]], int]:
-    """List whitelist entries grouped and paginated by subject.
-
-    The filter decides which subjects match.  Once a subject matches, all of
-    its rules are returned so category counts and the edit form stay complete.
-    Ordering uses the original creation time, therefore editing a subject does
-    not unexpectedly move it to the first page.
-    """
-
-    conditions = [
-        LimitPolicyOverride.mode.in_(
-            [limit_policy.MODE_UNLIMITED, limit_policy.MODE_SUPPRESS_ALERT]
-        )
-    ]
-    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)
-
-    matched_subjects = (
-        select(
-            LimitPolicyOverride.subject_type.label("subject_type"),
-            LimitPolicyOverride.subject_value.label("subject_value"),
-            func.min(LimitPolicyOverride.created_at).label("subject_created_at"),
-        )
-        .where(*conditions)
-        .group_by(
-            LimitPolicyOverride.subject_type,
-            LimitPolicyOverride.subject_value,
-        )
-        .subquery()
-    )
-    total = int(db.scalar(select(func.count()).select_from(matched_subjects)) or 0)
-    subjects = list(
-        db.execute(
-            select(
-                matched_subjects.c.subject_type,
-                matched_subjects.c.subject_value,
-                matched_subjects.c.subject_created_at,
-            )
-            .order_by(
-                matched_subjects.c.subject_created_at.desc(),
-                matched_subjects.c.subject_type,
-                matched_subjects.c.subject_value,
-            )
-            .offset(offset)
-            .limit(limit)
-        ).all()
-    )
-    if not subjects:
-        return [], total
-
-    pair_conditions = [
-        (
-            (LimitPolicyOverride.subject_type == item.subject_type)
-            & (LimitPolicyOverride.subject_value == item.subject_value)
-        )
-        for item in subjects
-    ]
-    rows = list(
-        db.scalars(
-            select(LimitPolicyOverride)
-            .where(
-                or_(*pair_conditions),
-                LimitPolicyOverride.mode.in_(
-                    [
-                        limit_policy.MODE_UNLIMITED,
-                        limit_policy.MODE_SUPPRESS_ALERT,
-                    ]
-                ),
-            )
-            .order_by(
-                LimitPolicyOverride.created_at.desc(),
-                LimitPolicyOverride.id.desc(),
-            )
-        ).all()
-    )
-    rows_by_subject: dict[tuple[str, str], list[LimitPolicyOverride]] = {}
-    for row in rows:
-        rows_by_subject.setdefault((row.subject_type, row.subject_value), []).append(row)
-
-    return [
-        (
-            item.subject_type,
-            item.subject_value,
-            rows_by_subject.get((item.subject_type, item.subject_value), []),
-            item.subject_created_at,
-        )
-        for item in subjects
-    ], total
-
-
-def rows_for_subject(
-    db: Session, *, subject_type: str, subject_value: str
-) -> list[LimitPolicyOverride]:
-    return list(
-        db.scalars(
-            select(LimitPolicyOverride)
-            .where(
-                LimitPolicyOverride.subject_type == subject_type,
-                LimitPolicyOverride.subject_value == subject_value,
-            )
-            .order_by(
-                LimitPolicyOverride.created_at.desc(),
-                LimitPolicyOverride.id.desc(),
-            )
-        ).all()
-    )
-
-
-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
-    limit_policy.set_global_limits(
-        db,
-        {rule.code: value},
-        admin_id=admin_id,
-        commit=False,
-    )
-    # 引导视频仍有一个旧的专用配置页。同步其结构化配置，保证两个入口读取
-    # 同一数值；业务判定仍统一走 limit_policy。
-    if rule.legacy_config_key and rule.legacy_json_field:
-        legacy = db.get(AppConfig, rule.legacy_config_key)
-        legacy_value = (
-            dict(legacy.value) if legacy is not None and isinstance(legacy.value, dict) else {}
-        )
-        legacy_value[rule.legacy_json_field] = value
-        if legacy is None:
-            legacy = AppConfig(
-                key=rule.legacy_config_key,
-                value=legacy_value,
-                updated_by_admin_id=admin_id,
-            )
-            db.add(legacy)
-        else:
-            legacy.value = legacy_value
-            legacy.updated_by_admin_id = admin_id
-        db.flush()
-    return before, value
-
-
-def restore_global(row: LimitPolicyOverride) -> LimitPolicyOverride:
-    """Stop applying the exception while retaining its history for audit."""
-
-    row.enabled = False
-    row.reset_at = None
-    return row

app/admin/routers/config.py

@@ -12,11 +12,9 @@ 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 import limit_policy
 from app.core.config_schema import CONFIG_DEFS
 from app.core.rewards import SIGNIN_CYCLE_LEN
 from app.models.admin import AdminUser
-from app.models.app_config import AppConfig
 from app.repositories import app_config
 
 router = APIRouter(
@@ -56,35 +54,7 @@ def _validate(key: str, value: Any) -> None:
             raise ValueError("需为布尔值")
 
 
-def _limit_item(
-    key: str,
-    values: dict[str, int],
-    bundle: AppConfig | None,
-) -> ConfigItemOut:
-    definition = CONFIG_DEFS[key]
-    rule_code = limit_policy.RULE_CODE_BY_CONFIG_KEY[key]
-    return ConfigItemOut(
-        key=key,
-        value=values[rule_code],
-        label=definition["label"],
-        group=definition["group"],
-        type=definition["type"],
-        help=definition.get("help"),
-        default=definition["default"],
-        overridden=bundle is not None,
-        updated_at=(
-            bundle.updated_at.isoformat() if bundle is not None else None
-        ),
-    )
-
-
 def _item(db, key: str) -> ConfigItemOut:
-    if key in limit_policy.RULE_CODE_BY_CONFIG_KEY:
-        return _limit_item(
-            key,
-            limit_policy.get_global_limits(db),
-            db.get(AppConfig, limit_policy.LIMIT_POLICY_GLOBAL_KEY),
-        )
     for item in app_config.list_all(db):
         if item["key"] == key:
             return ConfigItemOut(**item)
@@ -94,20 +64,11 @@ def _item(db, key: str) -> ConfigItemOut:
 @router.get("", response_model=list[ConfigItemOut], summary="所有可配项 + 当前值(不含 hidden)")
 def list_config(db: AdminDb) -> list[ConfigItemOut]:
     # hidden 项(已下线/由专用页管理,如福利页任务·里程碑·看广告调参、首页轮播数据源)不在本页渲染。
-    legacy_items = {
-        item["key"]: item for item in app_config.list_all(db)
-    }
-    values = limit_policy.get_global_limits(db)
-    bundle = db.get(AppConfig, limit_policy.LIMIT_POLICY_GLOBAL_KEY)
-    out: list[ConfigItemOut] = []
-    for key, definition in CONFIG_DEFS.items():
-        if definition.get("hidden"):
-            continue
-        if key in limit_policy.RULE_CODE_BY_CONFIG_KEY:
-            out.append(_limit_item(key, values, bundle))
-        else:
-            out.append(ConfigItemOut(**legacy_items[key]))
-    return out
+    return [
+        ConfigItemOut(**item)
+        for item in app_config.list_all(db)
+        if not CONFIG_DEFS[item["key"]].get("hidden")
+    ]
 
 
 @router.patch("/{key}", response_model=ConfigItemOut, summary="改某项配置(带审计)")
@@ -122,39 +83,11 @@ def update_config(
         raise HTTPException(status_code=404, detail="未知配置项")
     try:
         _validate(key, body.value)
-        rule_code = limit_policy.RULE_CODE_BY_CONFIG_KEY.get(key)
-        if rule_code is not None:
-            before = limit_policy.get_global_limits(db)[rule_code]
-            limit_policy.set_global_limits(
-                db,
-                {rule_code: body.value},
-                admin_id=admin.id,
-                commit=False,
-            )
-        else:
-            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 两项，避免旧入口写入后业务实际值不变。
-            limit_policy.set_global_limits(
-                db,
-                {
-                    "ad.reward_video.daily": body.value,
-                    "ad.feed.daily": body.value,
-                },
-                admin_id=admin.id,
-                commit=False,
-            )
     except ValueError as e:
-        db.rollback()
         raise HTTPException(status_code=400, detail=str(e)) from e
+
+    before = app_config.get_value(db, key)
+    app_config.set_value(db, key, body.value, admin_id=admin.id, commit=False)
     write_audit(
         db, admin, action="config.set", target_type="config", target_id=key,
         detail={"before": before, "after": body.value}, ip=get_client_ip(request), commit=False,

app/admin/routers/feedback.py

@@ -1,7 +1,6 @@
 """admin 反馈工单:列表筛选 + 审核采纳/拒绝(带金币发放与审计)。"""
 from __future__ import annotations
 
-import logging
 from datetime import datetime
 from typing import Annotated
 
@@ -26,8 +25,6 @@ from app.models.feedback import Feedback
 from app.repositories import wallet as wallet_repo
 from app.services import notification_events
 
-logger = logging.getLogger("shagua.admin.feedback")
-
 router = APIRouter(
     prefix="/admin/api/feedbacks",
     tags=["admin-feedback"],
@@ -49,24 +46,10 @@ def _approve_feedback(
     *,
     bulk: bool = False,
 ) -> FeedbackOut:
-    logger.info(
-        "feedback approve started feedback_id=%s admin_id=%s bulk=%s",
-        feedback_id, admin.id, bulk,
-    )
     fb = db.get(Feedback, feedback_id, with_for_update=True)
     if fb is None:
-        logger.warning(
-            "feedback approve rejected not found feedback_id=%s admin_id=%s bulk=%s",
-            feedback_id, admin.id, bulk,
-        )
         raise HTTPException(status_code=404, detail="反馈不存在")
-    if fb.status not in {"pending", "new"}:
-        logger.warning(
-            "feedback approve rejected invalid status feedback_id=%s user_id=%s "
-            "admin_id=%s status=%s bulk=%s",
-            feedback_id, fb.user_id, admin.id, fb.status, bulk,
-        )
-        _ensure_pending(fb)
+    _ensure_pending(fb)
 
     before = fb.status
     mutations.review_feedback(
@@ -108,18 +91,8 @@ def _approve_feedback(
     )
     db.commit()
     db.refresh(fb)
-    logger.info(
-        "feedback approve committed feedback_id=%s user_id=%s admin_id=%s "
-        "before=%s after=%s reward_coins=%s bulk=%s",
-        feedback_id, fb.user_id, admin.id, before, fb.status, payload.reward_coins, bulk,
-    )
     out = FeedbackOut.model_validate(fb)
     notification_events.notify_feedback_reward(db, fb)
-    logger.info(
-        "feedback approve notification dispatch returned feedback_id=%s user_id=%s "
-        "admin_id=%s notification_type=feedback_reward bulk=%s",
-        feedback_id, fb.user_id, admin.id, bulk,
-    )
     return out
 
 
@@ -132,24 +105,10 @@ def _reject_feedback(
     *,
     bulk: bool = False,
 ) -> FeedbackOut:
-    logger.info(
-        "feedback reject started feedback_id=%s admin_id=%s bulk=%s",
-        feedback_id, admin.id, bulk,
-    )
     fb = db.get(Feedback, feedback_id, with_for_update=True)
     if fb is None:
-        logger.warning(
-            "feedback reject rejected not found feedback_id=%s admin_id=%s bulk=%s",
-            feedback_id, admin.id, bulk,
-        )
         raise HTTPException(status_code=404, detail="反馈不存在")
-    if fb.status not in {"pending", "new"}:
-        logger.warning(
-            "feedback reject rejected invalid status feedback_id=%s user_id=%s "
-            "admin_id=%s status=%s bulk=%s",
-            feedback_id, fb.user_id, admin.id, fb.status, bulk,
-        )
-        _ensure_pending(fb)
+    _ensure_pending(fb)
 
     before = fb.status
     mutations.review_feedback(
@@ -183,18 +142,8 @@ def _reject_feedback(
     )
     db.commit()
     db.refresh(fb)
-    logger.info(
-        "feedback reject committed feedback_id=%s user_id=%s admin_id=%s "
-        "before=%s after=%s bulk=%s",
-        feedback_id, fb.user_id, admin.id, before, fb.status, bulk,
-    )
     out = FeedbackOut.model_validate(fb)
     notification_events.notify_feedback_reply(db, fb)
-    logger.info(
-        "feedback reject notification dispatch returned feedback_id=%s user_id=%s "
-        "admin_id=%s notification_type=feedback_reply bulk=%s",
-        feedback_id, fb.user_id, admin.id, bulk,
-    )
     return out
 
 
@@ -252,10 +201,6 @@ def bulk_approve_feedbacks(
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
 ) -> FeedbackBulkResult:
-    logger.info(
-        "feedback bulk approve started admin_id=%s item_count=%s",
-        admin.id, len(body.ids),
-    )
     results: list[FeedbackBulkItemResult] = []
     ip = get_client_ip(request)
     for feedback_id in body.ids:
@@ -264,24 +209,11 @@ def bulk_approve_feedbacks(
             results.append(FeedbackBulkItemResult(id=feedback_id, ok=True, status=out.status))
         except HTTPException as exc:
             db.rollback()
-            logger.warning(
-                "feedback bulk approve item failed feedback_id=%s admin_id=%s error=%s",
-                feedback_id, admin.id, exc.detail,
-            )
             results.append(FeedbackBulkItemResult(id=feedback_id, ok=False, error=str(exc.detail)))
         except Exception:  # noqa: BLE001 - 单笔失败不打断整批
             db.rollback()
-            logger.exception(
-                "feedback bulk approve item failed feedback_id=%s admin_id=%s",
-                feedback_id, admin.id,
-            )
             results.append(FeedbackBulkItemResult(id=feedback_id, ok=False, error="系统异常"))
-    result = _bulk_result(results)
-    logger.info(
-        "feedback bulk approve completed admin_id=%s total=%s success=%s failed=%s",
-        admin.id, result.total, result.success, result.failed,
-    )
-    return result
+    return _bulk_result(results)
 
 
 @router.post("/bulk/reject", response_model=FeedbackBulkResult, summary="批量拒绝采纳反馈")
@@ -291,10 +223,6 @@ def bulk_reject_feedbacks(
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
 ) -> FeedbackBulkResult:
-    logger.info(
-        "feedback bulk reject started admin_id=%s item_count=%s",
-        admin.id, len(body.ids),
-    )
     results: list[FeedbackBulkItemResult] = []
     ip = get_client_ip(request)
     for feedback_id in body.ids:
@@ -303,24 +231,11 @@ def bulk_reject_feedbacks(
             results.append(FeedbackBulkItemResult(id=feedback_id, ok=True, status=out.status))
         except HTTPException as exc:
             db.rollback()
-            logger.warning(
-                "feedback bulk reject item failed feedback_id=%s admin_id=%s error=%s",
-                feedback_id, admin.id, exc.detail,
-            )
             results.append(FeedbackBulkItemResult(id=feedback_id, ok=False, error=str(exc.detail)))
         except Exception:  # noqa: BLE001 - 单笔失败不打断整批
             db.rollback()
-            logger.exception(
-                "feedback bulk reject item failed feedback_id=%s admin_id=%s",
-                feedback_id, admin.id,
-            )
             results.append(FeedbackBulkItemResult(id=feedback_id, ok=False, error="系统异常"))
-    result = _bulk_result(results)
-    logger.info(
-        "feedback bulk reject completed admin_id=%s total=%s success=%s failed=%s",
-        admin.id, result.total, result.success, result.failed,
-    )
-    return result
+    return _bulk_result(results)
 
 
 @router.post("/{feedback_id}/handle", response_model=OkResponse, summary="标记反馈已处理")
@@ -343,16 +258,7 @@ def approve_feedback(
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
 ) -> FeedbackOut:
-    try:
-        return _approve_feedback(db, admin, feedback_id, payload, get_client_ip(request))
-    except HTTPException:
-        raise
-    except Exception:
-        logger.exception(
-            "feedback approve failed feedback_id=%s admin_id=%s",
-            feedback_id, admin.id,
-        )
-        raise
+    return _approve_feedback(db, admin, feedback_id, payload, get_client_ip(request))
 
 
 @router.post("/{feedback_id}/reject", response_model=FeedbackOut, summary="拒绝采纳反馈")
@@ -363,13 +269,4 @@ def reject_feedback(
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
 ) -> FeedbackOut:
-    try:
-        return _reject_feedback(db, admin, feedback_id, payload, get_client_ip(request))
-    except HTTPException:
-        raise
-    except Exception:
-        logger.exception(
-            "feedback reject failed feedback_id=%s admin_id=%s",
-            feedback_id, admin.id,
-        )
-        raise
+    return _reject_feedback(db, admin, feedback_id, payload, get_client_ip(request))

app/admin/routers/limit_whitelist.py

@@ -1,858 +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,
-    LimitSubjectEnabledPatch,
-    LimitSubjectList,
-    LimitSubjectOut,
-)
-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"))],
-)
-
-_WHITELIST_MODES = {
-    limit_policy.MODE_UNLIMITED,
-    limit_policy.MODE_SUPPRESS_ALERT,
-}
-
-
-def _subject_whitelist_rows(
-    db,
-    *,
-    subject_type: str,
-    subject_value: str,
-) -> list[LimitPolicyOverride]:
-    return [
-        row
-        for row in repo.rows_for_subject(
-            db,
-            subject_type=subject_type,
-            subject_value=subject_value,
-        )
-        if row.mode in _WHITELIST_MODES
-    ]
-
-
-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, populate_existing=True)
-    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")
-
-
-def _subject_out(
-    db,
-    *,
-    subject_type: str,
-    subject_value: str,
-    rows: list[LimitPolicyOverride],
-    created_at,
-) -> LimitSubjectOut:
-    items = [_out(db, row) for row in rows]
-    group_counts: dict[str, int] = {}
-    for item in items:
-        group_counts[item.rule_group] = group_counts.get(item.rule_group, 0) + 1
-    return LimitSubjectOut(
-        subject_type=subject_type,
-        subject_value=subject_value,
-        group_counts=group_counts,
-        total_rules=len(items),
-        items=items,
-        created_at=created_at,
-        updated_at=max(
-            (item.updated_at for item in items),
-            default=created_at,
-        ),
-    )
-
-
-@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.get(
-    "/subjects",
-    response_model=LimitSubjectList,
-    summary="按手机号或设备聚合读取白名单配置",
-)
-def list_override_subjects(
-    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(10, ge=1, le=100),
-) -> LimitSubjectList:
-    subjects, total = repo.list_subject_rows(
-        db,
-        subject_type=subject_type,
-        keyword=keyword,
-        rule_code=rule_code,
-        offset=offset,
-        limit=limit,
-    )
-    return LimitSubjectList(
-        items=[
-            _subject_out(
-                db,
-                subject_type=item_subject_type,
-                subject_value=subject_value,
-                rows=rows,
-                created_at=created_at,
-            )
-            for item_subject_type, subject_value, rows, created_at in subjects
-        ],
-        total=total,
-    )
-
-
-@router.patch(
-    "/subjects/enabled",
-    response_model=LimitSubjectOut,
-    summary="整体启用或停用一个手机号或设备的白名单",
-)
-def set_override_subject_enabled(
-    body: LimitSubjectEnabledPatch,
-    request: Request,
-    admin: CurrentAdmin,
-    db: AdminDb,
-) -> LimitSubjectOut:
-    try:
-        subject_value = limit_policy.validate_whitelist_subject(
-            body.subject_type, body.subject_value
-        )
-        repo.lock_subject(
-            db,
-            subject_type=body.subject_type,
-            subject_value=subject_value,
-        )
-        rows = _subject_whitelist_rows(
-            db,
-            subject_type=body.subject_type,
-            subject_value=subject_value,
-        )
-        if not rows:
-            raise HTTPException(status_code=404, detail="白名单主体不存在")
-        before = [_audit_payload(_out(db, row)) for row in rows]
-        for row in rows:
-            repo.update(
-                db,
-                row,
-                enabled=body.enabled,
-                starts_at=None,
-                expires_at=None,
-                reason=None,
-                fields_set={"enabled"},
-            )
-        for rule_code in {row.rule_code for row in rows}:
-            _reconcile_risk_rule(db, rule_code)
-        after = [_audit_payload(_out(db, row)) for row in rows]
-        write_audit(
-            db,
-            admin,
-            action="limit.override.subject_enabled",
-            target_type="limit_override_subject",
-            target_id=f"{body.subject_type}:{subject_value}",
-            detail={"before": before, "after": after},
-            ip=get_client_ip(request),
-            commit=False,
-        )
-        db.commit()
-    except HTTPException:
-        raise
-    except (KeyError, ValueError) as exc:
-        db.rollback()
-        raise HTTPException(status_code=400, detail=str(exc)) from exc
-
-    rows = [
-        row
-        for row in repo.rows_for_subject(
-            db,
-            subject_type=body.subject_type,
-            subject_value=subject_value,
-        )
-        if row.mode
-        in {
-            limit_policy.MODE_UNLIMITED,
-            limit_policy.MODE_SUPPRESS_ALERT,
-        }
-    ]
-    return _subject_out(
-        db,
-        subject_type=body.subject_type,
-        subject_value=subject_value,
-        rows=rows,
-        created_at=min(row.created_at for row in rows),
-    )
-
-
-@router.put(
-    "/subjects",
-    response_model=LimitSubjectOut,
-    summary="整体更新一个手机号或设备的白名单限制项",
-)
-def replace_override_subject(
-    body: LimitOverrideBulkWrite,
-    request: Request,
-    admin: CurrentAdmin,
-    db: AdminDb,
-) -> LimitSubjectOut:
-    current_rule_code = ""
-    try:
-        subject_value = limit_policy.validate_whitelist_subject(
-            body.subject_type, body.subject_value
-        )
-        repo.lock_subject(
-            db,
-            subject_type=body.subject_type,
-            subject_value=subject_value,
-        )
-        if body.subject_type == limit_policy.SUBJECT_DEVICE:
-            limit_policy.validate_device_rule_scope(body.rule_codes)
-
-        existing_rows = repo.rows_for_subject(
-            db,
-            subject_type=body.subject_type,
-            subject_value=subject_value,
-        )
-        subject_created_at = min(
-            (row.created_at for row in existing_rows),
-            default=None,
-        )
-        before = [_audit_payload(_out(db, row)) for row in existing_rows]
-        existing_by_rule = {row.rule_code: row for row in existing_rows}
-        selected_codes = set(body.rule_codes)
-        touched_rule_codes = set(selected_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
-            )
-            row = existing_by_rule.get(current_rule_code)
-            if row is None:
-                row = repo.create(
-                    db,
-                    subject_type=body.subject_type,
-                    subject_value=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,
-                )
-                # The table is ordered by the subject's original creation
-                # time.  If an edit replaces every rule, carry that timestamp
-                # to the new rows so the subject does not jump to the top.
-                if subject_created_at is not None:
-                    row.created_at = subject_created_at
-                continue
-
-            # Historical custom-value rows are converted to the only supported
-            # product modes when the administrator selects that rule again.
-            row.mode = mode
-            row.limit_value = None
-            row.reset_at = None
-            repo.update(
-                db,
-                row,
-                enabled=body.enabled,
-                starts_at=body.starts_at,
-                expires_at=body.expires_at,
-                reason=body.reason,
-                fields_set={"enabled", "starts_at", "expires_at", "reason"},
-            )
-
-        for row in existing_rows:
-            if row.rule_code not in selected_codes and row.mode in {
-                limit_policy.MODE_UNLIMITED,
-                limit_policy.MODE_SUPPRESS_ALERT,
-            }:
-                touched_rule_codes.add(row.rule_code)
-                db.delete(row)
-        db.flush()
-    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 rule_code in touched_rule_codes:
-        _reconcile_risk_rule(db, rule_code)
-    rows = [
-        row
-        for row in repo.rows_for_subject(
-            db,
-            subject_type=body.subject_type,
-            subject_value=subject_value,
-        )
-        if row.mode
-        in {
-            limit_policy.MODE_UNLIMITED,
-            limit_policy.MODE_SUPPRESS_ALERT,
-        }
-    ]
-    after = [_audit_payload(_out(db, row)) for row in rows]
-    write_audit(
-        db,
-        admin,
-        action="limit.override.subject_replace",
-        target_type="limit_override_subject",
-        target_id=f"{body.subject_type}:{subject_value}",
-        detail={"before": before, "after": after},
-        ip=get_client_ip(request),
-        commit=False,
-    )
-    db.commit()
-    rows = [
-        row
-        for row in repo.rows_for_subject(
-            db,
-            subject_type=body.subject_type,
-            subject_value=subject_value,
-        )
-        if row.mode
-        in {
-            limit_policy.MODE_UNLIMITED,
-            limit_policy.MODE_SUPPRESS_ALERT,
-        }
-    ]
-    return _subject_out(
-        db,
-        subject_type=body.subject_type,
-        subject_value=subject_value,
-        rows=rows,
-        created_at=min(row.created_at for row in rows),
-    )
-
-
-@router.post(
-    "",
-    response_model=LimitOverrideOut,
-    status_code=status.HTTP_201_CREATED,
-    summary="新增白名单配置",
-)
-def create_override(
-    body: LimitOverrideWrite,
-    request: Request,
-    admin: CurrentAdmin,
-    db: AdminDb,
-) -> LimitOverrideOut:
-    before: list[dict] = []
-    touched_rows: list[LimitPolicyOverride] = []
-    try:
-        subject_value = limit_policy.validate_whitelist_subject(
-            body.subject_type, body.subject_value
-        )
-        repo.lock_subject(
-            db,
-            subject_type=body.subject_type,
-            subject_value=subject_value,
-        )
-        existing_rows = _subject_whitelist_rows(
-            db,
-            subject_type=body.subject_type,
-            subject_value=subject_value,
-        )
-        if body.subject_type == limit_policy.SUBJECT_DEVICE:
-            limit_policy.validate_device_rule_scope(
-                [
-                    *(item.rule_code for item in existing_rows),
-                    body.rule_code,
-                ]
-            )
-        before = [_audit_payload(_out(db, item)) for item in existing_rows]
-        row = repo.create(
-            db,
-            subject_type=body.subject_type,
-            subject_value=subject_value,
-            rule_code=body.rule_code,
-            mode=body.mode,
-            enabled=body.enabled,
-            starts_at=body.starts_at,
-            expires_at=body.expires_at,
-            reason=body.reason,
-            limit_value=None,
-            admin_id=admin.id,
-        )
-        touched_rows = [*existing_rows, row]
-        if row.mode in _WHITELIST_MODES:
-            # Keep the legacy single-rule endpoint compatible without letting
-            # it create a second validity period for the same logical subject.
-            for existing in existing_rows:
-                repo.update(
-                    db,
-                    existing,
-                    enabled=body.enabled,
-                    starts_at=body.starts_at,
-                    expires_at=body.expires_at,
-                    reason=None,
-                    fields_set={"enabled", "starts_at", "expires_at"},
-                )
-    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
-    for rule_code in {item.rule_code for item in touched_rows}:
-        _reconcile_risk_rule(db, rule_code)
-    after_rows = _subject_whitelist_rows(
-        db,
-        subject_type=row.subject_type,
-        subject_value=row.subject_value,
-    )
-    write_audit(
-        db,
-        admin,
-        action="limit.override.create",
-        target_type="limit_override_subject",
-        target_id=f"{row.subject_type}:{row.subject_value}",
-        detail={
-            "before": before,
-            "after": [_audit_payload(_out(db, item)) for item in after_rows],
-        },
-        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] = []
-    before: list[dict] = []
-    current_rule_code = ""
-    try:
-        subject_value = limit_policy.validate_whitelist_subject(
-            body.subject_type, body.subject_value
-        )
-        repo.lock_subject(
-            db,
-            subject_type=body.subject_type,
-            subject_value=subject_value,
-        )
-        existing_rows = repo.rows_for_subject(
-            db,
-            subject_type=body.subject_type,
-            subject_value=subject_value,
-        )
-        whitelist_modes = {
-            limit_policy.MODE_UNLIMITED,
-            limit_policy.MODE_SUPPRESS_ALERT,
-        }
-        existing_whitelist_rows = [row for row in existing_rows if row.mode in whitelist_modes]
-        before = [_audit_payload(_out(db, row)) for row in existing_whitelist_rows]
-        existing_by_rule = {row.rule_code: row for row in existing_rows}
-        selected_codes = set(body.rule_codes)
-        if body.subject_type == limit_policy.SUBJECT_DEVICE:
-            limit_policy.validate_device_rule_scope(
-                list(
-                    {
-                        *(row.rule_code for row in existing_whitelist_rows),
-                        *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
-            )
-            row = existing_by_rule.get(current_rule_code)
-            if row is None:
-                row = repo.create(
-                    db,
-                    subject_type=body.subject_type,
-                    subject_value=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,
-                )
-            else:
-                row.mode = mode
-                row.limit_value = None
-                row.reset_at = None
-                repo.update(
-                    db,
-                    row,
-                    enabled=body.enabled,
-                    starts_at=body.starts_at,
-                    expires_at=body.expires_at,
-                    reason=body.reason,
-                    fields_set={
-                        "enabled",
-                        "starts_at",
-                        "expires_at",
-                        "reason",
-                    },
-                )
-
-        # 同一手机号或设备在产品上是一条白名单。追加限制项时保留原规则，
-        # 但统一使用最后一次配置的启用状态与有效期，避免一个主体出现多套时间。
-        rows = [
-            row
-            for row in repo.rows_for_subject(
-                db,
-                subject_type=body.subject_type,
-                subject_value=subject_value,
-            )
-            if row.mode in whitelist_modes
-        ]
-        for row in rows:
-            if row.rule_code in selected_codes:
-                continue
-            repo.update(
-                db,
-                row,
-                enabled=body.enabled,
-                starts_at=body.starts_at,
-                expires_at=body.expires_at,
-                reason=None,
-                fields_set={"enabled", "starts_at", "expires_at"},
-            )
-    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={
-            "before": before,
-            "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)
-    repo.lock_subject(
-        db,
-        subject_type=row.subject_type,
-        subject_value=row.subject_value,
-    )
-    row = _row_or_404(db, override_id)
-    subject_rows = (
-        _subject_whitelist_rows(
-            db,
-            subject_type=row.subject_type,
-            subject_value=row.subject_value,
-        )
-        if row.mode in _WHITELIST_MODES
-        else [row]
-    )
-    before = [_audit_payload(_out(db, item)) for item in subject_rows]
-    fields_set = set(body.model_fields_set)
-    try:
-        period_fields = {"enabled", "starts_at", "expires_at"}
-        if row.mode in _WHITELIST_MODES and fields_set & period_fields:
-            new_enabled = body.enabled if "enabled" in fields_set else row.enabled
-            new_starts = body.starts_at if "starts_at" in fields_set else row.starts_at
-            new_expires = body.expires_at if "expires_at" in fields_set else row.expires_at
-            for item in subject_rows:
-                item_fields = set(period_fields)
-                if item.id == row.id and "reason" in fields_set:
-                    item_fields.add("reason")
-                repo.update(
-                    db,
-                    item,
-                    enabled=new_enabled,
-                    starts_at=new_starts,
-                    expires_at=new_expires,
-                    reason=body.reason if item.id == row.id else None,
-                    fields_set=item_fields,
-                )
-        else:
-            repo.update(
-                db,
-                row,
-                **body.model_dump(),
-                fields_set=fields_set,
-            )
-    except (KeyError, ValueError) as exc:
-        db.rollback()
-        raise HTTPException(status_code=400, detail=str(exc)) from exc
-    for rule_code in {item.rule_code for item in subject_rows}:
-        _reconcile_risk_rule(db, rule_code)
-    after_rows = [_audit_payload(_out(db, item)) for item in subject_rows]
-    write_audit(
-        db,
-        admin,
-        action="limit.override.update",
-        target_type="limit_override_subject",
-        target_id=f"{row.subject_type}:{row.subject_value}",
-        detail={"before": before, "after": after_rows},
-        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)
-    repo.lock_subject(
-        db,
-        subject_type=row.subject_type,
-        subject_value=row.subject_value,
-    )
-    row = _row_or_404(db, override_id)
-    subject_rows = (
-        _subject_whitelist_rows(
-            db,
-            subject_type=row.subject_type,
-            subject_value=row.subject_value,
-        )
-        if row.mode in _WHITELIST_MODES
-        else [row]
-    )
-    before = [_audit_payload(_out(db, item)) for item in subject_rows]
-    for item in subject_rows:
-        repo.restore_global(item)
-    for rule_code in {item.rule_code for item in subject_rows}:
-        _reconcile_risk_rule(db, rule_code)
-    after = [_audit_payload(_out(db, item)) for item in subject_rows]
-    write_audit(
-        db,
-        admin,
-        action="limit.override.restore_global",
-        target_type="limit_override_subject",
-        target_id=f"{row.subject_type}:{row.subject_value}",
-        detail={"before": before, "after": 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)
-    repo.lock_subject(
-        db,
-        subject_type=row.subject_type,
-        subject_value=row.subject_value,
-    )
-    row = _row_or_404(db, override_id)
-    before = _audit_payload(_out(db, row))
-    target_id = str(row.id)
-    rule_code = row.rule_code
-    db.delete(row)
-    db.flush()
-    _reconcile_risk_rule(db, rule_code)
-    write_audit(
-        db,
-        admin,
-        action="limit.override.delete",
-        target_type="limit_override",
-        target_id=target_id,
-        detail={"before": before},
-        ip=get_client_ip(request),
-        commit=False,
-    )
-    db.commit()

app/admin/routers/risk_monitor.py

@@ -23,8 +23,13 @@ from app.admin.schemas.risk_monitor import (
     RiskResetResponse,
     RiskRuleConfig,
 )
-from app.core import limit_policy
+from app.core.config_schema import (
+    RISK_COMPARE_DAILY_THRESHOLD_KEY,
+    RISK_ONECLICK_DAILY_THRESHOLD_KEY,
+    RISK_SMS_HOURLY_THRESHOLD_KEY,
+)
 from app.models.risk import RiskIncident, SubjectRestriction
+from app.repositories import app_config
 from app.repositories import risk as risk_repo
 
 router = APIRouter(
@@ -78,16 +83,15 @@ def update_rules(
 ) -> RiskRuleConfig:
     before = _rule_config(db).model_dump()
     after = body.model_dump()
-    limit_policy.set_global_limits(
-        db,
-        {
-            "risk.sms.hourly": body.sms_hourly_threshold,
-            "risk.oneclick.daily": body.oneclick_daily_threshold,
-            "risk.compare.daily": body.compare_daily_threshold,
-        },
-        admin_id=admin.id,
-        commit=False,
+    values = (
+        (RISK_SMS_HOURLY_THRESHOLD_KEY, body.sms_hourly_threshold),
+        (RISK_ONECLICK_DAILY_THRESHOLD_KEY, body.oneclick_daily_threshold),
+        (RISK_COMPARE_DAILY_THRESHOLD_KEY, body.compare_daily_threshold),
     )
+    for key, value in values:
+        app_config.set_value(
+            db, key, value, admin_id=admin.id, commit=False
+        )
 
     now = risk_repo.utcnow()
     risk_repo.reconcile_behavior_rule(

app/admin/schemas/limit_whitelist.py

@@ -1,150 +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 LimitSubjectEnabledPatch(BaseModel):
-    model_config = ConfigDict(extra="forbid")
-
-    subject_type: SubjectType
-    subject_value: str = Field(min_length=1, max_length=128)
-    enabled: bool
-
-
-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
-
-
-class LimitSubjectOut(BaseModel):
-    subject_type: str
-    subject_value: str
-    group_counts: dict[str, int]
-    total_rules: int
-    items: list[LimitOverrideOut]
-    created_at: datetime
-    updated_at: datetime
-
-
-class LimitSubjectList(BaseModel):
-    items: list[LimitSubjectOut]
-    total: int

app/admin/schemas/risk_monitor.py

@@ -21,9 +21,9 @@ class RiskMonitorSummary(BaseModel):
 
 
 class RiskRuleConfig(BaseModel):
-    sms_hourly_threshold: int = Field(ge=1, le=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):

app/api/v1/ad.py

@@ -18,7 +18,7 @@ import uuid
 from fastapi import APIRouter, Depends, HTTPException, Path, Request, status
 
 from app.api.deps import CurrentUser, DbSession
-from app.core import 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),
     )
 
 

app/api/v1/auth.py

@@ -16,7 +16,7 @@ from fastapi import APIRouter, HTTPException, Request
 from sqlalchemy.exc import IntegrityError
 
 from app.api.deps import CurrentUser, DbSession
-from app.core import 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(

app/api/v1/compare_record.py

@@ -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),
     )
 
 

app/core/config.py

@@ -88,8 +88,6 @@ class Settings(BaseSettings):
     # (OAuth 换 token 时 client_id 即 AppId)。发送走 v1 messages:send,成功码 80000000。
     HUAWEI_PUSH_APP_ID: str = ""
     HUAWEI_PUSH_APP_SECRET: str = ""
-    # 0=正式消息(默认，受正式消息频控)；1=测试消息(仅开发联调，勿用于生产)。
-    HUAWEI_PUSH_TARGET_USER_TYPE: int = Field(default=0, ge=0, le=1)
     HUAWEI_PUSH_TOKEN_ENDPOINT: str = "https://oauth-login.cloud.huawei.com/oauth2/v3/token"
     HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE: str = (
         "https://push-api.cloud.huawei.com/v1/{app_id}/messages:send"

app/core/config_schema.py

@@ -14,138 +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"
-LIMIT_POLICY_GLOBAL_KEY = "limit_policy_global"
 
 # 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",
@@ -182,8 +56,7 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
     },
     "ad_daily_limit": {
         "default": r.DAILY_AD_REWARD_LIMIT, "label": "看广告每日上限(次)",
-        "group": "看广告", "type": "int", "min": 1, "max": 100_000, "hidden": True,
-        "help": "历史共享上限；新配置由白名单页分别管理激励视频与 Draw 信息流。",
+        "group": "看广告", "type": "int", "help": "福利页激励视频每日可发奖次数上限,默认 500。",
     },
     "ad_max_coin": {
         "default": r.MAX_AD_REWARD_COIN, "label": "看广告单次金币上限",
@@ -195,8 +68,7 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
     },
     "ad_cooldown_sec": {
         "default": r.VIDEO_ROUND_COOLDOWN_SECONDS, "label": "广告关闭后冷却(秒)",
-        "group": "看广告", "type": "int", "min": 0, "max": 86_400,
-        "help": "点击退出广告后,下次点击观看前的冷却时间,默认 3 秒。",
+        "group": "看广告", "type": "int", "help": "点击退出广告后,下次点击观看前的冷却时间,默认 3 秒。",
     },
     "comparing_ad_enabled": {
         "default": True, "label": "比价/领券期信息流广告",
@@ -246,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,
@@ -266,7 +138,7 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
         "group": "风控",
         "type": "int",
         "min": 1,
-        "max": 100_000,
+        "max": 100,
         "hidden": True,
         "help": "同一账户在北京时间同一自然日内发起比价达到该次数时告警。",
     },

app/core/limit_policy.py

@@ -1,629 +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 delete, 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,
-    LIMIT_POLICY_GLOBAL_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}
-RULE_CODE_BY_CONFIG_KEY = {rule.config_key: rule.code for rule in RULES}
-LIMIT_CONFIG_KEYS = tuple(RULE_CODE_BY_CONFIG_KEY)
-
-
-@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 default_global_limits() -> dict[str, int]:
-    """Return the complete 16-rule default snapshot keyed by rule code."""
-    return {rule.code: rule.default_limit for rule in RULES}
-
-
-def _normalise_global_limits(value: object) -> dict[str, int]:
-    """Merge a stored JSON object with safe code defaults.
-
-    The migration and every admin write persist all rules.  Defaults are still
-    merged here so a manually damaged/older partial JSON cannot take the
-    service down after deployment.
-    """
-
-    values = default_global_limits()
-    if not isinstance(value, dict):
-        return values
-    for rule_code, raw in value.items():
-        rule = RULE_MAP.get(str(rule_code))
-        if rule is None or isinstance(raw, bool):
-            continue
-        try:
-            parsed = int(raw)
-        except (TypeError, ValueError):
-            continue
-        if rule.min_value <= parsed <= rule.max_value:
-            values[rule.code] = parsed
-    return values
-
-
-def _legacy_global_limit(db: Session, rule: RuleDefinition) -> tuple[int, str]:
-    """Read the pre-bundle representation while upgrading old/test databases."""
-
-    row = db.get(AppConfig, rule.config_key)
-    if row is not None:
-        return int(row.value), "legacy-key"
-
-    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 get_global_limits(db: Session) -> dict[str, int]:
-    """Read the complete global-limit JSON, with a pre-migration fallback."""
-
-    row = db.get(AppConfig, LIMIT_POLICY_GLOBAL_KEY)
-    if row is not None:
-        return _normalise_global_limits(row.value)
-    return {rule.code: _legacy_global_limit(db, rule)[0] for rule in RULES}
-
-
-def set_global_limits(
-    db: Session,
-    updates: dict[str, int],
-    *,
-    admin_id: int,
-    commit: bool = True,
-) -> dict[str, int]:
-    """Atomically update selected rules inside the single complete JSON row."""
-
-    parsed_updates: dict[str, int] = {}
-    for rule_code, raw_value in updates.items():
-        rule = get_rule(rule_code)
-        value = int(raw_value)
-        if not rule.min_value <= value <= rule.max_value:
-            raise ValueError(
-                f"limit for {rule_code} must be between "
-                f"{rule.min_value} and {rule.max_value}"
-            )
-        parsed_updates[rule.code] = value
-
-    row = db.scalar(
-        select(AppConfig)
-        .where(AppConfig.key == LIMIT_POLICY_GLOBAL_KEY)
-        .with_for_update()
-    )
-    values = (
-        _normalise_global_limits(row.value)
-        if row is not None
-        else {rule.code: _legacy_global_limit(db, rule)[0] for rule in RULES}
-    )
-    values.update(parsed_updates)
-
-    if row is None:
-        row = AppConfig(
-            key=LIMIT_POLICY_GLOBAL_KEY,
-            value=values,
-            updated_by_admin_id=admin_id,
-        )
-        db.add(row)
-    else:
-        row.value = dict(values)
-        row.updated_by_admin_id = admin_id
-
-    # Once the bundle exists, stale sparse rows must not become a second source
-    # of truth.  The data migration performs the same cleanup for production.
-    db.execute(delete(AppConfig).where(AppConfig.key.in_(LIMIT_CONFIG_KEYS)))
-    if commit:
-        db.commit()
-        db.refresh(row)
-    else:
-        db.flush()
-    return values
-
-
-def _global_limit(db: Session, rule: RuleDefinition) -> tuple[int, str]:
-    row = db.get(AppConfig, LIMIT_POLICY_GLOBAL_KEY)
-    if row is not None:
-        return _normalise_global_limits(row.value)[rule.code], "configured"
-    return _legacy_global_limit(db, rule)
-
-
-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] = []
-    values = get_global_limits(db)
-    for rule in RULES:
-        out.append(
-            {
-                "code": rule.code,
-                "label": rule.label,
-                "group": rule.group,
-                "global_limit": values[rule.code],
-                "default_limit": rule.default_limit,
-                "window_label": rule.window_label,
-                "subject_types": list(rule.subject_types),
-                "allowed_modes": list(rule.allowed_modes),
-                "min_value": rule.min_value,
-                "max_value": rule.max_value,
-                "supports_reset": rule.supports_reset,
-                "alert_only": rule.alert_only,
-            }
-        )
-    return out
-
-
-def validate_override(
-    rule: RuleDefinition,
-    *,
-    subject_type: str,
-    mode: str,
-    limit_value: int | None,
-    starts_at: datetime | None,
-    expires_at: datetime | None,
-) -> None:
-    if subject_type not in rule.subject_types:
-        raise ValueError("该规则不支持此主体类型")
-    if mode not in rule.allowed_modes:
-        raise ValueError("该规则不支持此策略模式")
-    if limit_value is not None:
-        raise ValueError("白名单不支持覆盖指定值")
-    if mode in {MODE_UNLIMITED, MODE_SUPPRESS_ALERT} and expires_at is None:
-        raise ValueError("临时白名单必须设置失效时间")
-    if starts_at and expires_at and _aware(expires_at) <= _aware(starts_at):
-        raise ValueError("失效时间必须晚于生效时间")
-    if (
-        mode in {MODE_UNLIMITED, MODE_SUPPRESS_ALERT}
-        and expires_at is not None
-        and _aware(expires_at) <= datetime.now(UTC)
-    ):
-        raise ValueError("临时白名单的失效时间必须晚于当前时间")
-    if mode == MODE_SUPPRESS_ALERT and not rule.alert_only:
-        raise ValueError("免告警只支持风控监控的三项规则")
-
-
-def active_overrides(
-    db: Session,
-    *,
-    subject_type: str | None = None,
-    keyword: str | None = None,
-) -> Iterable[LimitPolicyOverride]:
-    stmt = select(LimitPolicyOverride)
-    if subject_type:
-        stmt = stmt.where(LimitPolicyOverride.subject_type == subject_type)
-    if keyword:
-        stmt = stmt.where(LimitPolicyOverride.subject_value.ilike(f"%{keyword.strip()}%"))
-    return db.execute(
-        stmt.order_by(LimitPolicyOverride.updated_at.desc(), LimitPolicyOverride.id.desc())
-    ).scalars()

app/core/ratelimit.py

@@ -75,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)

app/integrations/sms/__init__.py

@@ -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)

app/integrations/sms/aliyun.py

@@ -35,7 +35,7 @@ _GC_THRESHOLD = 10000                    # 超此阈值,send 时顺手清老于
 
 # 发码错误码 → (HTTP 码, 用户提示)。未列出的一律 503(供应商不可用)。
 _SEND_ERRORS: dict[str, tuple[int, str]] = {
-    "MOBILE_NUMBER_ILLEGAL": (400, "请输入有效的手机号"),
+    "MOBILE_NUMBER_ILLEGAL": (400, "手机号无效"),
     "BUSINESS_LIMIT_CONTROL": (429, "今日发送次数过多,请明天再试"),
     "FREQUENCY_FAIL": (429, "发送过于频繁,请稍后再试"),
 }
@@ -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

app/integrations/sms/chuanglan.py

@@ -53,7 +53,7 @@ _GC_THRESHOLD = 10000  # 任一内存 dict 超此阈值,send 时顺手清过期
 # 发码错误码(创蓝 `code`)→ (HTTP 码, 用户提示)。未列出的一律 503(供应商不可用)。
 _SEND_ERRORS: dict[str, tuple[int, str]] = {
     "103": (429, "发送过于频繁，请稍后再试"),   # 提交速度过快
-    "107": (400, "请输入有效的手机号"),                 # 手机号码错误
+    "107": (400, "手机号无效"),                 # 手机号码错误
 }
 # 需运维介入的配置/开通/余额类错误:打 critical 日志(仍归 503)。
 _SEND_CRITICAL_CODES = frozenset({
@@ -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")):

app/integrations/sms/jiguang.py

@@ -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")):
@@ -198,5 +185,5 @@ def _send_via_jiguang(phone: str, code: str) -> None:
     if ecode == 50009:  # 极光侧超频
         raise SmsError("发送过于频繁,请稍后再试", status_code=429)
     if ecode == 50006:  # 手机号无效(schema 已挡格式,这里多是空号/停机)
-        raise SmsError("请输入有效的手机号", status_code=400)
+        raise SmsError("手机号无效", status_code=400)
     raise SmsError(f"短信发送失败(code={ecode})", status_code=503)

app/integrations/vendor_push.py

@@ -22,7 +22,7 @@ import uuid
 from collections.abc import Callable
 from dataclasses import dataclass
 from typing import Any
-from urllib.parse import quote, urlsplit
+from urllib.parse import quote
 
 import httpx
 
@@ -31,7 +31,6 @@ from app.core.config import settings
 logger = logging.getLogger("shagua.vendor_push")
 
 TYPE_ACCESSIBILITY_DISABLED = "accessibility_disabled"
-DATA_EVENT_NOTIFICATION_CREATED = "notification_created"
 SUPPORTED_VENDORS = frozenset({"honor", "huawei", "vivo", "xiaomi", "oppo"})
 
 # vendor key → 中文名(测试/配置状态接口展示用)
@@ -70,18 +69,6 @@ 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:
@@ -128,16 +115,8 @@ def send_notification(
 
     if mock:
         logger.info(
-            "vendor push mock vendor=%s request=%s",
-            vendor,
-            _log_summary(
-                {
-                    "push_token": token,
-                    "title": title,
-                    "body": body,
-                    "extras": extras,
-                }
-            ),
+            "[mock push] vendor=%s token=%s... title=%s body=%s extras=%s",
+            vendor, token[:12], title, body, extras,
         )
         return {
             "mock": True,
@@ -154,28 +133,7 @@ def send_notification(
         "xiaomi": _send_xiaomi,
         "oppo": _send_oppo,
     }
-    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
+    return dispatch[vendor](token, title, body, extras)
 
 
 def send_accessibility_disabled(
@@ -195,188 +153,19 @@ def send_accessibility_disabled(
     )
 
 
-def send_data_event(
-    push_vendor: str,
-    push_token: str,
-    *,
-    event: str,
-    notification_id: str,
-    mock: bool = False,
-) -> dict[str, Any]:
-    """发送无界面的轻量事件；当前只允许站内消息创建事件。"""
-    vendor = normalize_vendor(push_vendor)
-    token = push_token.strip() if push_token else ""
-    if not vendor or vendor not in SUPPORTED_VENDORS:
-        raise VendorPushError(f"unsupported push vendor: {push_vendor}")
-    if not token:
-        raise VendorPushError("push token is empty")
-    if event != DATA_EVENT_NOTIFICATION_CREATED:
-        raise VendorPushError(f"unsupported data event: {event}")
-    if not notification_id:
-        raise VendorPushError("notification_id is empty")
-
-    payload = {"event": event, "notificationId": str(notification_id)}
-    if mock:
-        return {"mock": True, "vendor": vendor, "payload": payload}
-
-    # vivo 的通知已设置 foregroundShow=false：App 在前台时必走
-    # onForegroundMessageArrived，正好就是本事件需要的刷新信号；不重复占用一次推送配额。
-    if vendor == "vivo":
-        return {"skipped": True, "reason": "foreground notification callback"}
-
-    # OPush 当前只支持通知栏消息，没有服务端透传单推接口。
-    # 客户端在 OPPO 首页可见时轮询未读数；这里必须安全跳过，不能请求不存在的
-    # /message/transparent/unicast（该地址会稳定返回 HTTP 404）。
-    if vendor == "oppo":
-        return {"skipped": True, "reason": "oppo does not support data messages"}
-
-    dispatch: dict[str, Callable[[str, dict[str, str]], dict[str, Any]]] = {
-        "honor": _send_honor_data,
-        "huawei": _send_huawei_data,
-        "xiaomi": _send_xiaomi_data,
-    }
-    return dispatch[vendor](token, payload)
-
-
 def _require(value: str, name: str) -> str:
     if not value:
         raise VendorPushError(f"{name} not configured")
     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 _vendor_response_failed(vendor: str, data: dict[str, Any]) -> bool:
-    """识别 HTTP 200 中明确的厂商业务失败，避免将失败请求记录成 success。"""
-    if data.get("error") or data.get("success") is False:
-        return True
-    if vendor == "xiaomi":
-        code = data.get("code")
-        result = str(data.get("result", "ok")).lower()
-        return code not in (0, "0", None) or result not in ("ok", "success")
-    if vendor == "oppo" and data.get("code") is not None:
-        return int(data["code"]) != 0
-    if vendor == "vivo" and data.get("result") is not None:
-        return int(data["result"]) != 0
-    if vendor == "honor" and data.get("code") is not None:
-        return int(data["code"]) != 200
-    if vendor == "huawei" and data.get("code") is not None:
-        return str(data["code"]) != "80000000"
-    return False
-
-
 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,
@@ -385,85 +174,41 @@ 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 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),
-        )
+        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:
-        data = resp.json()
+        return 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
-    vendor_failed = _vendor_response_failed(vendor, data)
-    log = logger.warning if vendor_failed else logger.info
-    log(
-        "vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
-        "outcome=%s http_status=%s request=%s response=%s elapsed_ms=%.1f",
-        vendor,
-        operation,
-        method,
-        _endpoint_for_log(url),
-        "vendor_error" if vendor_failed else "success",
-        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]:
-    return _request_json(
-        method,
-        url,
-        vendor=vendor,
-        operation=operation,
-        expected_status=expected_status,
-        **kwargs,
-    )
+    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
 
 
 def _cache_get(key: str) -> str | None:
@@ -489,8 +234,6 @@ 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,
@@ -500,22 +243,16 @@ def _honor_access_token() -> str:
     )
     token = data.get("access_token")
     if not token:
-        raise VendorPushError(f"honor auth failed: {_raw_log_summary(data)}")
+        raise VendorPushError(f"honor auth failed: {data}")
     return _cache_put(cache_key, str(token), data.get("expires_in"))
 
 
 def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> dict[str, Any]:
     app_id = _require(settings.HONOR_PUSH_APP_ID, "HONOR_PUSH_APP_ID")
     access_token = _honor_access_token()
-    click_action: dict[str, Any]
-    if extras.get("notificationId"):
-        # type=3 只负责打开首页，不保证 data 会变成目标 Activity 的 extras。消息中心通知必须
-        # 用自定义页面(type=1)+ intent URI，把 notificationId/type/feedbackId 等直接带给
-        # MainActivity；否则华为/荣耀系统能展示通知，但用户点击后客户端收不到任何导航参数。
-        click_action = {"type": 1, "intent": _click_intent_uri(extras)}
-    else:
-        click_action = {"type": 3}
     payload = {
+        # clickAction type=3(打开应用首页)时,荣耀点击会把 data JSON 的键值对注入启动 intent 的
+        # extras(与 HMS 同机制)→ MainActivity.consumeNavTarget 读 notif_id/notif_type 直达落地。
         "data": json.dumps(_click_extras(extras), ensure_ascii=False),
         "notification": {"title": title, "body": body},
         "android": {
@@ -524,7 +261,7 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
             "notification": {
                 "title": title,
                 "body": body,
-                "clickAction": click_action,
+                "clickAction": {"type": 3},
                 "importance": "NORMAL",
             },
         },
@@ -533,8 +270,6 @@ 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",
@@ -544,28 +279,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: {_raw_log_summary(data)}")
-    return data
-
-
-def _send_honor_data(token: str, payload: dict[str, str]) -> dict[str, Any]:
-    app_id = _require(settings.HONOR_PUSH_APP_ID, "HONOR_PUSH_APP_ID")
-    access_token = _honor_access_token()
-    data = _request_json(
-        "POST",
-        settings.HONOR_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
-        vendor="honor",
-        operation="send_data_event",
-        json={"data": json.dumps(payload, ensure_ascii=False), "token": [token]},
-        headers={
-            "Content-Type": "application/json; charset=UTF-8",
-            "Authorization": f"Bearer {access_token}",
-            "timestamp": str(int(time.time() * 1000)),
-        },
-    )
-    code = data.get("code")
-    if code is not None and int(code) != 200:
-        raise VendorPushError(f"honor data push failed: {_raw_log_summary(data)}")
+        raise VendorPushError(f"honor push failed: {data}")
     return data
 
 
@@ -580,8 +294,6 @@ 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,
@@ -591,7 +303,7 @@ def _huawei_access_token() -> str:
     )
     token = data.get("access_token")
     if not token:
-        raise VendorPushError(f"huawei auth failed: {_raw_log_summary(data)}")
+        raise VendorPushError(f"huawei auth failed: {data}")
     return _cache_put(cache_key, str(token), data.get("expires_in"))
 
 
@@ -600,24 +312,18 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d
     '80100000' 为部分成功(单 token 场景仍视为失败,错误里带原始响应便于排障)。"""
     app_id = _require(settings.HUAWEI_PUSH_APP_ID, "HUAWEI_PUSH_APP_ID")
     access_token = _huawei_access_token()
-    click_action: dict[str, Any]
-    if extras.get("notificationId"):
-        # type=3 只打开首页，Mate 20/EMUI 不会把 message.data 自动拆成启动 Intent extras。
-        # type=1 的自定义 intent 才能稳定携带反馈记录 id，且冷启动/onNewIntent 都走同一路由。
-        click_action = {"type": 1, "intent": _click_intent_uri(extras)}
-    else:
-        click_action = {"type": 3}
     payload = {
         "validate_only": False,
         "message": {
+            # click_action type=3(打开应用首页)时,HMS 点击会把 data JSON 的键值对注入启动 intent
+            # 的 extras → MainActivity.consumeNavTarget 读 notif_id/notif_type 直达落地。
             "data": json.dumps(_click_extras(extras), ensure_ascii=False),
             "android": {
-                "target_user_type": settings.HUAWEI_PUSH_TARGET_USER_TYPE,
                 "ttl": f"{settings.PUSH_TIME_TO_LIVE_SEC}s",
                 "notification": {
                     "title": title,
                     "body": body,
-                    "click_action": click_action,
+                    "click_action": {"type": 3},
                     "importance": "NORMAL",
                 },
             },
@@ -627,8 +333,6 @@ 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",
@@ -636,32 +340,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: {_raw_log_summary(data)}")
-    return data
-
-
-def _send_huawei_data(token: str, payload: dict[str, str]) -> dict[str, Any]:
-    app_id = _require(settings.HUAWEI_PUSH_APP_ID, "HUAWEI_PUSH_APP_ID")
-    access_token = _huawei_access_token()
-    data = _request_json(
-        "POST",
-        settings.HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
-        vendor="huawei",
-        operation="send_data_event",
-        json={
-            "validate_only": False,
-            "message": {
-                "data": json.dumps(payload, ensure_ascii=False),
-                "token": [token],
-            },
-        },
-        headers={
-            "Content-Type": "application/json; charset=UTF-8",
-            "Authorization": f"Bearer {access_token}",
-        },
-    )
-    if str(data.get("code", "")) != "80000000":
-        raise VendorPushError(f"huawei data push failed: {_raw_log_summary(data)}")
+        raise VendorPushError(f"huawei push failed: {data}")
     return data
 
 
@@ -678,8 +357,6 @@ 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,
@@ -689,10 +366,10 @@ def _vivo_auth_token() -> str:
         headers={"Content-Type": "application/json"},
     )
     if int(data.get("result", -1)) != 0:
-        raise VendorPushError(f"vivo auth failed: {_raw_log_summary(data)}")
+        raise VendorPushError(f"vivo auth failed: {data}")
     token = data.get("authToken")
     if not token:
-        raise VendorPushError(f"vivo auth missing authToken: {_raw_log_summary(data)}")
+        raise VendorPushError(f"vivo auth missing authToken: {data}")
     return _cache_put(cache_key, str(token), 24 * 3600)
 
 
@@ -708,22 +385,15 @@ def _send_vivo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
         "timeToLive": settings.PUSH_TIME_TO_LIVE_SEC,
         "requestId": uuid.uuid4().hex,
         "pushMode": settings.VIVO_PUSH_MODE,
-        # vivo SDK 4.1.5 只有在前台展示关闭时才调用
-        # OpenClientPushMessageReceiver.onForegroundMessageArrived。
-        # App 在该回调中刷新服务端未读数并补发一条本地通知；后台/锁屏时仍由
-        # vivo 系统展示，因而任何场景都只会出现一条通知。
-        "foregroundShow": False,
-        # vivo 官方 VPush 角标字段：离线收到通知时先由桌面自动 +1；
-        # App 启动/同步后再由统一未读数通过系统 API 精确校准。
-        "addBadge": True,
         "clientCustomMap": extras,
     }
-    # vivo Push SDK 480+ 已不再回调 skipType=3；必须使用 skipType=4 的完整 Intent URI。
-    # URI 同时包含 data/scheme、显式 component 和 S. extras，OriginOS 才会把业务参数
-    # 原样交给 MainActivity（仅靠隐式 deeplink 会打开 App，但可能剥掉 extras）。
+    # 点击落地:消息中心推送(带 notificationId)→ skipType=4 + skipContent=intent uri,由 vivo
+    # 系统直启 MainActivity 并携带 S. extras(与小米 notify_effect=2 同机制)。不依赖客户端
+    # VivoPushReceiver.onNotificationMessageClicked 里的后台 startActivity——Android 10+ BAL
+    # 会静默拦掉,receiver 路径仅作兜底。无 notificationId 的召回类保持 skipType=1 仅打开首页。
     if extras.get("notificationId"):
         payload["skipType"] = 4
-        payload["skipContent"] = _vivo_click_intent_uri(extras)
+        payload["skipContent"] = _click_intent_uri(extras)
     else:
         payload["skipType"] = 1
     if settings.VIVO_PUSH_CATEGORY:
@@ -731,37 +401,14 @@ 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",
             "authToken": auth_token,
         },
     )
-    # 10089 = 当前 vivo 应用尚未开通 VPush「离线自动角标」能力。
-    # 角标只是通知中心未读数的镜像，不能因此让整条通知发送失败：去掉 addBadge
-    # 原样重试，通知到达/应用同步后仍由客户端系统 API 按统一未读数精确设置。
-    if int(data.get("result", -1)) == 10089:
-        logger.warning("vivo VPush badge not enabled (10089); retrying without addBadge")
-        fallback_payload = dict(payload)
-        fallback_payload.pop("addBadge", None)
-        fallback_payload["requestId"] = uuid.uuid4().hex
-        data = _request_json(
-            "POST",
-            settings.VIVO_PUSH_SEND_ENDPOINT,
-            vendor="vivo",
-            operation="send_notification",
-            json=fallback_payload,
-            headers={
-                "Content-Type": "application/json",
-                "authToken": auth_token,
-            },
-        )
-        if int(data.get("result", -1)) == 0:
-            data = {**data, "badgeFallback": True}
     if int(data.get("result", -1)) != 0:
-        raise VendorPushError(f"vivo push failed: {_raw_log_summary(data)}")
+        raise VendorPushError(f"vivo push failed: {data}")
     return data
 
 
@@ -799,43 +446,14 @@ 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: {_raw_log_summary(data)}")
+        raise VendorPushError(f"xiaomi push failed: {data}")
     if str(data.get("result", "ok")).lower() not in ("ok", "success"):
-        raise VendorPushError(f"xiaomi push failed: {_raw_log_summary(data)}")
-    return data
-
-
-def _send_xiaomi_data(token: str, payload: dict[str, str]) -> dict[str, Any]:
-    app_secret = _require(settings.XIAOMI_PUSH_APP_SECRET, "XIAOMI_PUSH_APP_SECRET")
-    form = {
-        "registration_id": token,
-        "restricted_package_name": settings.ANDROID_PACKAGE_NAME,
-        "payload": json.dumps(payload, ensure_ascii=False),
-        "pass_through": "1",
-        "time_to_live": str(settings.PUSH_TIME_TO_LIVE_SEC * 1000),
-    }
-    if settings.XIAOMI_PUSH_CHANNEL_ID.strip():
-        form["extra.channel_id"] = settings.XIAOMI_PUSH_CHANNEL_ID.strip()
-    data = _request_form(
-        "POST",
-        settings.XIAOMI_PUSH_SEND_ENDPOINT,
-        vendor="xiaomi",
-        operation="send_data_event",
-        data=form,
-        headers={"Authorization": f"key={app_secret}"},
-    )
-    code = data.get("code")
-    if code not in (0, "0", None):
-        raise VendorPushError(f"xiaomi data push failed: {_raw_log_summary(data)}")
-    if str(data.get("result", "ok")).lower() not in ("ok", "success"):
-        raise VendorPushError(f"xiaomi data push failed: {_raw_log_summary(data)}")
+        raise VendorPushError(f"xiaomi push failed: {data}")
     return data
 
 
@@ -872,24 +490,6 @@ def _click_intent_uri(extras: dict[str, str]) -> str:
     return ";".join(parts)
 
 
-def _vivo_click_intent_uri(extras: dict[str, str]) -> str:
-    """vivo skipType=4 使用官方要求的完整 intent deeplink 形态。
-
-    `?#Intent` 中的问号不能省；OriginOS 对缺少它的 URI 会退化为“仅打开应用”，
-    不把 S. 参数交给目标 Activity。
-    """
-    pkg = settings.ANDROID_PACKAGE_NAME
-    parts = [
-        "intent://push/detail?#Intent",
-        "scheme=shaguabijia",
-        f"component={pkg}/{pkg}.MainActivity",
-        "launchFlags=0x14000000",
-    ]
-    parts += [f"S.{key}={quote(str(value), safe='')}" for key, value in _click_extras(extras).items()]
-    parts.append("end")
-    return ";".join(parts)
-
-
 def _xiaomi_template_param(title: str, alert: str) -> str:
     rendered = (
         settings.XIAOMI_PUSH_TEMPLATE_PARAM_JSON
@@ -922,8 +522,6 @@ 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,
@@ -932,10 +530,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: {_raw_log_summary(data)}")
+        raise VendorPushError(f"oppo auth failed: {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: {_raw_log_summary(data)}")
+        raise VendorPushError(f"oppo auth missing auth_token: {data}")
     return _cache_put(cache_key, str(token), 24 * 3600)
 
 
@@ -950,11 +548,6 @@ def _send_oppo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
         "off_line_ttl": ttl_hours,
         "action_parameters": json.dumps(_click_extras(extras), ensure_ascii=False),
     }
-    if extras.get("notificationId"):
-        # 只有已落库的站内消息才计入角标；保活提醒等临时推送不应留下无法消除的未读数。
-        # 客户端打开/已读后会按服务端真实未读总数覆盖，避免长期漂移。
-        notification["badge_operation_type"] = 1
-        notification["badge_message_count"] = 1
     # 点击落地:OPPO SDK 没有点击回调,参数只能靠服务端点击动作配置送达——action_parameters 的
     # 键值对仅在 click_action_type=1/4 时才会注入目标 Activity 的 intent extras(type=0「启动应用」
     # 会忽略它,extras 全丢 → 点了没反应,与小米 notify_effect=1 同款坑)。
@@ -980,8 +573,6 @@ 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),
@@ -989,5 +580,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: {_raw_log_summary(data)}")
+        raise VendorPushError(f"oppo push failed: {data}")
     return data

app/models/__init__.py

@@ -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

app/models/limit_policy.py

@@ -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,
-    )

app/repositories/ad_feed_reward.py

@@ -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,

app/repositories/ad_reward.py

@@ -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,

app/repositories/comparison.py

@@ -375,16 +375,6 @@ def _derive_from_platforms(
     rows = [p for p in (platforms or []) if isinstance(p, dict)]
     src = next((p for p in rows if p.get("role") == "source"), None)
     best = next((p for p in rows if p.get("is_best")), None)
-    if best is None:
-        # pricebot 没标 is_best(如全平台 has_dish_diff「相似替换/仅供参考」→ 不认定权威最低价)
-        # 但仍有有价行 → 兜底取有价行里最低价当参考 best,避免记录级 best_*/saved 整条落 NULL
-        # (否则首页价 0.00 / 记录页无最低红框 / 省额丢失 /「累计发现可省」漏计)。**候选含源**:
-        # 源常年全菜、价可信,源本身最便宜时 best 回落源(saved=0、is_source_best=True),与 _derive
-        # 「全目标缺菜回落源、不虚报省」同一语义;若排除源强选更贵目标,saved 会变负、倒扣「累计
-        # 发现可省」(get_stats 对 saved_amount_cents 求和不带 >0 过滤)。
-        priced = [p for p in rows if p.get("price") is not None]
-        if priced:
-            best = min(priced, key=lambda p: (p["price"], p.get("display_order") or 0))
     source_price_cents = _yuan_to_cents(src.get("price")) if src else None
     best_price_cents = _yuan_to_cents(best.get("price")) if best else None
     saved_amount_cents = None
@@ -444,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.
 
@@ -471,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(
@@ -490,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(
@@ -503,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(

app/repositories/guide_video.py

@@ -10,7 +10,7 @@ from sqlalchemy import func, select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
-from app.core import limit_policy, media, rewards
+from app.core import media, rewards
 from app.core.config import settings
 from app.models.app_config import AppConfig
 from app.models.guide_video import GuideVideoPlay
@@ -25,9 +25,7 @@ _KEY_BY_SCENE = {
 BIZ_TYPE = "guide_video"
 CIRCLE_COUNT = 10
 PLAN_TTL = timedelta(minutes=10)
-# 0 表示全局暂停播放；白名单页与旧引导视频配置页必须接受同一口径，
-# 否则在白名单页设为 0 后，旧页面连金币/开关等无关字段也无法保存。
-MIN_PLAYS = 0
+MIN_PLAYS = 1
 MAX_PLAYS_LIMIT = 50
 MIN_REWARD_COIN = 10
 REWARD_COIN_LIMIT = 10_000
@@ -97,10 +95,6 @@ def _public_config(cfg: dict[str, Any]) -> dict[str, Any]:
 def get_config(db: Session, scene: str = "coupon") -> dict[str, Any]:
     row = db.get(AppConfig, _config_key(scene))
     cfg = _public_config(_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
 
@@ -159,27 +153,7 @@ def update_config(
     raw["max_plays"] = candidate_plays
     raw["reward_coin"] = candidate_reward
     raw["config_version"] = int(raw.get("config_version") or 0) + 1
-    after = _write(
-        db,
-        raw,
-        scene=scene,
-        admin_id=admin_id,
-        commit=False,
-    )
-    if scene == "coupon" and max_plays is not None:
-        # 兼容仍调用旧专用接口的客户端/脚本，并把统一策略全局值一并更新。
-        limit_policy.set_global_limits(
-            db,
-            {"guide.video.lifetime": candidate_plays},
-            admin_id=admin_id,
-            commit=False,
-        )
-        # set_global_limits 已 flush；即使由 admin 路由统一在外层提交，也要把
-        # 同一事务内的最新统一策略值写进审计 after 快照。
-        after = get_config(db, scene)
-    if commit:
-        db.commit()
-        after = get_config(db, scene)
+    after = _write(db, raw, scene=scene, admin_id=admin_id, commit=commit)
     return before, after
 
 
@@ -217,52 +191,20 @@ def set_video(
     return before, after
 
 
-def used_plays(
-    db: Session,
-    user_id: int,
-    scene: str = "coupon",
-    *,
-    reset_at: datetime | None = None,
-) -> int:
-    stmt = select(func.count()).select_from(GuideVideoPlay).where(
-        GuideVideoPlay.user_id == user_id,
-        GuideVideoPlay.scene == scene,
-        GuideVideoPlay.status != "prepared",
+def used_plays(db: Session, user_id: int, scene: str = "coupon") -> int:
+    return int(
+        db.execute(
+            select(func.count()).select_from(GuideVideoPlay).where(
+                GuideVideoPlay.user_id == user_id,
+                GuideVideoPlay.scene == scene,
+                GuideVideoPlay.status != "prepared",
+            )
+        ).scalar_one()
     )
-    if reset_at is not None:
-        # guide_video_play 使用北京时间 naive 墙钟；白名单重置点是带时区时间。
-        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 _effective_quota(
-    db: Session,
-    user_id: int,
-    scene: str,
-    cfg: dict[str, Any],
-) -> tuple[int | None, datetime | None]:
-    if scene != "coupon":
-        return int(cfg.get("max_plays") or 0), None
-    policy = limit_policy.resolve_for_user(
-        db, "guide.video.lifetime", user_id
-    )
-    return policy.limit, policy.reset_at
-
-
-def _remaining(maximum: int | None, used: int) -> int:
-    # API 字段保持整数兼容；不限时使用足够大的展示值，不参与服务端判定。
-    exposed_maximum = maximum if maximum is not None else 1_000_000
-    return max(0, exposed_maximum - used)
-
-
-def _prepare_miss(
-    scene: str,
-    reason: str,
-    cfg: dict[str, Any],
-    used: int,
-    maximum: int | None,
-) -> dict[str, Any]:
+def _prepare_miss(scene: str, reason: str, cfg: dict[str, Any], used: int) -> dict[str, Any]:
+    maximum = int(cfg.get("max_plays") or 0)
     return {
         "should_play": False,
         "reason": reason,
@@ -276,27 +218,23 @@ def _prepare_miss(
         "reward_coin": int(cfg.get("reward_coin") or 0),
         "reward_per_circle": int(cfg.get("reward_coin") or 0) // CIRCLE_COUNT,
         "seq": used,
-        "remaining": _remaining(maximum, used),
+        "remaining": max(0, maximum - used),
         "expires_at": None,
     }
 
 
 def prepare_play(db: Session, user_id: int, *, scene: str = "coupon") -> dict[str, Any]:
     cfg = get_config(db, scene)
-    maximum, reset_at = _effective_quota(db, user_id, scene, cfg)
-    used = used_plays(db, user_id, scene, reset_at=reset_at)
+    used = used_plays(db, user_id, scene)
     video_url = str(cfg.get("video_url") or "").strip()
     duration = int(cfg.get("duration_ms") or 0)
+    maximum = int(cfg.get("max_plays") or 0)
     if not cfg.get("enabled"):
-        return _prepare_miss(scene, "disabled", cfg, used, maximum)
+        return _prepare_miss(scene, "disabled", cfg, used)
     if not video_url or cfg.get("analysis_status") != "valid" or duration <= 0:
-        return _prepare_miss(
-            scene, "video_unavailable", cfg, used, maximum
-        )
-    if maximum is not None and used >= maximum:
-        return _prepare_miss(
-            scene, "play_limit_reached", cfg, used, maximum
-        )
+        return _prepare_miss(scene, "video_unavailable", cfg, used)
+    if used >= maximum:
+        return _prepare_miss(scene, "play_limit_reached", cfg, used)
 
     now = _now()
     play = GuideVideoPlay(
@@ -330,7 +268,7 @@ def prepare_play(db: Session, user_id: int, *, scene: str = "coupon") -> dict[st
         "reward_coin": play.coin,
         "reward_per_circle": play.coin // CIRCLE_COUNT,
         "seq": used + 1,
-        "remaining": _remaining(maximum, used + 1),
+        "remaining": max(0, maximum - used - 1),
         "expires_at": play.expires_at.isoformat(),
     }
 
@@ -344,12 +282,7 @@ def _find_play(db: Session, user_id: int, token: str) -> GuideVideoPlay | None:
     ).scalar_one_or_none()
 
 
-def _start_out(
-    play: GuideVideoPlay,
-    maximum: int | None,
-    used: int,
-    status: str,
-) -> dict[str, Any]:
+def _start_out(play: GuideVideoPlay, maximum: int, status: str) -> dict[str, Any]:
     assert play.started_at is not None and play.seq is not None and play.video_url
     return {
         "started": True,
@@ -364,7 +297,7 @@ def _start_out(
         "reward_coin": play.coin,
         "reward_per_circle": play.coin // CIRCLE_COUNT,
         "seq": play.seq,
-        "remaining": _remaining(maximum, used),
+        "remaining": max(0, maximum - play.seq),
         "started_at": play.started_at.isoformat(),
     }
 
@@ -374,14 +307,9 @@ def start_play(db: Session, user_id: int, *, play_token: str) -> dict[str, Any]:
     if play is None:
         raise PlayStateError("play_not_found", "播放计划不存在")
     cfg = get_config(db, play.scene)
-    maximum, reset_at = _effective_quota(
-        db, user_id, play.scene, cfg
-    )
-    used = used_plays(
-        db, user_id, play.scene, reset_at=reset_at
-    )
+    maximum = int(cfg["max_plays"])
     if play.status in {"started", "completed"}:
-        return _start_out(play, maximum, used, "already_started")
+        return _start_out(play, maximum, "already_started")
     if play.status != "prepared":
         raise PlayStateError("play_not_found", "播放计划不可用")
     now = _now()
@@ -395,11 +323,10 @@ def start_play(db: Session, user_id: int, *, play_token: str) -> dict[str, Any]:
             "config_changed", "视频配置已变化，请重新获取",
             reprepare_required=True,
         )
-    if maximum is not None and used >= maximum:
+    used = used_plays(db, user_id, play.scene)
+    if used >= maximum:
         raise PlayStateError("play_limit_reached", "播放次数已用完")
-    # seq 在数据库中是账号+场景生命周期唯一值；白名单重置只重置额度，
-    # 不能从 1 重新编号，否则会与历史记录冲突。
-    play.seq = used_plays(db, user_id, play.scene) + 1
+    play.seq = used + 1
     play.status = "started"
     play.started_at = now
     try:
@@ -408,7 +335,7 @@ def start_play(db: Session, user_id: int, *, play_token: str) -> dict[str, Any]:
     except IntegrityError as exc:
         db.rollback()
         raise PlayStateError("play_limit_reached", "并发起播冲突，请重新获取") from exc
-    return _start_out(play, maximum, used + 1, "started")
+    return _start_out(play, maximum, "started")
 
 
 def _coin_balance(db: Session, user_id: int) -> int:

app/repositories/phone_rebind.py

@@ -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))
 
 

app/repositories/risk.py

@@ -9,10 +9,15 @@ 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,
+    RISK_SMS_HOURLY_THRESHOLD_KEY,
+)
 from app.models.app_config import AppConfig
 from app.models.comparison import ComparisonRecord
 from app.models.risk import BehaviorEvent, RiskIncident, SubjectRestriction
+from app.repositories import app_config
 
 CN_TZ = ZoneInfo("Asia/Shanghai")
 
@@ -36,6 +41,7 @@ class RuleSpec:
     code: str
     event_type: str
     subject_type: str
+    threshold_key: str
     window: str
     count_outcomes: tuple[str, ...]
 
@@ -45,6 +51,7 @@ RULES: dict[str, RuleSpec] = {
         code=RULE_SMS_HOURLY,
         event_type=EVENT_SMS_SEND,
         subject_type="device",
+        threshold_key=RISK_SMS_HOURLY_THRESHOLD_KEY,
         window="hour",
         count_outcomes=("success",),
     ),
@@ -52,23 +59,22 @@ RULES: dict[str, RuleSpec] = {
         code=RULE_ONECLICK_DAILY,
         event_type=EVENT_ONECLICK_LOGIN,
         subject_type="device",
+        threshold_key=RISK_ONECLICK_DAILY_THRESHOLD_KEY,
         window="day",
         count_outcomes=("success", "failed"),
     ),
 }
 
-RISK_LIMIT_RULE_CODES: dict[str, str] = {
-    RULE_SMS_HOURLY: "risk.sms.hourly",
-    RULE_ONECLICK_DAILY: "risk.oneclick.daily",
-    RULE_COMPARE_DAILY: "risk.compare.daily",
+RULE_THRESHOLD_KEYS: dict[str, str] = {
+    RULE_SMS_HOURLY: RISK_SMS_HOURLY_THRESHOLD_KEY,
+    RULE_ONECLICK_DAILY: RISK_ONECLICK_DAILY_THRESHOLD_KEY,
+    RULE_COMPARE_DAILY: RISK_COMPARE_DAILY_THRESHOLD_KEY,
 }
 
 
 def get_rule_threshold(db: Session, rule_code: str) -> int:
     """读取规则当前阈值；配置表为空时回退上线前的 5/20/100 默认值。"""
-    return limit_policy.resolve(
-        db, RISK_LIMIT_RULE_CODES[rule_code]
-    ).global_limit
+    return int(app_config.get_value(db, RULE_THRESHOLD_KEYS[rule_code]))
 
 
 def utcnow() -> datetime:
@@ -251,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,
@@ -306,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
@@ -316,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(
@@ -404,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)
@@ -439,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,
@@ -520,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,
@@ -531,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(

app/schemas/compare_record.py

@@ -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):

app/services/notification_events.py

@@ -85,10 +85,6 @@ def _dispatch(
     push_vars: dict[str, str] | None = None,
 ) -> Notification | None:
     """落一条站内消息并向该用户设备直推。返回落库行;去重命中/失败返回 None。"""
-    logger.info(
-        "notification dispatch started user_id=%s type=%s dedup_key=%s",
-        user_id, type_key, dedup_key,
-    )
     try:
         row = notif_repo.create_notification(
             db,
@@ -115,10 +111,6 @@ def _dispatch(
             logger.exception("rollback after notification failure also failed")
         return None
 
-    logger.info(
-        "notification created user_id=%s type=%s notification_id=%s dedup_key=%s",
-        user_id, type_key, row.id, dedup_key,
-    )
     _push_to_user_devices(db, row, push_vars)
     return row
 
@@ -132,95 +124,29 @@ def _push_to_user_devices(db: Session, row: Notification, push_vars: dict[str, s
         extras.update({str(k): str(v) for k, v in (row.extra or {}).items()})
         extras["notificationId"] = str(row.id)
 
-        targets = device_repo.list_push_targets(db, user_id=row.user_id)
-        logger.info(
-            "push targets resolved user_id=%s type=%s notification_id=%s target_count=%s",
-            row.user_id, row.type, row.id, len(targets),
-        )
-        if not targets:
-            logger.warning(
-                "push skipped no targets user_id=%s type=%s notification_id=%s",
-                row.user_id, row.type, row.id,
-            )
-            return
-
-        sent = failed = skipped = data_sent = data_failed = 0
-        for dev in targets:
+        for dev in device_repo.list_push_targets(db, user_id=row.user_id):
             vendor = vendor_push.normalize_vendor(dev.push_vendor)
             if not vendor or vendor not in vendor_push.SUPPORTED_VENDORS:
-                skipped += 1
-                logger.warning(
-                    "push target skipped unsupported vendor user_id=%s type=%s "
-                    "notification_id=%s device_id=%s raw_vendor=%s normalized_vendor=%s",
-                    row.user_id, row.type, row.id, dev.device_id, dev.push_vendor, vendor,
-                )
                 continue
-            missing = vendor_push.missing_settings(vendor)
-            if missing:
-                skipped += 1
+            if vendor_push.missing_settings(vendor):
                 # 本地/测试环境凭据不齐 → 只落站内消息,不发真推送(与 push/vendors 的报缺口径一致)
-                logger.warning(
-                    "push target skipped vendor not configured user_id=%s type=%s "
-                    "notification_id=%s device_id=%s vendor=%s missing_settings=%s",
-                    row.user_id, row.type, row.id, dev.device_id, vendor, missing,
+                logger.info(
+                    "skip push (vendor %s not configured) user_id=%s type=%s",
+                    vendor, row.user_id, row.type,
                 )
                 continue
-            logger.info(
-                "push send started user_id=%s type=%s notification_id=%s "
-                "device_id=%s vendor=%s",
-                row.user_id, row.type, row.id, dev.device_id, vendor,
-            )
             try:
-                response = vendor_push.send_notification(
+                vendor_push.send_notification(
                     vendor, dev.push_token, title=title, body=body, extras=extras
                 )
-                if vendor == "huawei" and row.type in {"feedback_reply", "feedback_reward"}:
-                    # 华为反馈推送联调日志：保留厂商返回码/requestId 等排障信息，
-                    # 请求本身的 token、Authorization 和应用密钥不会进入 response。
-                    logger.info(
-                        "huawei feedback push response user_id=%s type=%s "
-                        "notification_id=%s response=%s",
-                        row.user_id, row.type, row.id, response,
-                    )
                 logger.info(
-                    "push sent user_id=%s type=%s vendor=%s notification_id=%s device_id=%s",
-                    row.user_id, row.type, vendor, row.id, dev.device_id,
-                )
-                sent += 1
-            except vendor_push.VendorPushError as e:
-                failed += 1
-                logger.warning(
-                    "push failed user_id=%s type=%s vendor=%s notification_id=%s "
-                    "device_id=%s error=%s",
-                    row.user_id, row.type, vendor, row.id, dev.device_id, e,
-                )
-            try:
-                data_response = vendor_push.send_data_event(
-                    vendor,
-                    dev.push_token,
-                    event=vendor_push.DATA_EVENT_NOTIFICATION_CREATED,
-                    notification_id=str(row.id),
-                )
-                data_sent += 1
-                logger.info(
-                    "push data event completed user_id=%s type=%s vendor=%s "
-                    "notification_id=%s device_id=%s response=%s",
-                    row.user_id, row.type, vendor, row.id, dev.device_id, data_response,
+                    "push sent user_id=%s type=%s vendor=%s notification_id=%s",
+                    row.user_id, row.type, vendor, row.id,
                 )
             except vendor_push.VendorPushError as e:
-                data_failed += 1
-                # 透传只负责前台铃铛实时刷新，失败不能影响通知栏消息或站内消息。
                 logger.warning(
-                    "push data event failed user_id=%s type=%s vendor=%s notification_id=%s "
-                    "device_id=%s error=%s",
-                    row.user_id, row.type, vendor, row.id, dev.device_id, e,
+                    "push failed user_id=%s type=%s vendor=%s: %s", row.user_id, row.type, vendor, e
                 )
-        logger.info(
-            "push dispatch completed user_id=%s type=%s notification_id=%s targets=%s "
-            "sent=%s failed=%s skipped=%s data_sent=%s data_failed=%s",
-            row.user_id, row.type, row.id, len(targets),
-            sent, failed, skipped, data_sent, data_failed,
-        )
     except Exception:  # noqa: BLE001 — 渲染/查设备等意外失败同样不外抛
         logger.exception("push notification failed user_id=%s type=%s", row.user_id, row.type)
 

tests/test_admin_config.py

@@ -68,12 +68,12 @@ def test_list_config(admin_client: TestClient, token: str) -> None:
     r = admin_client.get("/admin/api/config", headers=_auth(token))
     assert r.status_code == 200, r.text
     items = {i["key"]: i for i in r.json()}
-    # 非 hidden 项照常返回；广告次数上限迁到「白名单」统一配置。
-    assert "signin_rewards" in items and "ad_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"] == [
@@ -124,50 +124,6 @@ def test_update_ad_limit_takes_effect(admin_client: TestClient, token: str) -> N
         db.close()
 
 
-def test_list_config_reads_limit_values_from_global_bundle(
-    admin_client: TestClient,
-    token: str,
-) -> None:
-    changed = admin_client.patch(
-        "/admin/api/config/ad_cooldown_sec",
-        json={"value": 17},
-        headers=_auth(token),
-    )
-    assert changed.status_code == 200, changed.text
-
-    items = {
-        item["key"]: item
-        for item in admin_client.get(
-            "/admin/api/config",
-            headers=_auth(token),
-        ).json()
-    }
-    assert items["ad_cooldown_sec"]["value"] == 17
-    assert items["ad_cooldown_sec"]["overridden"] is True
-
-
-@pytest.mark.parametrize(
-    ("key", "value"),
-    [
-        ("ad_daily_limit", 0),
-        ("ad_daily_limit", 100_001),
-        ("ad_cooldown_sec", 86_401),
-    ],
-)
-def test_update_limit_config_rejects_out_of_range_values(
-    admin_client: TestClient,
-    token: str,
-    key: str,
-    value: int,
-) -> None:
-    response = admin_client.patch(
-        f"/admin/api/config/{key}",
-        json={"value": value},
-        headers=_auth(token),
-    )
-    assert response.status_code == 400, response.text
-
-
 def test_update_bool_config(admin_client: TestClient, token: str) -> None:
     # 提现自动对账开关默认 True
     items = {

tests/test_admin_roles.py

@@ -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",
     }
 
 

tests/test_compare_harvest.py

@@ -140,75 +140,6 @@ def test_harvest_done_failed_derives_fail_reason(client) -> None:
         assert rec.information == "比价过程出错，请稍后重试"  # 原文案仍留存
 
 
-def _done_params_platforms_no_isbest() -> dict:
-    """id 3304 型:done 帧 platforms 全平台「相似替换/仅供参考」(has_dish_diff),pricebot
-    一个 is_best 都没标,但有有价目标(美团 57.8 < 源淘宝闪购 60.8)。"""
-    return {
-        "platforms": [
-            {"role": "source", "platform_id": "eleme", "platform_name": "淘宝闪购",
-             "package": "me.ele", "price": 60.8, "is_best": False, "has_dish_diff": False,
-             "store_name": "窑鸡王", "items": [{"name": "招牌窑鸡 整只-香辣", "qty": 1}]},
-            {"role": "target", "platform_id": "meituan_waimai", "platform_name": "美团外卖",
-             "package": "com.sankuai.meituan.takeoutnew", "price": 57.8, "is_best": False,
-             "has_dish_diff": True},
-            {"role": "target", "platform_id": "jd_waimai_standalone", "platform_name": "京东外卖",
-             "package": "com.jd.waimai", "price": 74.9, "is_best": False, "has_dish_diff": True},
-        ],
-        "information": "美团更便宜（含相似商品替换）",
-    }
-
-
-def test_harvest_done_platforms_no_isbest_falls_back_to_cheapest_target(client) -> None:
-    """回归(id 3304):platforms 全无 is_best(全平台相似替换)但有有价目标 → best 应兜底取
-    有价目标里最低那家,不能让 best_*/saved 整条落 NULL(否则首页价 0.00 / 记录页无最低红框 /
-    省额丢失)。"""
-    tid = _tid()
-    with SessionLocal() as db:
-        crud.harvest_running(db, trace_id=tid, user_id=None)
-        rec, _ = crud.harvest_done(db, trace_id=tid, user_id=None,
-                                   done_params=_done_params_platforms_no_isbest())
-        assert rec.status == "success"
-        assert rec.best_platform_id == "meituan_waimai"   # 有价目标里最低
-        assert rec.best_price_cents == 5780               # 57.8 元
-        assert rec.source_price_cents == 6080             # 源淘宝闪购 60.8
-        assert rec.saved_amount_cents == 300              # 60.8 - 57.8
-        assert rec.is_source_best is False                # 兜底选的是目标,非源
-
-
-def _done_params_no_isbest_source_cheapest() -> dict:
-    """无 is_best 且源本身最便宜:源淘宝闪购 50.0 < 全部 dish-diff 目标(美团 57.8 / 京东 74.9)。
-    此时不能强选更贵的目标当 best(否则 saved 变负、污染「累计发现可省」),应回落源。"""
-    return {
-        "platforms": [
-            {"role": "source", "platform_id": "eleme", "platform_name": "淘宝闪购",
-             "package": "me.ele", "price": 50.0, "is_best": False, "has_dish_diff": False,
-             "store_name": "窑鸡王", "items": [{"name": "招牌窑鸡 整只-香辣", "qty": 1}]},
-            {"role": "target", "platform_id": "meituan_waimai", "platform_name": "美团外卖",
-             "package": "com.sankuai.meituan.takeoutnew", "price": 57.8, "is_best": False,
-             "has_dish_diff": True},
-            {"role": "target", "platform_id": "jd_waimai_standalone", "platform_name": "京东外卖",
-             "package": "com.jd.waimai", "price": 74.9, "is_best": False, "has_dish_diff": True},
-        ],
-        "information": "源平台已是最低（其余为相似替换）",
-    }
-
-
-def test_harvest_done_no_isbest_source_cheapest_falls_back_to_source(client) -> None:
-    """回归:无 is_best 且源最便宜 → best 回落源(与 _derive「全目标缺菜回落源、不虚报省」同语义),
-    saved=0、is_source_best=True,绝不因强选更贵目标而让 saved 变负、倒扣「累计发现可省」。"""
-    tid = _tid()
-    with SessionLocal() as db:
-        crud.harvest_running(db, trace_id=tid, user_id=None)
-        rec, _ = crud.harvest_done(db, trace_id=tid, user_id=None,
-                                   done_params=_done_params_no_isbest_source_cheapest())
-        assert rec.status == "success"
-        assert rec.best_platform_id == "eleme"            # 回落到源(源最便宜)
-        assert rec.best_price_cents == 5000               # 源 50.0
-        assert rec.source_price_cents == 5000
-        assert rec.saved_amount_cents == 0                # 没省到,绝不为负
-        assert rec.is_source_best is True                 # 源就是最便宜
-
-
 def test_harvest_abort_cancels_running(client) -> None:
     tid = _tid()
     with SessionLocal() as db:

tests/test_push_center.py

@@ -7,9 +7,7 @@ mock/真发/变量覆盖/站内联动/设备反查/参数校验。厂商 HTTP
 from __future__ import annotations
 
 import json
-import logging
 
-import httpx
 import pytest
 from fastapi.testclient import TestClient
 
@@ -112,99 +110,6 @@ 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"

tests/test_risk_monitor.py

@@ -10,7 +10,11 @@ from fastapi.testclient import TestClient
 from app.admin.main import admin_app
 from app.admin.repositories import admin_user as admin_repo
 from app.api.v1 import compare as compare_api
-from app.core.config_schema import LIMIT_POLICY_GLOBAL_KEY
+from app.core.config_schema import (
+    RISK_COMPARE_DAILY_THRESHOLD_KEY,
+    RISK_ONECLICK_DAILY_THRESHOLD_KEY,
+    RISK_SMS_HOURLY_THRESHOLD_KEY,
+)
 from app.core.security import issue_token_pair
 from app.db.session import SessionLocal
 from app.models.app_config import AppConfig
@@ -24,7 +28,9 @@ from app.repositories import user as user_repo
 @pytest.fixture(autouse=True)
 def _reset_risk_rule_config():
     keys = (
-        LIMIT_POLICY_GLOBAL_KEY,
+        RISK_SMS_HOURLY_THRESHOLD_KEY,
+        RISK_ONECLICK_DAILY_THRESHOLD_KEY,
+        RISK_COMPARE_DAILY_THRESHOLD_KEY,
         risk_repo.RISK_RESET_BASELINES_KEY,
     )
     with SessionLocal() as db:
@@ -527,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,
             },