Compare commits

..

2 Commits

Author SHA1 Message Date
unknown 84d426631b 优化:拆分历史节省金额批量回填 2026-07-28 20:11:30 +08:00
unknown 42c34fb398 修复:统一大盘平均节省金额口径
成功但未节省的比价记录纳入平均值分母,无法计算的记录保持排除。新增历史数据迁移,按完整购物篮规则排除漏菜平台并回填最优平台与节省金额。
2026-07-28 16:35:37 +08:00
53 changed files with 733 additions and 4066 deletions
@@ -1,26 +0,0 @@
"""merge comparison platforms + fail_reason heads
Revision ID: 6d2309208549
Revises: comparison_platforms_col, comparison_record_fail_reason
Create Date: 2026-07-29 01:48:41.868083
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '6d2309208549'
down_revision: Union[str, Sequence[str], None] = ('comparison_platforms_col', 'comparison_record_fail_reason')
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
pass
def downgrade() -> None:
pass
@@ -1,47 +0,0 @@
"""add platforms unified array column to comparison_record
展示模型统一数组(pricebot done.params.platforms 原样存): 每平台一行、自带
status/is_best/display, 记录页据此直接渲染, 不再靠 comparison_results + 客户端合并 + 前端派生。
纯新增列, 老记录为空 → 前端回退老 comparison_results。
Revision ID: comparison_platforms_col
Revises: user_manual_risk_fields
"""
from __future__ import annotations
from collections.abc import Sequence
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
from alembic import op
revision: str = "comparison_platforms_col"
down_revision: str | None = "user_manual_risk_fields"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
def upgrade() -> None:
# 幂等: 线上为了提前给历史数据补 platforms(2026-07-29), 已手动
# `ALTER TABLE comparison_record ADD COLUMN IF NOT EXISTS platforms jsonb
# NOT NULL DEFAULT '[]'::jsonb`(与本 migration 定义一致)。列已存在时跳过,
# 否则上线 alembic upgrade head 会撞 DuplicateColumn 直接部署失败。
bind = op.get_bind()
cols = {c["name"] for c in sa.inspect(bind).get_columns("comparison_record")}
if "platforms" in cols:
return
with op.batch_alter_table("comparison_record") as batch_op:
batch_op.add_column(
sa.Column(
"platforms", _JSON, nullable=False,
server_default=sa.text("'[]'"),
)
)
def downgrade() -> None:
with op.batch_alter_table("comparison_record") as batch_op:
batch_op.drop_column("platforms")
@@ -1,31 +0,0 @@
"""guide video play count is independent for coupon and comparison
Revision ID: guide_video_scene_unique
Revises: 6d2309208549
"""
from alembic import op
revision = "guide_video_scene_unique"
down_revision = "6d2309208549"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.drop_index("uq_guide_video_play_user_seq", table_name="guide_video_play")
op.create_index(
"uq_guide_video_play_user_scene_seq",
"guide_video_play",
["user_id", "scene", "seq"],
unique=True,
)
def downgrade() -> None:
op.drop_index("uq_guide_video_play_user_scene_seq", table_name="guide_video_play")
op.create_index(
"uq_guide_video_play_user_seq",
"guide_video_play",
["user_id", "seq"],
unique=True,
)
-128
View File
@@ -1,128 +0,0 @@
"""add per-subject limit policy whitelist
Revision ID: limit_policy_whitelist
Revises: guide_video_scene_unique
"""
from __future__ import annotations
from collections.abc import Sequence
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
from alembic import op
revision: str = "limit_policy_whitelist"
down_revision: str | Sequence[str] | None = "guide_video_scene_unique"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
_PAGE = "limit-whitelist"
_DEFAULT_ROLES = ("operator", "tech")
def _role_table() -> sa.TableClause:
return sa.table(
"admin_role",
sa.column("name", sa.String),
sa.column("pages", _JSON),
)
def _add_default_role_permissions() -> None:
"""Grant the page without replacing existing role customisations."""
role = _role_table()
conn = op.get_bind()
rows = conn.execute(
sa.select(role.c.name, role.c.pages).where(
role.c.name.in_(_DEFAULT_ROLES)
)
).all()
for name, pages in rows:
current_pages = list(pages or [])
if _PAGE not in current_pages:
conn.execute(
role.update()
.where(role.c.name == name)
.values(pages=[*current_pages, _PAGE])
)
def _remove_default_role_permissions() -> None:
role = _role_table()
conn = op.get_bind()
rows = conn.execute(
sa.select(role.c.name, role.c.pages).where(
role.c.name.in_(_DEFAULT_ROLES)
)
).all()
for name, pages in rows:
current_pages = list(pages or [])
if _PAGE in current_pages:
conn.execute(
role.update()
.where(role.c.name == name)
.values(
pages=[page for page in current_pages if page != _PAGE]
)
)
def upgrade() -> None:
op.create_table(
"limit_policy_override",
sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
sa.Column("subject_type", sa.String(length=16), nullable=False),
sa.Column("subject_value", sa.String(length=128), nullable=False),
sa.Column("rule_code", sa.String(length=64), nullable=False),
sa.Column("mode", sa.String(length=24), nullable=False),
sa.Column("limit_value", sa.Integer(), nullable=True),
sa.Column(
"enabled", sa.Boolean(), server_default=sa.true(), nullable=False
),
sa.Column("starts_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("reset_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("reason", sa.String(length=256), nullable=True),
sa.Column("created_by_admin_id", sa.Integer(), nullable=True),
sa.Column(
"created_at",
sa.DateTime(timezone=True),
server_default=sa.func.now(),
nullable=False,
),
sa.Column(
"updated_at",
sa.DateTime(timezone=True),
server_default=sa.func.now(),
nullable=False,
),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint(
"subject_type",
"subject_value",
"rule_code",
name="uq_limit_policy_subject_rule",
),
)
op.create_index(
"ix_limit_policy_lookup",
"limit_policy_override",
["subject_type", "subject_value", "rule_code", "enabled"],
unique=False,
)
op.create_index(
"ix_limit_policy_expires",
"limit_policy_override",
["expires_at"],
unique=False,
)
_add_default_role_permissions()
def downgrade() -> None:
_remove_default_role_permissions()
op.drop_index("ix_limit_policy_expires", table_name="limit_policy_override")
op.drop_index("ix_limit_policy_lookup", table_name="limit_policy_override")
op.drop_table("limit_policy_override")
-2
View File
@@ -31,7 +31,6 @@ from app.admin.routers.feedback import router as feedback_router
from app.admin.routers.feedback_qr import router as feedback_qr_router
from app.admin.routers.guide_video import router as guide_video_router
from app.admin.routers.huawei_review import router as huawei_review_router
from app.admin.routers.limit_whitelist import router as limit_whitelist_router
from app.admin.routers.onboarding import router as onboarding_router
from app.admin.routers.ops_marquee_seed import router as ops_marquee_seed_router
from app.admin.routers.ops_stat_config import router as ops_stat_config_router
@@ -104,7 +103,6 @@ admin_app.include_router(wallet_router)
admin_app.include_router(withdraw_router)
admin_app.include_router(price_report_router)
admin_app.include_router(risk_monitor_router)
admin_app.include_router(limit_whitelist_router)
admin_app.include_router(feedback_router)
admin_app.include_router(event_logs_router)
admin_app.include_router(analytics_health_router)
+1 -3
View File
@@ -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",
]},
]
+5 -5
View File
@@ -43,7 +43,7 @@ _KNOWN_PROD_BUSINESS_CODE_IDS = frozenset({"104098712", "104099389"})
_TEST_BUSINESS_CODE_IDS = frozenset({"104127529", "104127626", "104137445"})
def business_code_ids(db: Session, app_env: str | None) -> set[str]:
def _business_code_ids(db: Session, app_env: str | None) -> set[str]:
"""返回指定应用环境下可用于业务收益对账的 GroMore 聚合代码位。"""
prod_config = app_config.get_ad_config(db)
prod_ids = set(_KNOWN_PROD_BUSINESS_CODE_IDS) | {
@@ -320,10 +320,10 @@ def ad_revenue_report(
# 业务口径仅保留正式配置/测试业务链路实际使用的代码位。穿山甲“全量”还包含广告测试
# demo、插屏等没有客户端收益上报的曝光,两边直接比较会天然产生假差额。
business_ids: set[str] | None = None
business_code_ids: set[str] | None = None
if revenue_scope == "business":
business_ids = business_code_ids(db, app_env)
events = [e for e in events if e.get("our_code_id") in business_ids]
business_code_ids = _business_code_ids(db, app_env)
events = [e for e in events if e.get("our_code_id") in business_code_ids]
# 排序:time=按时间倒序(新→旧);ecpm=按 eCPM 数值倒序(eCPM 原值是字符串「分」,转数值排;
# 纯发奖行用其发奖采用的 eCPM,缺失/非法计 0 排末尾)。
@@ -381,7 +381,7 @@ def ad_revenue_report(
date_from=date_from,
date_to=date_to,
app_env=app_env,
our_code_ids=business_ids,
our_code_ids=business_code_ids,
)
if pangle_aggs:
by_date = {a["date"]: a for a in pangle_aggs}
-455
View File
@@ -1,455 +0,0 @@
"""CRUD and presentation helpers for limit policy overrides."""
from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import String, cast, func, or_, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from app.core import limit_policy
from app.models.app_config import AppConfig
from app.models.comparison import ComparisonRecord
from app.models.limit_policy import LimitPolicyOverride
from app.models.risk import BehaviorEvent
from app.models.user import User
from app.repositories import app_config
from app.repositories import risk as risk_repo
class DuplicateOverrideError(Exception):
pass
_EVENT_SOURCE_LABELS = {
risk_repo.EVENT_SMS_SEND: "短信验证码",
risk_repo.EVENT_SMS_LOGIN: "短信登录",
risk_repo.EVENT_ONECLICK_LOGIN: "一键登录",
}
def _matching_users(keyword: str):
pattern = f"%{keyword}%"
return select(User.id).where(
or_(
cast(User.id, String).ilike(pattern),
User.username.ilike(pattern),
User.phone.ilike(pattern),
User.nickname.ilike(pattern),
)
)
def _user_maps(
db: Session, *, user_ids: set[int], phones: set[str]
) -> tuple[dict[int, User], dict[str, User]]:
conditions = []
if user_ids:
conditions.append(User.id.in_(user_ids))
if phones:
conditions.append(User.phone.in_(phones))
if not conditions:
return {}, {}
users = list(db.scalars(select(User).where(or_(*conditions))).all())
return {user.id: user for user in users}, {user.phone: user for user in users}
def _behavior_device_candidates(
db: Session, *, keyword: str | None, limit: int
) -> list[dict]:
conditions = [
BehaviorEvent.subject_type == limit_policy.SUBJECT_DEVICE,
BehaviorEvent.subject_id != "",
BehaviorEvent.subject_id.not_like(
f"{limit_policy.LEGACY_IP_DEVICE_PREFIX}%"
),
]
if keyword:
value = keyword.strip()
pattern = f"%{value}%"
matched_users = _matching_users(value)
conditions.append(
or_(
BehaviorEvent.subject_id.ilike(pattern),
BehaviorEvent.device_id.ilike(pattern),
BehaviorEvent.device_model.ilike(pattern),
BehaviorEvent.phone.ilike(pattern),
cast(BehaviorEvent.user_id, String).ilike(pattern),
BehaviorEvent.user_id.in_(matched_users),
BehaviorEvent.phone.in_(
select(User.phone).where(User.id.in_(_matching_users(value)))
),
)
)
ranked = (
select(
BehaviorEvent.subject_id.label("device_id"),
BehaviorEvent.event_type,
BehaviorEvent.user_id,
BehaviorEvent.phone,
BehaviorEvent.device_model,
BehaviorEvent.occurred_at.label("last_active_at"),
func.row_number()
.over(
partition_by=BehaviorEvent.subject_id,
order_by=(
BehaviorEvent.occurred_at.desc(),
BehaviorEvent.id.desc(),
),
)
.label("rank"),
)
.where(*conditions)
.subquery()
)
rows = db.execute(
select(ranked)
.where(ranked.c.rank == 1)
.order_by(ranked.c.last_active_at.desc(), ranked.c.device_id.asc())
.limit(limit)
).all()
by_id, by_phone = _user_maps(
db,
user_ids={row.user_id for row in rows if row.user_id is not None},
phones={row.phone for row in rows if row.phone},
)
result = []
for row in rows:
user = by_id.get(row.user_id) if row.user_id is not None else None
user = user or (by_phone.get(row.phone) if row.phone else None)
result.append(
{
"device_id": row.device_id,
"source": row.event_type,
"source_label": _EVENT_SOURCE_LABELS.get(
row.event_type, "登录/鉴权流水"
),
"user_id": user.id if user else row.user_id,
"username": user.username if user else None,
"phone": user.phone if user else row.phone,
"nickname": user.nickname if user else None,
"device_model": row.device_model,
"last_active_at": row.last_active_at,
}
)
return result
def _comparison_device_candidates(
db: Session, *, keyword: str | None, limit: int
) -> list[dict]:
conditions = [
ComparisonRecord.device_id.is_not(None),
ComparisonRecord.device_id != "",
]
if keyword:
value = keyword.strip()
pattern = f"%{value}%"
conditions.append(
or_(
ComparisonRecord.device_id.ilike(pattern),
ComparisonRecord.device_model.ilike(pattern),
cast(ComparisonRecord.user_id, String).ilike(pattern),
ComparisonRecord.user_id.in_(_matching_users(value)),
)
)
ranked = (
select(
ComparisonRecord.device_id,
ComparisonRecord.user_id,
ComparisonRecord.device_model,
ComparisonRecord.created_at.label("last_active_at"),
func.row_number()
.over(
partition_by=ComparisonRecord.device_id,
order_by=(
ComparisonRecord.created_at.desc(),
ComparisonRecord.id.desc(),
),
)
.label("rank"),
)
.where(*conditions)
.subquery()
)
rows = db.execute(
select(ranked)
.where(ranked.c.rank == 1)
.order_by(ranked.c.last_active_at.desc(), ranked.c.device_id.asc())
.limit(limit)
).all()
by_id, _ = _user_maps(
db,
user_ids={row.user_id for row in rows if row.user_id is not None},
phones=set(),
)
return [
{
"device_id": row.device_id,
"source": "comparison_record",
"source_label": "比价记录",
"user_id": user.id if user else row.user_id,
"username": user.username if user else None,
"phone": user.phone if user else None,
"nickname": user.nickname if user else None,
"device_model": row.device_model,
"last_active_at": row.last_active_at,
}
for row in rows
for user in [by_id.get(row.user_id) if row.user_id is not None else None]
]
def list_device_candidates(
db: Session,
*,
rule_code: str,
keyword: str | None = None,
limit: int = 30,
) -> list[dict]:
rule = limit_policy.get_rule(rule_code)
if limit_policy.SUBJECT_DEVICE not in rule.subject_types:
raise ValueError("当前限制项不支持设备白名单")
if limit_policy.device_source_scope(rule_code) == "comparison":
return _comparison_device_candidates(db, keyword=keyword, limit=limit)
return _behavior_device_candidates(db, keyword=keyword, limit=limit)
def _aware(value: datetime | None) -> datetime | None:
if value is None:
return None
return value.replace(tzinfo=UTC) if value.tzinfo is None else value
def status_of(row: LimitPolicyOverride, *, now: datetime | None = None) -> str:
now = _aware(now) or datetime.now(UTC)
if not row.enabled:
return "disabled"
if _aware(row.starts_at) and _aware(row.starts_at) > now:
return "scheduled"
if _aware(row.expires_at) and _aware(row.expires_at) <= now:
return "expired"
return "active"
def to_dict(db: Session, row: LimitPolicyOverride) -> dict:
rule = limit_policy.get_rule(row.rule_code)
effective = limit_policy.resolve(
db,
row.rule_code,
phone=row.subject_value if row.subject_type == "phone" else None,
device=row.subject_value if row.subject_type == "device" else None,
)
return {
"id": row.id,
"subject_type": row.subject_type,
"subject_value": row.subject_value,
"rule_code": row.rule_code,
"rule_label": rule.label,
"rule_group": rule.group,
"mode": row.mode,
"limit_value": row.limit_value,
"global_limit": effective.global_limit,
"effective_limit": effective.limit,
"enabled": row.enabled,
"starts_at": row.starts_at,
"expires_at": row.expires_at,
"reset_at": row.reset_at,
"reason": row.reason,
"status": status_of(row),
"created_by_admin_id": row.created_by_admin_id,
"created_at": row.created_at,
"updated_at": row.updated_at,
}
def list_rows(
db: Session,
*,
subject_type: str | None = None,
keyword: str | None = None,
rule_code: str | None = None,
offset: int = 0,
limit: int = 100,
) -> tuple[list[LimitPolicyOverride], int]:
stmt = select(LimitPolicyOverride)
count_stmt = select(func.count(LimitPolicyOverride.id))
conditions = []
if subject_type:
conditions.append(LimitPolicyOverride.subject_type == subject_type)
if keyword:
conditions.append(
LimitPolicyOverride.subject_value.ilike(f"%{keyword.strip()}%")
)
if rule_code:
conditions.append(LimitPolicyOverride.rule_code == rule_code)
if conditions:
stmt = stmt.where(*conditions)
count_stmt = count_stmt.where(*conditions)
total = int(db.scalar(count_stmt) or 0)
rows = list(
db.execute(
stmt.order_by(
LimitPolicyOverride.created_at.desc(),
LimitPolicyOverride.id.desc(),
)
.offset(offset)
.limit(limit)
).scalars()
)
return rows, total
def create(
db: Session,
*,
subject_type: str,
subject_value: str,
rule_code: str,
mode: str,
limit_value: int | None,
enabled: bool,
starts_at: datetime | None,
expires_at: datetime | None,
reason: str,
admin_id: int,
reactivate_inactive: bool = False,
) -> LimitPolicyOverride:
rule = limit_policy.get_rule(rule_code)
subject_value = limit_policy.validate_whitelist_subject(
subject_type, subject_value
)
limit_policy.validate_override(
rule,
subject_type=subject_type,
mode=mode,
limit_value=limit_value,
starts_at=starts_at,
expires_at=expires_at,
)
existing = db.execute(
select(LimitPolicyOverride).where(
LimitPolicyOverride.subject_type == subject_type,
LimitPolicyOverride.subject_value == subject_value,
LimitPolicyOverride.rule_code == rule_code,
)
).scalar_one_or_none()
if existing is not None:
if (
not reactivate_inactive
or status_of(existing) not in {"disabled", "expired"}
):
raise DuplicateOverrideError
# “恢复全局”与自然过期都会保留原行供审计。再次加入白名单时
# 复用该行,避免唯一键让批量新增永久失败;完整变更仍写 admin 审计日志。
existing.mode = mode
existing.limit_value = (
limit_value if mode == limit_policy.MODE_OVERRIDE else None
)
existing.enabled = enabled
existing.starts_at = starts_at
existing.expires_at = expires_at
existing.reset_at = None
existing.reason = reason.strip() or None
db.flush()
return existing
row = LimitPolicyOverride(
subject_type=subject_type,
subject_value=subject_value,
rule_code=rule_code,
mode=mode,
limit_value=limit_value if mode == limit_policy.MODE_OVERRIDE else None,
enabled=enabled,
starts_at=starts_at,
expires_at=expires_at,
reason=reason.strip() or None,
created_by_admin_id=admin_id,
)
db.add(row)
try:
db.flush()
except IntegrityError as exc:
db.rollback()
raise DuplicateOverrideError from exc
return row
def update(
db: Session,
row: LimitPolicyOverride,
*,
enabled: bool | None,
starts_at: datetime | None,
expires_at: datetime | None,
reason: str | None,
fields_set: set[str],
) -> LimitPolicyOverride:
new_starts = starts_at if "starts_at" in fields_set else row.starts_at
new_expires = expires_at if "expires_at" in fields_set else row.expires_at
new_enabled = enabled if "enabled" in fields_set and enabled is not None else row.enabled
rule = limit_policy.get_rule(row.rule_code)
if row.mode == limit_policy.MODE_OVERRIDE and new_enabled:
raise ValueError("历史覆盖配置只能停用、恢复全局或删除")
if new_enabled:
limit_policy.validate_override(
rule,
subject_type=row.subject_type,
mode=row.mode,
limit_value=None,
starts_at=new_starts,
expires_at=new_expires,
)
if "enabled" in fields_set and enabled is not None:
row.enabled = enabled
if "starts_at" in fields_set:
row.starts_at = starts_at
if "expires_at" in fields_set:
row.expires_at = expires_at
if "reason" in fields_set:
row.reason = reason.strip() if reason else None
db.flush()
return row
def update_global_limit(
db: Session, rule_code: str, value: int, *, admin_id: int
) -> tuple[int, int]:
rule = limit_policy.get_rule(rule_code)
if not rule.min_value <= value <= rule.max_value:
raise ValueError(
f"限制值必须在 {rule.min_value}{rule.max_value} 之间"
)
before = limit_policy.resolve(db, rule_code).global_limit
app_config.set_value(db, rule.config_key, value, admin_id=admin_id, commit=False)
# 引导视频仍有一个旧的专用配置页。同步其结构化配置,保证两个入口读取
# 同一数值;业务判定仍统一走 limit_policy。
if rule.legacy_config_key and rule.legacy_json_field:
legacy = db.get(AppConfig, rule.legacy_config_key)
legacy_value = (
dict(legacy.value)
if legacy is not None and isinstance(legacy.value, dict)
else {}
)
legacy_value[rule.legacy_json_field] = value
if legacy is None:
legacy = AppConfig(
key=rule.legacy_config_key,
value=legacy_value,
updated_by_admin_id=admin_id,
)
db.add(legacy)
else:
legacy.value = legacy_value
legacy.updated_by_admin_id = admin_id
db.flush()
return before, value
def restore_global(row: LimitPolicyOverride) -> LimitPolicyOverride:
"""Stop applying the exception while retaining its history for audit."""
row.enabled = False
row.reset_at = None
return row
+14 -57
View File
@@ -14,7 +14,6 @@ from sqlalchemy.orm import Session
from app.core import rewards
from app.core.config import settings
from app.models.ad_ecpm import AdEcpmRecord
from app.models.ad_feed_reward import AdFeedRewardRecord
from app.models.ad_reward import AdRewardRecord
from app.models.admin import AdminAuditLog
@@ -1258,15 +1257,11 @@ def user_reward_stats(
date_from: datetime | None = None,
date_to: datetime | None = None,
withdraw_source: str | None = None,
app_env: str | None = None,
revenue_scope: str = "all",
feed_scene: str | None = None,
) -> dict:
"""提现详情「用户统计区」10 项。窗口作用于除「现金余额」外的所有项(余额是当前快照)。
口径:激励视频/信息流奖励数量只统计 granted;数量——视频按条数、信息流按份数(unit_count 累加)
平均 Draw eCPM 与广告收益报表一致:基于 ad_ecpm_record 的全部 draw/feed 展示记录计算,
不以是否发奖为筛选条件。各「提现」= 该来源累计金币折现。
口径:激励视频/信息流只统计 granted;数量——视频按条数、信息流按份数(unit_count 累加);
平均 eCPM 用原始分值(分/千次)按记录取算术平均;各「提现」= 该来源累计金币折现。
传统任务 = 窗口内正向金币中,排除广告(reward_video/feed_ad_reward)与人工调整后的折现。
"""
withdraw_source_conds = (
@@ -1301,68 +1296,30 @@ def user_reward_stats(
# 只投影本统计实际使用的列。避免滚动发布或旧本地库尚未补齐无关新列时,
# SQLAlchemy 因 select(ORM) 自动展开整表字段而让提现详情整体 500。
business_ids: set[str] | None = None
if revenue_scope == "business":
# 与广告收益报表共用正式/测试业务代码位集合,避免两个页面随配置切换后再次漂移。
from app.admin.repositories.ad_revenue import business_code_ids
business_ids = business_code_ids(db, app_env)
rv_conds = [
AdRewardRecord.user_id == user_id,
AdRewardRecord.reward_scene == "reward_video",
AdRewardRecord.status == "granted",
*_window_conds(AdRewardRecord.created_at, date_from, date_to),
]
if app_env is not None:
rv_conds.append(AdRewardRecord.app_env == app_env)
if business_ids is not None:
rv_conds.append(AdRewardRecord.our_code_id.in_(business_ids))
rv = db.execute(
select(AdRewardRecord.ecpm_raw, AdRewardRecord.coin).where(
*rv_conds,
AdRewardRecord.user_id == user_id,
AdRewardRecord.reward_scene == "reward_video",
AdRewardRecord.status == "granted",
*_window_conds(AdRewardRecord.created_at, date_from, date_to),
)
).all()
rv_ecpms = [rewards.parse_ecpm_fen(r.ecpm_raw) for r in rv if r.ecpm_raw]
rv_coins = sum(r.coin for r in rv)
feed_reward_conds = [
AdFeedRewardRecord.user_id == user_id,
AdFeedRewardRecord.status == "granted",
*_window_conds(AdFeedRewardRecord.created_at, date_from, date_to),
]
if app_env is not None:
feed_reward_conds.append(AdFeedRewardRecord.app_env == app_env)
if feed_scene is not None:
feed_reward_conds.append(AdFeedRewardRecord.feed_scene == feed_scene)
if business_ids is not None:
feed_reward_conds.append(AdFeedRewardRecord.our_code_id.in_(business_ids))
feed_rewards = db.execute(
feed = db.execute(
select(
AdFeedRewardRecord.unit_count,
AdFeedRewardRecord.ecpm_raw,
AdFeedRewardRecord.coin,
).where(
*feed_reward_conds,
AdFeedRewardRecord.user_id == user_id,
AdFeedRewardRecord.status == "granted",
*_window_conds(AdFeedRewardRecord.created_at, date_from, date_to),
)
).all()
feed_coins = sum(f.coin for f in feed_rewards)
feed_impression_conds = [
AdEcpmRecord.user_id == user_id,
AdEcpmRecord.ad_type.in_(("draw", "feed")),
*_window_conds(AdEcpmRecord.created_at, date_from, date_to),
]
if app_env is not None:
feed_impression_conds.append(AdEcpmRecord.app_env == app_env)
if feed_scene is not None:
feed_impression_conds.append(AdEcpmRecord.feed_scene == feed_scene)
if business_ids is not None:
feed_impression_conds.append(AdEcpmRecord.our_code_id.in_(business_ids))
feed_impressions = db.execute(
select(AdEcpmRecord.ecpm_raw).where(*feed_impression_conds)
).all()
# 与 ad_revenue.category_stats 相同:每次展示权重相同,非法原值按 parse_ecpm_fen 记 0。
feed_ecpms = [rewards.parse_ecpm_fen(row.ecpm_raw) for row in feed_impressions]
feed_ecpms = [rewards.parse_ecpm_fen(f.ecpm_raw) for f in feed if f.ecpm_raw]
feed_coins = sum(f.coin for f in feed)
trad_coins = db.execute(
select(func.coalesce(func.sum(CoinTransaction.amount), 0)).where(
@@ -1381,7 +1338,7 @@ def user_reward_stats(
"reward_video_count": len(rv),
"reward_video_avg_ecpm": round(sum(rv_ecpms) / len(rv_ecpms), 2) if rv_ecpms else 0.0,
"reward_video_cash_cents": _coins_to_cents(rv_coins),
"feed_count": int(sum(f.unit_count for f in feed_rewards)),
"feed_count": int(sum(f.unit_count for f in feed)),
"feed_avg_ecpm": round(sum(feed_ecpms) / len(feed_ecpms), 2) if feed_ecpms else 0.0,
"feed_cash_cents": _coins_to_cents(feed_coins),
}
+17 -7
View File
@@ -298,21 +298,31 @@ def dashboard_overview(
if period_comparison_success_denominator > 0
else None
)
period_saved_positive_count = _count(
# “平均节省金额”表示一次可计算的成功比价平均能省多少钱:
# 原平台已经最便宜时 saved_amount_cents=0,也必须进入分母;否则只统计
# “确实省到钱”的记录会系统性抬高大盘指标。源价/最优价缺失导致的 NULL
# 无法判断实际节省额,仍排除在分母之外。
period_saved_calculable_count = _count(
ComparisonRecord,
*period_comparison_conds,
ComparisonRecord.status == "success",
ComparisonRecord.saved_amount_cents > 0,
ComparisonRecord.saved_amount_cents.is_not(None),
)
period_saved_positive_sum = _sum(
ComparisonRecord.saved_amount_cents,
period_saved_sum = _sum(
case(
(
ComparisonRecord.saved_amount_cents > 0,
ComparisonRecord.saved_amount_cents,
),
else_=0,
),
*period_comparison_conds,
ComparisonRecord.status == "success",
ComparisonRecord.saved_amount_cents > 0,
ComparisonRecord.saved_amount_cents.is_not(None),
)
period_avg_saved_cents = (
round(period_saved_positive_sum / period_saved_positive_count)
if period_saved_positive_count
round(period_saved_sum / period_saved_calculable_count)
if period_saved_calculable_count
else None
)
period_avg_duration_ms = db.execute(
+1 -22
View File
@@ -12,11 +12,7 @@ from fastapi import APIRouter, Depends, HTTPException, Request
from app.admin.audit import write_audit
from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
from app.admin.schemas.config import ConfigItemOut, ConfigUpdateRequest
from app.core.config_schema import (
AD_FEED_DAILY_LIMIT_KEY,
AD_REWARD_VIDEO_DAILY_LIMIT_KEY,
CONFIG_DEFS,
)
from app.core.config_schema import CONFIG_DEFS
from app.core.rewards import SIGNIN_CYCLE_LEN
from app.models.admin import AdminUser
from app.repositories import app_config
@@ -92,23 +88,6 @@ def update_config(
before = app_config.get_value(db, key)
app_config.set_value(db, key, body.value, admin_id=admin.id, commit=False)
if key == "ad_daily_limit":
# 旧系统配置接口过去只有一个广告日上限。仍有人直接调用时,同时同步
# 新的激励视频/Draw 两项,避免旧入口写入后业务实际值不变。
app_config.set_value(
db,
AD_REWARD_VIDEO_DAILY_LIMIT_KEY,
body.value,
admin_id=admin.id,
commit=False,
)
app_config.set_value(
db,
AD_FEED_DAILY_LIMIT_KEY,
body.value,
admin_id=admin.id,
commit=False,
)
write_audit(
db, admin, action="config.set", target_type="config", target_id=key,
detail={"before": before, "after": body.value}, ip=get_client_ip(request), commit=False,
+14 -35
View File
@@ -9,7 +9,7 @@ client_max_body_size,见 shaguabijia-admin-web/deploy/nginx/admin.shaguabijia.co
"""
from __future__ import annotations
from typing import Annotated, Literal
from typing import Annotated
from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile
@@ -26,21 +26,15 @@ router = APIRouter(
dependencies=[Depends(get_current_admin)],
)
GuideScene = Literal["coupon", "comparison"]
def _out(db: AdminDb, scene: GuideScene) -> GuideVideoConfigOut:
def _out(db: AdminDb) -> GuideVideoConfigOut:
"""配置 + 播放统计合成响应(四个写接口都以最新状态返回,前端一次同步到位)。"""
return GuideVideoConfigOut(
scene=scene,
**guide_video.get_config(db, scene),
**guide_video.play_stats(db, scene),
)
return GuideVideoConfigOut(**guide_video.get_config(db), **guide_video.play_stats(db))
@router.get("", response_model=GuideVideoConfigOut, summary="新手引导视频配置(领券浮层)")
def get_config(db: AdminDb, scene: GuideScene = "coupon") -> GuideVideoConfigOut:
return _out(db, scene)
def get_config(db: AdminDb) -> GuideVideoConfigOut:
return _out(db)
@router.patch("", response_model=GuideVideoConfigOut, summary="改开关/次数/金币(带审计)")
@@ -49,24 +43,21 @@ def update_config(
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
scene: GuideScene = "coupon",
) -> GuideVideoConfigOut:
before, after = guide_video.update_config(
db,
enabled=body.enabled,
max_plays=body.max_plays,
reward_coin=body.reward_coin,
scene=scene,
admin_id=admin.id,
commit=False,
)
write_audit(
db, admin, action="guide_video.update", target_type="guide_video", target_id=None,
detail={"scene": scene, "before": before, "after": after},
ip=get_client_ip(request), commit=False,
detail={"before": before, "after": after}, ip=get_client_ip(request), commit=False,
)
db.commit()
return _out(db, scene)
return _out(db)
@router.post("/video", response_model=GuideVideoConfigOut, summary="上传新手引导视频(MP4,带审计)")
@@ -74,31 +65,23 @@ async def upload_video(
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
file: Annotated[UploadFile, File()],
scene: GuideScene = "coupon",
file: UploadFile = File(...),
) -> GuideVideoConfigOut:
data = await file.read()
try:
url = media.save_guide_video(data)
except media.MediaError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
before, after = guide_video.set_video(
db, url, scene=scene, admin_id=admin.id, commit=False
)
before, after = guide_video.set_video(db, url, admin_id=admin.id, commit=False)
write_audit(
db, admin, action="guide_video.set_video", target_type="guide_video", target_id=None,
detail={
"scene": scene,
"before": before.get("video_url"),
"after": url,
"bytes": len(data),
},
detail={"before": before.get("video_url"), "after": url, "bytes": len(data)},
ip=get_client_ip(request), commit=False,
)
db.commit()
# 提交成功后再删旧片,避免新片没落库就把旧片丢了
media.delete_guide_video(before.get("video_url"))
return _out(db, scene)
return _out(db)
@router.delete("/video", response_model=GuideVideoConfigOut, summary="移除新手引导视频(带审计)")
@@ -106,17 +89,13 @@ def delete_video(
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
scene: GuideScene = "coupon",
) -> GuideVideoConfigOut:
"""移除后 /guide-video/start 一律返回 should_play=false,领券浮层回到「只放广告」。"""
before, after = guide_video.set_video(
db, None, scene=scene, admin_id=admin.id, commit=False
)
before, after = guide_video.set_video(db, None, admin_id=admin.id, commit=False)
write_audit(
db, admin, action="guide_video.delete_video", target_type="guide_video", target_id=None,
detail={"scene": scene, "before": before.get("video_url")},
ip=get_client_ip(request), commit=False,
detail={"before": before.get("video_url")}, ip=get_client_ip(request), commit=False,
)
db.commit()
media.delete_guide_video(before.get("video_url"))
return _out(db, scene)
return _out(db)
-361
View File
@@ -1,361 +0,0 @@
"""Unified limit rules and per-phone/device whitelist overrides."""
from __future__ import annotations
from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
from app.admin.audit import write_audit
from app.admin.deps import AdminDb, CurrentAdmin, get_client_ip, require_page
from app.admin.repositories import limit_whitelist as repo
from app.admin.schemas.limit_whitelist import (
DeviceCandidateOut,
GlobalLimitUpdate,
LimitOverrideBulkWrite,
LimitOverrideList,
LimitOverrideOut,
LimitOverridePatch,
LimitOverrideWrite,
LimitRuleOut,
)
from app.core import limit_policy
from app.models.limit_policy import LimitPolicyOverride
from app.repositories import risk as risk_repo
router = APIRouter(
prefix="/admin/api/limit-whitelist",
tags=["admin-limit-whitelist"],
dependencies=[Depends(require_page("limit-whitelist"))],
)
def _reconcile_risk_rule(db, rule_code: str) -> None:
now = risk_repo.utcnow()
if rule_code == "risk.sms.hourly":
risk_repo.reconcile_behavior_rule(
db,
rule_code=risk_repo.RULE_SMS_HOURLY,
at=now,
commit=False,
)
elif rule_code == "risk.oneclick.daily":
risk_repo.reconcile_behavior_rule(
db,
rule_code=risk_repo.RULE_ONECLICK_DAILY,
at=now,
commit=False,
)
elif rule_code == "risk.compare.daily":
risk_repo.reconcile_compare_rule(db, at=now, commit=False)
def _row_or_404(db, override_id: int) -> LimitPolicyOverride:
row = db.get(LimitPolicyOverride, override_id)
if row is None:
raise HTTPException(status_code=404, detail="白名单配置不存在")
return row
def _out(db, row: LimitPolicyOverride) -> LimitOverrideOut:
return LimitOverrideOut(**repo.to_dict(db, row))
def _audit_payload(value: LimitOverrideOut) -> dict:
return value.model_dump(mode="json")
@router.get("/rules", response_model=list[LimitRuleOut], summary="读取所有限制规则")
def list_rules(db: AdminDb) -> list[LimitRuleOut]:
return [LimitRuleOut(**item) for item in limit_policy.rule_catalog(db)]
@router.patch(
"/rules/{rule_code}",
response_model=LimitRuleOut,
summary="修改规则的全局限制值",
)
def update_rule(
rule_code: str,
body: GlobalLimitUpdate,
request: Request,
admin: CurrentAdmin,
db: AdminDb,
) -> LimitRuleOut:
try:
before, after = repo.update_global_limit(
db, rule_code, body.value, admin_id=admin.id
)
except KeyError as exc:
raise HTTPException(status_code=404, detail="未知限制规则") from exc
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
_reconcile_risk_rule(db, rule_code)
write_audit(
db,
admin,
action="limit.rule.update",
target_type="limit_rule",
target_id=rule_code,
detail={"before": before, "after": after},
ip=get_client_ip(request),
commit=False,
)
db.commit()
item = next(
item for item in limit_policy.rule_catalog(db) if item["code"] == rule_code
)
return LimitRuleOut(**item)
@router.get(
"/device-candidates",
response_model=list[DeviceCandidateOut],
summary="按限制项搜索可加入白名单的设备",
)
def list_device_candidates(
db: AdminDb,
rule_code: str = Query(..., min_length=1, max_length=64),
keyword: str | None = Query(None, max_length=128),
limit: int = Query(30, ge=1, le=50),
) -> list[DeviceCandidateOut]:
try:
rows = repo.list_device_candidates(
db,
rule_code=rule_code,
keyword=keyword,
limit=limit,
)
except KeyError as exc:
raise HTTPException(status_code=404, detail="未知限制规则") from exc
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
return [DeviceCandidateOut(**row) for row in rows]
@router.get("", response_model=LimitOverrideList, summary="读取白名单配置")
def list_overrides(
db: AdminDb,
subject_type: str | None = Query(None, pattern="^(phone|device)$"),
keyword: str | None = Query(None, max_length=128),
rule_code: str | None = Query(None, max_length=64),
offset: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=500),
) -> LimitOverrideList:
rows, total = repo.list_rows(
db,
subject_type=subject_type,
keyword=keyword,
rule_code=rule_code,
offset=offset,
limit=limit,
)
return LimitOverrideList(items=[_out(db, row) for row in rows], total=total)
@router.post(
"",
response_model=LimitOverrideOut,
status_code=status.HTTP_201_CREATED,
summary="新增白名单配置",
)
def create_override(
body: LimitOverrideWrite,
request: Request,
admin: CurrentAdmin,
db: AdminDb,
) -> LimitOverrideOut:
try:
row = repo.create(
db,
**body.model_dump(),
limit_value=None,
admin_id=admin.id,
)
except KeyError as exc:
raise HTTPException(status_code=404, detail="未知限制规则") from exc
except repo.DuplicateOverrideError as exc:
raise HTTPException(
status_code=409,
detail="该手机号或设备已配置此规则,请编辑现有配置",
) from exc
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
_reconcile_risk_rule(db, row.rule_code)
result = _out(db, row)
write_audit(
db,
admin,
action="limit.override.create",
target_type="limit_override",
target_id=str(row.id),
detail={"after": _audit_payload(result)},
ip=get_client_ip(request),
commit=False,
)
db.commit()
return _out(db, row)
@router.post(
"/bulk",
response_model=list[LimitOverrideOut],
status_code=status.HTTP_201_CREATED,
summary="批量新增临时不限或免告警白名单",
)
def create_overrides_bulk(
body: LimitOverrideBulkWrite,
request: Request,
admin: CurrentAdmin,
db: AdminDb,
) -> list[LimitOverrideOut]:
rows: list[LimitPolicyOverride] = []
current_rule_code = ""
try:
if body.subject_type == limit_policy.SUBJECT_DEVICE:
limit_policy.validate_device_rule_scope(body.rule_codes)
for current_rule_code in body.rule_codes:
rule = limit_policy.get_rule(current_rule_code)
mode = (
limit_policy.MODE_SUPPRESS_ALERT
if rule.alert_only
else limit_policy.MODE_UNLIMITED
)
rows.append(
repo.create(
db,
subject_type=body.subject_type,
subject_value=body.subject_value,
rule_code=current_rule_code,
mode=mode,
limit_value=None,
enabled=body.enabled,
starts_at=body.starts_at,
expires_at=body.expires_at,
reason=body.reason,
admin_id=admin.id,
reactivate_inactive=True,
)
)
except KeyError as exc:
raise HTTPException(status_code=404, detail="未知限制规则") from exc
except repo.DuplicateOverrideError as exc:
rule = limit_policy.get_rule(current_rule_code)
raise HTTPException(
status_code=409,
detail=f"{rule.label}”已有白名单配置,请编辑或重置现有配置",
) from exc
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
for row in rows:
_reconcile_risk_rule(db, row.rule_code)
results = [_out(db, row) for row in rows]
write_audit(
db,
admin,
action="limit.override.bulk_create",
target_type="limit_override",
target_id=",".join(str(row.id) for row in rows),
detail={"after": [_audit_payload(item) for item in results]},
ip=get_client_ip(request),
commit=False,
)
db.commit()
return [_out(db, row) for row in rows]
@router.patch(
"/{override_id}",
response_model=LimitOverrideOut,
summary="编辑白名单配置",
)
def update_override(
override_id: int,
body: LimitOverridePatch,
request: Request,
admin: CurrentAdmin,
db: AdminDb,
) -> LimitOverrideOut:
row = _row_or_404(db, override_id)
before = _audit_payload(_out(db, row))
try:
repo.update(
db,
row,
**body.model_dump(),
fields_set=set(body.model_fields_set),
)
except (KeyError, ValueError) as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
_reconcile_risk_rule(db, row.rule_code)
after = _out(db, row)
write_audit(
db,
admin,
action="limit.override.update",
target_type="limit_override",
target_id=str(row.id),
detail={"before": before, "after": _audit_payload(after)},
ip=get_client_ip(request),
commit=False,
)
db.commit()
return _out(db, row)
@router.post(
"/{override_id}/reset",
response_model=LimitOverrideOut,
summary="停用例外配置并恢复全局策略",
)
def restore_global_policy(
override_id: int,
request: Request,
admin: CurrentAdmin,
db: AdminDb,
) -> LimitOverrideOut:
row = _row_or_404(db, override_id)
before = _audit_payload(_out(db, row))
repo.restore_global(row)
_reconcile_risk_rule(db, row.rule_code)
after = _out(db, row)
write_audit(
db,
admin,
action="limit.override.restore_global",
target_type="limit_override",
target_id=str(row.id),
detail={"before": before, "after": _audit_payload(after)},
ip=get_client_ip(request),
commit=False,
)
db.commit()
return _out(db, row)
@router.delete(
"/{override_id}",
status_code=status.HTTP_204_NO_CONTENT,
summary="删除白名单配置",
)
def delete_override(
override_id: int,
request: Request,
admin: CurrentAdmin,
db: AdminDb,
) -> None:
row = _row_or_404(db, override_id)
before = _audit_payload(_out(db, row))
target_id = str(row.id)
rule_code = row.rule_code
db.delete(row)
db.flush()
_reconcile_risk_rule(db, rule_code)
write_audit(
db,
admin,
action="limit.override.delete",
target_type="limit_override",
target_id=target_id,
detail={"before": before},
ip=get_client_ip(request),
commit=False,
)
db.commit()
-8
View File
@@ -88,11 +88,6 @@ def get_user_reward_stats(
withdraw_source: Annotated[
str | None, Query(pattern="^(coin_cash|invite_cash)$")
] = None,
app_env: Annotated[str | None, Query(pattern="^(prod|test)$")] = None,
revenue_scope: Annotated[str, Query(pattern="^(business|all)$")] = "all",
feed_scene: Annotated[
str | None, Query(pattern="^(comparison|coupon|welfare)$")
] = None,
) -> UserRewardStats:
"""提现详情抽屉「用户统计区」。date_from/date_to 都不传 = 注册至今(全量)。"""
if not user_repo.user_exists(db, user_id):
@@ -104,9 +99,6 @@ def get_user_reward_stats(
date_from=date_from,
date_to=date_to,
withdraw_source=withdraw_source,
app_env=app_env,
revenue_scope=revenue_scope,
feed_scene=feed_scene,
)
)
-1
View File
@@ -7,7 +7,6 @@ from app.repositories.guide_video import MAX_PLAYS_LIMIT, REWARD_COIN_LIMIT
class GuideVideoConfigOut(BaseModel):
scene: str
enabled: bool
video_url: str | None = None # 相对地址 /media/guide_video/xxx.mp4;未配片 = None
max_plays: int
-127
View File
@@ -1,127 +0,0 @@
"""Admin contracts for global limit rules and per-subject overrides."""
from __future__ import annotations
from datetime import UTC, datetime
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field, model_validator
SubjectType = Literal["phone", "device"]
PolicyMode = Literal["unlimited", "suppress_alert"]
class LimitRuleOut(BaseModel):
code: str
label: str
group: str
global_limit: int
default_limit: int
window_label: str
subject_types: list[str]
allowed_modes: list[str]
min_value: int
max_value: int
supports_reset: bool
alert_only: bool
class GlobalLimitUpdate(BaseModel):
value: int = Field(ge=0, le=1_000_000)
class DeviceCandidateOut(BaseModel):
device_id: str
source: str
source_label: str
user_id: int | None = None
username: str | None = None
phone: str | None = None
nickname: str | None = None
device_model: str | None = None
last_active_at: datetime
class LimitOverrideWrite(BaseModel):
model_config = ConfigDict(extra="forbid")
subject_type: SubjectType
subject_value: str = Field(min_length=1, max_length=128)
rule_code: str = Field(min_length=1, max_length=64)
mode: PolicyMode
enabled: bool = True
starts_at: datetime | None = None
expires_at: datetime | None = None
reason: str = Field("", max_length=256)
@model_validator(mode="after")
def validate_time_range(self):
if self.starts_at and self.expires_at and self.expires_at <= self.starts_at:
raise ValueError("失效时间必须晚于生效时间")
return self
class LimitOverrideBulkWrite(BaseModel):
model_config = ConfigDict(extra="forbid")
subject_type: SubjectType
subject_value: str = Field(min_length=1, max_length=128)
rule_codes: list[str] = Field(min_length=1, max_length=32)
enabled: bool = True
starts_at: datetime | None = None
expires_at: datetime
reason: str = Field("", max_length=256)
@model_validator(mode="after")
def validate_bulk_request(self):
self.rule_codes = list(dict.fromkeys(self.rule_codes))
starts_at = (
self.starts_at.replace(tzinfo=UTC)
if self.starts_at and self.starts_at.tzinfo is None
else self.starts_at
)
expires_at = (
self.expires_at.replace(tzinfo=UTC)
if self.expires_at.tzinfo is None
else self.expires_at
)
if starts_at and expires_at <= starts_at:
raise ValueError("失效时间必须晚于生效时间")
if expires_at <= datetime.now(UTC):
raise ValueError("失效时间必须晚于当前时间")
return self
class LimitOverridePatch(BaseModel):
model_config = ConfigDict(extra="forbid")
enabled: bool | None = None
starts_at: datetime | None = None
expires_at: datetime | None = None
reason: str | None = Field(None, max_length=256)
class LimitOverrideOut(BaseModel):
id: int
subject_type: str
subject_value: str
rule_code: str
rule_label: str
rule_group: str
mode: str
limit_value: int | None
global_limit: int
effective_limit: int | None
enabled: bool
starts_at: datetime | None
expires_at: datetime | None
reset_at: datetime | None
reason: str | None
status: str
created_by_admin_id: int | None
created_at: datetime
updated_at: datetime
class LimitOverrideList(BaseModel):
items: list[LimitOverrideOut]
total: int
+2 -2
View File
@@ -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):
+1 -1
View File
@@ -60,7 +60,7 @@ class UserRewardStats(BaseModel):
reward_video_avg_ecpm: float # 平均激励视频 eCPM(分/千次)
reward_video_cash_cents: int # 激励视频提现(金币折现)
feed_count: int # 累计信息流广告数(granted 份数,unit_count 累加)
feed_avg_ecpm: float # 全部 Draw/feed 实际展示的平均 eCPM(分/千次,含未发奖展示)
feed_avg_ecpm: float # 平均信息流广告 eCPM(分/千次)
feed_cash_cents: int # 信息流广告提现(金币折现)
+2 -13
View File
@@ -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),
)
+26 -185
View File
@@ -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(
+3 -11
View File
@@ -108,7 +108,7 @@ def _harvest_done_blocking(
def _harvest_abort_blocking(
trace_id: str, status_hint: str, reason: str | None, trace_url: str | None,
) -> int | None:
) -> None:
with SessionLocal() as db:
rec = crud_compare.harvest_abort(
db, trace_id=trace_id, status=status_hint, reason=reason, trace_url=trace_url,
@@ -118,7 +118,6 @@ def _harvest_abort_blocking(
extra={"phase": "harvest_abort",
"status": (rec.status if rec else None), "reason": reason},
)
return rec.id if rec is not None else None
async def _forward(
@@ -292,10 +291,7 @@ async def trace_epilogue(
@router.post("/trace/finalize", summary="比价 trace 收尾上云 (透传 + 夭折落库)")
async def trace_finalize(
request: Request,
background_tasks: BackgroundTasks,
user: OptionalUser,
db: DbSession,
request: Request, user: OptionalUser, db: DbSession
) -> dict[str, Any]:
_ensure_compare_allowed(user, db)
# 用户终止 / Phase1 未识别没到 done 帧: pricebot 打包半截上云返回 {trace_url};
@@ -306,16 +302,12 @@ async def trace_finalize(
request, "/api/trace/finalize", user, harvest_first_frame=False,
)
try:
record_id = await run_in_threadpool(
await run_in_threadpool(
_harvest_abort_blocking, trace_id,
(meta.get("status") or "cancelled"),
(meta.get("reason") or meta.get("information")),
(resp.get("trace_url") if isinstance(resp, dict) else None),
)
if record_id is not None:
background_tasks.add_task(
backfill_comparison_llm_cost, record_id, trace_id
)
except Exception as e: # noqa: BLE001
logger.warning("harvest_abort failed trace=%s: %s", trace_id, e)
return resp
+4 -22
View File
@@ -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),
)
+4 -130
View File
@@ -14,137 +14,12 @@ from app.core import rewards as r
RISK_SMS_HOURLY_THRESHOLD_KEY = "risk_sms_hourly_threshold"
RISK_ONECLICK_DAILY_THRESHOLD_KEY = "risk_oneclick_daily_threshold"
RISK_COMPARE_DAILY_THRESHOLD_KEY = "risk_compare_daily_threshold"
COMPARE_DAILY_LIMIT_KEY = "compare_daily_limit"
SMS_SEND_HOURLY_LIMIT_KEY = "sms_send_hourly_limit"
SMS_SEND_DAILY_LIMIT_KEY = "sms_send_daily_limit"
SMS_LOGIN_HOURLY_LIMIT_KEY = "sms_login_hourly_limit"
WECHAT_BIND_SMS_HOURLY_LIMIT_KEY = "wechat_bind_sms_hourly_limit"
WECHAT_CONFLICT_HOURLY_LIMIT_KEY = "wechat_conflict_hourly_limit"
SMS_PHONE_COOLDOWN_SECONDS_KEY = "sms_phone_cooldown_seconds"
SMS_CODE_MAX_FAILED_ATTEMPTS_KEY = "sms_code_max_failed_attempts"
AD_REWARD_VIDEO_DAILY_LIMIT_KEY = "ad_reward_video_daily_limit"
AD_FEED_DAILY_LIMIT_KEY = "ad_feed_daily_limit"
PHONE_REBIND_DAYS_KEY = "phone_rebind_days"
GUIDE_VIDEO_MAX_PLAYS_KEY = "guide_video_max_plays"
# type 约定(给前端渲染编辑控件用):int / int_list / dict_str_int / bool / enum
# hidden=True:仍是合法可配项(业务照常 get_value / admin 可经专用端点读写),但**不在通用
# 「系统配置」页渲染**(admin/routers/config.py:list_config 按此过滤)。用于把已下线/已改由
# 专用页管理的项从福利页 Tab 收起,同时保留后端默认值与写入能力。
CONFIG_DEFS: dict[str, dict[str, Any]] = {
COMPARE_DAILY_LIMIT_KEY: {
"default": 100,
"label": "账号每日比价次数上限",
"group": "限制策略",
"type": "int",
"min": 1,
"max": 100_000,
"hidden": True,
"help": "北京时间自然日内每个账号可发起的比价次数。",
},
SMS_SEND_HOURLY_LIMIT_KEY: {
"default": 5,
"label": "短信每小时发送上限",
"group": "限制策略",
"type": "int",
"min": 1,
"max": 10_000,
"hidden": True,
"help": "同一设备和 IP 在固定 1 小时窗口内成功发送短信的次数。",
},
SMS_SEND_DAILY_LIMIT_KEY: {
"default": 20,
"label": "短信 24 小时发送上限",
"group": "限制策略",
"type": "int",
"min": 1,
"max": 100_000,
"hidden": True,
"help": "同一设备和 IP 在固定 24 小时窗口内成功发送短信的次数。",
},
SMS_LOGIN_HOURLY_LIMIT_KEY: {
"default": 5,
"label": "短信登录每小时尝试上限",
"group": "限制策略",
"type": "int",
"min": 1,
"max": 10_000,
"hidden": True,
"help": "成功与失败均计入。",
},
WECHAT_BIND_SMS_HOURLY_LIMIT_KEY: {
"default": 5,
"label": "微信短信绑定每小时尝试上限",
"group": "限制策略",
"type": "int",
"min": 1,
"max": 10_000,
"hidden": True,
},
WECHAT_CONFLICT_HOURLY_LIMIT_KEY: {
"default": 5,
"label": "微信冲突处理每小时尝试上限",
"group": "限制策略",
"type": "int",
"min": 1,
"max": 10_000,
"hidden": True,
"help": "继续登录与重新绑定共享此额度。",
},
SMS_PHONE_COOLDOWN_SECONDS_KEY: {
"default": 60,
"label": "同手机号短信发送冷却秒数",
"group": "限制策略",
"type": "int",
"min": 0,
"max": 86_400,
"hidden": True,
},
SMS_CODE_MAX_FAILED_ATTEMPTS_KEY: {
"default": 5,
"label": "单验证码最大失败次数",
"group": "限制策略",
"type": "int",
"min": 1,
"max": 100,
"hidden": True,
},
AD_REWARD_VIDEO_DAILY_LIMIT_KEY: {
"default": r.DAILY_AD_REWARD_LIMIT,
"label": "激励视频每日发奖次数",
"group": "限制策略",
"type": "int",
"min": 1,
"max": 100_000,
"hidden": True,
},
AD_FEED_DAILY_LIMIT_KEY: {
"default": r.DAILY_AD_REWARD_LIMIT,
"label": "Draw 信息流每日发奖次数",
"group": "限制策略",
"type": "int",
"min": 1,
"max": 100_000,
"hidden": True,
},
PHONE_REBIND_DAYS_KEY: {
"default": 30,
"label": "手机/微信换绑冷却天数",
"group": "限制策略",
"type": "int",
"min": 0,
"max": 3650,
"hidden": True,
},
GUIDE_VIDEO_MAX_PLAYS_KEY: {
"default": 3,
"label": "领券引导视频最大播放次数",
"group": "限制策略",
"type": "int",
"min": 0,
"max": 50,
"hidden": True,
},
"signin_rewards": {
"default": list(r.SIGNIN_REWARDS), "label": "签到 7 天金币档位",
"group": "签到", "type": "int_list",
@@ -181,8 +56,7 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
},
"ad_daily_limit": {
"default": r.DAILY_AD_REWARD_LIMIT, "label": "看广告每日上限(次)",
"group": "看广告", "type": "int", "hidden": True,
"help": "历史共享上限;新配置由白名单页分别管理激励视频与 Draw 信息流。",
"group": "看广告", "type": "int", "help": "福利页激励视频每日可发奖次数上限,默认 500。",
},
"ad_max_coin": {
"default": r.MAX_AD_REWARD_COIN, "label": "看广告单次金币上限",
@@ -244,9 +118,9 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
"group": "风控",
"type": "int",
"min": 1,
"max": 100_000,
"max": 5,
"hidden": True,
"help": "同一设备在北京时间同一自然小时内成功下发短信达到该次数时告警。",
"help": "同一设备在北京时间同一自然小时内成功下发短信达到该次数时告警;不得高于现有每小时 5 次的发送上限",
},
RISK_ONECLICK_DAILY_THRESHOLD_KEY: {
"default": 20,
@@ -264,7 +138,7 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
"group": "风控",
"type": "int",
"min": 1,
"max": 100_000,
"max": 100,
"hidden": True,
"help": "同一账户在北京时间同一自然日内发起比价达到该次数时告警。",
},
-525
View File
@@ -1,525 +0,0 @@
"""Unified global limits and per-phone/device policy overrides.
The registry is the single source of truth for the whitelist page. Existing
constants remain as backwards-compatible defaults, while business call sites
resolve an effective value here.
"""
from __future__ import annotations
from collections.abc import Iterable
from dataclasses import dataclass
from datetime import UTC, datetime
from sqlalchemy import or_, select
from sqlalchemy.orm import Session
from app.core.config_schema import (
AD_FEED_DAILY_LIMIT_KEY,
AD_REWARD_VIDEO_DAILY_LIMIT_KEY,
COMPARE_DAILY_LIMIT_KEY,
GUIDE_VIDEO_MAX_PLAYS_KEY,
PHONE_REBIND_DAYS_KEY,
RISK_COMPARE_DAILY_THRESHOLD_KEY,
RISK_ONECLICK_DAILY_THRESHOLD_KEY,
RISK_SMS_HOURLY_THRESHOLD_KEY,
SMS_CODE_MAX_FAILED_ATTEMPTS_KEY,
SMS_LOGIN_HOURLY_LIMIT_KEY,
SMS_PHONE_COOLDOWN_SECONDS_KEY,
SMS_SEND_DAILY_LIMIT_KEY,
SMS_SEND_HOURLY_LIMIT_KEY,
WECHAT_BIND_SMS_HOURLY_LIMIT_KEY,
WECHAT_CONFLICT_HOURLY_LIMIT_KEY,
)
from app.models.app_config import AppConfig
from app.models.limit_policy import LimitPolicyOverride
from app.models.user import User
from app.repositories import app_config
MODE_INHERIT = "inherit"
MODE_OVERRIDE = "override"
MODE_UNLIMITED = "unlimited"
MODE_SUPPRESS_ALERT = "suppress_alert"
SUBJECT_PHONE = "phone"
SUBJECT_DEVICE = "device"
SUBJECT_TYPES = (SUBJECT_PHONE, SUBJECT_DEVICE)
SUBJECT_PRECEDENCE = {SUBJECT_DEVICE: 0, SUBJECT_PHONE: 1}
LEGACY_IP_DEVICE_PREFIX = "legacy-ip:"
@dataclass(frozen=True)
class RuleDefinition:
code: str
label: str
group: str
config_key: str
default_limit: int
window_label: str
subject_types: tuple[str, ...]
min_value: int = 1
max_value: int = 100_000
allow_unlimited: bool = True
supports_reset: bool = True
alert_only: bool = False
legacy_config_key: str | None = None
legacy_json_field: str | None = None
@property
def allowed_modes(self) -> tuple[str, ...]:
if self.alert_only:
return (MODE_SUPPRESS_ALERT,)
return (MODE_UNLIMITED,) if self.allow_unlimited else ()
RULES: tuple[RuleDefinition, ...] = (
RuleDefinition(
"compare.start.daily",
"每日发起比价次数",
"比价",
COMPARE_DAILY_LIMIT_KEY,
100,
"北京时间自然日",
SUBJECT_TYPES,
),
RuleDefinition(
"sms.send.hourly",
"短信每小时成功发送次数",
"短信与登录",
SMS_SEND_HOURLY_LIMIT_KEY,
5,
"固定 1 小时窗口",
SUBJECT_TYPES,
max_value=10_000,
),
RuleDefinition(
"sms.send.daily",
"短信 24 小时成功发送次数",
"短信与登录",
SMS_SEND_DAILY_LIMIT_KEY,
20,
"固定 24 小时窗口",
SUBJECT_TYPES,
),
RuleDefinition(
"sms.phone.cooldown",
"同手机号短信发送冷却",
"短信与登录",
SMS_PHONE_COOLDOWN_SECONDS_KEY,
60,
"",
(SUBJECT_PHONE,),
min_value=0,
max_value=86_400,
),
RuleDefinition(
"sms.code.failed_attempts",
"单验证码最大失败次数",
"短信与登录",
SMS_CODE_MAX_FAILED_ATTEMPTS_KEY,
5,
"单个验证码",
(SUBJECT_PHONE,),
max_value=100,
),
RuleDefinition(
"sms.login.hourly",
"短信登录每小时尝试次数",
"短信与登录",
SMS_LOGIN_HOURLY_LIMIT_KEY,
5,
"固定 1 小时窗口",
SUBJECT_TYPES,
max_value=10_000,
),
RuleDefinition(
"wechat.bind.hourly",
"微信短信绑定每小时尝试次数",
"短信与登录",
WECHAT_BIND_SMS_HOURLY_LIMIT_KEY,
5,
"固定 1 小时窗口",
SUBJECT_TYPES,
max_value=10_000,
),
RuleDefinition(
"wechat.conflict.hourly",
"微信冲突处理每小时尝试次数",
"短信与登录",
WECHAT_CONFLICT_HOURLY_LIMIT_KEY,
5,
"固定 1 小时窗口",
SUBJECT_TYPES,
max_value=10_000,
),
RuleDefinition(
"ad.reward_video.daily",
"激励视频每日发奖次数",
"广告",
AD_REWARD_VIDEO_DAILY_LIMIT_KEY,
500,
"北京时间自然日",
(SUBJECT_PHONE,),
legacy_config_key="ad_daily_limit",
),
RuleDefinition(
"ad.feed.daily",
"Draw 信息流每日发奖次数",
"广告",
AD_FEED_DAILY_LIMIT_KEY,
500,
"北京时间自然日",
(SUBJECT_PHONE,),
legacy_config_key="ad_daily_limit",
),
RuleDefinition(
"ad.reward_video.cooldown",
"激励视频发奖后冷却秒数",
"广告",
"ad_cooldown_sec",
3,
"",
(SUBJECT_PHONE,),
min_value=0,
max_value=86_400,
),
RuleDefinition(
"guide.video.lifetime",
"领券引导视频最大播放次数",
"引导与账号",
GUIDE_VIDEO_MAX_PLAYS_KEY,
3,
"账号生命周期",
(SUBJECT_PHONE,),
min_value=0,
max_value=50,
legacy_config_key="coupon_guide_video",
legacy_json_field="max_plays",
),
RuleDefinition(
"phone.rebind.days",
"手机/微信换绑冷却天数",
"引导与账号",
PHONE_REBIND_DAYS_KEY,
30,
"自然日",
(SUBJECT_PHONE,),
min_value=0,
max_value=3650,
),
RuleDefinition(
"risk.sms.hourly",
"短信设备每小时告警",
"风控免告警",
RISK_SMS_HOURLY_THRESHOLD_KEY,
5,
"北京时间自然小时",
(SUBJECT_DEVICE,),
max_value=100_000,
allow_unlimited=False,
alert_only=True,
),
RuleDefinition(
"risk.oneclick.daily",
"一键登录设备每日告警",
"风控免告警",
RISK_ONECLICK_DAILY_THRESHOLD_KEY,
20,
"北京时间自然日",
(SUBJECT_DEVICE,),
max_value=100_000,
allow_unlimited=False,
alert_only=True,
),
RuleDefinition(
"risk.compare.daily",
"比价账号每日告警",
"风控免告警",
RISK_COMPARE_DAILY_THRESHOLD_KEY,
100,
"北京时间自然日",
(SUBJECT_PHONE,),
max_value=100_000,
allow_unlimited=False,
alert_only=True,
),
)
RULE_MAP = {rule.code: rule for rule in RULES}
@dataclass(frozen=True)
class EffectiveLimit:
rule_code: str
global_limit: int
limit: int | None
mode: str
suppressed: bool
override_id: int | None
matched_subject_type: str | None
matched_subject_value: str | None
reset_at: datetime | None
bucket_version: str
@property
def unlimited(self) -> bool:
return self.limit is None
def normalize_subject(subject_type: str, value: str) -> str:
value = (value or "").strip()
if subject_type == SUBJECT_PHONE:
value = "".join(ch for ch in value if ch.isdigit())
if subject_type not in SUBJECT_TYPES:
raise ValueError(f"unsupported subject type: {subject_type}")
if not value:
raise ValueError("subject value is empty")
return value[:128]
def validate_whitelist_subject(subject_type: str, value: str) -> str:
"""Normalize a whitelist subject and reject unsafe pseudo-devices.
Old clients without a device ID are grouped by public IP for rate limiting.
That fallback remains valid at runtime, but it is not a stable, unique
device identity and must never be persisted as a device whitelist target.
"""
normalized = normalize_subject(subject_type, value)
if (
subject_type == SUBJECT_DEVICE
and normalized.startswith(LEGACY_IP_DEVICE_PREFIX)
):
raise ValueError(
"旧客户端未上报真实设备 ID,不能加入设备白名单,请升级客户端后重试"
)
return normalized
def get_rule(rule_code: str) -> RuleDefinition:
try:
return RULE_MAP[rule_code]
except KeyError as exc:
raise ValueError(f"unknown rule: {rule_code}") from exc
def device_source_scope(rule_code: str) -> str:
"""返回设备候选数据所属命名空间,防止一个设备 ID 跨来源误套规则。"""
rule = get_rule(rule_code)
if SUBJECT_DEVICE not in rule.subject_types:
raise ValueError("当前限制项不支持设备白名单")
return "comparison" if rule_code == "compare.start.daily" else "auth"
def validate_device_rule_scope(rule_codes: Iterable[str]) -> None:
scopes = {device_source_scope(rule_code) for rule_code in rule_codes}
if len(scopes) > 1:
raise ValueError("比价设备限制与短信/登录设备限制不能在同一白名单中混选")
def _global_limit(db: Session, rule: RuleDefinition) -> tuple[int, str]:
row = db.get(AppConfig, rule.config_key)
if row is not None:
return int(row.value), "configured"
if rule.legacy_config_key:
legacy = db.get(AppConfig, rule.legacy_config_key)
if legacy is not None:
value = legacy.value
if rule.legacy_json_field:
value = value.get(rule.legacy_json_field) if isinstance(value, dict) else None
if value is not None:
return int(value), "legacy"
try:
return int(app_config.get_value(db, rule.config_key)), "default"
except KeyError:
return rule.default_limit, "default"
def _aware(value: datetime | None) -> datetime | None:
if value is None:
return None
return value.replace(tzinfo=UTC) if value.tzinfo is None else value
def _matching_overrides(
db: Session,
rule: RuleDefinition,
subjects: dict[str, str | None],
now: datetime,
) -> list[LimitPolicyOverride]:
pairs: list[tuple[str, str]] = []
for subject_type in rule.subject_types:
raw = subjects.get(subject_type)
if raw:
pairs.append((subject_type, normalize_subject(subject_type, raw)))
if not pairs:
return []
clauses = [
(
(LimitPolicyOverride.subject_type == subject_type)
& (LimitPolicyOverride.subject_value == subject_value)
)
for subject_type, subject_value in pairs
]
rows = list(
db.execute(
select(LimitPolicyOverride).where(
LimitPolicyOverride.rule_code == rule.code,
LimitPolicyOverride.enabled.is_(True),
or_(*clauses),
)
).scalars()
)
active = [
row
for row in rows
if (_aware(row.starts_at) is None or _aware(row.starts_at) <= now)
and (_aware(row.expires_at) is None or _aware(row.expires_at) > now)
]
return sorted(active, key=lambda row: SUBJECT_PRECEDENCE[row.subject_type])
def resolve(
db: Session,
rule_code: str,
*,
phone: str | None = None,
device: str | None = None,
now: datetime | None = None,
) -> EffectiveLimit:
"""Resolve global config plus the most specific active override.
Device overrides win over phone overrides when both match.
"""
rule = get_rule(rule_code)
now = _aware(now) or datetime.now(UTC)
global_limit, global_version = _global_limit(db, rule)
matches = _matching_overrides(
db, rule, {SUBJECT_PHONE: phone, SUBJECT_DEVICE: device}, now
)
row = matches[0] if matches else None
if row is None:
return EffectiveLimit(
rule.code,
global_limit,
global_limit,
MODE_INHERIT,
False,
None,
None,
None,
None,
global_version,
)
limit: int | None = global_limit
suppressed = False
if row.mode == MODE_OVERRIDE:
limit = int(row.limit_value) if row.limit_value is not None else global_limit
elif row.mode == MODE_UNLIMITED:
limit = None
elif row.mode == MODE_SUPPRESS_ALERT:
suppressed = True
# 编辑限制值/备注不应隐式清空计数;只有显式“重置状态”才切换桶。
reset_version = _aware(row.reset_at)
version = (
f"{global_version}:o:{row.id}:"
f"{reset_version.isoformat() if reset_version else '0'}"
)
return EffectiveLimit(
rule.code,
global_limit,
limit,
row.mode,
suppressed,
row.id,
row.subject_type,
row.subject_value,
_aware(row.reset_at),
version,
)
def resolve_for_user(
db: Session,
rule_code: str,
user_id: int,
*,
device: str | None = None,
) -> EffectiveLimit:
user = db.get(User, user_id)
return resolve(
db,
rule_code,
phone=user.phone if user is not None else None,
device=device,
)
def rule_catalog(db: Session) -> list[dict]:
out: list[dict] = []
for rule in RULES:
value, _ = _global_limit(db, rule)
out.append(
{
"code": rule.code,
"label": rule.label,
"group": rule.group,
"global_limit": value,
"default_limit": rule.default_limit,
"window_label": rule.window_label,
"subject_types": list(rule.subject_types),
"allowed_modes": list(rule.allowed_modes),
"min_value": rule.min_value,
"max_value": rule.max_value,
"supports_reset": rule.supports_reset,
"alert_only": rule.alert_only,
}
)
return out
def validate_override(
rule: RuleDefinition,
*,
subject_type: str,
mode: str,
limit_value: int | None,
starts_at: datetime | None,
expires_at: datetime | None,
) -> None:
if subject_type not in rule.subject_types:
raise ValueError("该规则不支持此主体类型")
if mode not in rule.allowed_modes:
raise ValueError("该规则不支持此策略模式")
if limit_value is not None:
raise ValueError("白名单不支持覆盖指定值")
if mode in {MODE_UNLIMITED, MODE_SUPPRESS_ALERT} and expires_at is None:
raise ValueError("临时白名单必须设置失效时间")
if starts_at and expires_at and _aware(expires_at) <= _aware(starts_at):
raise ValueError("失效时间必须晚于生效时间")
if (
mode in {MODE_UNLIMITED, MODE_SUPPRESS_ALERT}
and expires_at is not None
and _aware(expires_at) <= datetime.now(UTC)
):
raise ValueError("临时白名单的失效时间必须晚于当前时间")
if mode == MODE_SUPPRESS_ALERT and not rule.alert_only:
raise ValueError("免告警只支持风控监控的三项规则")
def active_overrides(
db: Session,
*,
subject_type: str | None = None,
keyword: str | None = None,
) -> Iterable[LimitPolicyOverride]:
stmt = select(LimitPolicyOverride)
if subject_type:
stmt = stmt.where(LimitPolicyOverride.subject_type == subject_type)
if keyword:
stmt = stmt.where(LimitPolicyOverride.subject_value.ilike(f"%{keyword.strip()}%"))
return db.execute(
stmt.order_by(LimitPolicyOverride.updated_at.desc(), LimitPolicyOverride.id.desc())
).scalars()
+6 -19
View File
@@ -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)
+4 -15
View File
@@ -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)
+8 -29
View File
@@ -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
+6 -19
View File
@@ -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")):
+6 -19
View File
@@ -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")):
-1
View File
@@ -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
-4
View File
@@ -113,10 +113,6 @@ class ComparisonRecord(Base):
items: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
# 逐平台对比 [{platform_id, platform_name, package, price, is_source, rank, coupon_saved, coupon_name, applied_coupons}](price/coupon_saved 单位:元,原样存;coupon_name=优惠来源名;applied_coupons=[{name,amount}] 多券明细)
comparison_results: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
# 展示模型统一数组(pricebot done.params.platforms 原样存): 每平台一行、自带
# status/is_best/display/display_order, 记录页据此直接渲染, 不再靠 comparison_results
# + 客户端合并 + 前端派生。老记录/旧客户端为空 → 前端回退老 comparison_results 渲染。
platforms: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
# 目标平台未找到、跳过的菜名
skipped_dish_names: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
# 客户端上报的原始 payload(calibration + done.params 全量),未来取数兜底
+2 -8
View File
@@ -1,7 +1,7 @@
"""新手引导视频播放记录(领券浮层前 N 次用它替代广告)。
产品规则(2026-07 拍板):新用户点一键自动领取后的等候浮层,** 3 **不放广告,
改放运营后台上传的引导视频;默认每次固定 100 金币,中途关闭也算看完照发
改放运营后台上传的引导视频;每次固定 120 金币,中途关闭也算看完照发
口径:
- **计次按账号**(user_id),与设备无关 换设备不重新送 3
@@ -38,13 +38,7 @@ class GuideVideoPlay(Base):
# start_play 捕获 IntegrityError 降级成"这次不放视频"。
# 用 unique Index 而非 UniqueConstraint:与迁移里的 create_index 对齐(SQLite 加约束
# 要整表重建),autogenerate 才不会每次报一条假 diff。
Index(
"uq_guide_video_play_user_scene_seq",
"user_id",
"scene",
"seq",
unique=True,
),
Index("uq_guide_video_play_user_seq", "user_id", "seq", unique=True),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
-77
View File
@@ -1,77 +0,0 @@
"""Per-subject limit policy overrides used by the admin whitelist page."""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import (
Boolean,
DateTime,
Index,
Integer,
String,
UniqueConstraint,
func,
true,
)
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
class LimitPolicyOverride(Base):
"""One rule override for one phone or device.
``reset_at`` is a non-destructive usage baseline. Business records and
security events remain intact; quota readers only count rows at or after
this timestamp.
"""
__tablename__ = "limit_policy_override"
__table_args__ = (
UniqueConstraint(
"subject_type",
"subject_value",
"rule_code",
name="uq_limit_policy_subject_rule",
),
Index(
"ix_limit_policy_lookup",
"subject_type",
"subject_value",
"rule_code",
"enabled",
),
Index("ix_limit_policy_expires", "expires_at"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
subject_type: Mapped[str] = mapped_column(String(16), nullable=False)
subject_value: Mapped[str] = mapped_column(String(128), nullable=False)
rule_code: Mapped[str] = mapped_column(String(64), nullable=False)
# 产品只保留“临时不限/免告警”。即使有内部脚本绕过 API 直接建 ORM
# 对象,也不能再悄悄落成已经下线的 override 模式。
mode: Mapped[str] = mapped_column(String(24), nullable=False, default="unlimited")
limit_value: Mapped[int | None] = mapped_column(Integer, nullable=True)
enabled: Mapped[bool] = mapped_column(
Boolean, nullable=False, default=True, server_default=true()
)
starts_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True
)
expires_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True
)
reset_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True
)
reason: Mapped[str | None] = mapped_column(String(256), nullable=True)
created_by_admin_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), nullable=False
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
onupdate=func.now(),
nullable=False,
)
+5 -36
View File
@@ -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,
+17 -82
View File
@@ -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,
+13 -114
View File
@@ -164,9 +164,8 @@ def _derive(payload: ComparisonRecordIn) -> dict:
is_source_best = best.is_source if best is not None else None
# status:优先 pricebot record_status(区分 below_minimum/store_closed) → 客户端显式 status
# → 兜底"非源且有价"=success/否则 failed。record_status 让"未满起送"不再塌缩成 failed。
status = payload.record_status or payload.status
# status:客户端显式给了就用;否则有"非源且有价"的结果=success,否则 failed
status = payload.status
if status is None:
has_valid_target = any(
(not r.is_source) and r.price is not None for r in results
@@ -198,35 +197,16 @@ def upsert_record(
灰度期老客户端 POST /compare/record 走这条,与后端 harvest trace_id reconcile;
新客户端不再 POST(改由 compare.py 透传壳 harvest 落库)
"""
# 单源派生: 与 harvest_done 一致, payload 带 platforms 时从它派生(唯一真相源
# _derive_from_platforms), 老客户端不带 platforms 时回退 _derive(从 comparison_results)。
if payload.platforms:
derived = _derive_from_platforms(payload.platforms, payload.record_status)
# 对齐 _derive 返回键(#189 fail_reason): 两路径 fields 键集一致, 覆盖已有行时不残留旧值
derived["fail_reason"] = (
_derive_fail_display(payload.information, payload.platform_results or {})
if derived["status"] == "failed"
else None
)
# 单源派生取自 platforms 源行(常无源平台元数据/店名)→ 空则用 payload 兜底不丢字段。
# 下面 fields 不再显式写这四个键, 统一由 derived 提供(否则 dict(store_name=..., **derived)
# 与 _derive_from_platforms 同名键撞键 TypeError)。
for _k in ("store_name", "source_platform_id", "source_platform_name", "source_package"):
if not derived.get(_k):
derived[_k] = getattr(payload, _k)
else:
derived = _derive(payload)
# _derive 只从 comparison_results 派生, 不含源平台四件套 / store_name → 从 payload 补,
# 与上面 platforms 分支键集对齐(fields 统一靠 **derived 提供这些列)。
for _k in ("store_name", "source_platform_id", "source_platform_name", "source_package"):
derived[_k] = getattr(payload, _k)
derived = _derive(payload)
items = [it.model_dump(exclude_none=True) for it in payload.items]
fields = dict(
device_id=payload.device_id,
business_type=payload.business_type,
store_name=payload.store_name,
product_names=_product_names_from_items(items),
# store_name / source_platform_id / source_platform_name / source_package 统一由
# derived 提供(见上方两分支补齐), 不在此显式写 —— 否则与 _derive_from_platforms 撞键。
source_platform_id=payload.source_platform_id,
source_platform_name=payload.source_platform_name,
source_package=payload.source_package,
information=payload.information,
best_deeplink=payload.best_deeplink,
trace_url=payload.trace_url,
@@ -234,7 +214,6 @@ def upsert_record(
skipped_dish_count=payload.skipped_dish_count,
items=items,
comparison_results=[r.model_dump() for r in payload.comparison_results],
platforms=list(payload.platforms or []),
skipped_dish_names=list(payload.skipped_dish_names),
# 客户端环境 / 性能(debug,客户端上报;旧客户端为 None)
device_model=payload.device_model,
@@ -307,8 +286,7 @@ def upsert_record(
def _derive_from_results(
results: list[dict], platform_results: dict | None = None,
record_status: str | None = None,
results: list[dict], platform_results: dict | None = None
) -> dict:
"""从 done 帧 comparison_results(pricebot 原始 dict 列表)派生结构化列。
等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象
@@ -356,53 +334,7 @@ def _derive_from_results(
"saved_amount_cents": saved_amount_cents,
"is_source_best": best.get("is_source") if best else None,
"store_name": (src_row or {}).get("store_name") or None,
# 记录级结局: 优先用 pricebot 下发的 record_status(区分 below_minimum/store_closed,
# 不再把"未满起送"塌缩成 failed → 记录页不再误报"网络开小差"); 旧 pricebot 未下发时
# 回退老的 success/failed 二态派生, 向后兼容。
"status": record_status or ("success" if has_valid_target else "failed"),
}
def _derive_from_platforms(
platforms: list, record_status: str | None = None,
) -> dict:
"""从 done 帧 platforms(每平台一行、渲染就绪)派生结构化列——**单一真相源**。
best_* 直接取 platforms is_best 的那一行source_* role=source ,与前端读的
platforms 天然一致(不再像 _derive_from_results 那样从 comparison_results 二次评最优,
消除"标量列 vs platforms"双源不一致)platforms 非空时优先走这里; pricebot
platforms 时调用方回退 _derive_from_results(向后兼容)"""
rows = [p for p in (platforms or []) if isinstance(p, dict)]
src = next((p for p in rows if p.get("role") == "source"), None)
best = next((p for p in rows if p.get("is_best")), None)
source_price_cents = _yuan_to_cents(src.get("price")) if src else None
best_price_cents = _yuan_to_cents(best.get("price")) if best else None
saved_amount_cents = None
if source_price_cents is not None and best_price_cents is not None:
saved_amount_cents = source_price_cents - best_price_cents
has_valid_target = any(
p.get("role") != "source" and p.get("price") is not None for p in rows
)
# store_name: 优先源行; recompare 场景源平台自己当目标、源行被目标覆盖(pricebot
# _build_platform_rows 有意去重, platforms 无 role=source 行)→ 回退 best 行 → 首个有店名
# 的行(显示现场实际比到的店), 免得记录页店名空掉兜底显示成"比价"。正常比价有源行不走回退。
store_name = (
(src or {}).get("store_name")
or (best or {}).get("store_name")
or next((p.get("store_name") for p in rows if p.get("store_name")), None)
)
return {
"source_platform_id": (src or {}).get("platform_id"),
"source_platform_name": (src or {}).get("platform_name"),
"source_package": (src or {}).get("package"),
"source_price_cents": source_price_cents,
"best_platform_id": (best or {}).get("platform_id"),
"best_platform_name": (best or {}).get("platform_name"),
"best_price_cents": best_price_cents,
"saved_amount_cents": saved_amount_cents,
"is_source_best": (best.get("role") == "source") if best else None,
"store_name": store_name or None,
"status": record_status or ("success" if has_valid_target else "failed"),
"status": "success" if has_valid_target else "failed",
}
@@ -434,8 +366,6 @@ def reserve_daily_start(
business_type: str = "food",
device_id: str | None = None,
now: datetime | None = None,
limit: int | None = DAILY_COMPARE_START_LIMIT,
reset_at: datetime | None = None,
) -> tuple[ComparisonRecord, int]:
"""Atomically reserve one of a user's 100 Beijing-day comparison starts.
@@ -461,11 +391,6 @@ def reserve_daily_start(
if existing_at.tzinfo is not None:
existing_at = existing_at.astimezone(CN_TZ).replace(tzinfo=None)
day_start = existing_at.replace(hour=0, minute=0, second=0, microsecond=0)
if reset_at is not None:
reset_start = reset_at
if reset_start.tzinfo is not None:
reset_start = reset_start.astimezone(CN_TZ).replace(tzinfo=None)
day_start = max(day_start, reset_start)
day_end = day_start + timedelta(days=1)
used = db.scalar(
select(func.count(ComparisonRecord.id)).where(
@@ -480,11 +405,6 @@ def reserve_daily_start(
if current.tzinfo is not None:
current = current.astimezone(CN_TZ).replace(tzinfo=None)
day_start = current.replace(hour=0, minute=0, second=0, microsecond=0)
if reset_at is not None:
reset_start = reset_at
if reset_start.tzinfo is not None:
reset_start = reset_start.astimezone(CN_TZ).replace(tzinfo=None)
day_start = max(day_start, reset_start)
day_end = day_start + timedelta(days=1)
used = db.scalar(
select(func.count(ComparisonRecord.id)).where(
@@ -493,7 +413,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(
@@ -573,29 +493,7 @@ def harvest_done(
返回 (记录, 是否本次****落成 success)供调用方据此幂等发一次邀请奖
行不存在(理论上帧0已建;防御)则新建"""
results = done_params.get("comparison_results") or []
# 展示模型统一数组(pricebot 新增, 每平台一行自带 status/is_best): 原样存, 记录页据此直渲染。
# record_status: 记录级结局(success/below_minimum/store_closed/failed), 覆盖老二态派生。
platforms = done_params.get("platforms") or []
record_status = done_params.get("record_status")
# 单源派生: platforms(含 pricebot 权威 is_best)是唯一真相源, best_*/source_*/saved/status
# 全从它取 → 与前端读的 platforms 天然一致; 菜品也取 platforms 源行。老 pricebot 无
# platforms 时回退从 comparison_results 派生(向后兼容)。
if platforms:
derived = _derive_from_platforms(platforms, record_status)
# 菜品优先源行; recompare 无源行 → 回退 best 行 → 首个有菜品的行(同 store_name 回退)
_item_row = (
next((p for p in platforms if isinstance(p, dict) and p.get("role") == "source"), None)
or next((p for p in platforms if isinstance(p, dict) and p.get("is_best")), None)
or next((p for p in platforms if isinstance(p, dict) and p.get("items")), None)
)
items = (_item_row or {}).get("items") or []
else:
derived = _derive_from_results(
results, done_params.get("platform_results"), record_status
)
# pricebot 已把源单菜品塞进 comparison_results[源行].items
items = next((r.get("items") or [] for r in results if r.get("is_source")), [])
# 失败展示原因(#189): platforms / results 两个派生分支的 status 都可能 failed, 统一在此算
derived = _derive_from_results(results, done_params.get("platform_results"))
fail_reason = (
_derive_fail_display(
done_params.get("information"), done_params.get("platform_results")
@@ -603,6 +501,8 @@ def harvest_done(
if derived["status"] == "failed"
else None
)
# 菜品:pricebot 已把源单菜品塞进 comparison_results[源行].items
items = next((r.get("items") or [] for r in results if r.get("is_source")), [])
fields = dict(
business_type=business_type or "food",
information=done_params.get("information") or None,
@@ -614,7 +514,6 @@ def harvest_done(
skipped_dish_count=done_params.get("skipped_dish_count"),
skipped_dish_names=list(done_params.get("skipped_dish_names") or []),
comparison_results=results,
platforms=platforms,
items=items,
product_names=_product_names_from_items(items),
raw_payload=done_params,
+38 -134
View File
@@ -1,8 +1,8 @@
"""新手引导视频:运营配置读写 + 播放计次 + 发币。
**配置**(开关 / 视频地址 / 前几次 / 每次金币)按场景分别作为一个 JSON 存进通用
app_config 领券使用 coupon_guide_video比价使用 comparison_guide_video
写法完全对齐 feedback_qr不进入系统配置页的通用列表
**配置**(开关 / 视频地址 / 前几次 / 每次金币)整体作为一个 JSON 存进通用 app_config
(key=coupon_guide_video),写法完全对齐 feedback_qr 不进 CONFIG_DEFS,所以不会污染
系统配置页的通用列表,由本模块独占维护
**计次**按账号(user_id)**开播即计数**:客户端每次要展示领券等候浮层时调
`/api/v1/guide-video/start`,命中则当场写一行 guide_video_play(status='playing')
@@ -25,16 +25,12 @@ from sqlalchemy import func, select, update
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from app.core import limit_policy, rewards
from app.core import rewards
from app.models.app_config import AppConfig
from app.models.guide_video import GuideVideoPlay
from app.repositories import wallet as crud_wallet
SCENES = ("coupon", "comparison")
_KEY_BY_SCENE = {
"coupon": "coupon_guide_video",
"comparison": "comparison_guide_video",
}
_KEY = "coupon_guide_video"
#: 金币流水 biz_type。客户端收益明细按它显示「新手引导视频奖励」。
BIZ_TYPE = "guide_video"
@@ -45,7 +41,7 @@ _DEFAULTS: dict[str, Any] = {
"enabled": True,
"video_url": None, # None/空 = 未配片 → 不下发,浮层照旧放广告
"max_plays": 3, # 每个账号前 N 次浮层放引导视频
"reward_coin": 100, # 每次固定金币
"reward_coin": 120, # 每次固定金币
}
_FIELDS = tuple(_DEFAULTS.keys())
@@ -69,38 +65,19 @@ def _merge(raw: Any) -> dict[str, Any]:
return out
def _config_key(scene: str) -> str:
if scene not in _KEY_BY_SCENE:
raise ValueError(f"unsupported guide video scene: {scene}")
return _KEY_BY_SCENE[scene]
def get_config(db: Session, scene: str = "coupon") -> dict[str, Any]:
def get_config(db: Session) -> dict[str, Any]:
"""完整配置 + updated_at(admin 读 / 业务读共用)。"""
row = db.get(AppConfig, _config_key(scene))
row = db.get(AppConfig, _KEY)
cfg = _merge(row.value if row is not None else None)
if scene == "coupon":
# 白名单里的引导视频规则只对应领券场景;比价仍使用自己的独立配置。
cfg["max_plays"] = limit_policy.resolve(
db, "guide.video.lifetime"
).global_limit
cfg["updated_at"] = row.updated_at.isoformat() if row is not None and row.updated_at else None
return cfg
def _write(
db: Session,
value: dict[str, Any],
*,
scene: str,
admin_id: int,
commit: bool,
) -> dict[str, Any]:
def _write(db: Session, value: dict[str, Any], *, admin_id: int, commit: bool) -> dict[str, Any]:
"""整体覆写该行(value 须为完整字段 dict),返回合并后的完整配置(含 updated_at)。"""
key = _config_key(scene)
row = db.get(AppConfig, key)
row = db.get(AppConfig, _KEY)
if row is None:
row = AppConfig(key=key, value=value, updated_by_admin_id=admin_id)
row = AppConfig(key=_KEY, value=value, updated_by_admin_id=admin_id)
db.add(row)
else:
row.value = value # 整体重新赋值,SQLAlchemy 才侦测得到变更
@@ -121,12 +98,11 @@ def update_config(
enabled: bool | None = None,
max_plays: int | None = None,
reward_coin: int | None = None,
scene: str = "coupon",
admin_id: int,
commit: bool = True,
) -> tuple[dict[str, Any], dict[str, Any]]:
"""改开关 / 次数 / 金币(只改传了的字段;视频走 set_video)。返回 (before, after) 供审计。"""
row = db.get(AppConfig, _config_key(scene))
row = db.get(AppConfig, _KEY)
before = _merge(row.value if row is not None else None)
new_value = {k: before[k] for k in _FIELDS}
if enabled is not None:
@@ -135,91 +111,45 @@ def update_config(
new_value["max_plays"] = max(0, min(int(max_plays), MAX_PLAYS_LIMIT))
if reward_coin is not None:
new_value["reward_coin"] = max(0, min(int(reward_coin), REWARD_COIN_LIMIT))
after = _write(
db,
new_value,
scene=scene,
admin_id=admin_id,
commit=False,
)
if scene == "coupon" and max_plays is not None:
# 兼容仍调用旧专用接口的客户端/脚本,并把统一策略全局值一并更新。
from app.core.config_schema import GUIDE_VIDEO_MAX_PLAYS_KEY
from app.repositories import app_config
app_config.set_value(
db,
GUIDE_VIDEO_MAX_PLAYS_KEY,
new_value["max_plays"],
admin_id=admin_id,
commit=False,
)
if commit:
db.commit()
after = get_config(db, scene)
after = _write(db, new_value, admin_id=admin_id, commit=commit)
return before, after
def set_video(
db: Session,
video_url: str | None,
*,
scene: str = "coupon",
admin_id: int,
commit: bool = True,
db: Session, video_url: str | None, *, admin_id: int, commit: bool = True
) -> tuple[dict[str, Any], dict[str, Any]]:
"""设置/清空引导视频地址。返回 (before, after);before['video_url'] 供调用方删旧文件。"""
row = db.get(AppConfig, _config_key(scene))
row = db.get(AppConfig, _KEY)
before = _merge(row.value if row is not None else None)
new_value = {k: before[k] for k in _FIELDS}
new_value["video_url"] = video_url
after = _write(
db,
new_value,
scene=scene,
admin_id=admin_id,
commit=commit,
)
after = _write(db, new_value, admin_id=admin_id, commit=commit)
return before, after
# ===== 播放计次 =====
def used_plays(
db: Session,
user_id: int,
*,
scene: str = "coupon",
reset_at: datetime | None = None,
) -> int:
def used_plays(db: Session, user_id: int) -> int:
"""该账号已用掉的引导视频次数(开播即算,含未发币的)。"""
stmt = select(func.count()).select_from(GuideVideoPlay).where(
GuideVideoPlay.user_id == user_id,
GuideVideoPlay.scene == scene,
)
if reset_at is not None:
# 该表的既有写入口使用北京时间 naive 墙钟,重置基线必须转换成
# 同一存储口径再比较;单独改成 UTC 会让存量/增量记录偏移 8 小时。
reset_value = reset_at.astimezone(rewards.CN_TZ).replace(tzinfo=None)
stmt = stmt.where(GuideVideoPlay.started_at >= reset_value)
return int(db.execute(stmt).scalar_one())
def play_stats(db: Session, scene: str = "coupon") -> dict[str, int]:
"""全站播放统计(admin 页展示):总播放次数 / 其中已发币次数。"""
total = int(
return int(
db.execute(
select(func.count()).select_from(GuideVideoPlay).where(
GuideVideoPlay.scene == scene
GuideVideoPlay.user_id == user_id
)
).scalar_one()
)
def play_stats(db: Session) -> dict[str, int]:
"""全站播放统计(admin 页展示):总播放次数 / 其中已发币次数。"""
total = int(
db.execute(select(func.count()).select_from(GuideVideoPlay)).scalar_one()
)
granted = int(
db.execute(
select(func.count()).select_from(GuideVideoPlay).where(
GuideVideoPlay.status == "granted",
GuideVideoPlay.scene == scene,
GuideVideoPlay.status == "granted"
)
).scalar_one()
)
@@ -238,22 +168,11 @@ def start_play(
reward_coin 播完/中途关闭都发的固定金币
seq / remaining 第几次 / 发完这次还剩几次(仅展示与排查用)
"""
_config_key(scene)
cfg = get_config(db, scene)
cfg = get_config(db)
video_url = (cfg.get("video_url") or "").strip()
policy = (
limit_policy.resolve_for_user(db, "guide.video.lifetime", user_id)
if scene == "coupon"
else None
)
max_plays = policy.limit if policy is not None else int(cfg.get("max_plays") or 0)
exposed_max_plays = max_plays if max_plays is not None else 1_000_000
max_plays = int(cfg.get("max_plays") or 0)
reward_coin = int(cfg.get("reward_coin") or 0)
used = (
used_plays(db, user_id, scene=scene, reset_at=policy.reset_at)
if policy is not None and policy.reset_at is not None
else used_plays(db, user_id, scene=scene)
)
used = used_plays(db, user_id)
def _miss(used_now: int) -> dict[str, Any]:
return {
@@ -262,29 +181,20 @@ def start_play(
"play_token": "",
"reward_coin": reward_coin,
"seq": used_now,
"remaining": max(0, exposed_max_plays - used_now),
"remaining": max(0, max_plays - used_now),
}
if (
not cfg.get("enabled")
or not video_url
or (max_plays is not None and (max_plays <= 0 or used >= max_plays))
):
if not cfg.get("enabled") or not video_url or max_plays <= 0 or used >= max_plays:
return _miss(used)
# ``seq`` remains scene 内 lifetime-monotonic because
# (user_id, scene, seq) is unique.
# A whitelist reset only changes the quota baseline; reusing seq=1 would
# collide with historical rows and make every post-reset start fail.
seq = used_plays(db, user_id, scene=scene) + 1
seq = used + 1
play = GuideVideoPlay(
user_id=user_id,
play_token=uuid.uuid4().hex,
scene=scene,
seq=seq,
video_url=video_url,
# 固化本次承诺发放的金币,避免运营改价后已开播记录按新价结算。
coin=reward_coin,
coin=0,
status="playing",
completed=0,
started_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
@@ -300,19 +210,14 @@ def start_play(
db.flush()
except IntegrityError:
db.rollback()
used_now = (
used_plays(db, user_id, scene=scene, reset_at=policy.reset_at)
if policy is not None and policy.reset_at is not None
else used_plays(db, user_id, scene=scene)
)
return _miss(used_now)
return _miss(used_plays(db, user_id))
return {
"should_play": True,
"video_url": video_url,
"play_token": play.play_token,
"reward_coin": reward_coin,
"seq": seq,
"remaining": max(0, exposed_max_plays - (used + 1)),
"remaining": max(0, max_plays - seq),
}
@@ -335,9 +240,8 @@ def grant_play(
重复上报返回 granted=False + 已发金币(客户端据此不重复累加 toast 金额)
"""
token = (play_token or "").strip()
# 金币额度取开播时由服务端固化的值,不信客户端,也不受后续配置变更影响
play = _find_play(db, user_id, token)
coin = int(play.coin if play is not None else 0)
# 金币额度以**服务端配置**为准,不信客户端(客户端只上报"播完/关闭")
coin = int(get_config(db).get("reward_coin") or 0)
# 幂等核心:把 status 放进 WHERE 做条件更新(compare-and-set),而不是"先读再判再写"。
# 「播完」与「✕ 关闭」抢跑、或客户端超时重试时,两个请求会都读到 status='playing',
+7 -29
View File
@@ -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))
+23 -85
View File
@@ -9,7 +9,6 @@ from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from app.core import limit_policy
from app.core.config_schema import (
RISK_COMPARE_DAILY_THRESHOLD_KEY,
RISK_ONECLICK_DAILY_THRESHOLD_KEY,
@@ -258,34 +257,22 @@ def _upsert_incident(
def evaluate_behavior_rule(
db: Session,
*,
rule_code: str,
subject_id: str,
at: datetime,
threshold: int | None = None,
subject_reset_at: datetime | None = None,
db: Session, *, rule_code: str, subject_id: str, at: datetime
) -> RiskIncident | None:
spec = RULES[rule_code]
effective_threshold = threshold or get_rule_threshold(db, rule_code)
threshold = get_rule_threshold(db, rule_code)
window_key, window_start, end = _window_bounds(at, spec.window)
reset_at = get_rule_reset_at(db, rule_code)
baselines = [value for value in (reset_at, subject_reset_at) if value is not None]
start = max(window_start, *baselines) if baselines else window_start
start = max(window_start, reset_at) if reset_at else window_start
count, first_at, last_at, triggered_at = _event_stats(
db,
spec,
subject_id=subject_id,
start=start,
end=end,
threshold=effective_threshold,
threshold=threshold,
)
if (
count < effective_threshold
or first_at is None
or last_at is None
or triggered_at is None
):
if count < threshold or first_at is None or last_at is None or triggered_at is None:
return None
return _upsert_incident(
db,
@@ -313,6 +300,7 @@ def reconcile_behavior_rule(
"""按当前阈值重算短信/一键登录当前窗口,并收起已不再命中的待处理告警。"""
spec = RULES[rule_code]
current = at or utcnow()
threshold = get_rule_threshold(db, rule_code)
window_key, window_start, end = _window_bounds(current, spec.window)
reset_at = get_rule_reset_at(db, rule_code)
start = max(window_start, reset_at) if reset_at else window_start
@@ -323,39 +311,23 @@ def reconcile_behavior_rule(
BehaviorEvent.occurred_at >= start,
BehaviorEvent.occurred_at < end,
)
subject_rows = db.execute(
qualifying_rows = db.execute(
select(
BehaviorEvent.subject_id,
func.max(BehaviorEvent.occurred_at),
func.max(BehaviorEvent.phone),
)
.where(*filters)
.group_by(BehaviorEvent.subject_id)
.having(func.count(BehaviorEvent.id) >= threshold)
).all()
qualifying: set[str] = set()
policy_code = {
RULE_SMS_HOURLY: "risk.sms.hourly",
RULE_ONECLICK_DAILY: "risk.oneclick.daily",
}[rule_code]
for subject_id, last_at, phone in subject_rows:
policy = limit_policy.resolve(
db,
policy_code,
phone=phone,
device=str(subject_id),
)
if policy.suppressed or policy.limit is None:
continue
incident = evaluate_behavior_rule(
qualifying = {str(subject_id) for subject_id, _ in qualifying_rows}
for subject_id, last_at in qualifying_rows:
evaluate_behavior_rule(
db,
rule_code=rule_code,
subject_id=str(subject_id),
at=last_at or current,
threshold=policy.limit,
subject_reset_at=policy.reset_at,
)
if incident is not None:
qualifying.add(str(subject_id))
open_incidents = db.scalars(
select(RiskIncident).where(
@@ -411,29 +383,7 @@ def record_behavior_event(
db.add(event)
db.flush()
if evaluate_rule:
policy_code = {
RULE_SMS_HOURLY: "risk.sms.hourly",
RULE_ONECLICK_DAILY: "risk.oneclick.daily",
}.get(evaluate_rule)
policy = (
limit_policy.resolve(
db,
policy_code,
phone=phone,
device=device_id or subject_id,
)
if policy_code
else None
)
if policy is None or not policy.suppressed:
evaluate_behavior_rule(
db,
rule_code=evaluate_rule,
subject_id=subject_id,
at=at,
threshold=policy.limit if policy else None,
subject_reset_at=policy.reset_at if policy else None,
)
evaluate_behavior_rule(db, rule_code=evaluate_rule, subject_id=subject_id, at=at)
if commit:
db.commit()
db.refresh(event)
@@ -446,33 +396,20 @@ def sync_compare_incident(
user_id: int,
at: datetime,
threshold: int | None = None,
device_id: str | None = None,
commit: bool = True,
) -> RiskIncident | None:
policy = limit_policy.resolve_for_user(
db,
"risk.compare.daily",
user_id,
device=device_id,
)
if policy.suppressed:
return None
effective_threshold = threshold or policy.limit or get_rule_threshold(
db, RULE_COMPARE_DAILY
)
effective_threshold = threshold or get_rule_threshold(db, RULE_COMPARE_DAILY)
# comparison_record 的既有写入口统一落“北京时间 naive”时间;这里必须沿用同一
# 口径,否则 SQLite/PG session timezone 不同时会把凌晨记录算到前一天。
local = at.astimezone(CN_TZ).replace(tzinfo=None) if at.tzinfo else at
window_start = local.replace(hour=0, minute=0, second=0, microsecond=0)
end = window_start + timedelta(days=1)
window_key = window_start.strftime("%Y-%m-%d")
global_reset_at = get_rule_reset_at(db, RULE_COMPARE_DAILY)
baselines = [
value.astimezone(CN_TZ).replace(tzinfo=None)
for value in (global_reset_at, policy.reset_at)
if value is not None
]
start = max(window_start, *baselines) if baselines else window_start
reset_at = get_rule_reset_at(db, RULE_COMPARE_DAILY)
reset_local = (
reset_at.astimezone(CN_TZ).replace(tzinfo=None) if reset_at else None
)
start = max(window_start, reset_local) if reset_local else window_start
filters = (
ComparisonRecord.user_id == user_id,
ComparisonRecord.created_at >= start,
@@ -527,6 +464,7 @@ def reconcile_compare_rule(
reset_at.astimezone(CN_TZ).replace(tzinfo=None) if reset_at else None
)
start = max(window_start, reset_local) if reset_local else window_start
threshold = get_rule_threshold(db, RULE_COMPARE_DAILY)
rows = db.execute(
select(
ComparisonRecord.user_id,
@@ -538,17 +476,17 @@ def reconcile_compare_rule(
ComparisonRecord.created_at < end,
)
.group_by(ComparisonRecord.user_id)
.having(func.count(ComparisonRecord.id) >= threshold)
).all()
qualifying: set[str] = set()
qualifying = {str(user_id) for user_id, _ in rows}
for user_id, last_at in rows:
incident = sync_compare_incident(
sync_compare_incident(
db,
user_id=int(user_id),
at=last_at or current,
threshold=threshold,
commit=False,
)
if incident is not None:
qualifying.add(str(user_id))
open_incidents = db.scalars(
select(RiskIncident).where(
+2 -12
View File
@@ -107,13 +107,6 @@ class ComparisonRecordIn(BaseModel):
# 明细
items: list[ComparisonItemIn] = Field(default_factory=list)
comparison_results: list[ComparisonResultIn] = Field(default_factory=list)
# 展示模型统一数组(pricebot done.params.platforms 原样透传): 每平台一行、自带
# status/is_best/display/display_order,记录页据此直渲染。宽松 list[dict] 存(结构由
# pricebot 定,server 只原样落库),前端读它、老记录空时回退 comparison_results。
platforms: list[dict] = Field(default_factory=list)
# 记录级结局(pricebot 下发): success/below_minimum/store_closed/failed。让"未满起送"不再
# 被塌缩成 failed。_derive 优先用它、其次客户端 status、再兜底二态派生。
record_status: str | None = None
# 逐平台结局摘要(含失败平台的细分原因 status: store_not_found/items_not_found/below_minimum/
# unsupported/...)。来自 done.params.platform_results,客户端透传;落 raw_payload(不单列),
# admin「卡在哪一步」从这里读。dict{platform_id: {...}} 宽松存(结构由 pricebot 定——是
@@ -184,9 +177,6 @@ class ComparisonRecordOut(BaseModel):
fail_reason: str | None = None
items: list = []
comparison_results: list = []
# 展示模型统一数组(每平台一行、自带 status/is_best/display/display_order): 记录页据此
# 直渲染, 不再靠 comparison_results + 前端派生。老记录为空 → 前端回退 comparison_results。
platforms: list = []
skipped_dish_names: list = []
total_ms: int | None = None
# 「已下单」(店级):该店名在该用户真实下单(source='compare')里出现过即 True。
@@ -222,9 +212,9 @@ class CompareStartReserveIn(BaseModel):
class CompareStartReserveOut(BaseModel):
limit: int | None
limit: int
used: int
remaining: int | None
remaining: int
class CompareStatsOut(BaseModel):
+2 -4
View File
@@ -1,15 +1,13 @@
"""新手引导视频(领券等候浮层前 N 次替代广告)的客户端请求/响应契约。"""
from __future__ import annotations
from typing import Literal
from pydantic import BaseModel, Field
class GuideVideoStartIn(BaseModel):
"""开播询问。领券与比价分别使用独立配置和独立次数"""
"""开播询问。scene 目前只有 coupon(领券浮层);预留给日后比价等场景"""
scene: Literal["coupon", "comparison"] = "coupon"
scene: str = Field(default="coupon", max_length=16)
class GuideVideoStartOut(BaseModel):
+1 -1
View File
@@ -135,7 +135,7 @@ def repair_missing_comparison_llm_costs(
select(ComparisonRecord.id, ComparisonRecord.trace_id)
.where(
*date_conditions,
ComparisonRecord.status.in_(("success", "failed", "cancelled")),
ComparisonRecord.status.in_(("success", "failed")),
ComparisonRecord.llm_cost_yuan.is_(None),
)
.order_by(ComparisonRecord.created_at.desc(), ComparisonRecord.id.desc())
+201
View File
@@ -0,0 +1,201 @@
"""分批修正历史比价记录的最优平台与节省金额派生列。
该回填刻意不放进 Alembic生产启动迁移不应一次性读取并更新全部历史记录
调用方按主键游标分批执行每批独立事务默认 dry-run 时只计算差异不写库
"""
from __future__ import annotations
import json
from dataclasses import dataclass
from decimal import ROUND_HALF_UP, Decimal, InvalidOperation
from typing import Any
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from app.models.comparison import ComparisonRecord
_DERIVED_FIELDS = (
"source_price_cents",
"best_platform_id",
"best_platform_name",
"best_price_cents",
"saved_amount_cents",
"is_source_best",
)
@dataclass(frozen=True)
class SavingsBackfillBatch:
"""一批回填结果;last_id 可直接作为下一批 after_id。"""
scanned: int
changed: int
last_id: int
done: bool
def _json_value(value: Any, fallback: Any) -> Any:
if isinstance(value, str):
try:
return json.loads(value)
except (TypeError, ValueError):
return fallback
return value
def _decimal(value: Any) -> Decimal | None:
if value is None or isinstance(value, bool):
return None
try:
return Decimal(str(value))
except (InvalidOperation, TypeError, ValueError):
return None
def _yuan_to_cents(value: Any) -> int | None:
amount = _decimal(value)
if amount is None:
return None
return int((amount * 100).quantize(Decimal("1"), rounding=ROUND_HALF_UP))
def _rank(value: Any) -> Decimal:
rank = _decimal(value)
return rank if rank is not None else Decimal("1e18")
def derive_historical_savings(
source_price_cents: int | None,
comparison_results: Any,
raw_payload: Any,
) -> dict[str, Any] | None:
"""按当前完整购物篮规则,从历史原始数据重新派生结构化金额字段。"""
results = _json_value(comparison_results, [])
raw = _json_value(raw_payload, {})
if not isinstance(results, list):
return None
platform_results = raw.get("platform_results", {}) if isinstance(raw, dict) else {}
if not isinstance(platform_results, dict):
platform_results = {}
clean: list[tuple[dict[str, Any], Decimal]] = []
for result in results:
if not isinstance(result, dict):
continue
price = _decimal(result.get("price"))
if price is None:
continue
platform_id = result.get("platform_id")
platform_info = platform_results.get(platform_id) if platform_id else None
skipped_count = (
_decimal(platform_info.get("skipped_dish_count"))
if isinstance(platform_info, dict)
else None
)
if skipped_count is not None and skipped_count > 0:
continue
clean.append((result, price))
if not clean:
return None
best, best_price = min(
clean,
key=lambda item: (_rank(item[0].get("rank")), item[1]),
)
if source_price_cents is None:
source = next(
(
result
for result in results
if isinstance(result, dict)
and result.get("is_source")
and _decimal(result.get("price")) is not None
),
None,
)
source_price_cents = _yuan_to_cents(source.get("price")) if source else None
best_price_cents = _yuan_to_cents(best_price)
saved_amount_cents = (
source_price_cents - best_price_cents
if source_price_cents is not None and best_price_cents is not None
else None
)
return {
"source_price_cents": source_price_cents,
"best_platform_id": best.get("platform_id"),
"best_platform_name": best.get("platform_name"),
"best_price_cents": best_price_cents,
"saved_amount_cents": saved_amount_cents,
"is_source_best": bool(best.get("is_source")),
}
def latest_success_id(db: Session) -> int:
"""固定本次任务的扫描上界,避免执行期间新增记录让任务无限延长。"""
return int(
db.execute(
select(func.coalesce(func.max(ComparisonRecord.id), 0)).where(
ComparisonRecord.status == "success"
)
).scalar_one()
)
def repair_comparison_savings_batch(
db: Session,
*,
after_id: int,
max_id: int,
batch_size: int = 500,
apply: bool = False,
) -> SavingsBackfillBatch:
"""按 id 游标处理一批;apply=True 时仅更新实际发生变化的记录并提交。"""
if after_id < 0 or max_id < 0:
raise ValueError("after_id and max_id must be non-negative")
if batch_size <= 0:
raise ValueError("batch_size must be positive")
records = list(
db.scalars(
select(ComparisonRecord)
.where(
ComparisonRecord.status == "success",
ComparisonRecord.id > after_id,
ComparisonRecord.id <= max_id,
)
.order_by(ComparisonRecord.id)
.limit(batch_size)
)
)
changed = 0
for record in records:
derived = derive_historical_savings(
record.source_price_cents,
record.comparison_results,
record.raw_payload,
)
if derived is None:
continue
differs = any(
getattr(record, field) != derived[field] for field in _DERIVED_FIELDS
)
if not differs:
continue
changed += 1
if apply:
for field in _DERIVED_FIELDS:
setattr(record, field, derived[field])
if apply:
db.commit()
last_id = records[-1].id if records else after_id
return SavingsBackfillBatch(
scanned=len(records),
changed=changed,
last_id=last_id,
done=not records or len(records) < batch_size or last_id >= max_id,
)
+103
View File
@@ -0,0 +1,103 @@
"""分批回填历史比价节省金额。
默认只预览不写库
python -m scripts.backfill_comparison_savings
确认预览后实际执行
python -m scripts.backfill_comparison_savings --apply
中断后可从日志最后一个 last_id 继续
python -m scripts.backfill_comparison_savings --apply --after-id 12345
"""
from __future__ import annotations
import argparse
from app.db.session import SessionLocal
from app.services.comparison_savings_backfill import (
latest_success_id,
repair_comparison_savings_batch,
)
def _positive_int(value: str) -> int:
parsed = int(value)
if parsed <= 0:
raise argparse.ArgumentTypeError("必须为正整数")
return parsed
def _non_negative_int(value: str) -> int:
parsed = int(value)
if parsed < 0:
raise argparse.ArgumentTypeError("不能为负数")
return parsed
def main() -> None:
parser = argparse.ArgumentParser(
description="按完整购物篮口径分批回填历史比价节省金额(默认 dry-run)"
)
parser.add_argument("--apply", action="store_true", help="实际写库;默认仅预览")
parser.add_argument("--batch-size", type=_positive_int, default=500)
parser.add_argument(
"--after-id",
type=_non_negative_int,
default=0,
help="只处理大于该主键的记录,用于断点续跑",
)
parser.add_argument(
"--max-id",
type=_non_negative_int,
help="固定扫描上界;不传则启动时取成功记录最大主键",
)
parser.add_argument(
"--max-batches",
type=_positive_int,
help="本次最多处理多少批,便于灰度限量执行",
)
args = parser.parse_args()
with SessionLocal() as db:
max_id = args.max_id if args.max_id is not None else latest_success_id(db)
mode = "APPLY" if args.apply else "DRY-RUN"
after_id = args.after_id
scanned = 0
changed = 0
batches = 0
print(
f"comparison savings backfill mode={mode} "
f"after_id={after_id} max_id={max_id} batch_size={args.batch_size}"
)
while after_id < max_id:
with SessionLocal() as db:
result = repair_comparison_savings_batch(
db,
after_id=after_id,
max_id=max_id,
batch_size=args.batch_size,
apply=args.apply,
)
batches += 1
scanned += result.scanned
changed += result.changed
after_id = result.last_id
print(
f"batch={batches} scanned={result.scanned} changed={result.changed} "
f"last_id={result.last_id} done={result.done}"
)
if result.done or (
args.max_batches is not None and batches >= args.max_batches
):
break
print(
f"completed mode={mode} batches={batches} scanned={scanned} "
f"changed={changed} last_id={after_id} max_id={max_id}"
)
if __name__ == "__main__":
main()
+4 -4
View File
@@ -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"] == [
+49 -109
View File
@@ -120,6 +120,55 @@ def test_dashboard_period_comparison_is_aggregated_by_backend(
assert comparison["token_cost_total_yuan"] == pytest.approx(1.0)
def test_dashboard_average_saved_includes_zero_and_excludes_uncalculable(
admin_client: TestClient, admin_token: str
) -> None:
created_at = datetime(2037, 2, 15, 12)
db = SessionLocal()
try:
db.add_all(
[
ComparisonRecord(
trace_id="dashboard-saved-positive",
status="success",
saved_amount_cents=100,
created_at=created_at,
),
ComparisonRecord(
trace_id="dashboard-saved-source-best",
status="success",
saved_amount_cents=0,
created_at=created_at,
),
ComparisonRecord(
trace_id="dashboard-saved-uncalculable",
status="success",
saved_amount_cents=None,
created_at=created_at,
),
ComparisonRecord(
trace_id="dashboard-saved-failed",
status="failed",
saved_amount_cents=900,
created_at=created_at,
),
]
)
db.commit()
finally:
db.close()
response = admin_client.get(
"/admin/api/stats/overview",
params={"date_from": "2037-02-15", "date_to": "2037-02-15"},
headers=_auth(admin_token),
)
assert response.status_code == 200, response.text
# (100 + 0) / 2NULL 无法计算,failed 不属于成功比价。
assert response.json()["period"]["comparison"]["average_saved_cents"] == 50
def test_user_list_and_detail(admin_client: TestClient, admin_token: str) -> None:
uid = _seed_user_with_data("13800000002")
r = admin_client.get("/admin/api/users", headers=_auth(admin_token))
@@ -219,115 +268,6 @@ def test_user_reward_stats_can_scope_withdrawals_by_account(
assert invite.json()["cash_balance_cents"] == 456
def test_user_reward_stats_draw_ecpm_uses_all_filtered_impressions(
admin_client: TestClient, admin_token: str
) -> None:
"""Draw 平均 eCPM 应与收益报表一致,不能只平均成功发奖记录。"""
from app.models.ad_ecpm import AdEcpmRecord
from app.models.ad_feed_reward import AdFeedRewardRecord
uid = _seed_user_with_data("13800000024")
created_at = datetime(2038, 1, 15, 4, tzinfo=UTC)
db = SessionLocal()
try:
db.add_all(
[
AdFeedRewardRecord(
client_event_id="reward-stats-granted-high",
ad_session_id="reward-stats-granted-high",
user_id=uid,
reward_date="2038-01-15",
duration_seconds=10,
unit_count=1,
ecpm_raw="9000",
ad_type="draw",
feed_scene="coupon",
app_env="prod",
our_code_id="104098712",
coin=9,
status="granted",
created_at=created_at,
),
AdEcpmRecord(
user_id=uid,
ad_type="draw",
feed_scene="coupon",
ad_session_id="reward-stats-impression-low",
app_env="prod",
our_code_id="104098712",
ecpm_raw="1000",
report_date="2038-01-15",
created_at=created_at,
),
AdEcpmRecord(
user_id=uid,
ad_type="feed",
feed_scene="coupon",
ad_session_id="reward-stats-impression-mid",
app_env="prod",
our_code_id="104098712",
ecpm_raw="3000",
report_date="2038-01-15",
created_at=created_at,
),
# 同用户但不同场景/环境/非业务代码位,均不应进入本次详情筛选。
AdEcpmRecord(
user_id=uid,
ad_type="draw",
feed_scene="comparison",
ad_session_id="reward-stats-other-scene",
app_env="prod",
our_code_id="104098712",
ecpm_raw="7000",
report_date="2038-01-15",
created_at=created_at,
),
AdEcpmRecord(
user_id=uid,
ad_type="draw",
feed_scene="coupon",
ad_session_id="reward-stats-test-env",
app_env="test",
our_code_id="104127529",
ecpm_raw="8000",
report_date="2038-01-15",
created_at=created_at,
),
AdEcpmRecord(
user_id=uid,
ad_type="draw",
feed_scene="coupon",
ad_session_id="reward-stats-non-business",
app_env="prod",
our_code_id="demo-slot",
ecpm_raw="9000",
report_date="2038-01-15",
created_at=created_at,
),
]
)
db.commit()
finally:
db.close()
response = admin_client.get(
f"/admin/api/users/{uid}/reward-stats",
params={
"date_from": "2038-01-15T00:00:00Z",
"date_to": "2038-01-15T23:59:59Z",
"app_env": "prod",
"revenue_scope": "business",
"feed_scene": "coupon",
},
headers=_auth(admin_token),
)
assert response.status_code == 200, response.text
data = response.json()
assert data["feed_count"] == 1
# 全部真实展示 (1000 + 3000) / 2;不能返回成功发奖记录的 9000。
assert data["feed_avg_ecpm"] == 2000.0
def test_user_coin_record_sort_accepts_mixed_timezone_datetimes() -> None:
"""线上 PostgreSQL 返回 awareSQLite/历史转换可能返回 naive,二者必须可混排。"""
naive = datetime(2038, 1, 1, 8, 0)
+2 -7
View File
@@ -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",
}
+1 -4
View File
@@ -263,9 +263,7 @@ def test_trace_finalize_harvests_abort(client) -> None:
with SessionLocal() as db: # 先有 running 行(帧0建的)
crud.harvest_running(db, trace_id=tid, user_id=None)
p, _cap = _mock_pricebot({"trace_url": "https://price.shaguabijia.com/traces/fin/"})
with p, patch(
"app.api.v1.compare.backfill_comparison_llm_cost"
) as backfill:
with p:
r = client.post("/api/v1/trace/finalize",
json={"trace_id": tid, "status": "cancelled", "reason": "用户终止"})
assert r.status_code == 200
@@ -273,7 +271,6 @@ def test_trace_finalize_harvests_abort(client) -> None:
rec = _get(db, tid)
assert rec is not None and rec.status == "cancelled"
assert rec.trace_url.endswith("/fin/")
backfill.assert_called_once_with(rec.id, tid)
def test_price_step_binds_user_when_authed(client) -> None:
-4
View File
@@ -72,7 +72,6 @@ def test_backfill_retries_then_persists_cost(monkeypatch):
def test_repair_batch_only_targets_terminal_missing_rows(monkeypatch):
missing_id = _record("llm-repair-missing")
cancelled_id = _record("llm-repair-cancelled", status="cancelled")
running_id = _record("llm-repair-running", status="running")
calls = [
{
@@ -97,15 +96,12 @@ def test_repair_batch_only_targets_terminal_missing_rows(monkeypatch):
)
assert result["repaired"] >= 1
assert "llm-repair-missing" in seen
assert "llm-repair-cancelled" in seen
assert "llm-repair-running" not in seen
with SessionLocal() as db:
assert db.get(ComparisonRecord, missing_id).llm_cost_yuan is not None
assert db.get(ComparisonRecord, cancelled_id).llm_cost_yuan is not None
assert db.get(ComparisonRecord, running_id).llm_cost_yuan is None
finally:
_delete(missing_id)
_delete(cancelled_id)
_delete(running_id)
+136
View File
@@ -0,0 +1,136 @@
from __future__ import annotations
from datetime import datetime
from app.db.session import SessionLocal
from app.models.comparison import ComparisonRecord
from app.services.comparison_savings_backfill import (
latest_success_id,
repair_comparison_savings_batch,
)
def _record(trace_id: str, *, status: str = "success") -> ComparisonRecord:
return ComparisonRecord(
trace_id=trace_id,
status=status,
source_price_cents=4200,
best_platform_id="missing-dishes",
best_platform_name="Incomplete",
best_price_cents=2500,
saved_amount_cents=1700,
is_source_best=False,
comparison_results=[
{
"platform_id": "source",
"platform_name": "Source",
"price": 42,
"rank": 3,
"is_source": True,
},
{
"platform_id": "missing-dishes",
"platform_name": "Incomplete",
"price": 25,
"rank": 1,
"is_source": False,
},
{
"platform_id": "complete",
"platform_name": "Complete",
"price": 38.5,
"rank": 2,
"is_source": False,
},
],
raw_payload={
"platform_results": {
"missing-dishes": {"skipped_dish_count": 1},
"complete": {"skipped_dish_count": 0},
}
},
created_at=datetime(2037, 3, 1, 12),
)
def test_comparison_savings_backfill_is_batched_dry_run_and_idempotent() -> None:
db = SessionLocal()
try:
affected = _record("savings-backfill-affected")
unchanged = _record("savings-backfill-unchanged")
unchanged.best_platform_id = "complete"
unchanged.best_platform_name = "Complete"
unchanged.best_price_cents = 3850
unchanged.saved_amount_cents = 350
failed = _record("savings-backfill-failed", status="failed")
db.add_all([affected, unchanged, failed])
db.commit()
affected_id = affected.id
unchanged_id = unchanged.id
failed_id = failed.id
finally:
db.close()
db = SessionLocal()
try:
assert latest_success_id(db) >= unchanged_id
max_id = unchanged_id
affected = db.get(ComparisonRecord, affected_id)
assert affected is not None
preview = repair_comparison_savings_batch(
db,
after_id=affected_id - 1,
max_id=max_id,
batch_size=1,
)
assert preview.scanned == 1
assert preview.changed == 1
assert preview.last_id == affected_id
assert not preview.done
db.refresh(affected)
assert affected.best_platform_id == "missing-dishes"
assert affected.saved_amount_cents == 1700
applied = repair_comparison_savings_batch(
db,
after_id=affected_id - 1,
max_id=max_id,
batch_size=1,
apply=True,
)
assert applied.changed == 1
db.refresh(affected)
assert affected.best_platform_id == "complete"
assert affected.best_platform_name == "Complete"
assert affected.best_price_cents == 3850
assert affected.saved_amount_cents == 350
second_batch = repair_comparison_savings_batch(
db,
after_id=applied.last_id,
max_id=max_id,
batch_size=1,
apply=True,
)
assert second_batch.scanned == 1
assert second_batch.changed == 0
assert second_batch.last_id == unchanged_id
assert second_batch.done
rerun = repair_comparison_savings_batch(
db,
after_id=affected_id - 1,
max_id=max_id,
batch_size=10,
apply=True,
)
assert rerun.scanned == 2
assert rerun.changed == 0
failed_row = db.get(ComparisonRecord, failed_id)
assert failed_row is not None
assert failed_row.best_platform_id == "missing-dishes"
assert failed_row.saved_amount_cents == 1700
finally:
db.close()
+2 -102
View File
@@ -75,7 +75,7 @@ def _assert_ledger_balanced(user_id: int) -> None:
@pytest.fixture()
def guide_configured():
"""给全局配置塞一支片子(默认 max_plays=3 / reward_coin=100),用完还原成未配片。"""
"""给全局配置塞一支片子(默认 max_plays=3 / reward_coin=120),用完还原成未配片。"""
with SessionLocal() as db:
crud_guide.set_video(db, VIDEO_URL, admin_id=1)
cfg = crud_guide.get_config(db)
@@ -201,11 +201,7 @@ def test_start_loses_seq_race_degrades_to_ad(client, guide_configured, monkeypat
assert first["should_play"] is True and first["seq"] == 1
# 本次请求读到的是过期计数 → 仍会算出 seq=1
monkeypatch.setattr(
crud_guide,
"used_plays",
lambda db, user_id, *, scene="coupon", reset_at=None: 0,
)
monkeypatch.setattr(crud_guide, "used_plays", lambda db, user_id: 0)
with SessionLocal() as db:
result = crud_guide.start_play(db, uid)
@@ -265,99 +261,3 @@ def test_reward_loses_race_does_not_double_mint(client, guide_configured) -> Non
assert _coin_balance(uid) == coin, "一次播放只能发一次币"
assert len(_guide_txns(uid)) == 1, "一个 play_token 只能有一条金币流水"
_assert_ledger_balanced(uid)
def test_coupon_and_comparison_configs_and_counts_are_independent(client) -> None:
phone = "13920000008"
token = _login(client, phone)
uid = _user_id(phone)
comparison_url = "/media/guide_video/pytest_comparison.mp4"
with SessionLocal() as db:
coupon_before = crud_guide.get_config(db, "coupon")
comparison_before = crud_guide.get_config(db, "comparison")
crud_guide.set_video(db, VIDEO_URL, scene="coupon", admin_id=1)
crud_guide.update_config(
db,
scene="comparison",
enabled=True,
max_plays=1,
reward_coin=321,
admin_id=1,
)
crud_guide.set_video(
db,
comparison_url,
scene="comparison",
admin_id=1,
)
try:
coupon = client.post(
"/api/v1/guide-video/start",
json={"scene": "coupon"},
headers=_auth(token),
).json()
comparison = client.post(
"/api/v1/guide-video/start",
json={"scene": "comparison"},
headers=_auth(token),
).json()
comparison_blocked = client.post(
"/api/v1/guide-video/start",
json={"scene": "comparison"},
headers=_auth(token),
).json()
assert coupon["should_play"] is True
assert coupon["video_url"] == VIDEO_URL
assert coupon["seq"] == 1
assert comparison["should_play"] is True
assert comparison["video_url"] == comparison_url
assert comparison["reward_coin"] == 321
assert comparison["seq"] == 1
assert comparison_blocked["should_play"] is False
with SessionLocal() as db:
assert crud_guide.play_stats(db, "coupon")["total_plays"] >= 1
assert crud_guide.play_stats(db, "comparison")["total_plays"] == 1
rows = list(
db.scalars(
select(GuideVideoPlay)
.where(GuideVideoPlay.user_id == uid)
.order_by(GuideVideoPlay.scene)
)
)
assert [(row.scene, row.seq) for row in rows] == [
("comparison", 1),
("coupon", 1),
]
finally:
with SessionLocal() as db:
crud_guide.update_config(
db,
scene="coupon",
enabled=coupon_before["enabled"],
reward_coin=coupon_before["reward_coin"],
admin_id=1,
)
crud_guide.set_video(
db,
coupon_before["video_url"],
scene="coupon",
admin_id=1,
)
crud_guide.update_config(
db,
scene="comparison",
enabled=comparison_before["enabled"],
max_plays=comparison_before["max_plays"],
reward_coin=comparison_before["reward_coin"],
admin_id=1,
)
crud_guide.set_video(
db,
comparison_before["video_url"],
scene="comparison",
admin_id=1,
)
-943
View File
@@ -1,943 +0,0 @@
from __future__ import annotations
from copy import deepcopy
from datetime import UTC, datetime, timedelta
from uuid import uuid4
import pytest
from fastapi.testclient import TestClient
from app.admin.main import admin_app
from app.admin.repositories import admin_user as admin_repo
from app.admin.security import create_admin_token
from app.api.v1 import auth as auth_api
from app.core import limit_policy
from app.core.config_schema import (
AD_FEED_DAILY_LIMIT_KEY,
AD_REWARD_VIDEO_DAILY_LIMIT_KEY,
GUIDE_VIDEO_MAX_PLAYS_KEY,
PHONE_REBIND_DAYS_KEY,
SMS_PHONE_COOLDOWN_SECONDS_KEY,
)
from app.core.security import create_token
from app.db.session import SessionLocal
from app.main import app
from app.models.app_config import AppConfig
from app.models.comparison import ComparisonRecord
from app.models.limit_policy import LimitPolicyOverride
from app.models.risk import RiskIncident
from app.repositories import risk as risk_repo
from app.repositories import user as user_repo
def _snapshot_configs(keys: list[str]) -> dict[str, tuple[bool, object, int | None]]:
with SessionLocal() as db:
snapshot = {}
for key in keys:
row = db.get(AppConfig, key)
snapshot[key] = (
row is not None,
deepcopy(row.value) if row is not None else None,
row.updated_by_admin_id if row is not None else None,
)
return snapshot
def _restore_configs(snapshot: dict[str, tuple[bool, object, int | None]]) -> None:
with SessionLocal() as db:
for key, (existed, value, admin_id) in snapshot.items():
row = db.get(AppConfig, key)
if not existed:
if row is not None:
db.delete(row)
continue
if row is None:
db.add(
AppConfig(
key=key,
value=deepcopy(value),
updated_by_admin_id=admin_id,
)
)
else:
row.value = deepcopy(value)
row.updated_by_admin_id = admin_id
db.commit()
@pytest.fixture()
def admin_headers() -> dict[str, str]:
username = f"limit_admin_{uuid4().hex[:8]}"
with SessionLocal() as db:
admin = admin_repo.create_admin(
db,
username=username,
password="limit-admin-pass",
role="super_admin",
)
admin_id = admin.id
token, _ = create_admin_token(admin_id=admin_id, role="super_admin")
return {"Authorization": f"Bearer {token}"}
@pytest.fixture()
def custom_admin_headers() -> dict[str, str]:
username = f"limit_custom_{uuid4().hex[:8]}"
with SessionLocal() as db:
admin = admin_repo.create_admin(
db,
username=username,
password="limit-admin-pass",
role="custom",
pages_override=["limit-whitelist"],
)
admin_id = admin.id
token, _ = create_admin_token(admin_id=admin_id, role="custom")
return {"Authorization": f"Bearer {token}"}
def test_custom_admin_with_page_permission_can_manage_whitelist(
custom_admin_headers,
) -> None:
phone = f"139{int(uuid4().hex[:8], 16) % 100000000:08d}"
expires_at = (datetime.now(UTC) + timedelta(hours=2)).isoformat()
with TestClient(admin_app) as client:
created = client.post(
"/admin/api/limit-whitelist/bulk",
headers=custom_admin_headers,
json={
"subject_type": "phone",
"subject_value": phone,
"rule_codes": ["compare.start.daily"],
"expires_at": expires_at,
"reason": "custom role page permission",
},
)
assert created.status_code == 201, created.text
override_id = created.json()[0]["id"]
deleted = client.delete(
f"/admin/api/limit-whitelist/{override_id}",
headers=custom_admin_headers,
)
assert deleted.status_code == 204, deleted.text
def test_bulk_create_reactivates_existing_rows(admin_headers) -> None:
phone = f"139{int(uuid4().hex[:8], 16) % 100000000:08d}"
first_expiry = (datetime.now(UTC) + timedelta(hours=1)).isoformat()
second_expiry = (datetime.now(UTC) + timedelta(hours=3)).isoformat()
with TestClient(admin_app) as client:
first = client.post(
"/admin/api/limit-whitelist/bulk",
headers=admin_headers,
json={
"subject_type": "phone",
"subject_value": phone,
"rule_codes": ["compare.start.daily"],
"expires_at": first_expiry,
"reason": "first period",
},
)
assert first.status_code == 201, first.text
first_id = first.json()[0]["id"]
reset = client.post(
f"/admin/api/limit-whitelist/{first_id}/reset",
headers=admin_headers,
)
assert reset.status_code == 200, reset.text
recreated = client.post(
"/admin/api/limit-whitelist/bulk",
headers=admin_headers,
json={
"subject_type": "phone",
"subject_value": phone,
"rule_codes": [
"compare.start.daily",
"sms.phone.cooldown",
],
"expires_at": second_expiry,
"reason": "renewed period",
},
)
assert recreated.status_code == 201, recreated.text
rows = {item["rule_code"]: item for item in recreated.json()}
assert rows["compare.start.daily"]["id"] == first_id
assert rows["compare.start.daily"]["enabled"] is True
assert rows["compare.start.daily"]["status"] == "active"
assert rows["compare.start.daily"]["reason"] == "renewed period"
assert rows["sms.phone.cooldown"]["enabled"] is True
def test_admin_whitelist_crud_and_policy_precedence(admin_headers) -> None:
suffix = uuid4().hex[:10]
phone = f"139{int(suffix[:8], 16) % 100000000:08d}"
device = f"limit-device-{suffix}"
expires_at = (datetime.now(UTC) + timedelta(hours=2)).isoformat()
with TestClient(admin_app) as client:
rules = client.get(
"/admin/api/limit-whitelist/rules",
headers=admin_headers,
)
assert rules.status_code == 200, rules.text
assert {item["code"] for item in rules.json()} >= {
"compare.start.daily",
"sms.send.hourly",
"ad.reward_video.daily",
"risk.compare.daily",
}
rules_by_code = {item["code"]: item for item in rules.json()}
assert all(
"inherit" not in item["allowed_modes"] for item in rules.json()
)
assert all(
"override" not in item["allowed_modes"] for item in rules.json()
)
assert "unlimited" in rules_by_code["sms.phone.cooldown"]["allowed_modes"]
assert "unlimited" in rules_by_code["sms.code.failed_attempts"]["allowed_modes"]
assert "suppress_alert" in rules_by_code["risk.compare.daily"]["allowed_modes"]
removed_inherit = client.post(
"/admin/api/limit-whitelist",
headers=admin_headers,
json={
"subject_type": "phone",
"subject_value": phone,
"rule_code": "compare.start.daily",
"mode": "inherit",
"enabled": True,
},
)
assert removed_inherit.status_code == 422
invalid_alert = client.post(
"/admin/api/limit-whitelist",
headers=admin_headers,
json={
"subject_type": "phone",
"subject_value": phone,
"rule_code": "compare.start.daily",
"mode": "suppress_alert",
"enabled": True,
"reason": "invalid",
},
)
assert invalid_alert.status_code == 400
missing_expiry = client.post(
"/admin/api/limit-whitelist",
headers=admin_headers,
json={
"subject_type": "phone",
"subject_value": phone,
"rule_code": "compare.start.daily",
"mode": "unlimited",
"enabled": True,
"reason": "temporary QA",
},
)
assert missing_expiry.status_code == 400
expired_unlimited = client.post(
"/admin/api/limit-whitelist",
headers=admin_headers,
json={
"subject_type": "phone",
"subject_value": phone,
"rule_code": "compare.start.daily",
"mode": "unlimited",
"enabled": True,
"expires_at": "2020-01-01T00:00:00Z",
"reason": "expired temporary policy",
},
)
assert expired_unlimited.status_code == 400
assert "晚于当前时间" in expired_unlimited.json()["detail"]
removed_override = client.post(
"/admin/api/limit-whitelist",
headers=admin_headers,
json={
"subject_type": "phone",
"subject_value": phone,
"rule_code": "compare.start.daily",
"mode": "override",
"limit_value": 2,
"enabled": True,
"expires_at": expires_at,
},
)
assert removed_override.status_code == 422
phone_created = client.post(
"/admin/api/limit-whitelist",
headers=admin_headers,
json={
"subject_type": "phone",
"subject_value": phone,
"rule_code": "compare.start.daily",
"mode": "unlimited",
"enabled": True,
"expires_at": expires_at,
"reason": "temporary QA",
},
)
assert phone_created.status_code == 201, phone_created.text
phone_id = phone_created.json()["id"]
assert phone_created.json()["effective_limit"] is None
phone_updated = client.patch(
f"/admin/api/limit-whitelist/{phone_id}",
headers=admin_headers,
json={
"enabled": True,
"reason": "updated from admin",
},
)
assert phone_updated.status_code == 200, phone_updated.text
assert phone_updated.json()["effective_limit"] is None
assert phone_updated.json()["reason"] == "updated from admin"
with SessionLocal() as db:
persisted = db.get(LimitPolicyOverride, phone_id)
assert persisted is not None
assert persisted.mode == "unlimited"
assert persisted.limit_value is None
assert persisted.reason == "updated from admin"
removed_patch_fields = client.patch(
f"/admin/api/limit-whitelist/{phone_id}",
headers=admin_headers,
json={"mode": "override", "limit_value": 4},
)
assert removed_patch_fields.status_code == 422
inverted_time = client.patch(
f"/admin/api/limit-whitelist/{phone_id}",
headers=admin_headers,
json={
"starts_at": "2030-01-02T00:00:00Z",
"expires_at": "2030-01-01T00:00:00Z",
},
)
assert inverted_time.status_code == 400
assert "晚于生效时间" in inverted_time.json()["detail"]
duplicate = client.post(
"/admin/api/limit-whitelist",
headers=admin_headers,
json={
"subject_type": "phone",
"subject_value": phone,
"rule_code": "compare.start.daily",
"mode": "unlimited",
"enabled": True,
"expires_at": expires_at,
},
)
assert duplicate.status_code == 409
device_created = client.post(
"/admin/api/limit-whitelist",
headers=admin_headers,
json={
"subject_type": "device",
"subject_value": device,
"rule_code": "compare.start.daily",
"mode": "unlimited",
"enabled": True,
"expires_at": expires_at,
"reason": "device QA",
},
)
assert device_created.status_code == 201, device_created.text
device_id = device_created.json()["id"]
assert device_created.json()["effective_limit"] is None
with SessionLocal() as db:
effective = limit_policy.resolve(
db,
"compare.start.daily",
phone=phone,
device=device,
)
assert effective.unlimited is True
assert effective.matched_subject_type == "device"
restored = client.post(
f"/admin/api/limit-whitelist/{phone_id}/reset",
headers=admin_headers,
)
assert restored.status_code == 200, restored.text
assert restored.json()["enabled"] is False
assert restored.json()["status"] == "disabled"
assert restored.json()["effective_limit"] == restored.json()["global_limit"]
assert restored.json()["reset_at"] is None
with SessionLocal() as db:
effective = limit_policy.resolve(
db,
"compare.start.daily",
phone=phone,
device=None,
)
assert effective.override_id is None
assert effective.limit == effective.global_limit
ordered = client.get(
"/admin/api/limit-whitelist",
headers=admin_headers,
params={"rule_code": "compare.start.daily", "limit": 500},
)
assert ordered.status_code == 200, ordered.text
ordered_ids = [item["id"] for item in ordered.json()["items"]]
assert ordered_ids.index(device_id) < ordered_ids.index(phone_id)
listed = client.get(
"/admin/api/limit-whitelist",
headers=admin_headers,
params={"keyword": suffix[:5]},
)
assert listed.status_code == 200
assert listed.json()["total"] >= 1
deleted = client.delete(
f"/admin/api/limit-whitelist/{device_id}",
headers=admin_headers,
)
assert deleted.status_code == 204
client.delete(
f"/admin/api/limit-whitelist/{phone_id}",
headers=admin_headers,
)
def test_device_candidates_are_rule_aware_and_searchable(admin_headers) -> None:
suffix = uuid4().hex[:8]
phone = f"135{int(suffix, 16) % 100000000:08d}"
auth_device = f"android-id-{suffix}"
compare_device = f"device-PJZ110-{suffix}"
now = datetime.now(UTC).replace(microsecond=0)
with SessionLocal() as db:
user = user_repo.upsert_user_for_login(
db,
phone=phone,
register_channel="sms",
)
user.nickname = f"候选设备用户{suffix}"
risk_repo.record_behavior_event(
db,
event_type=risk_repo.EVENT_ONECLICK_LOGIN,
subject_type="device",
subject_id=auth_device,
user_id=user.id,
device_id=auth_device,
device_model="OPPO Find X8",
phone=phone,
outcome="success",
occurred_at=now,
)
db.add(
ComparisonRecord(
user_id=user.id,
device_id=compare_device,
trace_id=f"device-candidate-{suffix}",
device_model="PJZ110",
status="success",
items=[],
comparison_results=[],
skipped_dish_names=[],
created_at=now.replace(tzinfo=None),
)
)
db.commit()
user_id = user.id
username = user.username
with TestClient(admin_app) as client:
for keyword in (phone, str(user_id), username, "Find X8"):
response = client.get(
"/admin/api/limit-whitelist/device-candidates",
headers=admin_headers,
params={
"rule_code": "risk.oneclick.daily",
"keyword": keyword,
},
)
assert response.status_code == 200, response.text
item = next(
row for row in response.json() if row["device_id"] == auth_device
)
assert item["source_label"] == "一键登录"
assert item["user_id"] == user_id
assert item["phone"] == phone
assert item["device_model"] == "OPPO Find X8"
comparison = client.get(
"/admin/api/limit-whitelist/device-candidates",
headers=admin_headers,
params={
"rule_code": "compare.start.daily",
"keyword": "PJZ110",
},
)
assert comparison.status_code == 200, comparison.text
assert len(comparison.json()) == 1
comparison_item = comparison.json()[0]
assert comparison_item["device_id"] == compare_device
assert comparison_item["source"] == "comparison_record"
assert comparison_item["source_label"] == "比价记录"
assert comparison_item["user_id"] == user_id
assert comparison_item["username"] == username
assert comparison_item["phone"] == phone
assert comparison_item["nickname"] == f"候选设备用户{suffix}"
assert comparison_item["device_model"] == "PJZ110"
assert comparison_item["last_active_at"].startswith(
now.replace(tzinfo=None).isoformat()
)
phone_only = client.get(
"/admin/api/limit-whitelist/device-candidates",
headers=admin_headers,
params={"rule_code": "risk.compare.daily"},
)
assert phone_only.status_code == 400
assert "不支持设备白名单" in phone_only.json()["detail"]
def test_legacy_ip_is_not_a_device_candidate_or_valid_whitelist_subject(
admin_headers,
) -> None:
suffix = uuid4().hex[:8]
legacy_device = f"legacy-ip:203.0.113.{int(suffix[:2], 16) % 200 + 1}"
expires_at = (datetime.now(UTC) + timedelta(hours=2)).isoformat()
with SessionLocal() as db:
risk_repo.record_behavior_event(
db,
event_type=risk_repo.EVENT_SMS_SEND,
subject_type="device",
subject_id=legacy_device,
device_id=None,
phone=f"136{int(suffix, 16) % 100000000:08d}",
outcome="success",
)
with TestClient(admin_app) as client:
candidates = client.get(
"/admin/api/limit-whitelist/device-candidates",
headers=admin_headers,
params={
"rule_code": "sms.send.hourly",
"keyword": legacy_device,
},
)
assert candidates.status_code == 200, candidates.text
assert all(
row["device_id"] != legacy_device for row in candidates.json()
)
single = client.post(
"/admin/api/limit-whitelist",
headers=admin_headers,
json={
"subject_type": "device",
"subject_value": legacy_device,
"rule_code": "sms.send.hourly",
"mode": "unlimited",
"enabled": True,
"expires_at": expires_at,
},
)
assert single.status_code == 400, single.text
assert "未上报真实设备 ID" in single.json()["detail"]
bulk = client.post(
"/admin/api/limit-whitelist/bulk",
headers=admin_headers,
json={
"subject_type": "device",
"subject_value": legacy_device,
"rule_codes": ["sms.send.hourly", "sms.send.daily"],
"enabled": True,
"expires_at": expires_at,
},
)
assert bulk.status_code == 400, bulk.text
assert "未上报真实设备 ID" in bulk.json()["detail"]
def test_bulk_create_automatically_selects_unlimited_and_suppress_modes(
admin_headers,
) -> None:
suffix = uuid4().hex[:8]
phone = f"137{int(suffix, 16) % 100000000:08d}"
expires_at = (datetime.now(UTC) + timedelta(hours=2)).isoformat()
with TestClient(admin_app) as client:
created = client.post(
"/admin/api/limit-whitelist/bulk",
headers=admin_headers,
json={
"subject_type": "phone",
"subject_value": phone,
"rule_codes": [
"compare.start.daily",
"risk.compare.daily",
],
"expires_at": expires_at,
"reason": "批量白名单测试",
},
)
assert created.status_code == 201, created.text
rows = {item["rule_code"]: item for item in created.json()}
assert rows["compare.start.daily"]["mode"] == "unlimited"
assert rows["compare.start.daily"]["effective_limit"] is None
assert rows["risk.compare.daily"]["mode"] == "suppress_alert"
assert rows["risk.compare.daily"]["effective_limit"] == rows[
"risk.compare.daily"
]["global_limit"]
duplicate_batch = client.post(
"/admin/api/limit-whitelist/bulk",
headers=admin_headers,
json={
"subject_type": "phone",
"subject_value": phone,
"rule_codes": [
"compare.start.daily",
"sms.send.daily",
],
"expires_at": expires_at,
},
)
assert duplicate_batch.status_code == 409
with SessionLocal() as db:
rolled_back = (
db.query(LimitPolicyOverride)
.filter(
LimitPolicyOverride.subject_type == "phone",
LimitPolicyOverride.subject_value == phone,
LimitPolicyOverride.rule_code == "sms.send.daily",
)
.one_or_none()
)
assert rolled_back is None
for item in created.json():
deleted = client.delete(
f"/admin/api/limit-whitelist/{item['id']}",
headers=admin_headers,
)
assert deleted.status_code == 204
def test_device_bulk_rejects_mixed_candidate_namespaces(admin_headers) -> None:
expires_at = (datetime.now(UTC) + timedelta(hours=2)).isoformat()
with TestClient(admin_app) as client:
response = client.post(
"/admin/api/limit-whitelist/bulk",
headers=admin_headers,
json={
"subject_type": "device",
"subject_value": f"mixed-device-{uuid4().hex[:8]}",
"rule_codes": [
"compare.start.daily",
"sms.send.hourly",
],
"expires_at": expires_at,
},
)
assert response.status_code == 400
assert "不能在同一白名单中混选" in response.json()["detail"]
def test_legacy_ip_device_whitelist_matches_sms_policy(monkeypatch) -> None:
suffix = uuid4().hex[:8]
client_ip = f"203.0.113.{int(suffix[:2], 16) % 200 + 1}"
legacy_device = f"legacy-ip:{client_ip}"
phone = f"134{int(suffix, 16) % 100000000:08d}"
expires_at = datetime.now(UTC) + timedelta(hours=2)
with SessionLocal() as db:
row = LimitPolicyOverride(
subject_type="device",
subject_value=legacy_device,
rule_code="sms.send.hourly",
mode="unlimited",
enabled=True,
expires_at=expires_at,
)
db.add(row)
db.commit()
override_id = row.id
original_resolve = auth_api.limit_policy.resolve
matched_override_ids: list[int | None] = []
def spy_resolve(db, rule_code, **subjects):
result = original_resolve(db, rule_code, **subjects)
if rule_code == "sms.send.hourly":
matched_override_ids.append(result.override_id)
return result
monkeypatch.setattr(auth_api.limit_policy, "resolve", spy_resolve)
try:
with TestClient(app) as client:
response = client.post(
"/api/v1/auth/sms/send",
headers={"x-forwarded-for": client_ip},
json={"phone": phone},
)
assert response.status_code == 200, response.text
assert matched_override_ids == [override_id]
finally:
with SessionLocal() as db:
row = db.get(LimitPolicyOverride, override_id)
if row is not None:
db.delete(row)
db.commit()
def test_global_guide_limit_stays_in_sync_with_legacy_config(admin_headers) -> None:
snapshot = _snapshot_configs(
[
GUIDE_VIDEO_MAX_PLAYS_KEY,
"coupon_guide_video",
"comparison_guide_video",
]
)
try:
with TestClient(admin_app) as client:
changed = client.patch(
"/admin/api/limit-whitelist/rules/guide.video.lifetime",
headers=admin_headers,
json={"value": 7},
)
assert changed.status_code == 200, changed.text
assert changed.json()["global_limit"] == 7
legacy_page = client.get(
"/admin/api/guide-video",
headers=admin_headers,
)
assert legacy_page.status_code == 200, legacy_page.text
assert legacy_page.json()["max_plays"] == 7
assert legacy_page.json()["scene"] == "coupon"
comparison_changed = client.patch(
"/admin/api/guide-video",
headers=admin_headers,
params={"scene": "comparison"},
json={"max_plays": 9},
)
assert comparison_changed.status_code == 200, comparison_changed.text
assert comparison_changed.json()["scene"] == "comparison"
assert comparison_changed.json()["max_plays"] == 9
coupon_again = client.get(
"/admin/api/guide-video",
headers=admin_headers,
params={"scene": "coupon"},
)
assert coupon_again.status_code == 200, coupon_again.text
assert coupon_again.json()["max_plays"] == 7
finally:
_restore_configs(snapshot)
def test_legacy_ad_limit_update_syncs_split_global_rules(admin_headers) -> None:
snapshot = _snapshot_configs(
["ad_daily_limit", AD_REWARD_VIDEO_DAILY_LIMIT_KEY, AD_FEED_DAILY_LIMIT_KEY]
)
try:
with TestClient(admin_app) as client:
changed = client.patch(
"/admin/api/config/ad_daily_limit",
headers=admin_headers,
json={"value": 321},
)
assert changed.status_code == 200, changed.text
rules = client.get(
"/admin/api/limit-whitelist/rules",
headers=admin_headers,
)
by_code = {item["code"]: item for item in rules.json()}
assert by_code["ad.reward_video.daily"]["global_limit"] == 321
assert by_code["ad.feed.daily"]["global_limit"] == 321
finally:
_restore_configs(snapshot)
def test_zero_disables_cooldown_style_global_rules(admin_headers) -> None:
snapshot = _snapshot_configs(
[SMS_PHONE_COOLDOWN_SECONDS_KEY, PHONE_REBIND_DAYS_KEY]
)
try:
with TestClient(admin_app) as client:
for rule_code in ("sms.phone.cooldown", "phone.rebind.days"):
changed = client.patch(
f"/admin/api/limit-whitelist/rules/{rule_code}",
headers=admin_headers,
json={"value": 0},
)
assert changed.status_code == 200, changed.text
assert changed.json()["global_limit"] == 0
finally:
_restore_configs(snapshot)
def test_compare_start_uses_phone_override_and_reset_baseline() -> None:
suffix = uuid4().hex[:8]
phone = f"136{int(suffix, 16) % 100000000:08d}"
device = f"compare-policy-{suffix}"
with SessionLocal() as db:
user = user_repo.upsert_user_for_login(
db,
phone=phone,
register_channel="sms",
)
db.add(
LimitPolicyOverride(
subject_type="phone",
subject_value=phone,
rule_code="compare.start.daily",
mode="override",
limit_value=1,
enabled=True,
)
)
db.commit()
user_id = user.id
token, _ = create_token(user_id=user_id, token_type="access")
headers = {"Authorization": f"Bearer {token}"}
with TestClient(app) as client:
first = client.post(
"/api/v1/compare/start",
headers=headers,
json={
"trace_id": f"limit-policy-first-{suffix}",
"business_type": "food",
"device_id": device,
},
)
assert first.status_code == 200, first.text
assert first.json() == {"limit": 1, "used": 1, "remaining": 0}
blocked = client.post(
"/api/v1/compare/start",
headers=headers,
json={
"trace_id": f"limit-policy-blocked-{suffix}",
"business_type": "food",
"device_id": device,
},
)
assert blocked.status_code == 429
with SessionLocal() as db:
override = db.query(LimitPolicyOverride).filter(
LimitPolicyOverride.subject_type == "phone",
LimitPolicyOverride.subject_value == phone,
LimitPolicyOverride.rule_code == "compare.start.daily",
).one()
override.reset_at = datetime.now(UTC)
db.commit()
after_reset = client.post(
"/api/v1/compare/start",
headers=headers,
json={
"trace_id": f"limit-policy-reset-{suffix}",
"business_type": "food",
"device_id": device,
},
)
assert after_reset.status_code == 200, after_reset.text
assert after_reset.json() == {"limit": 1, "used": 1, "remaining": 0}
def test_suppress_alert_keeps_events_but_creates_no_incident() -> None:
suffix = uuid4().hex[:10]
device = f"suppress-device-{suffix}"
phone = f"138{int(suffix[:8], 16) % 100000000:08d}"
now = risk_repo.utcnow().replace(microsecond=0)
with SessionLocal() as db:
db.add(
LimitPolicyOverride(
subject_type="device",
subject_value=device,
rule_code="risk.sms.hourly",
mode="suppress_alert",
enabled=True,
)
)
db.commit()
for index in range(10):
risk_repo.record_behavior_event(
db,
event_type=risk_repo.EVENT_SMS_SEND,
subject_type="device",
subject_id=device,
device_id=device,
phone=phone,
outcome="success",
occurred_at=now + timedelta(seconds=index),
evaluate_rule=risk_repo.RULE_SMS_HOURLY,
)
with SessionLocal() as db:
incident = db.query(RiskIncident).filter(
RiskIncident.rule_code == risk_repo.RULE_SMS_HOURLY,
RiskIncident.subject_id == device,
).one_or_none()
assert incident is None
def test_adding_suppress_alert_resolves_existing_incident(admin_headers) -> None:
suffix = uuid4().hex[:10]
device = f"suppress-existing-{suffix}"
now = risk_repo.utcnow().replace(microsecond=0)
with SessionLocal() as db:
for index in range(5):
risk_repo.record_behavior_event(
db,
event_type=risk_repo.EVENT_SMS_SEND,
subject_type="device",
subject_id=device,
device_id=device,
phone=f"13700001{index:03d}",
outcome="success",
occurred_at=now + timedelta(seconds=index),
evaluate_rule=risk_repo.RULE_SMS_HOURLY,
)
incident = db.query(RiskIncident).filter(
RiskIncident.rule_code == risk_repo.RULE_SMS_HOURLY,
RiskIncident.subject_id == device,
).one()
assert incident.status == "open"
with TestClient(admin_app) as client:
created = client.post(
"/admin/api/limit-whitelist",
headers=admin_headers,
json={
"subject_type": "device",
"subject_value": device,
"rule_code": "risk.sms.hourly",
"mode": "suppress_alert",
"enabled": True,
"expires_at": (datetime.now(UTC) + timedelta(hours=2)).isoformat(),
"reason": "QA device",
},
)
assert created.status_code == 201, created.text
with SessionLocal() as db:
incident = db.query(RiskIncident).filter(
RiskIncident.rule_code == risk_repo.RULE_SMS_HOURLY,
RiskIncident.subject_id == device,
).one()
assert incident.status == "resolved"
assert incident.action_reason == risk_repo.AUTO_RESOLVED_REASON
+1 -1
View File
@@ -533,7 +533,7 @@ def test_admin_can_edit_rules_and_current_window_is_reconciled() -> None:
"/admin/api/risk-monitor/rules",
headers=headers,
json={
"sms_hourly_threshold": 100_001,
"sms_hourly_threshold": 21,
"oneclick_daily_threshold": 20,
"compare_daily_threshold": 100,
},