Compare commits

...

14 Commits

Author SHA1 Message Date
unknown c67407668e 合并:同步最新主分支 2026-07-30 20:21:29 +08:00
unknown fdbf7a893b 修复:完善白名单配置兼容与审计一致性 2026-07-30 20:21:16 +08:00
guke ef0ab9d95a fix(短信): 号码无效提示由「手机号无效」改为「请输入有效的手机号」 (#206)
极光/阿里云/创蓝三处 provider 的号码无效错误统一改成更友好的用户提示。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: guke <guke@autohome.com.cn>
Reviewed-on: #206
2026-07-30 19:15:45 +08:00
zuochenyong 06cd718610 修复之前错误配置的端口 (#203)
Co-authored-by: exinglang <exinglang@qq.com>
Reviewed-on: #203
Co-authored-by: zuochenyong <zuochenyong@wonderable.ai>
Co-committed-by: zuochenyong <zuochenyong@wonderable.ai>
2026-07-30 16:32:32 +08:00
unknown 125f0e1a37 Merge remote-tracking branch 'origin/main' into codex/limit-policy-whitelist
# Conflicts:
#	app/admin/routers/guide_video.py
#	app/models/guide_video.py
#	app/repositories/guide_video.py
#	app/schemas/guide_video.py
2026-07-30 15:01:20 +08:00
zuochenyong aa1a1240f2 fix(push): 完善厂商推送排障日志 (#202)
Co-authored-by: exinglang <exinglang@qq.com>
Reviewed-on: #202
Co-authored-by: zuochenyong <zuochenyong@wonderable.ai>
Co-committed-by: zuochenyong <zuochenyong@wonderable.ai>
2026-07-30 14:05:05 +08:00
guke b6ddb275f4 fix(comparison): 比价 best 无 is_best 时兜底回落「含源最低价」 (#201)
背景
_derive_from_platforms 是比价记录的单一真相源派生函数(pricebot done 帧带 platforms 时走它),正常路径直接取 platforms 里 is_best=true 的那行当最低价。

线上出现一类帧(如记录 id 3304):全平台 has_dish_diff(菜品「相似替换 / 价格仅供参考」),pricebot 认为无法认定权威最低价,于是一个 is_best 都不标。此时旧逻辑 best=None,导致 best_platform_id / best_price_cents / saved_amount_cents / is_source_best 整条落 NULL,连锁反应:

首页比价价显示 0.00
记录页无最低价红框
省额丢失、「累计发现可省」漏计
方案
无 is_best 时兜底:在有价行里取最低价当参考 best。

关键设计点 —— 候选池含源(而非仅目标):源平台常年全菜、价可信。若源本身最便宜(其余都是更贵的相似替换),则 best 回落到源、saved=0、is_source_best=True。这与仓库老派生函数 _derive 的既有语义(「全目标缺菜 → 回落源、不虚报省」)完全一致,三个派生函数行为对齐。

⚠️ 若像分支首个提交那样排除源、强选最低目标,当源最便宜时会选中更贵目标 → saved 变负,倒扣 get_stats 的「累计发现可省」(该聚合按 status='success' 求和、不带 >0 过滤)。第二个提交据此修正为含源。

影响面 / 兼容性
只影响「带 platforms 且无任何 is_best」这一条兜底分支;正常有 is_best 的路径不变。
老客户端不带 platforms → 走 _derive,不受影响。
纯派生逻辑,无 schema / 无迁移,回滚成本低。

---------

Co-authored-by: guke <guke@autohome.com.cn>
Reviewed-on: #201
2026-07-30 11:00:35 +08:00
zuochenyong ebacf01742 feat(新手引导): 支持十圈进度与逐圈发奖 (#200)
Co-authored-by: exinglang <exinglang@qq.com>
Reviewed-on: #200
Co-authored-by: zuochenyong <zuochenyong@wonderable.ai>
Co-committed-by: zuochenyong <zuochenyong@wonderable.ai>
2026-07-30 10:49:24 +08:00
zuochenyong 89f266419b fix(push): 完善厂商推送参数与消息通知 (#199)
Co-authored-by: exinglang <exinglang@qq.com>
Reviewed-on: #199
Co-authored-by: zuochenyong <zuochenyong@wonderable.ai>
Co-committed-by: zuochenyong <zuochenyong@wonderable.ai>
2026-07-30 10:49:08 +08:00
zuochenyong 675c7ecf81 fix(push): 完善厂商推送排障日志 (#198)
Co-authored-by: exinglang <exinglang@qq.com>
Reviewed-on: #198
Co-authored-by: zuochenyong <zuochenyong@wonderable.ai>
Co-committed-by: zuochenyong <zuochenyong@wonderable.ai>
2026-07-30 10:41:43 +08:00
unknown 3f2ec19ec2 功能:统一限制策略与白名单管理
新增统一限制规则、临时不限和风控免告警白名单,接入比价、短信登录、广告、引导视频与账号冷却等业务链路。补充设备选择、权限、审计、迁移及回归测试。
2026-07-29 19:07:51 +08:00
zuochenyong 53c3b7f60f feat(guide-video): 支持领券和比价独立视频奖励配置 (#196)
Co-authored-by: exinglang <exinglang@qq.com>
Reviewed-on: #196
Co-authored-by: zuochenyong <zuochenyong@wonderable.ai>
Co-committed-by: zuochenyong <zuochenyong@wonderable.ai>
2026-07-29 16:12:05 +08:00
marco 4bd4e66678 重构比价结果页取数据逻辑 (#195)
Reviewed-on: #195
2026-07-29 01:59:31 +08:00
linkeyu 90c6fe599a 修复:统一用户Draw信息流eCPM统计口径 (#190)
## 问题

业务收益详情的平均 Draw eCPM 仅平均成功发奖记录,会排除未发奖的真实展示,导致数值系统性偏高,且与广告收益页口径不一致。

## 修复

- `feed_avg_ecpm` 改为从 `ad_ecpm_record` 的全部 `draw/feed` 实际展示计算
- 成功发奖、未发奖展示均纳入,每次展示等权
- 日期、正式/测试环境、业务代码位、领券/比价场景支持与广告收益页对齐
- 奖励份数仍基于成功发奖表,不混用展示数据源
- 复用广告收益报表的业务代码位集合

## 线上数据复算

2026-07-25、正式业务、用户 #33:

- 旧口径(只看成功发奖):`29.9117 元/千次`
- 新口径(333 次真实展示):`19.9926 元/千次`
- 新值与广告收益报表一致

## 验证

- 新增成功/未发奖、场景、环境、业务代码位回归用例
- `tests/test_admin_read.py` + `tests/test_admin_ad_revenue_scope.py`:29 项全通过
- Ruff 改动文件检查通过

## 上线顺序

本 PR 需先于管理后台配套 PR 上线。

---------

Co-authored-by: guke <guke@wonderable.ai>
Co-authored-by: unknown <798648091@qq.com>
Reviewed-on: #190
Co-authored-by: linkeyu <linkeyu@wonderable.ai>
Co-committed-by: linkeyu <linkeyu@wonderable.ai>
2026-07-28 17:58:23 +08:00
59 changed files with 7118 additions and 607 deletions
+6
View File
@@ -175,3 +175,9 @@ PANGLE_REPORT_SITE_ID_TEST=5832303
# APPLOG_MAX_BATCH=500 # 单批最大条数(超 → 422;导入期常量,改需重启)
# APPLOG_MAX_BODY_BYTES=2097152 # 请求体上限 2MB(超 → 413;运行期可调)
# APPLOG_MAX_MSG_BYTES=8192 # 单条 msg 超此字节数截断
# ===== 华为 Push =====
HUAWEI_PUSH_APP_ID=
HUAWEI_PUSH_APP_SECRET=
# 0=正式消息(默认);1=测试消息(仅开发联调,生产必须保持 0)
HUAWEI_PUSH_TARGET_USER_TYPE=0
@@ -0,0 +1,26 @@
"""merge comparison platforms + fail_reason heads
Revision ID: 6d2309208549
Revises: comparison_platforms_col, comparison_record_fail_reason
Create Date: 2026-07-29 01:48:41.868083
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '6d2309208549'
down_revision: Union[str, Sequence[str], None] = ('comparison_platforms_col', 'comparison_record_fail_reason')
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
pass
def downgrade() -> None:
pass
@@ -0,0 +1,47 @@
"""add platforms unified array column to comparison_record
展示模型统一数组(pricebot done.params.platforms 原样存): 每平台一行、自带
status/is_best/display, 记录页据此直接渲染, 不再靠 comparison_results + 客户端合并 + 前端派生。
纯新增列, 老记录为空 → 前端回退老 comparison_results。
Revision ID: comparison_platforms_col
Revises: user_manual_risk_fields
"""
from __future__ import annotations
from collections.abc import Sequence
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
from alembic import op
revision: str = "comparison_platforms_col"
down_revision: str | None = "user_manual_risk_fields"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
def upgrade() -> None:
# 幂等: 线上为了提前给历史数据补 platforms(2026-07-29), 已手动
# `ALTER TABLE comparison_record ADD COLUMN IF NOT EXISTS platforms jsonb
# NOT NULL DEFAULT '[]'::jsonb`(与本 migration 定义一致)。列已存在时跳过,
# 否则上线 alembic upgrade head 会撞 DuplicateColumn 直接部署失败。
bind = op.get_bind()
cols = {c["name"] for c in sa.inspect(bind).get_columns("comparison_record")}
if "platforms" in cols:
return
with op.batch_alter_table("comparison_record") as batch_op:
batch_op.add_column(
sa.Column(
"platforms", _JSON, nullable=False,
server_default=sa.text("'[]'"),
)
)
def downgrade() -> None:
with op.batch_alter_table("comparison_record") as batch_op:
batch_op.drop_column("platforms")
@@ -0,0 +1,31 @@
"""guide video play count is independent for coupon and comparison
Revision ID: guide_video_scene_unique
Revises: 6d2309208549
"""
from alembic import op
revision = "guide_video_scene_unique"
down_revision = "6d2309208549"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.drop_index("uq_guide_video_play_user_seq", table_name="guide_video_play")
op.create_index(
"uq_guide_video_play_user_scene_seq",
"guide_video_play",
["user_id", "scene", "seq"],
unique=True,
)
def downgrade() -> None:
op.drop_index("uq_guide_video_play_user_scene_seq", table_name="guide_video_play")
op.create_index(
"uq_guide_video_play_user_seq",
"guide_video_play",
["user_id", "seq"],
unique=True,
)
@@ -0,0 +1,164 @@
"""guide video prepare/start/ten-circle reward state machine
Revision ID: guide_video_ten_circle_v2
Revises: guide_video_scene_unique
"""
from __future__ import annotations
import json
from pathlib import Path
import sqlalchemy as sa
from alembic import op
revision = "guide_video_ten_circle_v2"
down_revision = "guide_video_scene_unique"
branch_labels = None
depends_on = None
def _backfill_configs(connection) -> None:
from app.core import media
from app.core.config import settings
for key in ("coupon_guide_video", "comparison_guide_video"):
row = connection.execute(
sa.text("SELECT value FROM app_config WHERE key = :key"), {"key": key}
).first()
if not row:
continue
raw = row[0]
value = json.loads(raw) if isinstance(raw, str) else dict(raw or {})
if value.get("guide_video_v2_migrated") is True:
continue
maximum = int(value.get("max_plays", 3) or 0)
reward = int(value.get("reward_coin", 100) or 0)
video_url = str(value.get("video_url") or "")
error = None
analysis = None
if not 1 <= maximum <= 50:
error = "旧播放次数不符合 150 约束"
elif not 10 <= reward <= 10_000 or reward % 10:
error = "旧金币总价不符合 10~10000 且为 10 的倍数约束"
elif not video_url:
error = "尚未上传视频"
else:
prefix = f"{settings.MEDIA_URL_PREFIX}/guide_video/"
if not video_url.startswith(prefix):
error = "旧视频不是服务端托管文件,请重新上传"
else:
path = Path(settings.MEDIA_ROOT) / "guide_video" / video_url[len(prefix):]
if not path.is_file():
error = "旧视频文件缺失,请重新上传"
else:
try:
analysis = media._probe_guide_video(path)
except media.MediaError as exc:
error = str(exc)
value["config_version"] = int(value.get("config_version", 0) or 0) + 1
value["guide_video_v2_migrated"] = True
if analysis:
value.update(analysis)
else:
value.update(
enabled=False,
duration_ms=None,
video_codec=None,
audio_codec=None,
analysis_status="invalid" if video_url else "missing",
analysis_error=error,
)
statement = sa.text(
"UPDATE app_config SET value = :value WHERE key = :key"
).bindparams(sa.bindparam("value", type_=sa.JSON()))
connection.execute(statement, {"key": key, "value": value})
def _seal_legacy_plays(connection) -> None:
"""旧整笔发奖 token 永久封口;旧未发奖会话终止但仍保留计次行。"""
connection.execute(
sa.text(
"""
UPDATE guide_video_play
SET status = CASE WHEN status = 'granted' THEN 'legacy_completed' ELSE 'legacy_closed' END,
settled_circles = CASE WHEN status = 'granted' THEN 10 ELSE 0 END,
prepared_at = started_at,
expires_at = started_at
WHERE status IN ('granted', 'playing')
"""
)
)
def _prepare_downgrade(connection) -> None:
"""清除未起播计划,并把所有保留 token 封成旧版不可再领奖的 granted。"""
connection.execute(
sa.text("DELETE FROM guide_video_play WHERE status = 'prepared'")
)
connection.execute(
sa.text(
"""
UPDATE guide_video_play
SET status = 'granted',
completed = CASE WHEN settled_circles >= 10 THEN 1 ELSE completed END,
granted_at = CASE
WHEN settled_circles >= 10 AND granted_at IS NULL THEN started_at
ELSE granted_at
END
WHERE status IN (
'started', 'completed', 'legacy_completed', 'legacy_closed'
)
"""
)
)
# 防御未知/中间状态:宁可删除,也绝不让旧应用把它当 playing 整笔发奖。
connection.execute(
sa.text(
"""
DELETE FROM guide_video_play
WHERE seq IS NULL OR started_at IS NULL OR status != 'granted'
"""
)
)
def upgrade() -> None:
with op.batch_alter_table("guide_video_play") as batch:
batch.alter_column("seq", existing_type=sa.Integer(), nullable=True)
batch.alter_column("started_at", existing_type=sa.DateTime(timezone=True), nullable=True)
batch.add_column(sa.Column("duration_ms", sa.Integer(), nullable=False, server_default="0"))
batch.add_column(sa.Column("config_version", sa.Integer(), nullable=False, server_default="0"))
batch.add_column(sa.Column("settled_circles", sa.Integer(), nullable=False, server_default="0"))
batch.add_column(
sa.Column("prepared_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now())
)
batch.add_column(sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True))
batch.alter_column("status", existing_type=sa.String(length=16), type_=sa.String(length=24))
op.create_index("ix_guide_video_play_expires_at", "guide_video_play", ["expires_at"])
op.create_index(
"ux_coin_transaction_guide_video_ref",
"coin_transaction",
["user_id", "biz_type", "ref_id"],
unique=True,
sqlite_where=sa.text("biz_type = 'guide_video' AND ref_id IS NOT NULL"),
postgresql_where=sa.text("biz_type = 'guide_video' AND ref_id IS NOT NULL"),
)
connection = op.get_bind()
_seal_legacy_plays(connection)
_backfill_configs(connection)
def downgrade() -> None:
_prepare_downgrade(op.get_bind())
op.drop_index("ux_coin_transaction_guide_video_ref", table_name="coin_transaction")
op.drop_index("ix_guide_video_play_expires_at", table_name="guide_video_play")
with op.batch_alter_table("guide_video_play") as batch:
batch.alter_column("status", existing_type=sa.String(length=24), type_=sa.String(length=16))
batch.drop_column("expires_at")
batch.drop_column("prepared_at")
batch.drop_column("settled_circles")
batch.drop_column("config_version")
batch.drop_column("duration_ms")
batch.alter_column("started_at", existing_type=sa.DateTime(timezone=True), nullable=False)
batch.alter_column("seq", existing_type=sa.Integer(), nullable=False)
@@ -0,0 +1,193 @@
"""store all global limit values in one complete JSON document
Revision ID: limit_policy_global_bundle
Revises: limit_policy_whitelist
"""
from __future__ import annotations
from collections.abc import Sequence
from typing import Any
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
from alembic import op
revision: str = "limit_policy_global_bundle"
down_revision: str | Sequence[str] | None = "limit_policy_whitelist"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
_BUNDLE_KEY = "limit_policy_global"
# rule_code, old sparse key, default, legacy structured key, legacy JSON field
_RULES: tuple[tuple[str, str, int, str | None, str | None], ...] = (
("compare.start.daily", "compare_daily_limit", 100, None, None),
("sms.send.hourly", "sms_send_hourly_limit", 5, None, None),
("sms.send.daily", "sms_send_daily_limit", 20, None, None),
("sms.phone.cooldown", "sms_phone_cooldown_seconds", 60, None, None),
("sms.code.failed_attempts", "sms_code_max_failed_attempts", 5, None, None),
("sms.login.hourly", "sms_login_hourly_limit", 5, None, None),
("wechat.bind.hourly", "wechat_bind_sms_hourly_limit", 5, None, None),
("wechat.conflict.hourly", "wechat_conflict_hourly_limit", 5, None, None),
(
"ad.reward_video.daily",
"ad_reward_video_daily_limit",
500,
"ad_daily_limit",
None,
),
(
"ad.feed.daily",
"ad_feed_daily_limit",
500,
"ad_daily_limit",
None,
),
("ad.reward_video.cooldown", "ad_cooldown_sec", 3, None, None),
(
"guide.video.lifetime",
"guide_video_max_plays",
3,
"coupon_guide_video",
"max_plays",
),
("phone.rebind.days", "phone_rebind_days", 30, None, None),
("risk.sms.hourly", "risk_sms_hourly_threshold", 5, None, None),
(
"risk.oneclick.daily",
"risk_oneclick_daily_threshold",
20,
None,
None,
),
(
"risk.compare.daily",
"risk_compare_daily_threshold",
100,
None,
None,
),
)
def _table() -> sa.TableClause:
return sa.table(
"app_config",
sa.column("key", sa.String(64)),
sa.column("value", _JSON),
sa.column("updated_by_admin_id", sa.Integer),
sa.column("updated_at", sa.DateTime(timezone=True)),
)
def _row(conn, table, key: str):
return conn.execute(
sa.select(
table.c.value,
table.c.updated_by_admin_id,
).where(table.c.key == key)
).mappings().first()
def _int_or_none(value: Any) -> int | None:
if isinstance(value, bool):
return None
try:
return int(value)
except (TypeError, ValueError):
return None
def upgrade() -> None:
conn = op.get_bind()
table = _table()
values = {rule_code: default for rule_code, _, default, _, _ in _RULES}
# Old shared/structured values are the lowest-precedence compatibility
# source. Dedicated sparse keys override them.
for rule_code, _, _, legacy_key, legacy_field in _RULES:
if legacy_key is None:
continue
legacy = _row(conn, table, legacy_key)
if legacy is None:
continue
raw = legacy["value"]
if legacy_field is not None:
raw = raw.get(legacy_field) if isinstance(raw, dict) else None
parsed = _int_or_none(raw)
if parsed is not None:
values[rule_code] = parsed
for rule_code, sparse_key, _, _, _ in _RULES:
sparse = _row(conn, table, sparse_key)
parsed = _int_or_none(sparse["value"]) if sparse is not None else None
if parsed is not None:
values[rule_code] = parsed
# If a deployment already wrote the new key, preserve it over old keys.
bundle = _row(conn, table, _BUNDLE_KEY)
if bundle is not None and isinstance(bundle["value"], dict):
for rule_code, raw in bundle["value"].items():
if rule_code not in values:
continue
parsed = _int_or_none(raw)
if parsed is not None:
values[rule_code] = parsed
if bundle is None:
conn.execute(
table.insert().values(
key=_BUNDLE_KEY,
value=values,
updated_by_admin_id=None,
)
)
else:
conn.execute(
table.update()
.where(table.c.key == _BUNDLE_KEY)
.values(value=values, updated_at=sa.func.now())
)
sparse_keys = [sparse_key for _, sparse_key, _, _, _ in _RULES]
conn.execute(table.delete().where(table.c.key.in_(sparse_keys)))
def downgrade() -> None:
conn = op.get_bind()
table = _table()
bundle = _row(conn, table, _BUNDLE_KEY)
values = (
bundle["value"]
if bundle is not None and isinstance(bundle["value"], dict)
else {}
)
admin_id = bundle["updated_by_admin_id"] if bundle is not None else None
for rule_code, sparse_key, default, _, _ in _RULES:
value = _int_or_none(values.get(rule_code))
if value is None:
value = default
existing = _row(conn, table, sparse_key)
if existing is None:
conn.execute(
table.insert().values(
key=sparse_key,
value=value,
updated_by_admin_id=admin_id,
)
)
else:
conn.execute(
table.update()
.where(table.c.key == sparse_key)
.values(
value=value,
updated_by_admin_id=admin_id,
updated_at=sa.func.now(),
)
)
conn.execute(table.delete().where(table.c.key == _BUNDLE_KEY))
+128
View File
@@ -0,0 +1,128 @@
"""add per-subject limit policy whitelist
Revision ID: limit_policy_whitelist
Revises: guide_video_scene_unique
"""
from __future__ import annotations
from collections.abc import Sequence
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
from alembic import op
revision: str = "limit_policy_whitelist"
down_revision: str | Sequence[str] | None = "guide_video_ten_circle_v2"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
_PAGE = "limit-whitelist"
_DEFAULT_ROLES = ("operator", "tech")
def _role_table() -> sa.TableClause:
return sa.table(
"admin_role",
sa.column("name", sa.String),
sa.column("pages", _JSON),
)
def _add_default_role_permissions() -> None:
"""Grant the page without replacing existing role customisations."""
role = _role_table()
conn = op.get_bind()
rows = conn.execute(
sa.select(role.c.name, role.c.pages).where(
role.c.name.in_(_DEFAULT_ROLES)
)
).all()
for name, pages in rows:
current_pages = list(pages or [])
if _PAGE not in current_pages:
conn.execute(
role.update()
.where(role.c.name == name)
.values(pages=[*current_pages, _PAGE])
)
def _remove_default_role_permissions() -> None:
role = _role_table()
conn = op.get_bind()
rows = conn.execute(
sa.select(role.c.name, role.c.pages).where(
role.c.name.in_(_DEFAULT_ROLES)
)
).all()
for name, pages in rows:
current_pages = list(pages or [])
if _PAGE in current_pages:
conn.execute(
role.update()
.where(role.c.name == name)
.values(
pages=[page for page in current_pages if page != _PAGE]
)
)
def upgrade() -> None:
op.create_table(
"limit_policy_override",
sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
sa.Column("subject_type", sa.String(length=16), nullable=False),
sa.Column("subject_value", sa.String(length=128), nullable=False),
sa.Column("rule_code", sa.String(length=64), nullable=False),
sa.Column("mode", sa.String(length=24), nullable=False),
sa.Column("limit_value", sa.Integer(), nullable=True),
sa.Column(
"enabled", sa.Boolean(), server_default=sa.true(), nullable=False
),
sa.Column("starts_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("reset_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("reason", sa.String(length=256), nullable=True),
sa.Column("created_by_admin_id", sa.Integer(), nullable=True),
sa.Column(
"created_at",
sa.DateTime(timezone=True),
server_default=sa.func.now(),
nullable=False,
),
sa.Column(
"updated_at",
sa.DateTime(timezone=True),
server_default=sa.func.now(),
nullable=False,
),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint(
"subject_type",
"subject_value",
"rule_code",
name="uq_limit_policy_subject_rule",
),
)
op.create_index(
"ix_limit_policy_lookup",
"limit_policy_override",
["subject_type", "subject_value", "rule_code", "enabled"],
unique=False,
)
op.create_index(
"ix_limit_policy_expires",
"limit_policy_override",
["expires_at"],
unique=False,
)
_add_default_role_permissions()
def downgrade() -> None:
_remove_default_role_permissions()
op.drop_index("ix_limit_policy_expires", table_name="limit_policy_override")
op.drop_index("ix_limit_policy_lookup", table_name="limit_policy_override")
op.drop_table("limit_policy_override")
+3 -1
View File
@@ -1,6 +1,6 @@
"""Admin 后台 FastAPI app(独立进程)。
启动:uvicorn app.admin.main:admin_app --host 127.0.0.1 --port 8771
启动:uvicorn app.admin.main:admin_app --host 127.0.0.1 --port 8773
复用 App 的 DB/models/repositories/integrations;鉴权独立(admin JWT,见 app/admin/security.py)。
现有 app.main:app 不 import 本模块,两进程互不影响。
"""
@@ -31,6 +31,7 @@ from app.admin.routers.feedback import router as feedback_router
from app.admin.routers.feedback_qr import router as feedback_qr_router
from app.admin.routers.guide_video import router as guide_video_router
from app.admin.routers.huawei_review import router as huawei_review_router
from app.admin.routers.limit_whitelist import router as limit_whitelist_router
from app.admin.routers.onboarding import router as onboarding_router
from app.admin.routers.ops_marquee_seed import router as ops_marquee_seed_router
from app.admin.routers.ops_stat_config import router as ops_stat_config_router
@@ -103,6 +104,7 @@ admin_app.include_router(wallet_router)
admin_app.include_router(withdraw_router)
admin_app.include_router(price_report_router)
admin_app.include_router(risk_monitor_router)
admin_app.include_router(limit_whitelist_router)
admin_app.include_router(feedback_router)
admin_app.include_router(event_logs_router)
admin_app.include_router(analytics_health_router)
+3 -1
View File
@@ -40,6 +40,7 @@ PERMISSION_CATALOG: list[dict] = [
{"key": "analytics-health", "label": "埋点成功率"},
{"key": "event-logs", "label": "埋点日志"},
{"key": "audit-logs", "label": "审计日志"},
{"key": "limit-whitelist", "label": "白名单"},
]},
{"group": "其他", "pages": [
{"key": "admins", "label": "权限管理"},
@@ -58,13 +59,14 @@ BUILTIN_ROLES: list[dict] = [
{"name": "operator", "label": "运营", "pages": [
"dashboard", "coupon-data", "ad-revenue-report", "comparison-records",
"cps", "risk-monitor", "device-liveness", "price-reports", "feedbacks", "huawei-review",
"limit-whitelist",
]},
{"name": "finance", "label": "财务", "pages": [
"dashboard", "ad-revenue-report", "cps", "invite-withdraws", "withdraws",
]},
{"name": "tech", "label": "技术", "pages": [
"dashboard", "risk-monitor", "device-liveness", "analytics-health", "config", "ad-revenue", "huawei-review",
"event-logs", "audit-logs",
"event-logs", "audit-logs", "limit-whitelist",
]},
]
+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_code_ids: set[str] | None = None
business_ids: set[str] | None = None
if revenue_scope == "business":
business_code_ids = _business_code_ids(db, app_env)
events = [e for e in events if e.get("our_code_id") in business_code_ids]
business_ids = business_code_ids(db, app_env)
events = [e for e in events if e.get("our_code_id") in business_ids]
# 排序:time=按时间倒序(新→旧);ecpm=按 eCPM 数值倒序(eCPM 原值是字符串「分」,转数值排;
# 纯发奖行用其发奖采用的 eCPM,缺失/非法计 0 排末尾)。
@@ -381,7 +381,7 @@ def ad_revenue_report(
date_from=date_from,
date_to=date_to,
app_env=app_env,
our_code_ids=business_code_ids,
our_code_ids=business_ids,
)
if pangle_aggs:
by_date = {a["date"]: a for a in pangle_aggs}
+583
View File
@@ -0,0 +1,583 @@
"""CRUD and presentation helpers for limit policy overrides."""
from __future__ import annotations
from datetime import UTC, datetime
from hashlib import blake2b
from sqlalchemy import String, cast, func, or_, select, text
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from app.core import limit_policy
from app.models.app_config import AppConfig
from app.models.comparison import ComparisonRecord
from app.models.limit_policy import LimitPolicyOverride
from app.models.risk import BehaviorEvent
from app.models.user import User
from app.repositories import risk as risk_repo
class DuplicateOverrideError(Exception):
pass
def lock_subject(db: Session, *, subject_type: str, subject_value: str) -> None:
"""Serialize writes for one logical whitelist subject on PostgreSQL.
Row locks cannot protect a brand-new subject because there is no row to
lock yet. A transaction-scoped advisory lock closes that gap and also
serializes concurrent appends that touch different rule rows.
"""
bind = db.get_bind()
if bind.dialect.name != "postgresql":
return
identity = f"limit-whitelist:{subject_type}:{subject_value}".encode()
lock_id = int.from_bytes(
blake2b(identity, digest_size=8).digest(),
byteorder="big",
signed=True,
)
db.execute(
text("SELECT pg_advisory_xact_lock(:lock_id)"),
{"lock_id": lock_id},
)
_EVENT_SOURCE_LABELS = {
risk_repo.EVENT_SMS_SEND: "短信验证码",
risk_repo.EVENT_SMS_LOGIN: "短信登录",
risk_repo.EVENT_ONECLICK_LOGIN: "一键登录",
}
def _matching_users(keyword: str):
pattern = f"%{keyword}%"
return select(User.id).where(
or_(
cast(User.id, String).ilike(pattern),
User.username.ilike(pattern),
User.phone.ilike(pattern),
User.nickname.ilike(pattern),
)
)
def _user_maps(
db: Session, *, user_ids: set[int], phones: set[str]
) -> tuple[dict[int, User], dict[str, User]]:
conditions = []
if user_ids:
conditions.append(User.id.in_(user_ids))
if phones:
conditions.append(User.phone.in_(phones))
if not conditions:
return {}, {}
users = list(db.scalars(select(User).where(or_(*conditions))).all())
return {user.id: user for user in users}, {user.phone: user for user in users}
def _behavior_device_candidates(db: Session, *, keyword: str | None, limit: int) -> list[dict]:
conditions = [
BehaviorEvent.subject_type == limit_policy.SUBJECT_DEVICE,
BehaviorEvent.subject_id != "",
BehaviorEvent.subject_id.not_like(f"{limit_policy.LEGACY_IP_DEVICE_PREFIX}%"),
]
if keyword:
value = keyword.strip()
pattern = f"%{value}%"
matched_users = _matching_users(value)
conditions.append(
or_(
BehaviorEvent.subject_id.ilike(pattern),
BehaviorEvent.device_id.ilike(pattern),
BehaviorEvent.device_model.ilike(pattern),
BehaviorEvent.phone.ilike(pattern),
cast(BehaviorEvent.user_id, String).ilike(pattern),
BehaviorEvent.user_id.in_(matched_users),
BehaviorEvent.phone.in_(
select(User.phone).where(User.id.in_(_matching_users(value)))
),
)
)
ranked = (
select(
BehaviorEvent.subject_id.label("device_id"),
BehaviorEvent.event_type,
BehaviorEvent.user_id,
BehaviorEvent.phone,
BehaviorEvent.device_model,
BehaviorEvent.occurred_at.label("last_active_at"),
func.row_number()
.over(
partition_by=BehaviorEvent.subject_id,
order_by=(
BehaviorEvent.occurred_at.desc(),
BehaviorEvent.id.desc(),
),
)
.label("rank"),
)
.where(*conditions)
.subquery()
)
rows = db.execute(
select(ranked)
.where(ranked.c.rank == 1)
.order_by(ranked.c.last_active_at.desc(), ranked.c.device_id.asc())
.limit(limit)
).all()
by_id, by_phone = _user_maps(
db,
user_ids={row.user_id for row in rows if row.user_id is not None},
phones={row.phone for row in rows if row.phone},
)
result = []
for row in rows:
user = by_id.get(row.user_id) if row.user_id is not None else None
user = user or (by_phone.get(row.phone) if row.phone else None)
result.append(
{
"device_id": row.device_id,
"source": row.event_type,
"source_label": _EVENT_SOURCE_LABELS.get(row.event_type, "登录/鉴权流水"),
"user_id": user.id if user else row.user_id,
"username": user.username if user else None,
"phone": user.phone if user else row.phone,
"nickname": user.nickname if user else None,
"device_model": row.device_model,
"last_active_at": row.last_active_at,
}
)
return result
def _comparison_device_candidates(db: Session, *, keyword: str | None, limit: int) -> list[dict]:
conditions = [
ComparisonRecord.device_id.is_not(None),
ComparisonRecord.device_id != "",
]
if keyword:
value = keyword.strip()
pattern = f"%{value}%"
conditions.append(
or_(
ComparisonRecord.device_id.ilike(pattern),
ComparisonRecord.device_model.ilike(pattern),
cast(ComparisonRecord.user_id, String).ilike(pattern),
ComparisonRecord.user_id.in_(_matching_users(value)),
)
)
ranked = (
select(
ComparisonRecord.device_id,
ComparisonRecord.user_id,
ComparisonRecord.device_model,
ComparisonRecord.created_at.label("last_active_at"),
func.row_number()
.over(
partition_by=ComparisonRecord.device_id,
order_by=(
ComparisonRecord.created_at.desc(),
ComparisonRecord.id.desc(),
),
)
.label("rank"),
)
.where(*conditions)
.subquery()
)
rows = db.execute(
select(ranked)
.where(ranked.c.rank == 1)
.order_by(ranked.c.last_active_at.desc(), ranked.c.device_id.asc())
.limit(limit)
).all()
by_id, _ = _user_maps(
db,
user_ids={row.user_id for row in rows if row.user_id is not None},
phones=set(),
)
return [
{
"device_id": row.device_id,
"source": "comparison_record",
"source_label": "比价记录",
"user_id": user.id if user else row.user_id,
"username": user.username if user else None,
"phone": user.phone if user else None,
"nickname": user.nickname if user else None,
"device_model": row.device_model,
"last_active_at": row.last_active_at,
}
for row in rows
for user in [by_id.get(row.user_id) if row.user_id is not None else None]
]
def list_device_candidates(
db: Session,
*,
rule_code: str,
keyword: str | None = None,
limit: int = 30,
) -> list[dict]:
rule = limit_policy.get_rule(rule_code)
if limit_policy.SUBJECT_DEVICE not in rule.subject_types:
raise ValueError("当前限制项不支持设备白名单")
if limit_policy.device_source_scope(rule_code) == "comparison":
return _comparison_device_candidates(db, keyword=keyword, limit=limit)
return _behavior_device_candidates(db, keyword=keyword, limit=limit)
def _aware(value: datetime | None) -> datetime | None:
if value is None:
return None
return value.replace(tzinfo=UTC) if value.tzinfo is None else value
def status_of(row: LimitPolicyOverride, *, now: datetime | None = None) -> str:
now = _aware(now) or datetime.now(UTC)
if not row.enabled:
return "disabled"
if _aware(row.starts_at) and _aware(row.starts_at) > now:
return "scheduled"
if _aware(row.expires_at) and _aware(row.expires_at) <= now:
return "expired"
return "active"
def to_dict(db: Session, row: LimitPolicyOverride) -> dict:
rule = limit_policy.get_rule(row.rule_code)
effective = limit_policy.resolve(
db,
row.rule_code,
phone=row.subject_value if row.subject_type == "phone" else None,
device=row.subject_value if row.subject_type == "device" else None,
)
return {
"id": row.id,
"subject_type": row.subject_type,
"subject_value": row.subject_value,
"rule_code": row.rule_code,
"rule_label": rule.label,
"rule_group": rule.group,
"mode": row.mode,
"limit_value": row.limit_value,
"global_limit": effective.global_limit,
"effective_limit": effective.limit,
"enabled": row.enabled,
"starts_at": row.starts_at,
"expires_at": row.expires_at,
"reset_at": row.reset_at,
"reason": row.reason,
"status": status_of(row),
"created_by_admin_id": row.created_by_admin_id,
"created_at": row.created_at,
"updated_at": row.updated_at,
}
def list_rows(
db: Session,
*,
subject_type: str | None = None,
keyword: str | None = None,
rule_code: str | None = None,
offset: int = 0,
limit: int = 100,
) -> tuple[list[LimitPolicyOverride], int]:
stmt = select(LimitPolicyOverride)
count_stmt = select(func.count(LimitPolicyOverride.id))
conditions = []
if subject_type:
conditions.append(LimitPolicyOverride.subject_type == subject_type)
if keyword:
conditions.append(LimitPolicyOverride.subject_value.ilike(f"%{keyword.strip()}%"))
if rule_code:
conditions.append(LimitPolicyOverride.rule_code == rule_code)
if conditions:
stmt = stmt.where(*conditions)
count_stmt = count_stmt.where(*conditions)
total = int(db.scalar(count_stmt) or 0)
rows = list(
db.execute(
stmt.order_by(
LimitPolicyOverride.created_at.desc(),
LimitPolicyOverride.id.desc(),
)
.offset(offset)
.limit(limit)
).scalars()
)
return rows, total
def list_subject_rows(
db: Session,
*,
subject_type: str | None = None,
keyword: str | None = None,
rule_code: str | None = None,
offset: int = 0,
limit: int = 10,
) -> tuple[list[tuple[str, str, list[LimitPolicyOverride], datetime]], int]:
"""List whitelist entries grouped and paginated by subject.
The filter decides which subjects match. Once a subject matches, all of
its rules are returned so category counts and the edit form stay complete.
Ordering uses the original creation time, therefore editing a subject does
not unexpectedly move it to the first page.
"""
conditions = [
LimitPolicyOverride.mode.in_(
[limit_policy.MODE_UNLIMITED, limit_policy.MODE_SUPPRESS_ALERT]
)
]
if subject_type:
conditions.append(LimitPolicyOverride.subject_type == subject_type)
if keyword:
conditions.append(LimitPolicyOverride.subject_value.ilike(f"%{keyword.strip()}%"))
if rule_code:
conditions.append(LimitPolicyOverride.rule_code == rule_code)
matched_subjects = (
select(
LimitPolicyOverride.subject_type.label("subject_type"),
LimitPolicyOverride.subject_value.label("subject_value"),
func.min(LimitPolicyOverride.created_at).label("subject_created_at"),
)
.where(*conditions)
.group_by(
LimitPolicyOverride.subject_type,
LimitPolicyOverride.subject_value,
)
.subquery()
)
total = int(db.scalar(select(func.count()).select_from(matched_subjects)) or 0)
subjects = list(
db.execute(
select(
matched_subjects.c.subject_type,
matched_subjects.c.subject_value,
matched_subjects.c.subject_created_at,
)
.order_by(
matched_subjects.c.subject_created_at.desc(),
matched_subjects.c.subject_type,
matched_subjects.c.subject_value,
)
.offset(offset)
.limit(limit)
).all()
)
if not subjects:
return [], total
pair_conditions = [
(
(LimitPolicyOverride.subject_type == item.subject_type)
& (LimitPolicyOverride.subject_value == item.subject_value)
)
for item in subjects
]
rows = list(
db.scalars(
select(LimitPolicyOverride)
.where(
or_(*pair_conditions),
LimitPolicyOverride.mode.in_(
[
limit_policy.MODE_UNLIMITED,
limit_policy.MODE_SUPPRESS_ALERT,
]
),
)
.order_by(
LimitPolicyOverride.created_at.desc(),
LimitPolicyOverride.id.desc(),
)
).all()
)
rows_by_subject: dict[tuple[str, str], list[LimitPolicyOverride]] = {}
for row in rows:
rows_by_subject.setdefault((row.subject_type, row.subject_value), []).append(row)
return [
(
item.subject_type,
item.subject_value,
rows_by_subject.get((item.subject_type, item.subject_value), []),
item.subject_created_at,
)
for item in subjects
], total
def rows_for_subject(
db: Session, *, subject_type: str, subject_value: str
) -> list[LimitPolicyOverride]:
return list(
db.scalars(
select(LimitPolicyOverride)
.where(
LimitPolicyOverride.subject_type == subject_type,
LimitPolicyOverride.subject_value == subject_value,
)
.order_by(
LimitPolicyOverride.created_at.desc(),
LimitPolicyOverride.id.desc(),
)
).all()
)
def create(
db: Session,
*,
subject_type: str,
subject_value: str,
rule_code: str,
mode: str,
limit_value: int | None,
enabled: bool,
starts_at: datetime | None,
expires_at: datetime | None,
reason: str,
admin_id: int,
reactivate_inactive: bool = False,
) -> LimitPolicyOverride:
rule = limit_policy.get_rule(rule_code)
subject_value = limit_policy.validate_whitelist_subject(subject_type, subject_value)
limit_policy.validate_override(
rule,
subject_type=subject_type,
mode=mode,
limit_value=limit_value,
starts_at=starts_at,
expires_at=expires_at,
)
existing = db.execute(
select(LimitPolicyOverride).where(
LimitPolicyOverride.subject_type == subject_type,
LimitPolicyOverride.subject_value == subject_value,
LimitPolicyOverride.rule_code == rule_code,
)
).scalar_one_or_none()
if existing is not None:
if not reactivate_inactive or status_of(existing) not in {"disabled", "expired"}:
raise DuplicateOverrideError
# “恢复全局”与自然过期都会保留原行供审计。再次加入白名单时
# 复用该行,避免唯一键让批量新增永久失败;完整变更仍写 admin 审计日志。
existing.mode = mode
existing.limit_value = limit_value if mode == limit_policy.MODE_OVERRIDE else None
existing.enabled = enabled
existing.starts_at = starts_at
existing.expires_at = expires_at
existing.reset_at = None
existing.reason = reason.strip() or None
db.flush()
return existing
row = LimitPolicyOverride(
subject_type=subject_type,
subject_value=subject_value,
rule_code=rule_code,
mode=mode,
limit_value=limit_value if mode == limit_policy.MODE_OVERRIDE else None,
enabled=enabled,
starts_at=starts_at,
expires_at=expires_at,
reason=reason.strip() or None,
created_by_admin_id=admin_id,
)
db.add(row)
try:
db.flush()
except IntegrityError as exc:
db.rollback()
raise DuplicateOverrideError from exc
return row
def update(
db: Session,
row: LimitPolicyOverride,
*,
enabled: bool | None,
starts_at: datetime | None,
expires_at: datetime | None,
reason: str | None,
fields_set: set[str],
) -> LimitPolicyOverride:
new_starts = starts_at if "starts_at" in fields_set else row.starts_at
new_expires = expires_at if "expires_at" in fields_set else row.expires_at
new_enabled = enabled if "enabled" in fields_set and enabled is not None else row.enabled
rule = limit_policy.get_rule(row.rule_code)
if row.mode == limit_policy.MODE_OVERRIDE and new_enabled:
raise ValueError("历史覆盖配置只能停用、恢复全局或删除")
if new_enabled:
limit_policy.validate_override(
rule,
subject_type=row.subject_type,
mode=row.mode,
limit_value=None,
starts_at=new_starts,
expires_at=new_expires,
)
if "enabled" in fields_set and enabled is not None:
row.enabled = enabled
if "starts_at" in fields_set:
row.starts_at = starts_at
if "expires_at" in fields_set:
row.expires_at = expires_at
if "reason" in fields_set:
row.reason = reason.strip() if reason else None
db.flush()
return row
def update_global_limit(
db: Session, rule_code: str, value: int, *, admin_id: int
) -> tuple[int, int]:
rule = limit_policy.get_rule(rule_code)
if not rule.min_value <= value <= rule.max_value:
raise ValueError(f"限制值必须在 {rule.min_value}{rule.max_value} 之间")
before = limit_policy.resolve(db, rule_code).global_limit
limit_policy.set_global_limits(
db,
{rule.code: value},
admin_id=admin_id,
commit=False,
)
# 引导视频仍有一个旧的专用配置页。同步其结构化配置,保证两个入口读取
# 同一数值;业务判定仍统一走 limit_policy。
if rule.legacy_config_key and rule.legacy_json_field:
legacy = db.get(AppConfig, rule.legacy_config_key)
legacy_value = (
dict(legacy.value) if legacy is not None and isinstance(legacy.value, dict) else {}
)
legacy_value[rule.legacy_json_field] = value
if legacy is None:
legacy = AppConfig(
key=rule.legacy_config_key,
value=legacy_value,
updated_by_admin_id=admin_id,
)
db.add(legacy)
else:
legacy.value = legacy_value
legacy.updated_by_admin_id = admin_id
db.flush()
return before, value
def restore_global(row: LimitPolicyOverride) -> LimitPolicyOverride:
"""Stop applying the exception while retaining its history for audit."""
row.enabled = False
row.reset_at = None
return row
+57 -14
View File
@@ -14,6 +14,7 @@ from sqlalchemy.orm import Session
from app.core import rewards
from app.core.config import settings
from app.models.ad_ecpm import AdEcpmRecord
from app.models.ad_feed_reward import AdFeedRewardRecord
from app.models.ad_reward import AdRewardRecord
from app.models.admin import AdminAuditLog
@@ -1257,11 +1258,15 @@ def user_reward_stats(
date_from: datetime | None = None,
date_to: datetime | None = None,
withdraw_source: str | None = None,
app_env: str | None = None,
revenue_scope: str = "all",
feed_scene: str | None = None,
) -> dict:
"""提现详情「用户统计区」10 项。窗口作用于除「现金余额」外的所有项(余额是当前快照)。
口径:激励视频/信息流只统计 granted;数量——视频按条数、信息流按份数(unit_count 累加);
平均 eCPM 用原始分值(分/千次)按记录取算术平均;各「提现」= 该来源累计金币折现。
口径:激励视频/信息流奖励数量只统计 granted;数量——视频按条数、信息流按份数(unit_count 累加)
平均 Draw eCPM 与广告收益报表一致:基于 ad_ecpm_record 的全部 draw/feed 展示记录计算,
不以是否发奖为筛选条件。各「提现」= 该来源累计金币折现。
传统任务 = 窗口内正向金币中,排除广告(reward_video/feed_ad_reward)与人工调整后的折现。
"""
withdraw_source_conds = (
@@ -1296,30 +1301,68 @@ def user_reward_stats(
# 只投影本统计实际使用的列。避免滚动发布或旧本地库尚未补齐无关新列时,
# SQLAlchemy 因 select(ORM) 自动展开整表字段而让提现详情整体 500。
business_ids: set[str] | None = None
if revenue_scope == "business":
# 与广告收益报表共用正式/测试业务代码位集合,避免两个页面随配置切换后再次漂移。
from app.admin.repositories.ad_revenue import business_code_ids
business_ids = business_code_ids(db, app_env)
rv_conds = [
AdRewardRecord.user_id == user_id,
AdRewardRecord.reward_scene == "reward_video",
AdRewardRecord.status == "granted",
*_window_conds(AdRewardRecord.created_at, date_from, date_to),
]
if app_env is not None:
rv_conds.append(AdRewardRecord.app_env == app_env)
if business_ids is not None:
rv_conds.append(AdRewardRecord.our_code_id.in_(business_ids))
rv = db.execute(
select(AdRewardRecord.ecpm_raw, AdRewardRecord.coin).where(
AdRewardRecord.user_id == user_id,
AdRewardRecord.reward_scene == "reward_video",
AdRewardRecord.status == "granted",
*_window_conds(AdRewardRecord.created_at, date_from, date_to),
*rv_conds,
)
).all()
rv_ecpms = [rewards.parse_ecpm_fen(r.ecpm_raw) for r in rv if r.ecpm_raw]
rv_coins = sum(r.coin for r in rv)
feed = db.execute(
feed_reward_conds = [
AdFeedRewardRecord.user_id == user_id,
AdFeedRewardRecord.status == "granted",
*_window_conds(AdFeedRewardRecord.created_at, date_from, date_to),
]
if app_env is not None:
feed_reward_conds.append(AdFeedRewardRecord.app_env == app_env)
if feed_scene is not None:
feed_reward_conds.append(AdFeedRewardRecord.feed_scene == feed_scene)
if business_ids is not None:
feed_reward_conds.append(AdFeedRewardRecord.our_code_id.in_(business_ids))
feed_rewards = db.execute(
select(
AdFeedRewardRecord.unit_count,
AdFeedRewardRecord.ecpm_raw,
AdFeedRewardRecord.coin,
).where(
AdFeedRewardRecord.user_id == user_id,
AdFeedRewardRecord.status == "granted",
*_window_conds(AdFeedRewardRecord.created_at, date_from, date_to),
*feed_reward_conds,
)
).all()
feed_ecpms = [rewards.parse_ecpm_fen(f.ecpm_raw) for f in feed if f.ecpm_raw]
feed_coins = sum(f.coin for f in feed)
feed_coins = sum(f.coin for f in feed_rewards)
feed_impression_conds = [
AdEcpmRecord.user_id == user_id,
AdEcpmRecord.ad_type.in_(("draw", "feed")),
*_window_conds(AdEcpmRecord.created_at, date_from, date_to),
]
if app_env is not None:
feed_impression_conds.append(AdEcpmRecord.app_env == app_env)
if feed_scene is not None:
feed_impression_conds.append(AdEcpmRecord.feed_scene == feed_scene)
if business_ids is not None:
feed_impression_conds.append(AdEcpmRecord.our_code_id.in_(business_ids))
feed_impressions = db.execute(
select(AdEcpmRecord.ecpm_raw).where(*feed_impression_conds)
).all()
# 与 ad_revenue.category_stats 相同:每次展示权重相同,非法原值按 parse_ecpm_fen 记 0。
feed_ecpms = [rewards.parse_ecpm_fen(row.ecpm_raw) for row in feed_impressions]
trad_coins = db.execute(
select(func.coalesce(func.sum(CoinTransaction.amount), 0)).where(
@@ -1338,7 +1381,7 @@ def user_reward_stats(
"reward_video_count": len(rv),
"reward_video_avg_ecpm": round(sum(rv_ecpms) / len(rv_ecpms), 2) if rv_ecpms else 0.0,
"reward_video_cash_cents": _coins_to_cents(rv_coins),
"feed_count": int(sum(f.unit_count for f in feed)),
"feed_count": int(sum(f.unit_count for f in feed_rewards)),
"feed_avg_ecpm": round(sum(feed_ecpms) / len(feed_ecpms), 2) if feed_ecpms else 0.0,
"feed_cash_cents": _coins_to_cents(feed_coins),
}
+75 -8
View File
@@ -12,9 +12,11 @@ from fastapi import APIRouter, Depends, HTTPException, Request
from app.admin.audit import write_audit
from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
from app.admin.schemas.config import ConfigItemOut, ConfigUpdateRequest
from app.core import limit_policy
from app.core.config_schema import CONFIG_DEFS
from app.core.rewards import SIGNIN_CYCLE_LEN
from app.models.admin import AdminUser
from app.models.app_config import AppConfig
from app.repositories import app_config
router = APIRouter(
@@ -54,7 +56,35 @@ def _validate(key: str, value: Any) -> None:
raise ValueError("需为布尔值")
def _limit_item(
key: str,
values: dict[str, int],
bundle: AppConfig | None,
) -> ConfigItemOut:
definition = CONFIG_DEFS[key]
rule_code = limit_policy.RULE_CODE_BY_CONFIG_KEY[key]
return ConfigItemOut(
key=key,
value=values[rule_code],
label=definition["label"],
group=definition["group"],
type=definition["type"],
help=definition.get("help"),
default=definition["default"],
overridden=bundle is not None,
updated_at=(
bundle.updated_at.isoformat() if bundle is not None else None
),
)
def _item(db, key: str) -> ConfigItemOut:
if key in limit_policy.RULE_CODE_BY_CONFIG_KEY:
return _limit_item(
key,
limit_policy.get_global_limits(db),
db.get(AppConfig, limit_policy.LIMIT_POLICY_GLOBAL_KEY),
)
for item in app_config.list_all(db):
if item["key"] == key:
return ConfigItemOut(**item)
@@ -64,11 +94,20 @@ def _item(db, key: str) -> ConfigItemOut:
@router.get("", response_model=list[ConfigItemOut], summary="所有可配项 + 当前值(不含 hidden)")
def list_config(db: AdminDb) -> list[ConfigItemOut]:
# hidden 项(已下线/由专用页管理,如福利页任务·里程碑·看广告调参、首页轮播数据源)不在本页渲染。
return [
ConfigItemOut(**item)
for item in app_config.list_all(db)
if not CONFIG_DEFS[item["key"]].get("hidden")
]
legacy_items = {
item["key"]: item for item in app_config.list_all(db)
}
values = limit_policy.get_global_limits(db)
bundle = db.get(AppConfig, limit_policy.LIMIT_POLICY_GLOBAL_KEY)
out: list[ConfigItemOut] = []
for key, definition in CONFIG_DEFS.items():
if definition.get("hidden"):
continue
if key in limit_policy.RULE_CODE_BY_CONFIG_KEY:
out.append(_limit_item(key, values, bundle))
else:
out.append(ConfigItemOut(**legacy_items[key]))
return out
@router.patch("/{key}", response_model=ConfigItemOut, summary="改某项配置(带审计)")
@@ -83,11 +122,39 @@ def update_config(
raise HTTPException(status_code=404, detail="未知配置项")
try:
_validate(key, body.value)
rule_code = limit_policy.RULE_CODE_BY_CONFIG_KEY.get(key)
if rule_code is not None:
before = limit_policy.get_global_limits(db)[rule_code]
limit_policy.set_global_limits(
db,
{rule_code: body.value},
admin_id=admin.id,
commit=False,
)
else:
before = app_config.get_value(db, key)
app_config.set_value(
db,
key,
body.value,
admin_id=admin.id,
commit=False,
)
if key == "ad_daily_limit":
# 旧系统配置接口过去只有一个广告日上限。仍有人直接调用时,同时同步
# 新的激励视频/Draw 两项,避免旧入口写入后业务实际值不变。
limit_policy.set_global_limits(
db,
{
"ad.reward_video.daily": body.value,
"ad.feed.daily": body.value,
},
admin_id=admin.id,
commit=False,
)
except ValueError as e:
db.rollback()
raise HTTPException(status_code=400, detail=str(e)) from e
before = app_config.get_value(db, key)
app_config.set_value(db, key, body.value, admin_id=admin.id, commit=False)
write_audit(
db, admin, action="config.set", target_type="config", target_id=key,
detail={"before": before, "after": body.value}, ip=get_client_ip(request), commit=False,
+109 -6
View File
@@ -1,6 +1,7 @@
"""admin 反馈工单:列表筛选 + 审核采纳/拒绝(带金币发放与审计)。"""
from __future__ import annotations
import logging
from datetime import datetime
from typing import Annotated
@@ -25,6 +26,8 @@ from app.models.feedback import Feedback
from app.repositories import wallet as wallet_repo
from app.services import notification_events
logger = logging.getLogger("shagua.admin.feedback")
router = APIRouter(
prefix="/admin/api/feedbacks",
tags=["admin-feedback"],
@@ -46,10 +49,24 @@ def _approve_feedback(
*,
bulk: bool = False,
) -> FeedbackOut:
logger.info(
"feedback approve started feedback_id=%s admin_id=%s bulk=%s",
feedback_id, admin.id, bulk,
)
fb = db.get(Feedback, feedback_id, with_for_update=True)
if fb is None:
logger.warning(
"feedback approve rejected not found feedback_id=%s admin_id=%s bulk=%s",
feedback_id, admin.id, bulk,
)
raise HTTPException(status_code=404, detail="反馈不存在")
_ensure_pending(fb)
if fb.status not in {"pending", "new"}:
logger.warning(
"feedback approve rejected invalid status feedback_id=%s user_id=%s "
"admin_id=%s status=%s bulk=%s",
feedback_id, fb.user_id, admin.id, fb.status, bulk,
)
_ensure_pending(fb)
before = fb.status
mutations.review_feedback(
@@ -91,8 +108,18 @@ def _approve_feedback(
)
db.commit()
db.refresh(fb)
logger.info(
"feedback approve committed feedback_id=%s user_id=%s admin_id=%s "
"before=%s after=%s reward_coins=%s bulk=%s",
feedback_id, fb.user_id, admin.id, before, fb.status, payload.reward_coins, bulk,
)
out = FeedbackOut.model_validate(fb)
notification_events.notify_feedback_reward(db, fb)
logger.info(
"feedback approve notification dispatch returned feedback_id=%s user_id=%s "
"admin_id=%s notification_type=feedback_reward bulk=%s",
feedback_id, fb.user_id, admin.id, bulk,
)
return out
@@ -105,10 +132,24 @@ def _reject_feedback(
*,
bulk: bool = False,
) -> FeedbackOut:
logger.info(
"feedback reject started feedback_id=%s admin_id=%s bulk=%s",
feedback_id, admin.id, bulk,
)
fb = db.get(Feedback, feedback_id, with_for_update=True)
if fb is None:
logger.warning(
"feedback reject rejected not found feedback_id=%s admin_id=%s bulk=%s",
feedback_id, admin.id, bulk,
)
raise HTTPException(status_code=404, detail="反馈不存在")
_ensure_pending(fb)
if fb.status not in {"pending", "new"}:
logger.warning(
"feedback reject rejected invalid status feedback_id=%s user_id=%s "
"admin_id=%s status=%s bulk=%s",
feedback_id, fb.user_id, admin.id, fb.status, bulk,
)
_ensure_pending(fb)
before = fb.status
mutations.review_feedback(
@@ -142,8 +183,18 @@ def _reject_feedback(
)
db.commit()
db.refresh(fb)
logger.info(
"feedback reject committed feedback_id=%s user_id=%s admin_id=%s "
"before=%s after=%s bulk=%s",
feedback_id, fb.user_id, admin.id, before, fb.status, bulk,
)
out = FeedbackOut.model_validate(fb)
notification_events.notify_feedback_reply(db, fb)
logger.info(
"feedback reject notification dispatch returned feedback_id=%s user_id=%s "
"admin_id=%s notification_type=feedback_reply bulk=%s",
feedback_id, fb.user_id, admin.id, bulk,
)
return out
@@ -201,6 +252,10 @@ def bulk_approve_feedbacks(
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
) -> FeedbackBulkResult:
logger.info(
"feedback bulk approve started admin_id=%s item_count=%s",
admin.id, len(body.ids),
)
results: list[FeedbackBulkItemResult] = []
ip = get_client_ip(request)
for feedback_id in body.ids:
@@ -209,11 +264,24 @@ def bulk_approve_feedbacks(
results.append(FeedbackBulkItemResult(id=feedback_id, ok=True, status=out.status))
except HTTPException as exc:
db.rollback()
logger.warning(
"feedback bulk approve item failed feedback_id=%s admin_id=%s error=%s",
feedback_id, admin.id, exc.detail,
)
results.append(FeedbackBulkItemResult(id=feedback_id, ok=False, error=str(exc.detail)))
except Exception: # noqa: BLE001 - 单笔失败不打断整批
db.rollback()
logger.exception(
"feedback bulk approve item failed feedback_id=%s admin_id=%s",
feedback_id, admin.id,
)
results.append(FeedbackBulkItemResult(id=feedback_id, ok=False, error="系统异常"))
return _bulk_result(results)
result = _bulk_result(results)
logger.info(
"feedback bulk approve completed admin_id=%s total=%s success=%s failed=%s",
admin.id, result.total, result.success, result.failed,
)
return result
@router.post("/bulk/reject", response_model=FeedbackBulkResult, summary="批量拒绝采纳反馈")
@@ -223,6 +291,10 @@ def bulk_reject_feedbacks(
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
) -> FeedbackBulkResult:
logger.info(
"feedback bulk reject started admin_id=%s item_count=%s",
admin.id, len(body.ids),
)
results: list[FeedbackBulkItemResult] = []
ip = get_client_ip(request)
for feedback_id in body.ids:
@@ -231,11 +303,24 @@ def bulk_reject_feedbacks(
results.append(FeedbackBulkItemResult(id=feedback_id, ok=True, status=out.status))
except HTTPException as exc:
db.rollback()
logger.warning(
"feedback bulk reject item failed feedback_id=%s admin_id=%s error=%s",
feedback_id, admin.id, exc.detail,
)
results.append(FeedbackBulkItemResult(id=feedback_id, ok=False, error=str(exc.detail)))
except Exception: # noqa: BLE001 - 单笔失败不打断整批
db.rollback()
logger.exception(
"feedback bulk reject item failed feedback_id=%s admin_id=%s",
feedback_id, admin.id,
)
results.append(FeedbackBulkItemResult(id=feedback_id, ok=False, error="系统异常"))
return _bulk_result(results)
result = _bulk_result(results)
logger.info(
"feedback bulk reject completed admin_id=%s total=%s success=%s failed=%s",
admin.id, result.total, result.success, result.failed,
)
return result
@router.post("/{feedback_id}/handle", response_model=OkResponse, summary="标记反馈已处理")
@@ -258,7 +343,16 @@ def approve_feedback(
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
) -> FeedbackOut:
return _approve_feedback(db, admin, feedback_id, payload, get_client_ip(request))
try:
return _approve_feedback(db, admin, feedback_id, payload, get_client_ip(request))
except HTTPException:
raise
except Exception:
logger.exception(
"feedback approve failed feedback_id=%s admin_id=%s",
feedback_id, admin.id,
)
raise
@router.post("/{feedback_id}/reject", response_model=FeedbackOut, summary="拒绝采纳反馈")
@@ -269,4 +363,13 @@ def reject_feedback(
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
) -> FeedbackOut:
return _reject_feedback(db, admin, feedback_id, payload, get_client_ip(request))
try:
return _reject_feedback(db, admin, feedback_id, payload, get_client_ip(request))
except HTTPException:
raise
except Exception:
logger.exception(
"feedback reject failed feedback_id=%s admin_id=%s",
feedback_id, admin.id,
)
raise
+60 -32
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
from typing import Annotated, Literal
from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile
@@ -27,14 +27,16 @@ router = APIRouter(
)
def _out(db: AdminDb) -> GuideVideoConfigOut:
"""配置 + 播放统计合成响应(四个写接口都以最新状态返回,前端一次同步到位)。"""
return GuideVideoConfigOut(**guide_video.get_config(db), **guide_video.play_stats(db))
GuideScene = Literal["coupon", "comparison"]
def _out(db: AdminDb, scene: GuideScene) -> GuideVideoConfigOut:
return GuideVideoConfigOut(scene=scene, **guide_video.get_config(db, scene))
@router.get("", response_model=GuideVideoConfigOut, summary="新手引导视频配置(领券浮层)")
def get_config(db: AdminDb) -> GuideVideoConfigOut:
return _out(db)
def get_config(db: AdminDb, scene: GuideScene = "coupon") -> GuideVideoConfigOut:
return _out(db, scene)
@router.patch("", response_model=GuideVideoConfigOut, summary="改开关/次数/金币(带审计)")
@@ -43,21 +45,26 @@ 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,
admin_id=admin.id,
commit=False,
)
try:
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,
)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
write_audit(
db, admin, action="guide_video.update", target_type="guide_video", target_id=None,
detail={"before": before, "after": after}, ip=get_client_ip(request), commit=False,
detail={"scene": scene, "before": before, "after": after}, ip=get_client_ip(request), commit=False,
)
db.commit()
return _out(db)
return _out(db, scene)
@router.post("/video", response_model=GuideVideoConfigOut, summary="上传新手引导视频(MP4,带审计)")
@@ -65,23 +72,40 @@ async def upload_video(
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
file: UploadFile = File(...),
file: Annotated[UploadFile, File()],
scene: GuideScene = "coupon",
) -> GuideVideoConfigOut:
data = await file.read()
url: str | None = None
try:
url = media.save_guide_video(data)
url, analysis = media.save_guide_video(data)
except media.MediaError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
before, after = guide_video.set_video(db, url, admin_id=admin.id, commit=False)
write_audit(
db, admin, action="guide_video.set_video", target_type="guide_video", target_id=None,
detail={"before": before.get("video_url"), "after": url, "bytes": len(data)},
ip=get_client_ip(request), commit=False,
)
db.commit()
# 提交成功后再删旧片,避免新片没落库就把旧片丢了
media.delete_guide_video(before.get("video_url"))
return _out(db)
try:
before, after = guide_video.set_video(
db, url, analysis=analysis, scene=scene, admin_id=admin.id, commit=False
)
write_audit(
db, admin, action="guide_video.set_video", target_type="guide_video", target_id=None,
detail={
"scene": scene,
"before": before.get("video_url"),
"after": url,
"bytes": len(data),
"duration_ms": analysis["duration_ms"],
"video_codec": analysis["video_codec"],
"audio_codec": analysis["audio_codec"],
},
ip=get_client_ip(request), commit=False,
)
db.commit()
except Exception:
db.rollback()
media.delete_guide_video(url)
raise
# 旧文件仍可能被 prepare/已起播快照引用,不能在这里立即删除。
guide_video.cleanup_old_videos(db)
return _out(db, scene)
@router.delete("/video", response_model=GuideVideoConfigOut, summary="移除新手引导视频(带审计)")
@@ -89,13 +113,17 @@ def delete_video(
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
scene: GuideScene = "coupon",
) -> GuideVideoConfigOut:
"""移除后 /guide-video/start 一律返回 should_play=false,领券浮层回到「只放广告」。"""
before, after = guide_video.set_video(db, None, admin_id=admin.id, commit=False)
before, after = guide_video.set_video(
db, None, scene=scene, admin_id=admin.id, commit=False
)
write_audit(
db, admin, action="guide_video.delete_video", target_type="guide_video", target_id=None,
detail={"before": before.get("video_url")}, ip=get_client_ip(request), commit=False,
detail={"scene": scene, "before": before.get("video_url")}, ip=get_client_ip(request), commit=False,
)
db.commit()
media.delete_guide_video(before.get("video_url"))
return _out(db)
# 旧文件由保护期清理任务处理,避免破坏有效计划与已起播快照。
guide_video.cleanup_old_videos(db)
return _out(db, scene)
+858
View File
@@ -0,0 +1,858 @@
"""Unified limit rules and per-phone/device whitelist overrides."""
from __future__ import annotations
from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
from app.admin.audit import write_audit
from app.admin.deps import AdminDb, CurrentAdmin, get_client_ip, require_page
from app.admin.repositories import limit_whitelist as repo
from app.admin.schemas.limit_whitelist import (
DeviceCandidateOut,
GlobalLimitUpdate,
LimitOverrideBulkWrite,
LimitOverrideList,
LimitOverrideOut,
LimitOverridePatch,
LimitOverrideWrite,
LimitRuleOut,
LimitSubjectEnabledPatch,
LimitSubjectList,
LimitSubjectOut,
)
from app.core import limit_policy
from app.models.limit_policy import LimitPolicyOverride
from app.repositories import risk as risk_repo
router = APIRouter(
prefix="/admin/api/limit-whitelist",
tags=["admin-limit-whitelist"],
dependencies=[Depends(require_page("limit-whitelist"))],
)
_WHITELIST_MODES = {
limit_policy.MODE_UNLIMITED,
limit_policy.MODE_SUPPRESS_ALERT,
}
def _subject_whitelist_rows(
db,
*,
subject_type: str,
subject_value: str,
) -> list[LimitPolicyOverride]:
return [
row
for row in repo.rows_for_subject(
db,
subject_type=subject_type,
subject_value=subject_value,
)
if row.mode in _WHITELIST_MODES
]
def _reconcile_risk_rule(db, rule_code: str) -> None:
now = risk_repo.utcnow()
if rule_code == "risk.sms.hourly":
risk_repo.reconcile_behavior_rule(
db,
rule_code=risk_repo.RULE_SMS_HOURLY,
at=now,
commit=False,
)
elif rule_code == "risk.oneclick.daily":
risk_repo.reconcile_behavior_rule(
db,
rule_code=risk_repo.RULE_ONECLICK_DAILY,
at=now,
commit=False,
)
elif rule_code == "risk.compare.daily":
risk_repo.reconcile_compare_rule(db, at=now, commit=False)
def _row_or_404(db, override_id: int) -> LimitPolicyOverride:
row = db.get(LimitPolicyOverride, override_id, populate_existing=True)
if row is None:
raise HTTPException(status_code=404, detail="白名单配置不存在")
return row
def _out(db, row: LimitPolicyOverride) -> LimitOverrideOut:
return LimitOverrideOut(**repo.to_dict(db, row))
def _audit_payload(value: LimitOverrideOut) -> dict:
return value.model_dump(mode="json")
def _subject_out(
db,
*,
subject_type: str,
subject_value: str,
rows: list[LimitPolicyOverride],
created_at,
) -> LimitSubjectOut:
items = [_out(db, row) for row in rows]
group_counts: dict[str, int] = {}
for item in items:
group_counts[item.rule_group] = group_counts.get(item.rule_group, 0) + 1
return LimitSubjectOut(
subject_type=subject_type,
subject_value=subject_value,
group_counts=group_counts,
total_rules=len(items),
items=items,
created_at=created_at,
updated_at=max(
(item.updated_at for item in items),
default=created_at,
),
)
@router.get("/rules", response_model=list[LimitRuleOut], summary="读取所有限制规则")
def list_rules(db: AdminDb) -> list[LimitRuleOut]:
return [LimitRuleOut(**item) for item in limit_policy.rule_catalog(db)]
@router.patch(
"/rules/{rule_code}",
response_model=LimitRuleOut,
summary="修改规则的全局限制值",
)
def update_rule(
rule_code: str,
body: GlobalLimitUpdate,
request: Request,
admin: CurrentAdmin,
db: AdminDb,
) -> LimitRuleOut:
try:
before, after = repo.update_global_limit(db, rule_code, body.value, admin_id=admin.id)
except KeyError as exc:
raise HTTPException(status_code=404, detail="未知限制规则") from exc
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
_reconcile_risk_rule(db, rule_code)
write_audit(
db,
admin,
action="limit.rule.update",
target_type="limit_rule",
target_id=rule_code,
detail={"before": before, "after": after},
ip=get_client_ip(request),
commit=False,
)
db.commit()
item = next(item for item in limit_policy.rule_catalog(db) if item["code"] == rule_code)
return LimitRuleOut(**item)
@router.get(
"/device-candidates",
response_model=list[DeviceCandidateOut],
summary="按限制项搜索可加入白名单的设备",
)
def list_device_candidates(
db: AdminDb,
rule_code: str = Query(..., min_length=1, max_length=64),
keyword: str | None = Query(None, max_length=128),
limit: int = Query(30, ge=1, le=50),
) -> list[DeviceCandidateOut]:
try:
rows = repo.list_device_candidates(
db,
rule_code=rule_code,
keyword=keyword,
limit=limit,
)
except KeyError as exc:
raise HTTPException(status_code=404, detail="未知限制规则") from exc
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
return [DeviceCandidateOut(**row) for row in rows]
@router.get("", response_model=LimitOverrideList, summary="读取白名单配置")
def list_overrides(
db: AdminDb,
subject_type: str | None = Query(None, pattern="^(phone|device)$"),
keyword: str | None = Query(None, max_length=128),
rule_code: str | None = Query(None, max_length=64),
offset: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=500),
) -> LimitOverrideList:
rows, total = repo.list_rows(
db,
subject_type=subject_type,
keyword=keyword,
rule_code=rule_code,
offset=offset,
limit=limit,
)
return LimitOverrideList(items=[_out(db, row) for row in rows], total=total)
@router.get(
"/subjects",
response_model=LimitSubjectList,
summary="按手机号或设备聚合读取白名单配置",
)
def list_override_subjects(
db: AdminDb,
subject_type: str | None = Query(None, pattern="^(phone|device)$"),
keyword: str | None = Query(None, max_length=128),
rule_code: str | None = Query(None, max_length=64),
offset: int = Query(0, ge=0),
limit: int = Query(10, ge=1, le=100),
) -> LimitSubjectList:
subjects, total = repo.list_subject_rows(
db,
subject_type=subject_type,
keyword=keyword,
rule_code=rule_code,
offset=offset,
limit=limit,
)
return LimitSubjectList(
items=[
_subject_out(
db,
subject_type=item_subject_type,
subject_value=subject_value,
rows=rows,
created_at=created_at,
)
for item_subject_type, subject_value, rows, created_at in subjects
],
total=total,
)
@router.patch(
"/subjects/enabled",
response_model=LimitSubjectOut,
summary="整体启用或停用一个手机号或设备的白名单",
)
def set_override_subject_enabled(
body: LimitSubjectEnabledPatch,
request: Request,
admin: CurrentAdmin,
db: AdminDb,
) -> LimitSubjectOut:
try:
subject_value = limit_policy.validate_whitelist_subject(
body.subject_type, body.subject_value
)
repo.lock_subject(
db,
subject_type=body.subject_type,
subject_value=subject_value,
)
rows = _subject_whitelist_rows(
db,
subject_type=body.subject_type,
subject_value=subject_value,
)
if not rows:
raise HTTPException(status_code=404, detail="白名单主体不存在")
before = [_audit_payload(_out(db, row)) for row in rows]
for row in rows:
repo.update(
db,
row,
enabled=body.enabled,
starts_at=None,
expires_at=None,
reason=None,
fields_set={"enabled"},
)
for rule_code in {row.rule_code for row in rows}:
_reconcile_risk_rule(db, rule_code)
after = [_audit_payload(_out(db, row)) for row in rows]
write_audit(
db,
admin,
action="limit.override.subject_enabled",
target_type="limit_override_subject",
target_id=f"{body.subject_type}:{subject_value}",
detail={"before": before, "after": after},
ip=get_client_ip(request),
commit=False,
)
db.commit()
except HTTPException:
raise
except (KeyError, ValueError) as exc:
db.rollback()
raise HTTPException(status_code=400, detail=str(exc)) from exc
rows = [
row
for row in repo.rows_for_subject(
db,
subject_type=body.subject_type,
subject_value=subject_value,
)
if row.mode
in {
limit_policy.MODE_UNLIMITED,
limit_policy.MODE_SUPPRESS_ALERT,
}
]
return _subject_out(
db,
subject_type=body.subject_type,
subject_value=subject_value,
rows=rows,
created_at=min(row.created_at for row in rows),
)
@router.put(
"/subjects",
response_model=LimitSubjectOut,
summary="整体更新一个手机号或设备的白名单限制项",
)
def replace_override_subject(
body: LimitOverrideBulkWrite,
request: Request,
admin: CurrentAdmin,
db: AdminDb,
) -> LimitSubjectOut:
current_rule_code = ""
try:
subject_value = limit_policy.validate_whitelist_subject(
body.subject_type, body.subject_value
)
repo.lock_subject(
db,
subject_type=body.subject_type,
subject_value=subject_value,
)
if body.subject_type == limit_policy.SUBJECT_DEVICE:
limit_policy.validate_device_rule_scope(body.rule_codes)
existing_rows = repo.rows_for_subject(
db,
subject_type=body.subject_type,
subject_value=subject_value,
)
subject_created_at = min(
(row.created_at for row in existing_rows),
default=None,
)
before = [_audit_payload(_out(db, row)) for row in existing_rows]
existing_by_rule = {row.rule_code: row for row in existing_rows}
selected_codes = set(body.rule_codes)
touched_rule_codes = set(selected_codes)
for current_rule_code in body.rule_codes:
rule = limit_policy.get_rule(current_rule_code)
mode = (
limit_policy.MODE_SUPPRESS_ALERT if rule.alert_only else limit_policy.MODE_UNLIMITED
)
row = existing_by_rule.get(current_rule_code)
if row is None:
row = repo.create(
db,
subject_type=body.subject_type,
subject_value=subject_value,
rule_code=current_rule_code,
mode=mode,
limit_value=None,
enabled=body.enabled,
starts_at=body.starts_at,
expires_at=body.expires_at,
reason=body.reason,
admin_id=admin.id,
)
# The table is ordered by the subject's original creation
# time. If an edit replaces every rule, carry that timestamp
# to the new rows so the subject does not jump to the top.
if subject_created_at is not None:
row.created_at = subject_created_at
continue
# Historical custom-value rows are converted to the only supported
# product modes when the administrator selects that rule again.
row.mode = mode
row.limit_value = None
row.reset_at = None
repo.update(
db,
row,
enabled=body.enabled,
starts_at=body.starts_at,
expires_at=body.expires_at,
reason=body.reason,
fields_set={"enabled", "starts_at", "expires_at", "reason"},
)
for row in existing_rows:
if row.rule_code not in selected_codes and row.mode in {
limit_policy.MODE_UNLIMITED,
limit_policy.MODE_SUPPRESS_ALERT,
}:
touched_rule_codes.add(row.rule_code)
db.delete(row)
db.flush()
except KeyError as exc:
raise HTTPException(status_code=404, detail="未知限制规则") from exc
except repo.DuplicateOverrideError as exc:
rule = limit_policy.get_rule(current_rule_code)
raise HTTPException(
status_code=409,
detail=f"{rule.label}”已有白名单配置,请刷新后重试",
) from exc
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
for rule_code in touched_rule_codes:
_reconcile_risk_rule(db, rule_code)
rows = [
row
for row in repo.rows_for_subject(
db,
subject_type=body.subject_type,
subject_value=subject_value,
)
if row.mode
in {
limit_policy.MODE_UNLIMITED,
limit_policy.MODE_SUPPRESS_ALERT,
}
]
after = [_audit_payload(_out(db, row)) for row in rows]
write_audit(
db,
admin,
action="limit.override.subject_replace",
target_type="limit_override_subject",
target_id=f"{body.subject_type}:{subject_value}",
detail={"before": before, "after": after},
ip=get_client_ip(request),
commit=False,
)
db.commit()
rows = [
row
for row in repo.rows_for_subject(
db,
subject_type=body.subject_type,
subject_value=subject_value,
)
if row.mode
in {
limit_policy.MODE_UNLIMITED,
limit_policy.MODE_SUPPRESS_ALERT,
}
]
return _subject_out(
db,
subject_type=body.subject_type,
subject_value=subject_value,
rows=rows,
created_at=min(row.created_at for row in rows),
)
@router.post(
"",
response_model=LimitOverrideOut,
status_code=status.HTTP_201_CREATED,
summary="新增白名单配置",
)
def create_override(
body: LimitOverrideWrite,
request: Request,
admin: CurrentAdmin,
db: AdminDb,
) -> LimitOverrideOut:
before: list[dict] = []
touched_rows: list[LimitPolicyOverride] = []
try:
subject_value = limit_policy.validate_whitelist_subject(
body.subject_type, body.subject_value
)
repo.lock_subject(
db,
subject_type=body.subject_type,
subject_value=subject_value,
)
existing_rows = _subject_whitelist_rows(
db,
subject_type=body.subject_type,
subject_value=subject_value,
)
if body.subject_type == limit_policy.SUBJECT_DEVICE:
limit_policy.validate_device_rule_scope(
[
*(item.rule_code for item in existing_rows),
body.rule_code,
]
)
before = [_audit_payload(_out(db, item)) for item in existing_rows]
row = repo.create(
db,
subject_type=body.subject_type,
subject_value=subject_value,
rule_code=body.rule_code,
mode=body.mode,
enabled=body.enabled,
starts_at=body.starts_at,
expires_at=body.expires_at,
reason=body.reason,
limit_value=None,
admin_id=admin.id,
)
touched_rows = [*existing_rows, row]
if row.mode in _WHITELIST_MODES:
# Keep the legacy single-rule endpoint compatible without letting
# it create a second validity period for the same logical subject.
for existing in existing_rows:
repo.update(
db,
existing,
enabled=body.enabled,
starts_at=body.starts_at,
expires_at=body.expires_at,
reason=None,
fields_set={"enabled", "starts_at", "expires_at"},
)
except KeyError as exc:
raise HTTPException(status_code=404, detail="未知限制规则") from exc
except repo.DuplicateOverrideError as exc:
raise HTTPException(
status_code=409,
detail="该手机号或设备已配置此规则,请编辑现有配置",
) from exc
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
for rule_code in {item.rule_code for item in touched_rows}:
_reconcile_risk_rule(db, rule_code)
after_rows = _subject_whitelist_rows(
db,
subject_type=row.subject_type,
subject_value=row.subject_value,
)
write_audit(
db,
admin,
action="limit.override.create",
target_type="limit_override_subject",
target_id=f"{row.subject_type}:{row.subject_value}",
detail={
"before": before,
"after": [_audit_payload(_out(db, item)) for item in after_rows],
},
ip=get_client_ip(request),
commit=False,
)
db.commit()
return _out(db, row)
@router.post(
"/bulk",
response_model=list[LimitOverrideOut],
status_code=status.HTTP_201_CREATED,
summary="批量新增临时不限或免告警白名单",
)
def create_overrides_bulk(
body: LimitOverrideBulkWrite,
request: Request,
admin: CurrentAdmin,
db: AdminDb,
) -> list[LimitOverrideOut]:
rows: list[LimitPolicyOverride] = []
before: list[dict] = []
current_rule_code = ""
try:
subject_value = limit_policy.validate_whitelist_subject(
body.subject_type, body.subject_value
)
repo.lock_subject(
db,
subject_type=body.subject_type,
subject_value=subject_value,
)
existing_rows = repo.rows_for_subject(
db,
subject_type=body.subject_type,
subject_value=subject_value,
)
whitelist_modes = {
limit_policy.MODE_UNLIMITED,
limit_policy.MODE_SUPPRESS_ALERT,
}
existing_whitelist_rows = [row for row in existing_rows if row.mode in whitelist_modes]
before = [_audit_payload(_out(db, row)) for row in existing_whitelist_rows]
existing_by_rule = {row.rule_code: row for row in existing_rows}
selected_codes = set(body.rule_codes)
if body.subject_type == limit_policy.SUBJECT_DEVICE:
limit_policy.validate_device_rule_scope(
list(
{
*(row.rule_code for row in existing_whitelist_rows),
*body.rule_codes,
}
)
)
for current_rule_code in body.rule_codes:
rule = limit_policy.get_rule(current_rule_code)
mode = (
limit_policy.MODE_SUPPRESS_ALERT if rule.alert_only else limit_policy.MODE_UNLIMITED
)
row = existing_by_rule.get(current_rule_code)
if row is None:
row = repo.create(
db,
subject_type=body.subject_type,
subject_value=subject_value,
rule_code=current_rule_code,
mode=mode,
limit_value=None,
enabled=body.enabled,
starts_at=body.starts_at,
expires_at=body.expires_at,
reason=body.reason,
admin_id=admin.id,
)
else:
row.mode = mode
row.limit_value = None
row.reset_at = None
repo.update(
db,
row,
enabled=body.enabled,
starts_at=body.starts_at,
expires_at=body.expires_at,
reason=body.reason,
fields_set={
"enabled",
"starts_at",
"expires_at",
"reason",
},
)
# 同一手机号或设备在产品上是一条白名单。追加限制项时保留原规则,
# 但统一使用最后一次配置的启用状态与有效期,避免一个主体出现多套时间。
rows = [
row
for row in repo.rows_for_subject(
db,
subject_type=body.subject_type,
subject_value=subject_value,
)
if row.mode in whitelist_modes
]
for row in rows:
if row.rule_code in selected_codes:
continue
repo.update(
db,
row,
enabled=body.enabled,
starts_at=body.starts_at,
expires_at=body.expires_at,
reason=None,
fields_set={"enabled", "starts_at", "expires_at"},
)
except KeyError as exc:
raise HTTPException(status_code=404, detail="未知限制规则") from exc
except repo.DuplicateOverrideError as exc:
rule = limit_policy.get_rule(current_rule_code)
raise HTTPException(
status_code=409,
detail=f"{rule.label}”白名单配置发生并发更新,请刷新后重试",
) from exc
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
for row in rows:
_reconcile_risk_rule(db, row.rule_code)
results = [_out(db, row) for row in rows]
write_audit(
db,
admin,
action="limit.override.bulk_create",
target_type="limit_override",
target_id=",".join(str(row.id) for row in rows),
detail={
"before": before,
"after": [_audit_payload(item) for item in results],
},
ip=get_client_ip(request),
commit=False,
)
db.commit()
return [_out(db, row) for row in rows]
@router.patch(
"/{override_id}",
response_model=LimitOverrideOut,
summary="编辑白名单配置",
)
def update_override(
override_id: int,
body: LimitOverridePatch,
request: Request,
admin: CurrentAdmin,
db: AdminDb,
) -> LimitOverrideOut:
row = _row_or_404(db, override_id)
repo.lock_subject(
db,
subject_type=row.subject_type,
subject_value=row.subject_value,
)
row = _row_or_404(db, override_id)
subject_rows = (
_subject_whitelist_rows(
db,
subject_type=row.subject_type,
subject_value=row.subject_value,
)
if row.mode in _WHITELIST_MODES
else [row]
)
before = [_audit_payload(_out(db, item)) for item in subject_rows]
fields_set = set(body.model_fields_set)
try:
period_fields = {"enabled", "starts_at", "expires_at"}
if row.mode in _WHITELIST_MODES and fields_set & period_fields:
new_enabled = body.enabled if "enabled" in fields_set else row.enabled
new_starts = body.starts_at if "starts_at" in fields_set else row.starts_at
new_expires = body.expires_at if "expires_at" in fields_set else row.expires_at
for item in subject_rows:
item_fields = set(period_fields)
if item.id == row.id and "reason" in fields_set:
item_fields.add("reason")
repo.update(
db,
item,
enabled=new_enabled,
starts_at=new_starts,
expires_at=new_expires,
reason=body.reason if item.id == row.id else None,
fields_set=item_fields,
)
else:
repo.update(
db,
row,
**body.model_dump(),
fields_set=fields_set,
)
except (KeyError, ValueError) as exc:
db.rollback()
raise HTTPException(status_code=400, detail=str(exc)) from exc
for rule_code in {item.rule_code for item in subject_rows}:
_reconcile_risk_rule(db, rule_code)
after_rows = [_audit_payload(_out(db, item)) for item in subject_rows]
write_audit(
db,
admin,
action="limit.override.update",
target_type="limit_override_subject",
target_id=f"{row.subject_type}:{row.subject_value}",
detail={"before": before, "after": after_rows},
ip=get_client_ip(request),
commit=False,
)
db.commit()
return _out(db, row)
@router.post(
"/{override_id}/reset",
response_model=LimitOverrideOut,
summary="停用例外配置并恢复全局策略",
)
def restore_global_policy(
override_id: int,
request: Request,
admin: CurrentAdmin,
db: AdminDb,
) -> LimitOverrideOut:
row = _row_or_404(db, override_id)
repo.lock_subject(
db,
subject_type=row.subject_type,
subject_value=row.subject_value,
)
row = _row_or_404(db, override_id)
subject_rows = (
_subject_whitelist_rows(
db,
subject_type=row.subject_type,
subject_value=row.subject_value,
)
if row.mode in _WHITELIST_MODES
else [row]
)
before = [_audit_payload(_out(db, item)) for item in subject_rows]
for item in subject_rows:
repo.restore_global(item)
for rule_code in {item.rule_code for item in subject_rows}:
_reconcile_risk_rule(db, rule_code)
after = [_audit_payload(_out(db, item)) for item in subject_rows]
write_audit(
db,
admin,
action="limit.override.restore_global",
target_type="limit_override_subject",
target_id=f"{row.subject_type}:{row.subject_value}",
detail={"before": before, "after": after},
ip=get_client_ip(request),
commit=False,
)
db.commit()
return _out(db, row)
@router.delete(
"/{override_id}",
status_code=status.HTTP_204_NO_CONTENT,
summary="删除白名单配置",
)
def delete_override(
override_id: int,
request: Request,
admin: CurrentAdmin,
db: AdminDb,
) -> None:
row = _row_or_404(db, override_id)
repo.lock_subject(
db,
subject_type=row.subject_type,
subject_value=row.subject_value,
)
row = _row_or_404(db, override_id)
before = _audit_payload(_out(db, row))
target_id = str(row.id)
rule_code = row.rule_code
db.delete(row)
db.flush()
_reconcile_risk_rule(db, rule_code)
write_audit(
db,
admin,
action="limit.override.delete",
target_type="limit_override",
target_id=target_id,
detail={"before": before},
ip=get_client_ip(request),
commit=False,
)
db.commit()
+10 -14
View File
@@ -23,13 +23,8 @@ from app.admin.schemas.risk_monitor import (
RiskResetResponse,
RiskRuleConfig,
)
from app.core.config_schema import (
RISK_COMPARE_DAILY_THRESHOLD_KEY,
RISK_ONECLICK_DAILY_THRESHOLD_KEY,
RISK_SMS_HOURLY_THRESHOLD_KEY,
)
from app.core import limit_policy
from app.models.risk import RiskIncident, SubjectRestriction
from app.repositories import app_config
from app.repositories import risk as risk_repo
router = APIRouter(
@@ -83,15 +78,16 @@ def update_rules(
) -> RiskRuleConfig:
before = _rule_config(db).model_dump()
after = body.model_dump()
values = (
(RISK_SMS_HOURLY_THRESHOLD_KEY, body.sms_hourly_threshold),
(RISK_ONECLICK_DAILY_THRESHOLD_KEY, body.oneclick_daily_threshold),
(RISK_COMPARE_DAILY_THRESHOLD_KEY, body.compare_daily_threshold),
limit_policy.set_global_limits(
db,
{
"risk.sms.hourly": body.sms_hourly_threshold,
"risk.oneclick.daily": body.oneclick_daily_threshold,
"risk.compare.daily": body.compare_daily_threshold,
},
admin_id=admin.id,
commit=False,
)
for key, value in values:
app_config.set_value(
db, key, value, admin_id=admin.id, commit=False
)
now = risk_repo.utcnow()
risk_repo.reconcile_behavior_rule(
+8
View File
@@ -88,6 +88,11 @@ def get_user_reward_stats(
withdraw_source: Annotated[
str | None, Query(pattern="^(coin_cash|invite_cash)$")
] = None,
app_env: Annotated[str | None, Query(pattern="^(prod|test)$")] = None,
revenue_scope: Annotated[str, Query(pattern="^(business|all)$")] = "all",
feed_scene: Annotated[
str | None, Query(pattern="^(comparison|coupon|welfare)$")
] = None,
) -> UserRewardStats:
"""提现详情抽屉「用户统计区」。date_from/date_to 都不传 = 注册至今(全量)。"""
if not user_repo.user_exists(db, user_id):
@@ -99,6 +104,9 @@ def get_user_reward_stats(
date_from=date_from,
date_to=date_to,
withdraw_source=withdraw_source,
app_env=app_env,
revenue_scope=revenue_scope,
feed_scene=feed_scene,
)
)
+30 -11
View File
@@ -1,25 +1,44 @@
"""admin 新手引导视频配置 schemas(开关 / 视频地址 / 前几次 / 每次金币)"""
"""后台引导视频配置协议"""
from __future__ import annotations
from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, field_validator
from app.repositories.guide_video import MAX_PLAYS_LIMIT, REWARD_COIN_LIMIT
from app.repositories.guide_video import (
MAX_PLAYS_LIMIT,
MIN_PLAYS,
MIN_REWARD_COIN,
REWARD_COIN_LIMIT,
)
class GuideVideoConfigOut(BaseModel):
scene: str
enabled: bool
video_url: str | None = None # 相对地址 /media/guide_video/xxx.mp4;未配片 = None
video_url: str | None = None
max_plays: int
reward_coin: int
duration_ms: int | None = None
circle_count: int = 10
circle_duration_ms: float | None = None
reward_per_circle: int
video_codec: str | None = None
audio_codec: str | None = None
analysis_status: str
analysis_error: str | None = None
config_version: int
updated_at: str | None = None
# 只读统计,后台展示用:已有多少次播放、其中已发币多少次。
total_plays: int = 0
granted_plays: int = 0
class GuideVideoConfigUpdate(BaseModel):
"""部分更新:只改传入(非 None)字段。视频文件走 /video 上传接口。"""
enabled: bool | None = None
max_plays: int | None = Field(default=None, ge=0, le=MAX_PLAYS_LIMIT)
reward_coin: int | None = Field(default=None, ge=0, le=REWARD_COIN_LIMIT)
max_plays: int | None = Field(default=None, ge=MIN_PLAYS, le=MAX_PLAYS_LIMIT)
reward_coin: int | None = Field(
default=None, ge=MIN_REWARD_COIN, le=REWARD_COIN_LIMIT
)
@field_validator("reward_coin")
@classmethod
def reward_must_be_multiple_of_ten(cls, value: int | None) -> int | None:
if value is not None and value % 10:
raise ValueError("金币总价必须是 10 的倍数")
return value
+150
View File
@@ -0,0 +1,150 @@
"""Admin contracts for global limit rules and per-subject overrides."""
from __future__ import annotations
from datetime import UTC, datetime
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field, model_validator
SubjectType = Literal["phone", "device"]
PolicyMode = Literal["unlimited", "suppress_alert"]
class LimitRuleOut(BaseModel):
code: str
label: str
group: str
global_limit: int
default_limit: int
window_label: str
subject_types: list[str]
allowed_modes: list[str]
min_value: int
max_value: int
supports_reset: bool
alert_only: bool
class GlobalLimitUpdate(BaseModel):
value: int = Field(ge=0, le=1_000_000)
class DeviceCandidateOut(BaseModel):
device_id: str
source: str
source_label: str
user_id: int | None = None
username: str | None = None
phone: str | None = None
nickname: str | None = None
device_model: str | None = None
last_active_at: datetime
class LimitOverrideWrite(BaseModel):
model_config = ConfigDict(extra="forbid")
subject_type: SubjectType
subject_value: str = Field(min_length=1, max_length=128)
rule_code: str = Field(min_length=1, max_length=64)
mode: PolicyMode
enabled: bool = True
starts_at: datetime | None = None
expires_at: datetime | None = None
reason: str = Field("", max_length=256)
@model_validator(mode="after")
def validate_time_range(self):
if self.starts_at and self.expires_at and self.expires_at <= self.starts_at:
raise ValueError("失效时间必须晚于生效时间")
return self
class LimitOverrideBulkWrite(BaseModel):
model_config = ConfigDict(extra="forbid")
subject_type: SubjectType
subject_value: str = Field(min_length=1, max_length=128)
rule_codes: list[str] = Field(min_length=1, max_length=32)
enabled: bool = True
starts_at: datetime | None = None
expires_at: datetime
reason: str = Field("", max_length=256)
@model_validator(mode="after")
def validate_bulk_request(self):
self.rule_codes = list(dict.fromkeys(self.rule_codes))
starts_at = (
self.starts_at.replace(tzinfo=UTC)
if self.starts_at and self.starts_at.tzinfo is None
else self.starts_at
)
expires_at = (
self.expires_at.replace(tzinfo=UTC)
if self.expires_at.tzinfo is None
else self.expires_at
)
if starts_at and expires_at <= starts_at:
raise ValueError("失效时间必须晚于生效时间")
if expires_at <= datetime.now(UTC):
raise ValueError("失效时间必须晚于当前时间")
return self
class LimitOverridePatch(BaseModel):
model_config = ConfigDict(extra="forbid")
enabled: bool | None = None
starts_at: datetime | None = None
expires_at: datetime | None = None
reason: str | None = Field(None, max_length=256)
class LimitSubjectEnabledPatch(BaseModel):
model_config = ConfigDict(extra="forbid")
subject_type: SubjectType
subject_value: str = Field(min_length=1, max_length=128)
enabled: bool
class LimitOverrideOut(BaseModel):
id: int
subject_type: str
subject_value: str
rule_code: str
rule_label: str
rule_group: str
mode: str
limit_value: int | None
global_limit: int
effective_limit: int | None
enabled: bool
starts_at: datetime | None
expires_at: datetime | None
reset_at: datetime | None
reason: str | None
status: str
created_by_admin_id: int | None
created_at: datetime
updated_at: datetime
class LimitOverrideList(BaseModel):
items: list[LimitOverrideOut]
total: int
class LimitSubjectOut(BaseModel):
subject_type: str
subject_value: str
group_counts: dict[str, int]
total_rules: int
items: list[LimitOverrideOut]
created_at: datetime
updated_at: datetime
class LimitSubjectList(BaseModel):
items: list[LimitSubjectOut]
total: int
+2 -2
View File
@@ -21,9 +21,9 @@ class RiskMonitorSummary(BaseModel):
class RiskRuleConfig(BaseModel):
sms_hourly_threshold: int = Field(ge=1, le=5)
sms_hourly_threshold: int = Field(ge=1, le=100_000)
oneclick_daily_threshold: int = Field(ge=1, le=100_000)
compare_daily_threshold: int = Field(ge=1, le=100)
compare_daily_threshold: int = Field(ge=1, le=100_000)
class RiskIncidentItem(BaseModel):
+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 # 平均信息流广告 eCPM(分/千次)
feed_avg_ecpm: float # 全部 Draw/feed 实际展示的平均 eCPM(分/千次,含未发奖展示)
feed_cash_cents: int # 信息流广告提现(金币折现)
+13 -2
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 rewards
from app.core import limit_policy, rewards
from app.core.config import settings
from app.core.ratelimit import rate_limit
from app.integrations import pangle
@@ -413,12 +413,23 @@ def feed_reward(payload: FeedRewardIn, user: CurrentUser, db: DbSession) -> Feed
"feed ad reward user_id=%d event=%s status=%s units=%d coin=%d",
user.id, rec.client_event_id, rec.status, rec.unit_count, rec.coin,
)
feed_policy = limit_policy.resolve_for_user(db, "ad.feed.daily", user.id)
feed_limit = (
rewards.get_ad_daily_limit(db)
if feed_policy.override_id is None
and feed_policy.bucket_version == "default"
else feed_policy.limit
)
return FeedRewardOut(
granted=(rec.status == "granted"),
status=rec.status,
coin=rec.coin,
unit_count=rec.unit_count,
daily_limit=rewards.get_ad_daily_limit(db),
daily_limit=(
feed_limit
if feed_limit is not None
else limit_policy.get_rule("ad.feed.daily").max_value
),
)
+185 -26
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 test_account
from app.core import limit_policy, test_account
from app.core.ratelimit import (
RateLimitRule,
check_rate_limits,
@@ -69,6 +69,7 @@ SMS_LOGIN_MAX_PER_HOUR = 5
# 堵「换手机号绕开单号 60s 冷却」的洞 —— 冷却是单号维度,一机换号能绕开。
SMS_SEND_MAX_PER_HOUR_PER_DEVICE = 5 # 每小时上限
SMS_SEND_MAX_PER_DAY_PER_DEVICE = 20 # 每天上限(再叠一层日封顶,挡低频长时间轰炸)
UNLIMITED_VERIFY_ATTEMPTS = 2_147_483_647
def _client_ip(request: Request) -> str:
@@ -198,16 +199,53 @@ def sms_send(req: SmsSendRequest, request: Request, db: DbSession) -> SmsSendRes
# 补「换手机号绕开单号 60s 冷却」的洞(冷却是单号维度,一机换号能绕);设备维度按机器封顶,挡短信轰炸/烧钱。
# 关键:被单号 60s 冷却挡下的重发是「没真发、没烧钱」→ 不该占额度。故 check(先判)放在真发之前
# (超限直接 429、不真发),record(计数)只在 send_code 成功后调 —— 冷却/供应商失败抛 429 时直接返回、不计数。
hourly_policy = limit_policy.resolve(
db,
"sms.send.hourly",
phone=req.phone,
device=subject_id,
)
daily_policy = limit_policy.resolve(
db,
"sms.send.daily",
phone=req.phone,
device=subject_id,
)
cooldown_policy = limit_policy.resolve(
db,
"sms.phone.cooldown",
phone=req.phone,
)
hourly_limit = (
SMS_SEND_MAX_PER_HOUR_PER_DEVICE
if hourly_policy.override_id is None
and hourly_policy.bucket_version == "default"
else hourly_policy.limit
)
daily_limit = (
SMS_SEND_MAX_PER_DAY_PER_DEVICE
if daily_policy.override_id is None
and daily_policy.bucket_version == "default"
else daily_policy.limit
)
send_rules = [
RateLimitRule("sms-send-device", SMS_SEND_MAX_PER_HOUR_PER_DEVICE, 3600,
"操作过于频繁,请稍后再试"),
RateLimitRule("sms-send-device-daily", SMS_SEND_MAX_PER_DAY_PER_DEVICE, 86400,
"今日验证码发送次数过多,请明天再试"),
RateLimitRule("sms-send-device", hourly_limit, 3600,
"操作过于频繁,请稍后再试", hourly_policy.bucket_version),
RateLimitRule("sms-send-device-daily", daily_limit, 86400,
"今日验证码发送次数过多,请明天再试", daily_policy.bucket_version),
]
check_rate_limits(request, subject=req.device_id, rules=send_rules)
check_rate_limits(request, subject=subject_id, rules=send_rules)
try:
cooldown = send_code(req.phone)
cooldown = send_code(
req.phone,
cooldown_sec=(
None
if cooldown_policy.override_id is None
and cooldown_policy.bucket_version == "default"
else (0 if cooldown_policy.limit is None else cooldown_policy.limit)
),
)
except SmsError as e:
risk_repo.record_behavior_event(
db,
@@ -224,7 +262,7 @@ def sms_send(req: SmsSendRequest, request: Request, db: DbSession) -> SmsSendRes
raise HTTPException(status_code=e.status_code, detail=str(e)) from e
# 发码成功 → 两道闸各 +1(被单号冷却挡下的重发走不到这里,故不占额度)
record_rate_limits(request, subject=req.device_id, rules=send_rules)
record_rate_limits(request, subject=subject_id, rules=send_rules)
from app.core.config import settings # 局部 import 避免循环
@@ -283,17 +321,48 @@ def sms_login(req: SmsLoginRequest, request: Request, db: DbSession) -> TokenWit
# **之前** → 输错验证码的失败尝试也计数,才挡得住撞库/爆破(另有单码失败 SMS_MAX_VERIFY_ATTEMPTS 次即作废兜底)。
# ⚠️ 按设备而非手机号 → 一台机器换不同手机号刷登录也受限(防一机狂登多号);device_id 空(老客户端)时
# 退化为该 IP 下所有空设备聚一桶,仍受限。
login_policy = limit_policy.resolve(
db,
"sms.login.hourly",
phone=req.phone,
device=subject_id,
)
verify_policy = limit_policy.resolve(
db,
"sms.code.failed_attempts",
phone=req.phone,
)
login_limit = (
SMS_LOGIN_MAX_PER_HOUR
if login_policy.override_id is None
and login_policy.bucket_version == "default"
else login_policy.limit
)
enforce_rate_limit(
request,
scope="sms-login-device",
subject=req.device_id,
limit=SMS_LOGIN_MAX_PER_HOUR,
subject=subject_id,
limit=login_limit,
window_sec=3600,
detail="登录尝试过于频繁,请稍后再试",
bucket_suffix=login_policy.bucket_version,
)
try:
ok = verify_code(req.phone, req.code)
ok = verify_code(
req.phone,
req.code,
max_failed_attempts=(
None
if verify_policy.override_id is None
and verify_policy.bucket_version == "default"
else (
UNLIMITED_VERIFY_ATTEMPTS
if verify_policy.limit is None
else verify_policy.limit
)
),
)
except SmsError as e: # provider 校验降级(如阿里云接口异常)→ 原样透出其状态码(503),别误报「验证码错误」
raise HTTPException(status_code=e.status_code, detail=str(e)) from e
if not ok:
@@ -394,15 +463,25 @@ def _finish_wechat_bind(
未占用 → 新建微信账号(channel=wechat,昵称头像取微信)→ 签 token 登入。"""
existing = user_repo.get_user_by_phone(db, phone)
if existing is not None:
from app.core.config import settings # 局部 import,避免循环
ticket = create_conflict_ticket(
openid=openid,
wechat_nickname=wechat_nickname,
wechat_avatar_url=wechat_avatar_url,
phone=phone,
)
blocked = rebind_repo.rebound_within_days(db, phone, settings.PHONE_REBIND_LIMIT_DAYS)
rebind_policy = limit_policy.resolve(
db,
"phone.rebind.days",
phone=phone,
device=device_id,
)
rebind_days = rebind_policy.limit or 0
blocked = rebind_repo.rebound_within_days(
db,
phone,
rebind_days,
reset_at=rebind_policy.reset_at,
)
logger.info(
"wechat bind phone occupied phone=%s by user_id=%d has_wechat=%s",
mask_phone(phone), existing.id, bool(existing.wechat_openid),
@@ -418,7 +497,12 @@ def _finish_wechat_bind(
conflict_ticket=ticket,
rebind_available=not blocked,
rebind_blocked_days=(
rebind_repo.remaining_block_days(db, phone, settings.PHONE_REBIND_LIMIT_DAYS)
rebind_repo.remaining_block_days(
db,
phone,
rebind_days,
reset_at=rebind_policy.reset_at,
)
if blocked else 0
),
)
@@ -451,18 +535,50 @@ def wechat_bind_phone_sms(
except TokenError as e:
raise HTTPException(status_code=401, detail="授权已过期,请重新用微信登录") from e
subject_id = _device_subject(req.device_id, request)
# 防刷:同 sms/login,按 设备+IP 每小时限流(放在验证码校验之前,失败也计数)
bind_policy = limit_policy.resolve(
db,
"wechat.bind.hourly",
phone=req.phone,
device=subject_id,
)
verify_policy = limit_policy.resolve(
db,
"sms.code.failed_attempts",
phone=req.phone,
)
bind_limit = (
SMS_LOGIN_MAX_PER_HOUR
if bind_policy.override_id is None
and bind_policy.bucket_version == "default"
else bind_policy.limit
)
enforce_rate_limit(
request,
scope="wechat-bind-sms-device",
subject=req.device_id,
limit=SMS_LOGIN_MAX_PER_HOUR,
subject=subject_id,
limit=bind_limit,
window_sec=3600,
detail="登录尝试过于频繁,请稍后再试",
bucket_suffix=bind_policy.bucket_version,
)
try:
ok = verify_code(req.phone, req.code)
ok = verify_code(
req.phone,
req.code,
max_failed_attempts=(
None
if verify_policy.override_id is None
and verify_policy.bucket_version == "default"
else (
UNLIMITED_VERIFY_ATTEMPTS
if verify_policy.limit is None
else verify_policy.limit
)
),
)
except SmsError as e: # provider 校验降级(如阿里云接口异常)→ 原样透出其状态码(503),别误报「验证码错误」
raise HTTPException(status_code=e.status_code, detail=str(e)) from e
if not ok:
@@ -522,9 +638,23 @@ def wechat_conflict_continue(
except TokenError as e:
raise HTTPException(status_code=401, detail="操作超时,请重新用微信登录") from e
subject_id = _device_subject(req.device_id, request)
conflict_policy = limit_policy.resolve(
db,
"wechat.conflict.hourly",
phone=claims["phone"],
device=subject_id,
)
conflict_limit = (
SMS_LOGIN_MAX_PER_HOUR
if conflict_policy.override_id is None
and conflict_policy.bucket_version == "default"
else conflict_policy.limit
)
enforce_rate_limit(
request, scope="wechat-conflict-device", subject=req.device_id,
limit=SMS_LOGIN_MAX_PER_HOUR, window_sec=3600, detail="操作过于频繁,请稍后再试",
request, scope="wechat-conflict-device", subject=subject_id,
limit=conflict_limit, window_sec=3600, detail="操作过于频繁,请稍后再试",
bucket_suffix=conflict_policy.bucket_version,
)
user = user_repo.get_user_by_phone(db, claims["phone"])
@@ -562,21 +692,50 @@ def wechat_conflict_continue(
def wechat_conflict_rebind(
req: WechatConflictRebindRequest, request: Request, db: DbSession
) -> WechatBindResultResponse:
from app.core.config import settings # 局部 import,避免循环
try:
claims = decode_conflict_ticket(req.conflict_ticket)
except TokenError as e:
raise HTTPException(status_code=401, detail="操作超时,请重新用微信登录") from e
subject_id = _device_subject(req.device_id, request)
conflict_policy = limit_policy.resolve(
db,
"wechat.conflict.hourly",
phone=claims["phone"],
device=subject_id,
)
conflict_limit = (
SMS_LOGIN_MAX_PER_HOUR
if conflict_policy.override_id is None
and conflict_policy.bucket_version == "default"
else conflict_policy.limit
)
enforce_rate_limit(
request, scope="wechat-conflict-device", subject=req.device_id,
limit=SMS_LOGIN_MAX_PER_HOUR, window_sec=3600, detail="操作过于频繁,请稍后再试",
request, scope="wechat-conflict-device", subject=subject_id,
limit=conflict_limit, window_sec=3600, detail="操作过于频繁,请稍后再试",
bucket_suffix=conflict_policy.bucket_version,
)
phone = claims["phone"]
if rebind_repo.rebound_within_days(db, phone, settings.PHONE_REBIND_LIMIT_DAYS):
days = rebind_repo.remaining_block_days(db, phone, settings.PHONE_REBIND_LIMIT_DAYS)
rebind_policy = limit_policy.resolve(
db,
"phone.rebind.days",
phone=phone,
device=req.device_id,
)
rebind_days = rebind_policy.limit or 0
if rebind_repo.rebound_within_days(
db,
phone,
rebind_days,
reset_at=rebind_policy.reset_at,
):
days = rebind_repo.remaining_block_days(
db,
phone,
rebind_days,
reset_at=rebind_policy.reset_at,
)
raise HTTPException(status_code=409, detail=f"该手机号 {days} 天内已换绑过,暂不能再次换绑")
user = user_repo.rebind_account(
+22 -4
View File
@@ -16,6 +16,7 @@ import logging
from fastapi import APIRouter, BackgroundTasks, HTTPException, Query, status
from app.api.deps import CurrentUser, DbSession
from app.core import limit_policy
from app.repositories import comparison as crud_compare
from app.repositories import risk as risk_repo
from app.schemas.compare_record import (
@@ -53,17 +54,29 @@ def reserve_compare_start(
):
raise HTTPException(status_code=403, detail="账号存在异常,该功能暂不可用")
try:
policy = limit_policy.resolve(
db,
"compare.start.daily",
phone=user.phone,
device=payload.device_id,
)
rec, used = crud_compare.reserve_daily_start(
db,
user_id=user.id,
trace_id=payload.trace_id,
business_type=payload.business_type,
device_id=payload.device_id,
limit=policy.limit,
reset_at=policy.reset_at,
)
except crud_compare.DailyCompareStartLimitExceeded:
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail="今日已比价超过100次,请明天再试",
detail=(
f"今日已比价超过{policy.limit}次,请明天再试"
if policy.limit is not None
else "今日比价次数已达上限,请明天再试"
),
) from None
except crud_compare.ComparisonTraceOwnershipError:
raise HTTPException(
@@ -71,11 +84,16 @@ def reserve_compare_start(
detail="比价任务标识冲突,请重新发起",
) from None
# 风控阈值由后台动态配置,不能再只在固定的 100 次业务上限处同步。
risk_repo.sync_compare_incident(db, user_id=user.id, at=rec.created_at)
risk_repo.sync_compare_incident(
db,
user_id=user.id,
at=rec.created_at,
device_id=payload.device_id,
)
return CompareStartReserveOut(
limit=crud_compare.DAILY_COMPARE_START_LIMIT,
limit=policy.limit,
used=used,
remaining=max(crud_compare.DAILY_COMPARE_START_LIMIT - used, 0),
remaining=max(policy.limit - used, 0) if policy.limit is not None else None,
)
+42 -27
View File
@@ -1,22 +1,16 @@
"""新手引导视频(领券等候浮层前 N 次替代广告)。
路由前缀 `/api/v1/guide-video`(均需 Bearer):
POST /start 这次浮层放引导视频还是放广告?命中则**当场计次**并下发 play_token
POST /reward 播完 / 中途关闭都调,按 play_token 幂等发固定金币
发币额度以**服务端配置**为准(运营后台可改),客户端只报"播完/关闭",报不了金额,
所以被破解也刷不到超额金币;次数上限由 guide_video_play 行数(按账号)硬卡。
"""
"""引导视频 prepare/start/reward 客户端 API。"""
from __future__ import annotations
import logging
from fastapi import APIRouter, Depends
from fastapi import APIRouter, Depends, HTTPException
from app.api.deps import CurrentUser, DbSession
from app.core.ratelimit import rate_limit
from app.repositories import guide_video as crud_guide
from app.schemas.guide_video import (
GuideVideoPrepareIn,
GuideVideoPrepareOut,
GuideVideoRewardIn,
GuideVideoRewardOut,
GuideVideoStartIn,
@@ -24,25 +18,41 @@ from app.schemas.guide_video import (
)
logger = logging.getLogger("shagua.guide_video")
router = APIRouter(prefix="/api/v1/guide-video", tags=["guide-video"])
@router.post(
"/prepare",
response_model=GuideVideoPrepareOut,
dependencies=[Depends(rate_limit(60, 60, "guide-video-prepare"))],
)
def prepare(
payload: GuideVideoPrepareIn, user: CurrentUser, db: DbSession
) -> GuideVideoPrepareOut:
result = crud_guide.prepare_play(db, user.id, scene=payload.scene)
logger.info(
"guide video prepare user_id=%d scene=%s should_play=%s reason=%s",
user.id, payload.scene, result["should_play"], result["reason"],
)
return GuideVideoPrepareOut(**result)
@router.post(
"/start",
response_model=GuideVideoStartOut,
summary="领券浮层是否放新手引导视频(命中即计次)",
dependencies=[Depends(rate_limit(60, 60, "guide-video-start"))],
)
def start(payload: GuideVideoStartIn, user: CurrentUser, db: DbSession) -> GuideVideoStartOut:
"""开播即计数:返回 should_play=True 时服务端已写下这一次,客户端必须真的播。
没配视频 / 开关关 / 次数用完 → should_play=False,客户端照旧走广告链路(行为不变)。
"""
result = crud_guide.start_play(db, user.id, scene=payload.scene or "coupon")
def start(
payload: GuideVideoStartIn, user: CurrentUser, db: DbSession
) -> GuideVideoStartOut:
try:
result = crud_guide.start_play(db, user.id, play_token=payload.play_token)
except crud_guide.PlayStateError as exc:
status_code = 404 if exc.code == "play_not_found" else 409
raise HTTPException(status_code=status_code, detail=exc.detail()) from exc
logger.info(
"guide video start user_id=%d scene=%s should_play=%s seq=%d remaining=%d",
user.id, payload.scene, result["should_play"], result["seq"], result["remaining"],
"guide video start user_id=%d token=%s status=%s seq=%d",
user.id, payload.play_token[:12], result["status"], result["seq"],
)
return GuideVideoStartOut(**result)
@@ -50,15 +60,20 @@ def start(payload: GuideVideoStartIn, user: CurrentUser, db: DbSession) -> Guide
@router.post(
"/reward",
response_model=GuideVideoRewardOut,
summary="引导视频发金币(播完/中途关闭都发,play_token 幂等)",
dependencies=[Depends(rate_limit(60, 60, "guide-video-reward"))],
dependencies=[Depends(rate_limit(120, 60, "guide-video-reward"))],
)
def reward(payload: GuideVideoRewardIn, user: CurrentUser, db: DbSession) -> GuideVideoRewardOut:
result = crud_guide.grant_play(
db, user.id, play_token=payload.play_token, completed=payload.completed
def reward(
payload: GuideVideoRewardIn, user: CurrentUser, db: DbSession
) -> GuideVideoRewardOut:
result = crud_guide.grant_circle(
db,
user.id,
play_token=payload.play_token,
circle=payload.circle,
)
logger.info(
"guide video reward user_id=%d token=%s completed=%s granted=%s coin=%d",
user.id, payload.play_token[:12], payload.completed, result["granted"], result["coin"],
"guide video reward user_id=%d token=%s circle=%d status=%s granted=%s",
user.id, payload.play_token[:12], payload.circle,
result["status"], result["granted"],
)
return GuideVideoRewardOut(**result)
+3
View File
@@ -88,6 +88,8 @@ class Settings(BaseSettings):
# (OAuth 换 token 时 client_id 即 AppId)。发送走 v1 messages:send,成功码 80000000。
HUAWEI_PUSH_APP_ID: str = ""
HUAWEI_PUSH_APP_SECRET: str = ""
# 0=正式消息(默认,受正式消息频控);1=测试消息(仅开发联调,勿用于生产)。
HUAWEI_PUSH_TARGET_USER_TYPE: int = Field(default=0, ge=0, le=1)
HUAWEI_PUSH_TOKEN_ENDPOINT: str = "https://oauth-login.cloud.huawei.com/oauth2/v3/token"
HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE: str = (
"https://push-api.cloud.huawei.com/v1/{app_id}/messages:send"
@@ -436,6 +438,7 @@ class Settings(BaseSettings):
# 运营后台上传的新手引导视频上限。视频比图片大一个量级,单独一档;
# ⚠️ 改大时同步放宽网关 client_max_body_size(实测 QA 4MiB / prod 32MiB),否则 nginx 先挡下。
GUIDE_VIDEO_MAX_BYTES: int = 100 * 1024 * 1024 # 引导视频最大 100MB
FFPROBE_BINARY: str = "ffprobe"
# ===== 邀请好友 =====
# 分享落地页(二维码 / 分享链接指向这里;扫码 → 落地页 → 引导浏览器下载 APK)。
+133 -5
View File
@@ -14,12 +14,138 @@ from app.core import rewards as r
RISK_SMS_HOURLY_THRESHOLD_KEY = "risk_sms_hourly_threshold"
RISK_ONECLICK_DAILY_THRESHOLD_KEY = "risk_oneclick_daily_threshold"
RISK_COMPARE_DAILY_THRESHOLD_KEY = "risk_compare_daily_threshold"
COMPARE_DAILY_LIMIT_KEY = "compare_daily_limit"
SMS_SEND_HOURLY_LIMIT_KEY = "sms_send_hourly_limit"
SMS_SEND_DAILY_LIMIT_KEY = "sms_send_daily_limit"
SMS_LOGIN_HOURLY_LIMIT_KEY = "sms_login_hourly_limit"
WECHAT_BIND_SMS_HOURLY_LIMIT_KEY = "wechat_bind_sms_hourly_limit"
WECHAT_CONFLICT_HOURLY_LIMIT_KEY = "wechat_conflict_hourly_limit"
SMS_PHONE_COOLDOWN_SECONDS_KEY = "sms_phone_cooldown_seconds"
SMS_CODE_MAX_FAILED_ATTEMPTS_KEY = "sms_code_max_failed_attempts"
AD_REWARD_VIDEO_DAILY_LIMIT_KEY = "ad_reward_video_daily_limit"
AD_FEED_DAILY_LIMIT_KEY = "ad_feed_daily_limit"
PHONE_REBIND_DAYS_KEY = "phone_rebind_days"
GUIDE_VIDEO_MAX_PLAYS_KEY = "guide_video_max_plays"
LIMIT_POLICY_GLOBAL_KEY = "limit_policy_global"
# type 约定(给前端渲染编辑控件用):int / int_list / dict_str_int / bool / enum
# hidden=True:仍是合法可配项(业务照常 get_value / admin 可经专用端点读写),但**不在通用
# 「系统配置」页渲染**(admin/routers/config.py:list_config 按此过滤)。用于把已下线/已改由
# 专用页管理的项从福利页 Tab 收起,同时保留后端默认值与写入能力。
CONFIG_DEFS: dict[str, dict[str, Any]] = {
COMPARE_DAILY_LIMIT_KEY: {
"default": 100,
"label": "账号每日比价次数上限",
"group": "限制策略",
"type": "int",
"min": 1,
"max": 100_000,
"hidden": True,
"help": "北京时间自然日内每个账号可发起的比价次数。",
},
SMS_SEND_HOURLY_LIMIT_KEY: {
"default": 5,
"label": "短信每小时发送上限",
"group": "限制策略",
"type": "int",
"min": 1,
"max": 10_000,
"hidden": True,
"help": "同一设备和 IP 在固定 1 小时窗口内成功发送短信的次数。",
},
SMS_SEND_DAILY_LIMIT_KEY: {
"default": 20,
"label": "短信 24 小时发送上限",
"group": "限制策略",
"type": "int",
"min": 1,
"max": 100_000,
"hidden": True,
"help": "同一设备和 IP 在固定 24 小时窗口内成功发送短信的次数。",
},
SMS_LOGIN_HOURLY_LIMIT_KEY: {
"default": 5,
"label": "短信登录每小时尝试上限",
"group": "限制策略",
"type": "int",
"min": 1,
"max": 10_000,
"hidden": True,
"help": "成功与失败均计入。",
},
WECHAT_BIND_SMS_HOURLY_LIMIT_KEY: {
"default": 5,
"label": "微信短信绑定每小时尝试上限",
"group": "限制策略",
"type": "int",
"min": 1,
"max": 10_000,
"hidden": True,
},
WECHAT_CONFLICT_HOURLY_LIMIT_KEY: {
"default": 5,
"label": "微信冲突处理每小时尝试上限",
"group": "限制策略",
"type": "int",
"min": 1,
"max": 10_000,
"hidden": True,
"help": "继续登录与重新绑定共享此额度。",
},
SMS_PHONE_COOLDOWN_SECONDS_KEY: {
"default": 60,
"label": "同手机号短信发送冷却秒数",
"group": "限制策略",
"type": "int",
"min": 0,
"max": 86_400,
"hidden": True,
},
SMS_CODE_MAX_FAILED_ATTEMPTS_KEY: {
"default": 5,
"label": "单验证码最大失败次数",
"group": "限制策略",
"type": "int",
"min": 1,
"max": 100,
"hidden": True,
},
AD_REWARD_VIDEO_DAILY_LIMIT_KEY: {
"default": r.DAILY_AD_REWARD_LIMIT,
"label": "激励视频每日发奖次数",
"group": "限制策略",
"type": "int",
"min": 1,
"max": 100_000,
"hidden": True,
},
AD_FEED_DAILY_LIMIT_KEY: {
"default": r.DAILY_AD_REWARD_LIMIT,
"label": "Draw 信息流每日发奖次数",
"group": "限制策略",
"type": "int",
"min": 1,
"max": 100_000,
"hidden": True,
},
PHONE_REBIND_DAYS_KEY: {
"default": 30,
"label": "手机/微信换绑冷却天数",
"group": "限制策略",
"type": "int",
"min": 0,
"max": 3650,
"hidden": True,
},
GUIDE_VIDEO_MAX_PLAYS_KEY: {
"default": 3,
"label": "领券引导视频最大播放次数",
"group": "限制策略",
"type": "int",
"min": 0,
"max": 50,
"hidden": True,
},
"signin_rewards": {
"default": list(r.SIGNIN_REWARDS), "label": "签到 7 天金币档位",
"group": "签到", "type": "int_list",
@@ -56,7 +182,8 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
},
"ad_daily_limit": {
"default": r.DAILY_AD_REWARD_LIMIT, "label": "看广告每日上限(次)",
"group": "看广告", "type": "int", "help": "福利页激励视频每日可发奖次数上限,默认 500。",
"group": "看广告", "type": "int", "min": 1, "max": 100_000, "hidden": True,
"help": "历史共享上限;新配置由白名单页分别管理激励视频与 Draw 信息流。",
},
"ad_max_coin": {
"default": r.MAX_AD_REWARD_COIN, "label": "看广告单次金币上限",
@@ -68,7 +195,8 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
},
"ad_cooldown_sec": {
"default": r.VIDEO_ROUND_COOLDOWN_SECONDS, "label": "广告关闭后冷却(秒)",
"group": "看广告", "type": "int", "help": "点击退出广告后,下次点击观看前的冷却时间,默认 3 秒。",
"group": "看广告", "type": "int", "min": 0, "max": 86_400,
"help": "点击退出广告后,下次点击观看前的冷却时间,默认 3 秒。",
},
"comparing_ad_enabled": {
"default": True, "label": "比价/领券期信息流广告",
@@ -118,9 +246,9 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
"group": "风控",
"type": "int",
"min": 1,
"max": 5,
"max": 100_000,
"hidden": True,
"help": "同一设备在北京时间同一自然小时内成功下发短信达到该次数时告警;不得高于现有每小时 5 次的发送上限",
"help": "同一设备在北京时间同一自然小时内成功下发短信达到该次数时告警。",
},
RISK_ONECLICK_DAILY_THRESHOLD_KEY: {
"default": 20,
@@ -138,7 +266,7 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
"group": "风控",
"type": "int",
"min": 1,
"max": 100,
"max": 100_000,
"hidden": True,
"help": "同一账户在北京时间同一自然日内发起比价达到该次数时告警。",
},
+629
View File
@@ -0,0 +1,629 @@
"""Unified global limits and per-phone/device policy overrides.
The registry is the single source of truth for the whitelist page. Existing
constants remain as backwards-compatible defaults, while business call sites
resolve an effective value here.
"""
from __future__ import annotations
from collections.abc import Iterable
from dataclasses import dataclass
from datetime import UTC, datetime
from sqlalchemy import delete, or_, select
from sqlalchemy.orm import Session
from app.core.config_schema import (
AD_FEED_DAILY_LIMIT_KEY,
AD_REWARD_VIDEO_DAILY_LIMIT_KEY,
COMPARE_DAILY_LIMIT_KEY,
GUIDE_VIDEO_MAX_PLAYS_KEY,
LIMIT_POLICY_GLOBAL_KEY,
PHONE_REBIND_DAYS_KEY,
RISK_COMPARE_DAILY_THRESHOLD_KEY,
RISK_ONECLICK_DAILY_THRESHOLD_KEY,
RISK_SMS_HOURLY_THRESHOLD_KEY,
SMS_CODE_MAX_FAILED_ATTEMPTS_KEY,
SMS_LOGIN_HOURLY_LIMIT_KEY,
SMS_PHONE_COOLDOWN_SECONDS_KEY,
SMS_SEND_DAILY_LIMIT_KEY,
SMS_SEND_HOURLY_LIMIT_KEY,
WECHAT_BIND_SMS_HOURLY_LIMIT_KEY,
WECHAT_CONFLICT_HOURLY_LIMIT_KEY,
)
from app.models.app_config import AppConfig
from app.models.limit_policy import LimitPolicyOverride
from app.models.user import User
from app.repositories import app_config
MODE_INHERIT = "inherit"
MODE_OVERRIDE = "override"
MODE_UNLIMITED = "unlimited"
MODE_SUPPRESS_ALERT = "suppress_alert"
SUBJECT_PHONE = "phone"
SUBJECT_DEVICE = "device"
SUBJECT_TYPES = (SUBJECT_PHONE, SUBJECT_DEVICE)
SUBJECT_PRECEDENCE = {SUBJECT_DEVICE: 0, SUBJECT_PHONE: 1}
LEGACY_IP_DEVICE_PREFIX = "legacy-ip:"
@dataclass(frozen=True)
class RuleDefinition:
code: str
label: str
group: str
config_key: str
default_limit: int
window_label: str
subject_types: tuple[str, ...]
min_value: int = 1
max_value: int = 100_000
allow_unlimited: bool = True
supports_reset: bool = True
alert_only: bool = False
legacy_config_key: str | None = None
legacy_json_field: str | None = None
@property
def allowed_modes(self) -> tuple[str, ...]:
if self.alert_only:
return (MODE_SUPPRESS_ALERT,)
return (MODE_UNLIMITED,) if self.allow_unlimited else ()
RULES: tuple[RuleDefinition, ...] = (
RuleDefinition(
"compare.start.daily",
"每日发起比价次数",
"比价",
COMPARE_DAILY_LIMIT_KEY,
100,
"北京时间自然日",
SUBJECT_TYPES,
),
RuleDefinition(
"sms.send.hourly",
"短信每小时成功发送次数",
"短信与登录",
SMS_SEND_HOURLY_LIMIT_KEY,
5,
"固定 1 小时窗口",
SUBJECT_TYPES,
max_value=10_000,
),
RuleDefinition(
"sms.send.daily",
"短信 24 小时成功发送次数",
"短信与登录",
SMS_SEND_DAILY_LIMIT_KEY,
20,
"固定 24 小时窗口",
SUBJECT_TYPES,
),
RuleDefinition(
"sms.phone.cooldown",
"同手机号短信发送冷却",
"短信与登录",
SMS_PHONE_COOLDOWN_SECONDS_KEY,
60,
"",
(SUBJECT_PHONE,),
min_value=0,
max_value=86_400,
),
RuleDefinition(
"sms.code.failed_attempts",
"单验证码最大失败次数",
"短信与登录",
SMS_CODE_MAX_FAILED_ATTEMPTS_KEY,
5,
"单个验证码",
(SUBJECT_PHONE,),
max_value=100,
),
RuleDefinition(
"sms.login.hourly",
"短信登录每小时尝试次数",
"短信与登录",
SMS_LOGIN_HOURLY_LIMIT_KEY,
5,
"固定 1 小时窗口",
SUBJECT_TYPES,
max_value=10_000,
),
RuleDefinition(
"wechat.bind.hourly",
"微信短信绑定每小时尝试次数",
"短信与登录",
WECHAT_BIND_SMS_HOURLY_LIMIT_KEY,
5,
"固定 1 小时窗口",
SUBJECT_TYPES,
max_value=10_000,
),
RuleDefinition(
"wechat.conflict.hourly",
"微信冲突处理每小时尝试次数",
"短信与登录",
WECHAT_CONFLICT_HOURLY_LIMIT_KEY,
5,
"固定 1 小时窗口",
SUBJECT_TYPES,
max_value=10_000,
),
RuleDefinition(
"ad.reward_video.daily",
"激励视频每日发奖次数",
"广告",
AD_REWARD_VIDEO_DAILY_LIMIT_KEY,
500,
"北京时间自然日",
(SUBJECT_PHONE,),
legacy_config_key="ad_daily_limit",
),
RuleDefinition(
"ad.feed.daily",
"Draw 信息流每日发奖次数",
"广告",
AD_FEED_DAILY_LIMIT_KEY,
500,
"北京时间自然日",
(SUBJECT_PHONE,),
legacy_config_key="ad_daily_limit",
),
RuleDefinition(
"ad.reward_video.cooldown",
"激励视频发奖后冷却秒数",
"广告",
"ad_cooldown_sec",
3,
"",
(SUBJECT_PHONE,),
min_value=0,
max_value=86_400,
),
RuleDefinition(
"guide.video.lifetime",
"领券引导视频最大播放次数",
"引导与账号",
GUIDE_VIDEO_MAX_PLAYS_KEY,
3,
"账号生命周期",
(SUBJECT_PHONE,),
min_value=0,
max_value=50,
legacy_config_key="coupon_guide_video",
legacy_json_field="max_plays",
),
RuleDefinition(
"phone.rebind.days",
"手机/微信换绑冷却天数",
"引导与账号",
PHONE_REBIND_DAYS_KEY,
30,
"自然日",
(SUBJECT_PHONE,),
min_value=0,
max_value=3650,
),
RuleDefinition(
"risk.sms.hourly",
"短信设备每小时告警",
"风控免告警",
RISK_SMS_HOURLY_THRESHOLD_KEY,
5,
"北京时间自然小时",
(SUBJECT_DEVICE,),
max_value=100_000,
allow_unlimited=False,
alert_only=True,
),
RuleDefinition(
"risk.oneclick.daily",
"一键登录设备每日告警",
"风控免告警",
RISK_ONECLICK_DAILY_THRESHOLD_KEY,
20,
"北京时间自然日",
(SUBJECT_DEVICE,),
max_value=100_000,
allow_unlimited=False,
alert_only=True,
),
RuleDefinition(
"risk.compare.daily",
"比价账号每日告警",
"风控免告警",
RISK_COMPARE_DAILY_THRESHOLD_KEY,
100,
"北京时间自然日",
(SUBJECT_PHONE,),
max_value=100_000,
allow_unlimited=False,
alert_only=True,
),
)
RULE_MAP = {rule.code: rule for rule in RULES}
RULE_CODE_BY_CONFIG_KEY = {rule.config_key: rule.code for rule in RULES}
LIMIT_CONFIG_KEYS = tuple(RULE_CODE_BY_CONFIG_KEY)
@dataclass(frozen=True)
class EffectiveLimit:
rule_code: str
global_limit: int
limit: int | None
mode: str
suppressed: bool
override_id: int | None
matched_subject_type: str | None
matched_subject_value: str | None
reset_at: datetime | None
bucket_version: str
@property
def unlimited(self) -> bool:
return self.limit is None
def normalize_subject(subject_type: str, value: str) -> str:
value = (value or "").strip()
if subject_type == SUBJECT_PHONE:
value = "".join(ch for ch in value if ch.isdigit())
if subject_type not in SUBJECT_TYPES:
raise ValueError(f"unsupported subject type: {subject_type}")
if not value:
raise ValueError("subject value is empty")
return value[:128]
def validate_whitelist_subject(subject_type: str, value: str) -> str:
"""Normalize a whitelist subject and reject unsafe pseudo-devices.
Old clients without a device ID are grouped by public IP for rate limiting.
That fallback remains valid at runtime, but it is not a stable, unique
device identity and must never be persisted as a device whitelist target.
"""
normalized = normalize_subject(subject_type, value)
if (
subject_type == SUBJECT_DEVICE
and normalized.startswith(LEGACY_IP_DEVICE_PREFIX)
):
raise ValueError(
"旧客户端未上报真实设备 ID,不能加入设备白名单,请升级客户端后重试"
)
return normalized
def get_rule(rule_code: str) -> RuleDefinition:
try:
return RULE_MAP[rule_code]
except KeyError as exc:
raise ValueError(f"unknown rule: {rule_code}") from exc
def device_source_scope(rule_code: str) -> str:
"""返回设备候选数据所属命名空间,防止一个设备 ID 跨来源误套规则。"""
rule = get_rule(rule_code)
if SUBJECT_DEVICE not in rule.subject_types:
raise ValueError("当前限制项不支持设备白名单")
return "comparison" if rule_code == "compare.start.daily" else "auth"
def validate_device_rule_scope(rule_codes: Iterable[str]) -> None:
scopes = {device_source_scope(rule_code) for rule_code in rule_codes}
if len(scopes) > 1:
raise ValueError("比价设备限制与短信/登录设备限制不能在同一白名单中混选")
def default_global_limits() -> dict[str, int]:
"""Return the complete 16-rule default snapshot keyed by rule code."""
return {rule.code: rule.default_limit for rule in RULES}
def _normalise_global_limits(value: object) -> dict[str, int]:
"""Merge a stored JSON object with safe code defaults.
The migration and every admin write persist all rules. Defaults are still
merged here so a manually damaged/older partial JSON cannot take the
service down after deployment.
"""
values = default_global_limits()
if not isinstance(value, dict):
return values
for rule_code, raw in value.items():
rule = RULE_MAP.get(str(rule_code))
if rule is None or isinstance(raw, bool):
continue
try:
parsed = int(raw)
except (TypeError, ValueError):
continue
if rule.min_value <= parsed <= rule.max_value:
values[rule.code] = parsed
return values
def _legacy_global_limit(db: Session, rule: RuleDefinition) -> tuple[int, str]:
"""Read the pre-bundle representation while upgrading old/test databases."""
row = db.get(AppConfig, rule.config_key)
if row is not None:
return int(row.value), "legacy-key"
if rule.legacy_config_key:
legacy = db.get(AppConfig, rule.legacy_config_key)
if legacy is not None:
value = legacy.value
if rule.legacy_json_field:
value = value.get(rule.legacy_json_field) if isinstance(value, dict) else None
if value is not None:
return int(value), "legacy"
try:
return int(app_config.get_value(db, rule.config_key)), "default"
except KeyError:
return rule.default_limit, "default"
def get_global_limits(db: Session) -> dict[str, int]:
"""Read the complete global-limit JSON, with a pre-migration fallback."""
row = db.get(AppConfig, LIMIT_POLICY_GLOBAL_KEY)
if row is not None:
return _normalise_global_limits(row.value)
return {rule.code: _legacy_global_limit(db, rule)[0] for rule in RULES}
def set_global_limits(
db: Session,
updates: dict[str, int],
*,
admin_id: int,
commit: bool = True,
) -> dict[str, int]:
"""Atomically update selected rules inside the single complete JSON row."""
parsed_updates: dict[str, int] = {}
for rule_code, raw_value in updates.items():
rule = get_rule(rule_code)
value = int(raw_value)
if not rule.min_value <= value <= rule.max_value:
raise ValueError(
f"limit for {rule_code} must be between "
f"{rule.min_value} and {rule.max_value}"
)
parsed_updates[rule.code] = value
row = db.scalar(
select(AppConfig)
.where(AppConfig.key == LIMIT_POLICY_GLOBAL_KEY)
.with_for_update()
)
values = (
_normalise_global_limits(row.value)
if row is not None
else {rule.code: _legacy_global_limit(db, rule)[0] for rule in RULES}
)
values.update(parsed_updates)
if row is None:
row = AppConfig(
key=LIMIT_POLICY_GLOBAL_KEY,
value=values,
updated_by_admin_id=admin_id,
)
db.add(row)
else:
row.value = dict(values)
row.updated_by_admin_id = admin_id
# Once the bundle exists, stale sparse rows must not become a second source
# of truth. The data migration performs the same cleanup for production.
db.execute(delete(AppConfig).where(AppConfig.key.in_(LIMIT_CONFIG_KEYS)))
if commit:
db.commit()
db.refresh(row)
else:
db.flush()
return values
def _global_limit(db: Session, rule: RuleDefinition) -> tuple[int, str]:
row = db.get(AppConfig, LIMIT_POLICY_GLOBAL_KEY)
if row is not None:
return _normalise_global_limits(row.value)[rule.code], "configured"
return _legacy_global_limit(db, rule)
def _aware(value: datetime | None) -> datetime | None:
if value is None:
return None
return value.replace(tzinfo=UTC) if value.tzinfo is None else value
def _matching_overrides(
db: Session,
rule: RuleDefinition,
subjects: dict[str, str | None],
now: datetime,
) -> list[LimitPolicyOverride]:
pairs: list[tuple[str, str]] = []
for subject_type in rule.subject_types:
raw = subjects.get(subject_type)
if raw:
pairs.append((subject_type, normalize_subject(subject_type, raw)))
if not pairs:
return []
clauses = [
(
(LimitPolicyOverride.subject_type == subject_type)
& (LimitPolicyOverride.subject_value == subject_value)
)
for subject_type, subject_value in pairs
]
rows = list(
db.execute(
select(LimitPolicyOverride).where(
LimitPolicyOverride.rule_code == rule.code,
LimitPolicyOverride.enabled.is_(True),
or_(*clauses),
)
).scalars()
)
active = [
row
for row in rows
if (_aware(row.starts_at) is None or _aware(row.starts_at) <= now)
and (_aware(row.expires_at) is None or _aware(row.expires_at) > now)
]
return sorted(active, key=lambda row: SUBJECT_PRECEDENCE[row.subject_type])
def resolve(
db: Session,
rule_code: str,
*,
phone: str | None = None,
device: str | None = None,
now: datetime | None = None,
) -> EffectiveLimit:
"""Resolve global config plus the most specific active override.
Device overrides win over phone overrides when both match.
"""
rule = get_rule(rule_code)
now = _aware(now) or datetime.now(UTC)
global_limit, global_version = _global_limit(db, rule)
matches = _matching_overrides(
db, rule, {SUBJECT_PHONE: phone, SUBJECT_DEVICE: device}, now
)
row = matches[0] if matches else None
if row is None:
return EffectiveLimit(
rule.code,
global_limit,
global_limit,
MODE_INHERIT,
False,
None,
None,
None,
None,
global_version,
)
limit: int | None = global_limit
suppressed = False
if row.mode == MODE_OVERRIDE:
limit = int(row.limit_value) if row.limit_value is not None else global_limit
elif row.mode == MODE_UNLIMITED:
limit = None
elif row.mode == MODE_SUPPRESS_ALERT:
suppressed = True
# 编辑限制值/备注不应隐式清空计数;只有显式“重置状态”才切换桶。
reset_version = _aware(row.reset_at)
version = (
f"{global_version}:o:{row.id}:"
f"{reset_version.isoformat() if reset_version else '0'}"
)
return EffectiveLimit(
rule.code,
global_limit,
limit,
row.mode,
suppressed,
row.id,
row.subject_type,
row.subject_value,
_aware(row.reset_at),
version,
)
def resolve_for_user(
db: Session,
rule_code: str,
user_id: int,
*,
device: str | None = None,
) -> EffectiveLimit:
user = db.get(User, user_id)
return resolve(
db,
rule_code,
phone=user.phone if user is not None else None,
device=device,
)
def rule_catalog(db: Session) -> list[dict]:
out: list[dict] = []
values = get_global_limits(db)
for rule in RULES:
out.append(
{
"code": rule.code,
"label": rule.label,
"group": rule.group,
"global_limit": values[rule.code],
"default_limit": rule.default_limit,
"window_label": rule.window_label,
"subject_types": list(rule.subject_types),
"allowed_modes": list(rule.allowed_modes),
"min_value": rule.min_value,
"max_value": rule.max_value,
"supports_reset": rule.supports_reset,
"alert_only": rule.alert_only,
}
)
return out
def validate_override(
rule: RuleDefinition,
*,
subject_type: str,
mode: str,
limit_value: int | None,
starts_at: datetime | None,
expires_at: datetime | None,
) -> None:
if subject_type not in rule.subject_types:
raise ValueError("该规则不支持此主体类型")
if mode not in rule.allowed_modes:
raise ValueError("该规则不支持此策略模式")
if limit_value is not None:
raise ValueError("白名单不支持覆盖指定值")
if mode in {MODE_UNLIMITED, MODE_SUPPRESS_ALERT} and expires_at is None:
raise ValueError("临时白名单必须设置失效时间")
if starts_at and expires_at and _aware(expires_at) <= _aware(starts_at):
raise ValueError("失效时间必须晚于生效时间")
if (
mode in {MODE_UNLIMITED, MODE_SUPPRESS_ALERT}
and expires_at is not None
and _aware(expires_at) <= datetime.now(UTC)
):
raise ValueError("临时白名单的失效时间必须晚于当前时间")
if mode == MODE_SUPPRESS_ALERT and not rule.alert_only:
raise ValueError("免告警只支持风控监控的三项规则")
def active_overrides(
db: Session,
*,
subject_type: str | None = None,
keyword: str | None = None,
) -> Iterable[LimitPolicyOverride]:
stmt = select(LimitPolicyOverride)
if subject_type:
stmt = stmt.where(LimitPolicyOverride.subject_type == subject_type)
if keyword:
stmt = stmt.where(LimitPolicyOverride.subject_value.ilike(f"%{keyword.strip()}%"))
return db.execute(
stmt.order_by(LimitPolicyOverride.updated_at.desc(), LimitPolicyOverride.id.desc())
).scalars()
+76 -9
View File
@@ -10,8 +10,13 @@
"""
from __future__ import annotations
import json
import os
import secrets
import subprocess
from decimal import ROUND_HALF_UP, Decimal, InvalidOperation
from pathlib import Path
from typing import Any
from app.core.config import settings
@@ -81,29 +86,91 @@ def _sniff_video_ext(data: bytes) -> str | None:
"""按魔数判定视频类型,返回扩展名;非支持类型返回 None。
只认 MP4 家族(ISO BMFF):`....ftyp` 在偏移 4。Android ExoPlayer 与浏览器 <video>
都稳吃 H.264/AAC 的 mp4;放开 mkv/avi 只会让端上放不出来,不如在入口就挡掉。
支持 H.264 或 HEVC/H.265 视频及 AAC 音频;放开 mkv/avi 只会让端上放不出来,
不如在入口就挡掉。
"""
if len(data) >= 12 and data[4:8] == b"ftyp":
return ".mp4"
return None
def save_guide_video(data: bytes) -> str:
"""保存新手引导视频(运营后台上传的运营素材),返回相对 URL(`/media/guide_video/<file>`)。
def _probe_guide_video(path: Path) -> dict[str, Any]:
"""用 ffprobe 读取服务端可信的时长/编码信息。"""
try:
proc = subprocess.run(
[
settings.FFPROBE_BINARY,
"-v", "error",
"-show_format",
"-show_streams",
"-of", "json",
str(path),
],
capture_output=True,
text=True,
timeout=30,
check=False,
)
except FileNotFoundError as exc:
raise MediaError("服务器未安装 ffprobe,暂时无法分析视频") from exc
except (OSError, subprocess.TimeoutExpired) as exc:
raise MediaError("ffprobe 分析视频失败") from exc
if proc.returncode != 0:
raise MediaError("无法分析视频,请确认文件是有效的 MP4")
try:
payload = json.loads(proc.stdout)
streams = payload.get("streams") or []
video = next(s for s in streams if s.get("codec_type") == "video")
audio = next((s for s in streams if s.get("codec_type") == "audio"), None)
raw_duration = (
(payload.get("format") or {}).get("duration")
or video.get("duration")
)
duration_ms = int(
(Decimal(str(raw_duration)) * 1000).quantize(Decimal("1"), rounding=ROUND_HALF_UP)
)
except (StopIteration, TypeError, ValueError, InvalidOperation) as exc:
raise MediaError("视频缺少可识别的视频轨或时长") from exc
video_codec = str(video.get("codec_name") or "").lower()
audio_codec = str(audio.get("codec_name") or "").lower() if audio else None
if not 30_000 <= duration_ms <= 180_000:
raise MediaError("视频时长必须在 30180 秒之间")
if video_codec not in ("h264", "hevc", "h265"):
raise MediaError("视频编码必须为 H.264 或 HEVC/H.265")
if audio_codec not in (None, "aac"):
raise MediaError("音频编码必须为 AAC")
return {
"duration_ms": duration_ms,
"video_codec": video_codec,
"audio_codec": audio_codec,
"analysis_status": "valid",
"analysis_error": None,
}
与图片分开一套校验:体积上限走 [settings.GUIDE_VIDEO_MAX_BYTES],类型只认 MP4。
"""
def save_guide_video(data: bytes) -> tuple[str, dict[str, Any]]:
"""临时落盘、ffprobe 校验后原子发布,返回 URL 与分析结果。"""
if not data:
raise MediaError("空文件")
limit = settings.GUIDE_VIDEO_MAX_BYTES
if len(data) > limit:
raise MediaError(f"视频过大(上限 {limit // (1024 * 1024)}MB)")
if _sniff_video_ext(data) is None:
raise MediaError("仅支持 MP4 视频(H.264 编码)")
raise MediaError("仅支持 MP4 视频(H.264 或 HEVC/H.265 编码)")
fname = f"guide_{secrets.token_hex(8)}.mp4"
(_media_dir("guide_video") / fname).write_bytes(data)
return f"{settings.MEDIA_URL_PREFIX}/guide_video/{fname}"
directory = _media_dir("guide_video")
stem = f"guide_{secrets.token_hex(8)}"
temp_path = directory / f".{stem}.upload"
final_path = directory / f"{stem}.mp4"
try:
temp_path.write_bytes(data)
analysis = _probe_guide_video(temp_path)
os.replace(temp_path, final_path)
except Exception:
temp_path.unlink(missing_ok=True)
final_path.unlink(missing_ok=True)
raise
return f"{settings.MEDIA_URL_PREFIX}/guide_video/{final_path.name}", analysis
def save_cps_image(admin_id: int, data: bytes) -> str:
+19 -6
View File
@@ -75,10 +75,11 @@ def enforce_rate_limit(
request: Request,
scope: str,
subject: str,
limit: int,
limit: int | None,
window_sec: float,
*,
detail: str = "操作过于频繁,请稍后再试",
bucket_suffix: str = "",
) -> None:
"""在路由内部手动限流,按 (subject, 客户端 IP) 计数。
@@ -87,9 +88,9 @@ def enforce_rate_limit(
key = `scope:subject:client_ip`;同一 (subject, IP) 在 window_sec 内超过 limit 次 → 抛 429。
受 [settings.RATE_LIMIT_ENABLED] 总开关控制(与 [rate_limit] 一致)。
"""
if not settings.RATE_LIMIT_ENABLED:
if not settings.RATE_LIMIT_ENABLED or limit is None:
return
key = f"{scope}:{subject}:{_client_ip(request)}"
key = f"{scope}:{subject}:{_client_ip(request)}:{bucket_suffix}"
if not _hit(key, limit, window_sec):
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
@@ -112,9 +113,10 @@ class RateLimitRule(NamedTuple):
"""
scope: str
limit: int
limit: int | None
window_sec: float
detail: str = "操作过于频繁,请稍后再试"
bucket_suffix: str = ""
def _peek(key: str, limit: int, window_sec: float) -> bool:
@@ -150,7 +152,13 @@ def check_rate_limits(request: Request, subject: str, rules: list[RateLimitRule]
return
ip = _client_ip(request)
for rule in rules:
if not _peek(f"{rule.scope}:{subject}:{ip}", rule.limit, rule.window_sec):
if rule.limit is None:
continue
if not _peek(
f"{rule.scope}:{subject}:{ip}:{rule.bucket_suffix}",
rule.limit,
rule.window_sec,
):
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail=rule.detail,
@@ -167,4 +175,9 @@ def record_rate_limits(request: Request, subject: str, rules: list[RateLimitRule
return
ip = _client_ip(request)
for rule in rules:
_commit(f"{rule.scope}:{subject}:{ip}", rule.window_sec)
if rule.limit is None:
continue
_commit(
f"{rule.scope}:{subject}:{ip}:{rule.bucket_suffix}",
rule.window_sec,
)
+15 -4
View File
@@ -27,11 +27,22 @@ def _provider():
return _PROVIDERS.get(settings.SMS_PROVIDER, jiguang)
def send_code(phone: str) -> int:
def send_code(phone: str, *, cooldown_sec: int | None = None) -> int:
"""发送验证码,返回距下次可发的冷却秒数;失败抛 SmsError。委托给当前 provider。"""
return _provider().send_code(phone)
provider = _provider()
if cooldown_sec is None:
return provider.send_code(phone)
return provider.send_code(phone, cooldown_sec=cooldown_sec)
def verify_code(phone: str, code: str) -> bool:
def verify_code(
phone: str,
code: str,
*,
max_failed_attempts: int | None = None,
) -> bool:
"""校验验证码,返回是否通过;provider 异常降级抛 SmsError。委托给当前 provider。"""
return _provider().verify_code(phone, code)
provider = _provider()
if max_failed_attempts is None:
return provider.verify_code(phone, code)
return provider.verify_code(phone, code, max_failed_attempts=max_failed_attempts)
+30 -9
View File
@@ -35,7 +35,7 @@ _GC_THRESHOLD = 10000 # 超此阈值,send 时顺手清老于
# 发码错误码 → (HTTP 码, 用户提示)。未列出的一律 503(供应商不可用)。
_SEND_ERRORS: dict[str, tuple[int, str]] = {
"MOBILE_NUMBER_ILLEGAL": (400, "手机号无效"),
"MOBILE_NUMBER_ILLEGAL": (400, "请输入有效的手机号"),
"BUSINESS_LIMIT_CONTROL": (429, "今日发送次数过多,请明天再试"),
"FREQUENCY_FAIL": (429, "发送过于频繁,请稍后再试"),
}
@@ -47,19 +47,26 @@ _client = None # 惰性构建的 SDK client(模块级缓存)
# ============================ 对外:发码 / 校验 ============================
def send_code(phone: str) -> int:
def send_code(phone: str, *, cooldown_sec: int | None = None) -> int:
"""发送验证码(阿里云生成+下发)。
Returns: 距下次可发的秒数(= ALIYUN_SMS_INTERVAL_SEC,冷却由阿里云 Interval 侧执行)。
Raises: SmsError(手机号无效 400 / 过频·天级流控 429 / 未配置·未开通·其他 503)。
"""
effective_cooldown = (
settings.ALIYUN_SMS_INTERVAL_SEC if cooldown_sec is None else cooldown_sec
)
if settings.SMS_MOCK:
logger.info("[SMS-aliyun-MOCK] to %s**** (不真发)", phone[:3])
return settings.ALIYUN_SMS_INTERVAL_SEC
return effective_cooldown
if not settings.aliyun_sms_configured:
raise SmsError("短信服务未配置(缺阿里云凭证)", status_code=503)
result = _call_send(phone) # 传输/SDK 异常在内部抛 SmsError(503)
result = (
_call_send(phone)
if cooldown_sec is None
else _call_send(phone, cooldown_sec=effective_cooldown)
)
if result["success"] and result["code"] == "OK":
now = time.time()
@@ -68,7 +75,7 @@ def send_code(phone: str) -> int:
_verify_attempts.pop(phone, None) # 新码 = 新失败预算
_verify_seen.pop(phone, None)
logger.info("[SMS-aliyun] sent to %s****", phone[:3])
return settings.ALIYUN_SMS_INTERVAL_SEC
return effective_cooldown
code = result["code"]
logger.error("[SMS-aliyun] send failed code=%s msg=%s", code, result["message"])
@@ -78,7 +85,12 @@ def send_code(phone: str) -> int:
raise SmsError(msg, status_code=status)
def verify_code(phone: str, code: str) -> bool:
def verify_code(
phone: str,
code: str,
*,
max_failed_attempts: int | None = None,
) -> bool:
"""校验验证码(阿里云裁决)。
- **mock**:放行任意 N 位数字(provider 无关,同极光)。
@@ -92,8 +104,13 @@ def verify_code(phone: str, code: str) -> bool:
# 失败计数是 best-effort:网络调用不持锁(不能锁跨 IO),故并发下同号可能多放行个位数次。
# 无碍——API 层登录频控(设备+IP 5/时)是硬上限,阿里云码有效期 + DuplicatePolicy 亦兜底。
effective_max_attempts = (
settings.SMS_MAX_VERIFY_ATTEMPTS
if max_failed_attempts is None
else max_failed_attempts
)
with _lock:
if _verify_attempts.get(phone, 0) >= settings.SMS_MAX_VERIFY_ATTEMPTS:
if _verify_attempts.get(phone, 0) >= effective_max_attempts:
return False # 已作废:保持计数(直到 send_code 重置),与极光「达上限即作废」一致
result = _call_check(phone, code) # 传输/SDK 异常在内部抛 SmsError(503)
@@ -146,7 +163,7 @@ def _get_client():
return _client
def _call_send(phone: str) -> dict:
def _call_send(phone: str, *, cooldown_sec: int | None = None) -> dict:
"""调 SendSmsVerifyCode。返回归一化 {success, code, message};import/建 client/调用 任一失败抛 SmsError(503)。"""
valid_min = max(1, settings.ALIYUN_SMS_VALID_TIME_SEC // 60)
template_param = json.dumps({"code": "##code##", "min": str(valid_min)}, ensure_ascii=False)
@@ -160,7 +177,11 @@ def _call_send(phone: str) -> dict:
template_param=template_param,
code_length=settings.ALIYUN_SMS_CODE_LENGTH,
valid_time=settings.ALIYUN_SMS_VALID_TIME_SEC,
interval=settings.ALIYUN_SMS_INTERVAL_SEC,
interval=(
settings.ALIYUN_SMS_INTERVAL_SEC
if cooldown_sec is None
else cooldown_sec
),
scheme_name=settings.ALIYUN_SMS_SCHEME_NAME or None,
)
body = _get_client().send_sms_verify_code(req).body
+20 -7
View File
@@ -53,7 +53,7 @@ _GC_THRESHOLD = 10000 # 任一内存 dict 超此阈值,send 时顺手清过期
# 发码错误码(创蓝 `code`)→ (HTTP 码, 用户提示)。未列出的一律 503(供应商不可用)。
_SEND_ERRORS: dict[str, tuple[int, str]] = {
"103": (429, "发送过于频繁,请稍后再试"), # 提交速度过快
"107": (400, "手机号无效"), # 手机号码错误
"107": (400, "请输入有效的手机号"), # 手机号码错误
}
# 需运维介入的配置/开通/余额类错误:打 critical 日志(仍归 503)。
_SEND_CRITICAL_CODES = frozenset({
@@ -84,20 +84,23 @@ def _gc(now: float) -> None:
_last_sent.pop(p, None)
def send_code(phone: str) -> int:
def send_code(phone: str, *, cooldown_sec: int | None = None) -> int:
"""发送验证码。
Returns: 距下次可发的秒数(= SMS_SEND_INTERVAL_SEC)
Raises: SmsError(过频 429 / 手机号无效 400 / 供应商失败 503)
"""
now = time.time()
effective_cooldown = (
settings.SMS_SEND_INTERVAL_SEC if cooldown_sec is None else cooldown_sec
)
# --- lock 内:防刷检查 + 预占(防并发重复发烧钱)---
with _lock:
_gc(now) # 顺手清过期内存(超阈值才扫)
elapsed = now - _last_sent.get(phone, 0.0)
if elapsed < settings.SMS_SEND_INTERVAL_SEC:
remain = int(settings.SMS_SEND_INTERVAL_SEC - elapsed)
if elapsed < effective_cooldown:
remain = int(effective_cooldown - elapsed)
raise SmsError(f"发送过于频繁,请 {remain}s 后再试")
code = _gen_code()
@@ -122,10 +125,15 @@ def send_code(phone: str) -> int:
logger.exception("[SMS-chuanglan] send failed phone=%s****", phone[:3])
raise SmsError("验证码发送失败,请稍后重试", status_code=503) from e
return settings.SMS_SEND_INTERVAL_SEC
return effective_cooldown
def verify_code(phone: str, code: str) -> bool:
def verify_code(
phone: str,
code: str,
*,
max_failed_attempts: int | None = None,
) -> bool:
"""校验验证码。
- **mock 模式**:放行任意 N 位数字(测试/开发便利,不真校验)。
@@ -136,6 +144,11 @@ def verify_code(phone: str, code: str) -> bool:
logger.info("[SMS-chuanglan-MOCK] verify %s for %s****", "ok" if ok else "fail", phone[:3])
return ok
effective_max_attempts = (
settings.SMS_MAX_VERIFY_ATTEMPTS
if max_failed_attempts is None
else max_failed_attempts
)
with _lock:
rec = _codes.get(phone)
if rec is None:
@@ -143,7 +156,7 @@ def verify_code(phone: str, code: str) -> bool:
if time.time() > rec.expires_at:
_codes.pop(phone, None)
return False
if rec.attempts >= settings.SMS_MAX_VERIFY_ATTEMPTS:
if rec.attempts >= effective_max_attempts:
_codes.pop(phone, None) # 试错过多,作废
return False
if secrets.compare_digest(code.encode("utf-8"), rec.code.encode("utf-8")):
+20 -7
View File
@@ -70,20 +70,23 @@ def _gc(now: float) -> None:
_last_sent.pop(p, None)
def send_code(phone: str) -> int:
def send_code(phone: str, *, cooldown_sec: int | None = None) -> int:
"""发送验证码。
Returns: 距下次可发的秒数(= SMS_SEND_INTERVAL_SEC)
Raises: SmsError(过频 429 / 当日超限 429 / 供应商失败 503 / 手机号无效 400)
"""
now = time.time()
effective_cooldown = (
settings.SMS_SEND_INTERVAL_SEC if cooldown_sec is None else cooldown_sec
)
# --- lock 内:防刷检查 + 预占(防并发重复发烧钱)---
with _lock:
_gc(now) # 顺手清过期内存(超阈值才扫)
elapsed = now - _last_sent.get(phone, 0.0)
if elapsed < settings.SMS_SEND_INTERVAL_SEC:
remain = int(settings.SMS_SEND_INTERVAL_SEC - elapsed)
if elapsed < effective_cooldown:
remain = int(effective_cooldown - elapsed)
raise SmsError(f"发送过于频繁,请 {remain}s 后再试")
code = _gen_code()
@@ -108,10 +111,15 @@ def send_code(phone: str) -> int:
logger.exception("[SMS] send failed phone=%s****", phone[:3])
raise SmsError("验证码发送失败,请稍后重试", status_code=503) from e
return settings.SMS_SEND_INTERVAL_SEC
return effective_cooldown
def verify_code(phone: str, code: str) -> bool:
def verify_code(
phone: str,
code: str,
*,
max_failed_attempts: int | None = None,
) -> bool:
"""校验验证码。
- **mock 模式**:放行任意 N 位数字(测试/开发便利,不真校验)。
@@ -122,6 +130,11 @@ def verify_code(phone: str, code: str) -> bool:
logger.info("[SMS-MOCK] verify %s for %s****", "ok" if ok else "fail", phone[:3])
return ok
effective_max_attempts = (
settings.SMS_MAX_VERIFY_ATTEMPTS
if max_failed_attempts is None
else max_failed_attempts
)
with _lock:
rec = _codes.get(phone)
if rec is None:
@@ -129,7 +142,7 @@ def verify_code(phone: str, code: str) -> bool:
if time.time() > rec.expires_at:
_codes.pop(phone, None)
return False
if rec.attempts >= settings.SMS_MAX_VERIFY_ATTEMPTS:
if rec.attempts >= effective_max_attempts:
_codes.pop(phone, None) # 试错过多,作废
return False
if secrets.compare_digest(code.encode("utf-8"), rec.code.encode("utf-8")):
@@ -185,5 +198,5 @@ def _send_via_jiguang(phone: str, code: str) -> None:
if ecode == 50009: # 极光侧超频
raise SmsError("发送过于频繁,请稍后再试", status_code=429)
if ecode == 50006: # 手机号无效(schema 已挡格式,这里多是空号/停机)
raise SmsError("手机号无效", status_code=400)
raise SmsError("请输入有效的手机号", status_code=400)
raise SmsError(f"短信发送失败(code={ecode})", status_code=503)
+455 -46
View File
@@ -22,7 +22,7 @@ import uuid
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any
from urllib.parse import quote
from urllib.parse import quote, urlsplit
import httpx
@@ -31,6 +31,7 @@ from app.core.config import settings
logger = logging.getLogger("shagua.vendor_push")
TYPE_ACCESSIBILITY_DISABLED = "accessibility_disabled"
DATA_EVENT_NOTIFICATION_CREATED = "notification_created"
SUPPORTED_VENDORS = frozenset({"honor", "huawei", "vivo", "xiaomi", "oppo"})
# vendor key → 中文名(测试/配置状态接口展示用)
@@ -69,6 +70,18 @@ class _CachedToken:
_token_cache: dict[str, _CachedToken] = {}
_LOG_SUMMARY_MAX_CHARS = 1500
_LOG_STRING_MAX_CHARS = 200
_SENSITIVE_LOG_KEYS = {
"accesstoken",
"appkey",
"authtoken",
"authorization",
"clientsecret",
"mastersecret",
"sign",
}
def normalize_vendor(push_vendor: str | None) -> str | None:
if not push_vendor:
@@ -115,8 +128,16 @@ def send_notification(
if mock:
logger.info(
"[mock push] vendor=%s token=%s... title=%s body=%s extras=%s",
vendor, token[:12], title, body, extras,
"vendor push mock vendor=%s request=%s",
vendor,
_log_summary(
{
"push_token": token,
"title": title,
"body": body,
"extras": extras,
}
),
)
return {
"mock": True,
@@ -133,7 +154,28 @@ def send_notification(
"xiaomi": _send_xiaomi,
"oppo": _send_oppo,
}
return dispatch[vendor](token, title, body, extras)
started = time.perf_counter()
try:
result = dispatch[vendor](token, title, body, extras)
except Exception:
logger.exception(
"vendor push dispatch failed vendor=%s token=%s notification_type=%s "
"elapsed_ms=%.1f",
vendor,
token,
extras.get("type", ""),
_elapsed_ms(started),
)
raise
logger.info(
"vendor push dispatch succeeded vendor=%s token=%s notification_type=%s "
"elapsed_ms=%.1f",
vendor,
token,
extras.get("type", ""),
_elapsed_ms(started),
)
return result
def send_accessibility_disabled(
@@ -153,19 +195,188 @@ def send_accessibility_disabled(
)
def send_data_event(
push_vendor: str,
push_token: str,
*,
event: str,
notification_id: str,
mock: bool = False,
) -> dict[str, Any]:
"""发送无界面的轻量事件;当前只允许站内消息创建事件。"""
vendor = normalize_vendor(push_vendor)
token = push_token.strip() if push_token else ""
if not vendor or vendor not in SUPPORTED_VENDORS:
raise VendorPushError(f"unsupported push vendor: {push_vendor}")
if not token:
raise VendorPushError("push token is empty")
if event != DATA_EVENT_NOTIFICATION_CREATED:
raise VendorPushError(f"unsupported data event: {event}")
if not notification_id:
raise VendorPushError("notification_id is empty")
payload = {"event": event, "notificationId": str(notification_id)}
if mock:
return {"mock": True, "vendor": vendor, "payload": payload}
# vivo 的通知已设置 foregroundShow=falseApp 在前台时必走
# onForegroundMessageArrived,正好就是本事件需要的刷新信号;不重复占用一次推送配额。
if vendor == "vivo":
return {"skipped": True, "reason": "foreground notification callback"}
# OPush 当前只支持通知栏消息,没有服务端透传单推接口。
# 客户端在 OPPO 首页可见时轮询未读数;这里必须安全跳过,不能请求不存在的
# /message/transparent/unicast(该地址会稳定返回 HTTP 404)。
if vendor == "oppo":
return {"skipped": True, "reason": "oppo does not support data messages"}
dispatch: dict[str, Callable[[str, dict[str, str]], dict[str, Any]]] = {
"honor": _send_honor_data,
"huawei": _send_huawei_data,
"xiaomi": _send_xiaomi_data,
}
return dispatch[vendor](token, payload)
def _require(value: str, name: str) -> str:
if not value:
raise VendorPushError(f"{name} not configured")
return value
def _elapsed_ms(started: float) -> float:
return (time.perf_counter() - started) * 1000
def _redacted_value(value: Any) -> str:
raw = json.dumps(value, ensure_ascii=False, sort_keys=True, default=str)
digest = hashlib.sha256(raw.encode()).hexdigest()[:10]
return f"<redacted len={len(raw)} sha256={digest}>"
def _sanitize_for_log(value: Any, *, key: str = "", depth: int = 0) -> Any:
"""生成有排障价值、但不暴露服务端鉴权凭据的紧凑日志摘要。"""
normalized_key = "".join(char for char in key.lower() if char.isalnum())
if normalized_key in _SENSITIVE_LOG_KEYS or normalized_key.endswith("secret"):
return _redacted_value(value)
if depth >= 10:
return f"<{type(value).__name__}>"
if isinstance(value, dict):
items = list(value.items())
sanitized = {
str(item_key): _sanitize_for_log(item_value, key=str(item_key), depth=depth + 1)
for item_key, item_value in items[:30]
}
if len(items) > 30:
sanitized["<omitted_keys>"] = len(items) - 30
return sanitized
if isinstance(value, (list, tuple)):
items = list(value)
sanitized = [_sanitize_for_log(item, key=key, depth=depth + 1) for item in items[:20]]
if len(items) > 20:
sanitized.append(f"<omitted_items={len(items) - 20}>")
return sanitized
if isinstance(value, str):
if value.lstrip().startswith(("{", "[")):
try:
decoded = json.loads(value)
except ValueError:
pass
else:
return _sanitize_for_log(decoded, key=key, depth=depth + 1)
if len(value) > _LOG_STRING_MAX_CHARS:
return f"{value[:_LOG_STRING_MAX_CHARS]}…<len={len(value)}>"
return value
def _log_summary(value: Any) -> str:
rendered = json.dumps(
_sanitize_for_log(value),
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
default=str,
)
if len(rendered) > _LOG_SUMMARY_MAX_CHARS:
return f"{rendered[:_LOG_SUMMARY_MAX_CHARS]}…<len={len(rendered)}>"
return rendered
def _raw_log_summary(value: Any) -> str:
"""厂商响应摘要:保留原始字段和值,仅限制单条日志长度。"""
rendered = json.dumps(
value,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
default=str,
)
if len(rendered) > _LOG_SUMMARY_MAX_CHARS:
return f"{rendered[:_LOG_SUMMARY_MAX_CHARS]}…<len={len(rendered)}>"
return rendered
def _endpoint_for_log(url: str) -> str:
parsed = urlsplit(url)
return f"{parsed.scheme}://{parsed.netloc}{parsed.path}"
def _request_summary(kwargs: dict[str, Any]) -> str:
summary: dict[str, Any] = {}
if "json" in kwargs:
summary["body_type"] = "json"
summary["request_payload"] = kwargs["json"]
elif "data" in kwargs:
summary["body_type"] = "form"
summary["request_payload"] = kwargs["data"]
if kwargs.get("params"):
summary["params"] = kwargs["params"]
if kwargs.get("headers"):
summary["headers"] = kwargs["headers"]
return _log_summary(summary)
def _response_summary(resp: Any, parsed: Any | None = None) -> str:
if parsed is None:
try:
parsed = resp.json()
except ValueError:
pass
if parsed is not None:
return _raw_log_summary(parsed)
return _raw_log_summary(getattr(resp, "text", ""))
def _vendor_response_failed(vendor: str, data: dict[str, Any]) -> bool:
"""识别 HTTP 200 中明确的厂商业务失败,避免将失败请求记录成 success。"""
if data.get("error") or data.get("success") is False:
return True
if vendor == "xiaomi":
code = data.get("code")
result = str(data.get("result", "ok")).lower()
return code not in (0, "0", None) or result not in ("ok", "success")
if vendor == "oppo" and data.get("code") is not None:
return int(data["code"]) != 0
if vendor == "vivo" and data.get("result") is not None:
return int(data["result"]) != 0
if vendor == "honor" and data.get("code") is not None:
return int(data["code"]) != 200
if vendor == "huawei" and data.get("code") is not None:
return str(data["code"]) != "80000000"
return False
def _request_json(
method: str,
url: str,
*,
vendor: str,
operation: str,
expected_status: tuple[int, ...] = (200,),
**kwargs: Any,
) -> dict[str, Any]:
started = time.perf_counter()
request_summary = _request_summary(kwargs)
try:
resp = httpx.request(
method,
@@ -174,41 +385,85 @@ def _request_json(
**kwargs,
)
except httpx.HTTPError as e:
logger.error(
"vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
"outcome=network_error request=%s response=%s elapsed_ms=%.1f error=%s",
vendor,
operation,
method,
_endpoint_for_log(url),
request_summary,
"<no_response>",
_elapsed_ms(started),
type(e).__name__,
)
raise VendorPushError(f"push http error: {e}") from e
if resp.status_code not in expected_status:
logger.error("vendor push http failed url=%s http=%s body=%s", url, resp.status_code, resp.text[:500])
logger.error(
"vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
"outcome=http_error http_status=%s request=%s response=%s elapsed_ms=%.1f",
vendor,
operation,
method,
_endpoint_for_log(url),
resp.status_code,
request_summary,
_response_summary(resp),
_elapsed_ms(started),
)
raise VendorPushError(f"push http {resp.status_code}")
try:
return resp.json()
data = resp.json()
except ValueError as e:
logger.error(
"vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
"outcome=invalid_json http_status=%s request=%s response=%s elapsed_ms=%.1f",
vendor,
operation,
method,
_endpoint_for_log(url),
resp.status_code,
request_summary,
_response_summary(resp),
_elapsed_ms(started),
)
raise VendorPushError(f"push invalid json: {resp.text[:200]}") from e
vendor_failed = _vendor_response_failed(vendor, data)
log = logger.warning if vendor_failed else logger.info
log(
"vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
"outcome=%s http_status=%s request=%s response=%s elapsed_ms=%.1f",
vendor,
operation,
method,
_endpoint_for_log(url),
"vendor_error" if vendor_failed else "success",
resp.status_code,
request_summary,
_response_summary(resp, data),
_elapsed_ms(started),
)
return data
def _request_form(
method: str,
url: str,
*,
vendor: str,
operation: str,
expected_status: tuple[int, ...] = (200,),
**kwargs: Any,
) -> dict[str, Any]:
try:
resp = httpx.request(
method,
url,
timeout=settings.PUSH_REQUEST_TIMEOUT_SEC,
**kwargs,
)
except httpx.HTTPError as e:
raise VendorPushError(f"push http error: {e}") from e
if resp.status_code not in expected_status:
logger.error("vendor push http failed url=%s http=%s body=%s", url, resp.status_code, resp.text[:500])
raise VendorPushError(f"push http {resp.status_code}")
try:
return resp.json()
except ValueError as e:
raise VendorPushError(f"push invalid json: {resp.text[:200]}") from e
return _request_json(
method,
url,
vendor=vendor,
operation=operation,
expected_status=expected_status,
**kwargs,
)
def _cache_get(key: str) -> str | None:
@@ -234,6 +489,8 @@ def _honor_access_token() -> str:
data = _request_form(
"POST",
settings.HONOR_PUSH_TOKEN_ENDPOINT,
vendor="honor",
operation="authenticate",
data={
"grant_type": "client_credentials",
"client_id": client_id,
@@ -243,16 +500,22 @@ def _honor_access_token() -> str:
)
token = data.get("access_token")
if not token:
raise VendorPushError(f"honor auth failed: {data}")
raise VendorPushError(f"honor auth failed: {_raw_log_summary(data)}")
return _cache_put(cache_key, str(token), data.get("expires_in"))
def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> dict[str, Any]:
app_id = _require(settings.HONOR_PUSH_APP_ID, "HONOR_PUSH_APP_ID")
access_token = _honor_access_token()
click_action: dict[str, Any]
if extras.get("notificationId"):
# type=3 只负责打开首页,不保证 data 会变成目标 Activity 的 extras。消息中心通知必须
# 用自定义页面(type=1)+ intent URI,把 notificationId/type/feedbackId 等直接带给
# MainActivity;否则华为/荣耀系统能展示通知,但用户点击后客户端收不到任何导航参数。
click_action = {"type": 1, "intent": _click_intent_uri(extras)}
else:
click_action = {"type": 3}
payload = {
# clickAction type=3(打开应用首页)时,荣耀点击会把 data JSON 的键值对注入启动 intent 的
# extras(与 HMS 同机制)→ MainActivity.consumeNavTarget 读 notif_id/notif_type 直达落地。
"data": json.dumps(_click_extras(extras), ensure_ascii=False),
"notification": {"title": title, "body": body},
"android": {
@@ -261,7 +524,7 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
"notification": {
"title": title,
"body": body,
"clickAction": {"type": 3},
"clickAction": click_action,
"importance": "NORMAL",
},
},
@@ -270,6 +533,8 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
data = _request_json(
"POST",
settings.HONOR_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
vendor="honor",
operation="send_notification",
json=payload,
headers={
"Content-Type": "application/json; charset=UTF-8",
@@ -279,7 +544,28 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
)
code = data.get("code")
if code is not None and int(code) != 200:
raise VendorPushError(f"honor push failed: {data}")
raise VendorPushError(f"honor push failed: {_raw_log_summary(data)}")
return data
def _send_honor_data(token: str, payload: dict[str, str]) -> dict[str, Any]:
app_id = _require(settings.HONOR_PUSH_APP_ID, "HONOR_PUSH_APP_ID")
access_token = _honor_access_token()
data = _request_json(
"POST",
settings.HONOR_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
vendor="honor",
operation="send_data_event",
json={"data": json.dumps(payload, ensure_ascii=False), "token": [token]},
headers={
"Content-Type": "application/json; charset=UTF-8",
"Authorization": f"Bearer {access_token}",
"timestamp": str(int(time.time() * 1000)),
},
)
code = data.get("code")
if code is not None and int(code) != 200:
raise VendorPushError(f"honor data push failed: {_raw_log_summary(data)}")
return data
@@ -294,6 +580,8 @@ def _huawei_access_token() -> str:
data = _request_form(
"POST",
settings.HUAWEI_PUSH_TOKEN_ENDPOINT,
vendor="huawei",
operation="authenticate",
data={
"grant_type": "client_credentials",
"client_id": app_id,
@@ -303,7 +591,7 @@ def _huawei_access_token() -> str:
)
token = data.get("access_token")
if not token:
raise VendorPushError(f"huawei auth failed: {data}")
raise VendorPushError(f"huawei auth failed: {_raw_log_summary(data)}")
return _cache_put(cache_key, str(token), data.get("expires_in"))
@@ -312,18 +600,24 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d
'80100000' 为部分成功( token 场景仍视为失败,错误里带原始响应便于排障)"""
app_id = _require(settings.HUAWEI_PUSH_APP_ID, "HUAWEI_PUSH_APP_ID")
access_token = _huawei_access_token()
click_action: dict[str, Any]
if extras.get("notificationId"):
# type=3 只打开首页,Mate 20/EMUI 不会把 message.data 自动拆成启动 Intent extras。
# type=1 的自定义 intent 才能稳定携带反馈记录 id,且冷启动/onNewIntent 都走同一路由。
click_action = {"type": 1, "intent": _click_intent_uri(extras)}
else:
click_action = {"type": 3}
payload = {
"validate_only": False,
"message": {
# click_action type=3(打开应用首页)时,HMS 点击会把 data JSON 的键值对注入启动 intent
# 的 extras → MainActivity.consumeNavTarget 读 notif_id/notif_type 直达落地。
"data": json.dumps(_click_extras(extras), ensure_ascii=False),
"android": {
"target_user_type": settings.HUAWEI_PUSH_TARGET_USER_TYPE,
"ttl": f"{settings.PUSH_TIME_TO_LIVE_SEC}s",
"notification": {
"title": title,
"body": body,
"click_action": {"type": 3},
"click_action": click_action,
"importance": "NORMAL",
},
},
@@ -333,6 +627,8 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d
data = _request_json(
"POST",
settings.HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
vendor="huawei",
operation="send_notification",
json=payload,
headers={
"Content-Type": "application/json; charset=UTF-8",
@@ -340,7 +636,32 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d
},
)
if str(data.get("code", "")) != "80000000":
raise VendorPushError(f"huawei push failed: {data}")
raise VendorPushError(f"huawei push failed: {_raw_log_summary(data)}")
return data
def _send_huawei_data(token: str, payload: dict[str, str]) -> dict[str, Any]:
app_id = _require(settings.HUAWEI_PUSH_APP_ID, "HUAWEI_PUSH_APP_ID")
access_token = _huawei_access_token()
data = _request_json(
"POST",
settings.HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
vendor="huawei",
operation="send_data_event",
json={
"validate_only": False,
"message": {
"data": json.dumps(payload, ensure_ascii=False),
"token": [token],
},
},
headers={
"Content-Type": "application/json; charset=UTF-8",
"Authorization": f"Bearer {access_token}",
},
)
if str(data.get("code", "")) != "80000000":
raise VendorPushError(f"huawei data push failed: {_raw_log_summary(data)}")
return data
@@ -357,6 +678,8 @@ def _vivo_auth_token() -> str:
data = _request_json(
"POST",
settings.VIVO_PUSH_AUTH_ENDPOINT,
vendor="vivo",
operation="authenticate",
json={
"appId": app_id,
"appKey": app_key,
@@ -366,10 +689,10 @@ def _vivo_auth_token() -> str:
headers={"Content-Type": "application/json"},
)
if int(data.get("result", -1)) != 0:
raise VendorPushError(f"vivo auth failed: {data}")
raise VendorPushError(f"vivo auth failed: {_raw_log_summary(data)}")
token = data.get("authToken")
if not token:
raise VendorPushError(f"vivo auth missing authToken: {data}")
raise VendorPushError(f"vivo auth missing authToken: {_raw_log_summary(data)}")
return _cache_put(cache_key, str(token), 24 * 3600)
@@ -385,15 +708,22 @@ def _send_vivo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
"timeToLive": settings.PUSH_TIME_TO_LIVE_SEC,
"requestId": uuid.uuid4().hex,
"pushMode": settings.VIVO_PUSH_MODE,
# vivo SDK 4.1.5 只有在前台展示关闭时才调用
# OpenClientPushMessageReceiver.onForegroundMessageArrived。
# App 在该回调中刷新服务端未读数并补发一条本地通知;后台/锁屏时仍由
# vivo 系统展示,因而任何场景都只会出现一条通知。
"foregroundShow": False,
# vivo 官方 VPush 角标字段:离线收到通知时先由桌面自动 +1;
# App 启动/同步后再由统一未读数通过系统 API 精确校准。
"addBadge": True,
"clientCustomMap": extras,
}
# 点击落地:消息中心推送(带 notificationId)→ skipType=4 + skipContent=intent uri,由 vivo
# 系统直启 MainActivity 并携带 S. extras(与小米 notify_effect=2 同机制)。不依赖客户端
# VivoPushReceiver.onNotificationMessageClicked 里的后台 startActivity——Android 10+ BAL
# 会静默拦掉,receiver 路径仅作兜底。无 notificationId 的召回类保持 skipType=1 仅打开首页。
# vivo Push SDK 480+ 已不再回调 skipType=3;必须使用 skipType=4 的完整 Intent URI。
# URI 同时包含 data/scheme、显式 component 和 S. extrasOriginOS 才会把业务参数
# 原样交给 MainActivity(仅靠隐式 deeplink 会打开 App,但可能剥掉 extras)。
if extras.get("notificationId"):
payload["skipType"] = 4
payload["skipContent"] = _click_intent_uri(extras)
payload["skipContent"] = _vivo_click_intent_uri(extras)
else:
payload["skipType"] = 1
if settings.VIVO_PUSH_CATEGORY:
@@ -401,14 +731,37 @@ def _send_vivo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
data = _request_json(
"POST",
settings.VIVO_PUSH_SEND_ENDPOINT,
vendor="vivo",
operation="send_notification",
json=payload,
headers={
"Content-Type": "application/json",
"authToken": auth_token,
},
)
# 10089 = 当前 vivo 应用尚未开通 VPush「离线自动角标」能力。
# 角标只是通知中心未读数的镜像,不能因此让整条通知发送失败:去掉 addBadge
# 原样重试,通知到达/应用同步后仍由客户端系统 API 按统一未读数精确设置。
if int(data.get("result", -1)) == 10089:
logger.warning("vivo VPush badge not enabled (10089); retrying without addBadge")
fallback_payload = dict(payload)
fallback_payload.pop("addBadge", None)
fallback_payload["requestId"] = uuid.uuid4().hex
data = _request_json(
"POST",
settings.VIVO_PUSH_SEND_ENDPOINT,
vendor="vivo",
operation="send_notification",
json=fallback_payload,
headers={
"Content-Type": "application/json",
"authToken": auth_token,
},
)
if int(data.get("result", -1)) == 0:
data = {**data, "badgeFallback": True}
if int(data.get("result", -1)) != 0:
raise VendorPushError(f"vivo push failed: {data}")
raise VendorPushError(f"vivo push failed: {_raw_log_summary(data)}")
return data
@@ -446,14 +799,43 @@ def _send_xiaomi(token: str, title: str, body: str, extras: dict[str, str]) -> d
data = _request_form(
"POST",
settings.XIAOMI_PUSH_SEND_ENDPOINT,
vendor="xiaomi",
operation="send_notification",
data=form,
headers={"Authorization": f"key={app_secret}"},
)
code = data.get("code")
if code not in (0, "0", None):
raise VendorPushError(f"xiaomi push failed: {data}")
raise VendorPushError(f"xiaomi push failed: {_raw_log_summary(data)}")
if str(data.get("result", "ok")).lower() not in ("ok", "success"):
raise VendorPushError(f"xiaomi push failed: {data}")
raise VendorPushError(f"xiaomi push failed: {_raw_log_summary(data)}")
return data
def _send_xiaomi_data(token: str, payload: dict[str, str]) -> dict[str, Any]:
app_secret = _require(settings.XIAOMI_PUSH_APP_SECRET, "XIAOMI_PUSH_APP_SECRET")
form = {
"registration_id": token,
"restricted_package_name": settings.ANDROID_PACKAGE_NAME,
"payload": json.dumps(payload, ensure_ascii=False),
"pass_through": "1",
"time_to_live": str(settings.PUSH_TIME_TO_LIVE_SEC * 1000),
}
if settings.XIAOMI_PUSH_CHANNEL_ID.strip():
form["extra.channel_id"] = settings.XIAOMI_PUSH_CHANNEL_ID.strip()
data = _request_form(
"POST",
settings.XIAOMI_PUSH_SEND_ENDPOINT,
vendor="xiaomi",
operation="send_data_event",
data=form,
headers={"Authorization": f"key={app_secret}"},
)
code = data.get("code")
if code not in (0, "0", None):
raise VendorPushError(f"xiaomi data push failed: {_raw_log_summary(data)}")
if str(data.get("result", "ok")).lower() not in ("ok", "success"):
raise VendorPushError(f"xiaomi data push failed: {_raw_log_summary(data)}")
return data
@@ -490,6 +872,24 @@ def _click_intent_uri(extras: dict[str, str]) -> str:
return ";".join(parts)
def _vivo_click_intent_uri(extras: dict[str, str]) -> str:
"""vivo skipType=4 使用官方要求的完整 intent deeplink 形态。
`?#Intent` 中的问号不能省;OriginOS 对缺少它的 URI 会退化为“仅打开应用”,
不把 S. 参数交给目标 Activity
"""
pkg = settings.ANDROID_PACKAGE_NAME
parts = [
"intent://push/detail?#Intent",
"scheme=shaguabijia",
f"component={pkg}/{pkg}.MainActivity",
"launchFlags=0x14000000",
]
parts += [f"S.{key}={quote(str(value), safe='')}" for key, value in _click_extras(extras).items()]
parts.append("end")
return ";".join(parts)
def _xiaomi_template_param(title: str, alert: str) -> str:
rendered = (
settings.XIAOMI_PUSH_TEMPLATE_PARAM_JSON
@@ -522,6 +922,8 @@ def _oppo_auth_token() -> str:
data = _request_form(
"POST",
settings.OPPO_PUSH_AUTH_ENDPOINT,
vendor="oppo",
operation="authenticate",
data={
"app_key": app_key,
"timestamp": timestamp,
@@ -530,10 +932,10 @@ def _oppo_auth_token() -> str:
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
if int(data.get("code", -1)) != 0:
raise VendorPushError(f"oppo auth failed: {data}")
raise VendorPushError(f"oppo auth failed: {_raw_log_summary(data)}")
token = (data.get("data") or {}).get("auth_token") or data.get("auth_token")
if not token:
raise VendorPushError(f"oppo auth missing auth_token: {data}")
raise VendorPushError(f"oppo auth missing auth_token: {_raw_log_summary(data)}")
return _cache_put(cache_key, str(token), 24 * 3600)
@@ -548,6 +950,11 @@ def _send_oppo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
"off_line_ttl": ttl_hours,
"action_parameters": json.dumps(_click_extras(extras), ensure_ascii=False),
}
if extras.get("notificationId"):
# 只有已落库的站内消息才计入角标;保活提醒等临时推送不应留下无法消除的未读数。
# 客户端打开/已读后会按服务端真实未读总数覆盖,避免长期漂移。
notification["badge_operation_type"] = 1
notification["badge_message_count"] = 1
# 点击落地:OPPO SDK 没有点击回调,参数只能靠服务端点击动作配置送达——action_parameters 的
# 键值对仅在 click_action_type=1/4 时才会注入目标 Activity 的 intent extras(type=0「启动应用」
# 会忽略它,extras 全丢 → 点了没反应,与小米 notify_effect=1 同款坑)。
@@ -573,6 +980,8 @@ def _send_oppo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
data = _request_form(
"POST",
settings.OPPO_PUSH_SEND_ENDPOINT,
vendor="oppo",
operation="send_notification",
data={
"auth_token": auth_token,
"message": json.dumps(message, ensure_ascii=False),
@@ -580,5 +989,5 @@ def _send_oppo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
if int(data.get("code", -1)) != 0:
raise VendorPushError(f"oppo push failed: {data}")
raise VendorPushError(f"oppo push failed: {_raw_log_summary(data)}")
return data
+1
View File
@@ -36,6 +36,7 @@ from app.models.inactivity import ( # noqa: F401
from app.models.invite import InviteRelation # noqa: F401
from app.models.invite_fingerprint import InviteFingerprint # noqa: F401
from app.models.launch_confirm_sample import LaunchConfirmSample # noqa: F401
from app.models.limit_policy import LimitPolicyOverride # noqa: F401
from app.models.meituan_coupon import MeituanCoupon # noqa: F401
from app.models.notification import Notification # noqa: F401
from app.models.onboarding import OnboardingCompletion # noqa: F401
+4
View File
@@ -113,6 +113,10 @@ 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 全量),未来取数兜底
+25 -41
View File
@@ -1,20 +1,4 @@
"""新手引导视频播放记录(领券浮层前 N 次用它替代广告)。
产品规则(2026-07 拍板):新用户点一键自动领取后的等候浮层,** 3 **不放广告,
改放运营后台上传的引导视频;每次固定 120 金币,中途关闭也算看完照发
口径:
- **计次按账号**(user_id),与设备无关 换设备不重新送 3
- **开播即计数**:客户端每次要展示浮层时调 `/api/v1/guide-video/start`,服务端当场
写一行(status='playing')并返回 play_token;`COUNT(*)` 即已用次数用户中途 kill
App 也算用掉一次(产品选定口径,防反复进出刷金币)
- **发币幂等** play_token 定位 + `status='playing'` 条件更新:并发两次上报只有一次
改到行(另一次 rowcount=0),所以只发一次币光有 play_token 唯一键挡不住 发币走的是
UPDATE, INSERT,撞不到任何唯一键
- **次数上限** (user_id, seq) 唯一键兜底,防并发 /start 绕过 COUNT 判定(见下)
与广告收益(ad_feed_reward_record)彻底分离:引导视频不是广告,不该进广告收益报表
"""
"""引导视频播放计划、起播快照与逐圈结算状态。"""
from __future__ import annotations
from datetime import datetime
@@ -26,48 +10,48 @@ from app.db.base import Base
class GuideVideoPlay(Base):
"""一次引导视频播放一行。开播时建(status='playing'),发币后置 'granted'"""
__tablename__ = "guide_video_play"
__table_args__ = (
# 客户端幂等键:同一次播放重复上报奖励只发一次。
UniqueConstraint("play_token", name="uq_guide_video_play_token"),
# 次数上限的**硬约束**:start_play 是无锁 check-then-insert(读 COUNT 算 seq 再插),
# N 个并发 /start 会都读到同一个已用次数、算出同一个 seq,不拦就能各拿一个 token、
# 各发一次金币,3 次上限形同虚设(改包即可无限刷)。seq 唯一 → 并发同 seq 必撞,
# start_play 捕获 IntegrityError 降级成"这次不放视频"。
# 用 unique Index 而非 UniqueConstraint:与迁移里的 create_index 对齐(SQLite 加约束
# 要整表重建),autogenerate 才不会每次报一条假 diff。
Index("uq_guide_video_play_user_seq", "user_id", "seq", unique=True),
Index(
"uq_guide_video_play_user_scene_seq",
"user_id",
"scene",
"seq",
unique=True,
),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
user_id: Mapped[int] = mapped_column(
Integer, ForeignKey("user.id"), index=True, nullable=False
)
# 服务端生成下发给客户端的幂等键(uuid hex)。
play_token: Mapped[str] = mapped_column(String(64), nullable=False)
# 触发场景:目前只有 coupon(领券等候浮层);留字段以便日后比价等场景复用。
scene: Mapped[str] = mapped_column(String(16), nullable=False, default="coupon")
# 本账号第几次(1-based),= 建行时已有行数 + 1。日常判定仍以 COUNT 为准,但 (user_id, seq)
# 唯一键让并发 /start 只能成一个 —— 见 __table_args__。
seq: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
# 当次下发的视频地址(运营换片后能回溯用户当时看的是哪支)。
# prepare 不占次数,seq=NULLstart 成功才写入 1-based seq
seq: Mapped[int | None] = mapped_column(Integer, nullable=True)
video_url: Mapped[str | None] = mapped_column(String(512), nullable=True)
# 实发金币;未发时 0。
coin: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
# playing(已开播未发币) / granted(已发币)。
status: Mapped[str] = mapped_column(String(16), nullable=False, default="playing")
# 客户端上报时是否播完(true=自然播完 / false=中途关闭)。仅留痕:两者都发币。
duration_ms: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
config_version: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
settled_circles: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
# prepared / started / completed / legacy_completed / legacy_closed
status: Mapped[str] = mapped_column(String(24), nullable=False, default="prepared")
completed: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
started_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
prepared_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), nullable=False
)
expires_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), index=True, nullable=True
)
started_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), index=True, nullable=True
)
granted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
def __repr__(self) -> str: # pragma: no cover
return (
f"<GuideVideoPlay user={self.user_id} seq={self.seq} "
f"{self.status} coin={self.coin}>"
f"<GuideVideoPlay user={self.user_id} scene={self.scene} seq={self.seq} "
f"{self.status} circles={self.settled_circles}>"
)
+77
View File
@@ -0,0 +1,77 @@
"""Per-subject limit policy overrides used by the admin whitelist page."""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import (
Boolean,
DateTime,
Index,
Integer,
String,
UniqueConstraint,
func,
true,
)
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
class LimitPolicyOverride(Base):
"""One rule override for one phone or device.
``reset_at`` is a non-destructive usage baseline. Business records and
security events remain intact; quota readers only count rows at or after
this timestamp.
"""
__tablename__ = "limit_policy_override"
__table_args__ = (
UniqueConstraint(
"subject_type",
"subject_value",
"rule_code",
name="uq_limit_policy_subject_rule",
),
Index(
"ix_limit_policy_lookup",
"subject_type",
"subject_value",
"rule_code",
"enabled",
),
Index("ix_limit_policy_expires", "expires_at"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
subject_type: Mapped[str] = mapped_column(String(16), nullable=False)
subject_value: Mapped[str] = mapped_column(String(128), nullable=False)
rule_code: Mapped[str] = mapped_column(String(64), nullable=False)
# 产品只保留“临时不限/免告警”。即使有内部脚本绕过 API 直接建 ORM
# 对象,也不能再悄悄落成已经下线的 override 模式。
mode: Mapped[str] = mapped_column(String(24), nullable=False, default="unlimited")
limit_value: Mapped[int | None] = mapped_column(Integer, nullable=True)
enabled: Mapped[bool] = mapped_column(
Boolean, nullable=False, default=True, server_default=true()
)
starts_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True
)
expires_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True
)
reset_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True
)
reason: Mapped[str | None] = mapped_column(String(256), nullable=True)
created_by_admin_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), nullable=False
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
onupdate=func.now(),
nullable=False,
)
+9
View File
@@ -60,6 +60,15 @@ class CoinTransaction(Base):
sqlite_where=text("biz_type LIKE 'task%' AND ref_id IS NOT NULL"),
postgresql_where=text("biz_type LIKE 'task%' AND ref_id IS NOT NULL"),
),
Index(
"ux_coin_transaction_guide_video_ref",
"user_id",
"biz_type",
"ref_id",
unique=True,
sqlite_where=text("biz_type = 'guide_video' AND ref_id IS NOT NULL"),
postgresql_where=text("biz_type = 'guide_video' AND ref_id IS NOT NULL"),
),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+36 -5
View File
@@ -5,11 +5,13 @@
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from app.core import rewards
from app.core import limit_policy, rewards
from app.core.rewards import cn_today
from app.models.ad_feed_reward import AdFeedRewardRecord
from app.repositories import wallet as crud_wallet
@@ -29,8 +31,14 @@ def _find_by_event(db: Session, client_event_id: str) -> AdFeedRewardRecord | No
).scalar_one_or_none()
def _granted_today(db: Session, user_id: int, reward_date: str) -> int:
return db.execute(
def _granted_today(
db: Session,
user_id: int,
reward_date: str,
*,
reset_at: datetime | None = None,
) -> int:
stmt = (
select(func.count())
.select_from(AdFeedRewardRecord)
.where(
@@ -38,7 +46,10 @@ def _granted_today(db: Session, user_id: int, reward_date: str) -> int:
AdFeedRewardRecord.reward_date == reward_date,
AdFeedRewardRecord.status == "granted",
)
).scalar_one()
)
if reset_at is not None:
stmt = stmt.where(AdFeedRewardRecord.created_at >= reset_at)
return db.execute(stmt).scalar_one()
def granted_unit_total(db: Session, user_id: int) -> int:
@@ -121,7 +132,27 @@ def grant_feed_reward(
)
return _commit_record(db, rec, client_event_id)
if _granted_today(db, user_id, today) >= rewards.get_ad_daily_limit(db):
daily_policy = limit_policy.resolve_for_user(
db,
"ad.feed.daily",
user_id,
)
daily_limit = (
rewards.get_ad_daily_limit(db)
if daily_policy.override_id is None
and daily_policy.bucket_version == "default"
else daily_policy.limit
)
if (
daily_limit is not None
and _granted_today(
db,
user_id,
today,
reset_at=daily_policy.reset_at,
)
>= daily_limit
):
rec = AdFeedRewardRecord(
client_event_id=client_event_id,
user_id=user_id,
+82 -17
View File
@@ -10,13 +10,13 @@
"""
from __future__ import annotations
from datetime import datetime, timezone
from datetime import UTC, datetime
from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from app.core import rewards
from app.core import limit_policy, rewards
from app.core.ad_cooldown import compute_cooldown
from app.core.rewards import DAILY_AD_WATCH_SECONDS_LIMIT, cn_today
from app.models.ad_reward import AdRewardRecord
@@ -96,8 +96,14 @@ def round_coin_total(db: Session, user_id: int, boost_round_id: str) -> int:
)
def _granted_today(db: Session, user_id: int, reward_date: str) -> int:
return db.execute(
def _granted_today(
db: Session,
user_id: int,
reward_date: str,
*,
reset_at: datetime | None = None,
) -> int:
stmt = (
select(func.count())
.select_from(AdRewardRecord)
.where(
@@ -106,7 +112,10 @@ def _granted_today(db: Session, user_id: int, reward_date: str) -> int:
AdRewardRecord.status == "granted",
AdRewardRecord.reward_scene == "reward_video",
)
).scalar_one()
)
if reset_at is not None:
stmt = stmt.where(AdRewardRecord.created_at >= reset_at)
return db.execute(stmt).scalar_one()
def _granted_cumulative(db: Session, user_id: int) -> int:
@@ -165,7 +174,27 @@ def grant_ad_reward(
DAILY_AD_WATCH_SECONDS_LIMIT > 0
and watched_seconds_today(db, user_id, today=today) >= DAILY_AD_WATCH_SECONDS_LIMIT
)
over_count = _granted_today(db, user_id, today) >= rewards.get_ad_daily_limit(db)
daily_policy = limit_policy.resolve_for_user(
db,
"ad.reward_video.daily",
user_id,
)
daily_limit = (
rewards.get_ad_daily_limit(db)
if daily_policy.override_id is None
and daily_policy.bucket_version == "default"
else daily_policy.limit
)
over_count = (
daily_limit is not None
and _granted_today(
db,
user_id,
today,
reset_at=daily_policy.reset_at,
)
>= daily_limit
)
if over_time or over_count:
rec = AdRewardRecord(
trans_id=trans_id, user_id=user_id, coin=0, status="capped",
@@ -321,19 +350,24 @@ def _commit_record(db: Session, rec: AdRewardRecord, trans_id: str) -> AdRewardR
return rec
def _granted_times_today_desc(db: Session, user_id: int, reward_date: str) -> list[datetime]:
def _granted_times_today_desc(
db: Session,
user_id: int,
reward_date: str,
*,
reset_at: datetime | None = None,
) -> list[datetime]:
"""当日 status=granted 记录的 created_at,按时间倒序(最新在前)——冷却策略的输入数据。"""
return list(
db.execute(
select(AdRewardRecord.created_at)
.where(
stmt = select(AdRewardRecord.created_at).where(
AdRewardRecord.user_id == user_id,
AdRewardRecord.reward_date == reward_date,
AdRewardRecord.status == "granted",
AdRewardRecord.reward_scene == "reward_video",
)
.order_by(AdRewardRecord.created_at.desc())
).scalars()
if reset_at is not None:
stmt = stmt.where(AdRewardRecord.created_at >= reset_at)
return list(
db.execute(stmt.order_by(AdRewardRecord.created_at.desc())).scalars()
)
@@ -349,16 +383,47 @@ def today_status(
旧客户端兼容,当前 DAILY_AD_WATCH_SECONDS_LIMIT=0 表示不启用时长闸
"""
today = cn_today().isoformat()
granted_desc = _granted_times_today_desc(db, user_id, today)
daily_policy = limit_policy.resolve_for_user(
db,
"ad.reward_video.daily",
user_id,
)
cooldown_policy = limit_policy.resolve_for_user(
db,
"ad.reward_video.cooldown",
user_id,
)
daily_limit = (
rewards.get_ad_daily_limit(db)
if daily_policy.override_id is None
and daily_policy.bucket_version == "default"
else daily_policy.limit
)
cooldown_seconds = (
rewards.get_ad_cooldown_sec(db)
if cooldown_policy.override_id is None
and cooldown_policy.bucket_version == "default"
else (cooldown_policy.limit or 0)
)
granted_desc = _granted_times_today_desc(
db,
user_id,
today,
reset_at=daily_policy.reset_at,
)
state = compute_cooldown(
granted_desc,
datetime.now(timezone.utc),
datetime.now(UTC),
round_size=rewards.get_ad_round_count(db),
cooldown_seconds=rewards.get_ad_cooldown_sec(db),
cooldown_seconds=cooldown_seconds,
)
return (
len(granted_desc),
rewards.get_ad_daily_limit(db),
(
daily_limit
if daily_limit is not None
else limit_policy.get_rule("ad.reward_video.daily").max_value
),
0,
state.round_count,
state.cooldown_until,
+124 -13
View File
@@ -164,8 +164,9 @@ def _derive(payload: ComparisonRecordIn) -> dict:
is_source_best = best.is_source if best is not None else None
# status:客户端显式给了就用;否则有"非源且有价"的结果=success,否则 failed
status = payload.status
# status:优先 pricebot record_status(区分 below_minimum/store_closed) → 客户端显式 status
# → 兜底"非源且有价"=success/否则 failed。record_status 让"未满起送"不再塌缩成 failed。
status = payload.record_status or payload.status
if status is None:
has_valid_target = any(
(not r.is_source) and r.price is not None for r in results
@@ -197,16 +198,35 @@ def upsert_record(
灰度期老客户端 POST /compare/record 走这条,与后端 harvest trace_id reconcile;
新客户端不再 POST(改由 compare.py 透传壳 harvest 落库)
"""
derived = _derive(payload)
# 单源派生: 与 harvest_done 一致, payload 带 platforms 时从它派生(唯一真相源
# _derive_from_platforms), 老客户端不带 platforms 时回退 _derive(从 comparison_results)。
if payload.platforms:
derived = _derive_from_platforms(payload.platforms, payload.record_status)
# 对齐 _derive 返回键(#189 fail_reason): 两路径 fields 键集一致, 覆盖已有行时不残留旧值
derived["fail_reason"] = (
_derive_fail_display(payload.information, payload.platform_results or {})
if derived["status"] == "failed"
else None
)
# 单源派生取自 platforms 源行(常无源平台元数据/店名)→ 空则用 payload 兜底不丢字段。
# 下面 fields 不再显式写这四个键, 统一由 derived 提供(否则 dict(store_name=..., **derived)
# 与 _derive_from_platforms 同名键撞键 TypeError)。
for _k in ("store_name", "source_platform_id", "source_platform_name", "source_package"):
if not derived.get(_k):
derived[_k] = getattr(payload, _k)
else:
derived = _derive(payload)
# _derive 只从 comparison_results 派生, 不含源平台四件套 / store_name → 从 payload 补,
# 与上面 platforms 分支键集对齐(fields 统一靠 **derived 提供这些列)。
for _k in ("store_name", "source_platform_id", "source_platform_name", "source_package"):
derived[_k] = getattr(payload, _k)
items = [it.model_dump(exclude_none=True) for it in payload.items]
fields = dict(
device_id=payload.device_id,
business_type=payload.business_type,
store_name=payload.store_name,
product_names=_product_names_from_items(items),
source_platform_id=payload.source_platform_id,
source_platform_name=payload.source_platform_name,
source_package=payload.source_package,
# store_name / source_platform_id / source_platform_name / source_package 统一由
# derived 提供(见上方两分支补齐), 不在此显式写 —— 否则与 _derive_from_platforms 撞键。
information=payload.information,
best_deeplink=payload.best_deeplink,
trace_url=payload.trace_url,
@@ -214,6 +234,7 @@ def upsert_record(
skipped_dish_count=payload.skipped_dish_count,
items=items,
comparison_results=[r.model_dump() for r in payload.comparison_results],
platforms=list(payload.platforms or []),
skipped_dish_names=list(payload.skipped_dish_names),
# 客户端环境 / 性能(debug,客户端上报;旧客户端为 None)
device_model=payload.device_model,
@@ -286,7 +307,8 @@ def upsert_record(
def _derive_from_results(
results: list[dict], platform_results: dict | None = None
results: list[dict], platform_results: dict | None = None,
record_status: str | None = None,
) -> dict:
"""从 done 帧 comparison_results(pricebot 原始 dict 列表)派生结构化列。
等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象
@@ -334,7 +356,63 @@ def _derive_from_results(
"saved_amount_cents": saved_amount_cents,
"is_source_best": best.get("is_source") if best else None,
"store_name": (src_row or {}).get("store_name") or None,
"status": "success" if has_valid_target else "failed",
# 记录级结局: 优先用 pricebot 下发的 record_status(区分 below_minimum/store_closed,
# 不再把"未满起送"塌缩成 failed → 记录页不再误报"网络开小差"); 旧 pricebot 未下发时
# 回退老的 success/failed 二态派生, 向后兼容。
"status": record_status or ("success" if has_valid_target else "failed"),
}
def _derive_from_platforms(
platforms: list, record_status: str | None = None,
) -> dict:
"""从 done 帧 platforms(每平台一行、渲染就绪)派生结构化列——**单一真相源**。
best_* 直接取 platforms is_best 的那一行source_* role=source ,与前端读的
platforms 天然一致(不再像 _derive_from_results 那样从 comparison_results 二次评最优,
消除"标量列 vs platforms"双源不一致)platforms 非空时优先走这里; pricebot
platforms 时调用方回退 _derive_from_results(向后兼容)"""
rows = [p for p in (platforms or []) if isinstance(p, dict)]
src = next((p for p in rows if p.get("role") == "source"), None)
best = next((p for p in rows if p.get("is_best")), None)
if best is None:
# pricebot 没标 is_best(如全平台 has_dish_diff「相似替换/仅供参考」→ 不认定权威最低价)
# 但仍有有价行 → 兜底取有价行里最低价当参考 best,避免记录级 best_*/saved 整条落 NULL
# (否则首页价 0.00 / 记录页无最低红框 / 省额丢失 /「累计发现可省」漏计)。**候选含源**:
# 源常年全菜、价可信,源本身最便宜时 best 回落源(saved=0、is_source_best=True),与 _derive
# 「全目标缺菜回落源、不虚报省」同一语义;若排除源强选更贵目标,saved 会变负、倒扣「累计
# 发现可省」(get_stats 对 saved_amount_cents 求和不带 >0 过滤)。
priced = [p for p in rows if p.get("price") is not None]
if priced:
best = min(priced, key=lambda p: (p["price"], p.get("display_order") or 0))
source_price_cents = _yuan_to_cents(src.get("price")) if src else None
best_price_cents = _yuan_to_cents(best.get("price")) if best else None
saved_amount_cents = None
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"),
}
@@ -366,6 +444,8 @@ def reserve_daily_start(
business_type: str = "food",
device_id: str | None = None,
now: datetime | None = None,
limit: int | None = DAILY_COMPARE_START_LIMIT,
reset_at: datetime | None = None,
) -> tuple[ComparisonRecord, int]:
"""Atomically reserve one of a user's 100 Beijing-day comparison starts.
@@ -391,6 +471,11 @@ def reserve_daily_start(
if existing_at.tzinfo is not None:
existing_at = existing_at.astimezone(CN_TZ).replace(tzinfo=None)
day_start = existing_at.replace(hour=0, minute=0, second=0, microsecond=0)
if reset_at is not None:
reset_start = reset_at
if reset_start.tzinfo is not None:
reset_start = reset_start.astimezone(CN_TZ).replace(tzinfo=None)
day_start = max(day_start, reset_start)
day_end = day_start + timedelta(days=1)
used = db.scalar(
select(func.count(ComparisonRecord.id)).where(
@@ -405,6 +490,11 @@ def reserve_daily_start(
if current.tzinfo is not None:
current = current.astimezone(CN_TZ).replace(tzinfo=None)
day_start = current.replace(hour=0, minute=0, second=0, microsecond=0)
if reset_at is not None:
reset_start = reset_at
if reset_start.tzinfo is not None:
reset_start = reset_start.astimezone(CN_TZ).replace(tzinfo=None)
day_start = max(day_start, reset_start)
day_end = day_start + timedelta(days=1)
used = db.scalar(
select(func.count(ComparisonRecord.id)).where(
@@ -413,7 +503,7 @@ def reserve_daily_start(
ComparisonRecord.created_at < day_end,
)
) or 0
if used >= DAILY_COMPARE_START_LIMIT:
if limit is not None and used >= limit:
raise DailyCompareStartLimitExceeded
rec = ComparisonRecord(
@@ -493,7 +583,29 @@ def harvest_done(
返回 (记录, 是否本次****落成 success)供调用方据此幂等发一次邀请奖
行不存在(理论上帧0已建;防御)则新建"""
results = done_params.get("comparison_results") or []
derived = _derive_from_results(results, done_params.get("platform_results"))
# 展示模型统一数组(pricebot 新增, 每平台一行自带 status/is_best): 原样存, 记录页据此直渲染。
# record_status: 记录级结局(success/below_minimum/store_closed/failed), 覆盖老二态派生。
platforms = done_params.get("platforms") or []
record_status = done_params.get("record_status")
# 单源派生: platforms(含 pricebot 权威 is_best)是唯一真相源, best_*/source_*/saved/status
# 全从它取 → 与前端读的 platforms 天然一致; 菜品也取 platforms 源行。老 pricebot 无
# platforms 时回退从 comparison_results 派生(向后兼容)。
if platforms:
derived = _derive_from_platforms(platforms, record_status)
# 菜品优先源行; recompare 无源行 → 回退 best 行 → 首个有菜品的行(同 store_name 回退)
_item_row = (
next((p for p in platforms if isinstance(p, dict) and p.get("role") == "source"), None)
or next((p for p in platforms if isinstance(p, dict) and p.get("is_best")), None)
or next((p for p in platforms if isinstance(p, dict) and p.get("items")), None)
)
items = (_item_row or {}).get("items") or []
else:
derived = _derive_from_results(
results, done_params.get("platform_results"), record_status
)
# pricebot 已把源单菜品塞进 comparison_results[源行].items
items = next((r.get("items") or [] for r in results if r.get("is_source")), [])
# 失败展示原因(#189): platforms / results 两个派生分支的 status 都可能 failed, 统一在此算
fail_reason = (
_derive_fail_display(
done_params.get("information"), done_params.get("platform_results")
@@ -501,8 +613,6 @@ 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,
@@ -514,6 +624,7 @@ def harvest_done(
skipped_dish_count=done_params.get("skipped_dish_count"),
skipped_dish_names=list(done_params.get("skipped_dish_names") or []),
comparison_results=results,
platforms=platforms,
items=items,
product_names=_product_names_from_items(items),
raw_payload=done_params,
+463 -184
View File
@@ -1,95 +1,136 @@
"""新手引导视频:运营配置读写 + 播放计次 + 发币。
**配置**(开关 / 视频地址 / 前几次 / 每次金币)整体作为一个 JSON 存进通用 app_config
(key=coupon_guide_video),写法完全对齐 feedback_qr 不进 CONFIG_DEFS,所以不会污染
系统配置页的通用列表,由本模块独占维护
**计次**按账号(user_id)**开播即计数**:客户端每次要展示领券等候浮层时调
`/api/v1/guide-video/start`,命中则当场写一行 guide_video_play(status='playing')
已用次数 = 该账号的行数,达到 max_plays 后不再下发,客户端改放广告(原逻辑)
COUNT 判定本身无锁,真正卡住次数上限的是 (user_id, seq) 唯一键:并发 /start 只能成一个
**发币**幂等键是 play_token,落地方式是 `status='playing' 'granted'` **条件更新**:
同一次播放重复上报只入账一次(网络重试 / 关闭与播完同时触发都靠它挡住)
中途关闭也照发 产品拍板中途关闭也算看完
两处都是直接铸币的路径,改动前先看 `start_play` / `grant_play` 上的并发注释
"""
"""双场景引导视频配置与 prepare/start/reward 状态机。"""
from __future__ import annotations
import uuid
from datetime import datetime
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any
from sqlalchemy import func, select, update
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from app.core import rewards
from app.core import limit_policy, media, rewards
from app.core.config import settings
from app.models.app_config import AppConfig
from app.models.guide_video import GuideVideoPlay
from app.models.wallet import CoinAccount
from app.repositories import wallet as crud_wallet
_KEY = "coupon_guide_video"
#: 金币流水 biz_type。客户端收益明细按它显示「新手引导视频奖励」。
BIZ_TYPE = "guide_video"
# 默认值 = 「运营还没配」时的行为:video_url 为空 → 一律不下发引导视频,浮层维持现状(放广告)。
# 所以本功能上线后**不配视频就等于没上线**,不会影响存量用户。
_DEFAULTS: dict[str, Any] = {
"enabled": True,
"video_url": None, # None/空 = 未配片 → 不下发,浮层照旧放广告
"max_plays": 3, # 每个账号前 N 次浮层放引导视频
"reward_coin": 120, # 每次固定金币
SCENES = ("coupon", "comparison")
_KEY_BY_SCENE = {
"coupon": "coupon_guide_video",
"comparison": "comparison_guide_video",
}
_FIELDS = tuple(_DEFAULTS.keys())
# 后台可配范围的护栏:防手滑把次数/金币填成天文数字(配置直接决定发币)。
BIZ_TYPE = "guide_video"
CIRCLE_COUNT = 10
PLAN_TTL = timedelta(minutes=10)
# 0 表示全局暂停播放;白名单页与旧引导视频配置页必须接受同一口径,
# 否则在白名单页设为 0 后,旧页面连金币/开关等无关字段也无法保存。
MIN_PLAYS = 0
MAX_PLAYS_LIMIT = 50
MIN_REWARD_COIN = 10
REWARD_COIN_LIMIT = 10_000
_DEFAULTS: dict[str, Any] = {
"enabled": False,
"video_url": None,
"max_plays": 3,
"reward_coin": 100,
"duration_ms": None,
"video_codec": None,
"audio_codec": None,
"analysis_status": "missing",
"analysis_error": None,
"config_version": 0,
}
_FIELDS = tuple(_DEFAULTS)
# ===== 配置 =====
class PlayStateError(Exception):
def __init__(self, code: str, message: str, *, reprepare_required: bool = False):
super().__init__(message)
self.code = code
self.message = message
self.reprepare_required = reprepare_required
def detail(self) -> dict[str, Any]:
return {
"code": self.code,
"message": self.message,
"retryable": False,
"reprepare_required": self.reprepare_required,
}
def _now() -> datetime:
return datetime.now(rewards.CN_TZ).replace(tzinfo=None)
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 _merge(raw: Any) -> dict[str, Any]:
"""DB 里(可能不全的)dict 叠加到默认上,得到完整配置(4 个字段,无 updated_at)。"""
out = dict(_DEFAULTS)
if isinstance(raw, dict):
for k in _FIELDS:
v = raw.get(k)
if v is not None:
out[k] = v
for key in _FIELDS:
if key in raw and raw[key] is not None:
out[key] = raw[key]
return out
def get_config(db: Session) -> dict[str, Any]:
"""完整配置 + updated_at(admin 读 / 业务读共用)。"""
row = db.get(AppConfig, _KEY)
cfg = _merge(row.value if row is not None else None)
def _public_config(cfg: dict[str, Any]) -> dict[str, Any]:
duration = int(cfg.get("duration_ms") or 0)
reward = int(cfg.get("reward_coin") or 0)
return {
**cfg,
"duration_ms": duration or None,
"circle_count": CIRCLE_COUNT,
"circle_duration_ms": duration / CIRCLE_COUNT if duration else None,
"reward_per_circle": reward // CIRCLE_COUNT,
}
def get_config(db: Session, scene: str = "coupon") -> dict[str, Any]:
row = db.get(AppConfig, _config_key(scene))
cfg = _public_config(_merge(row.value if row is not None else None))
if scene == "coupon":
cfg["max_plays"] = limit_policy.resolve(
db, "guide.video.lifetime"
).global_limit
cfg["updated_at"] = row.updated_at.isoformat() if row is not None and row.updated_at else None
return cfg
def _write(db: Session, value: dict[str, Any], *, admin_id: int, commit: bool) -> dict[str, Any]:
"""整体覆写该行(value 须为完整字段 dict),返回合并后的完整配置(含 updated_at)。"""
row = db.get(AppConfig, _KEY)
def _validate_settings(max_plays: int, reward_coin: int) -> None:
if not MIN_PLAYS <= max_plays <= MAX_PLAYS_LIMIT:
raise ValueError(f"播放次数必须在 {MIN_PLAYS}{MAX_PLAYS_LIMIT} 之间")
if not MIN_REWARD_COIN <= reward_coin <= REWARD_COIN_LIMIT:
raise ValueError(f"金币总价必须在 {MIN_REWARD_COIN}{REWARD_COIN_LIMIT} 之间")
if reward_coin % CIRCLE_COUNT:
raise ValueError("金币总价必须是 10 的倍数")
def _write(
db: Session, value: dict[str, Any], *, scene: str, admin_id: int, commit: bool
) -> dict[str, Any]:
key = _config_key(scene)
row = db.get(AppConfig, key)
if row is None:
row = AppConfig(key=_KEY, value=value, updated_by_admin_id=admin_id)
row = AppConfig(key=key, value=value, updated_by_admin_id=admin_id)
db.add(row)
else:
row.value = value # 整体重新赋值,SQLAlchemy 才侦测得到变更
row.value = value
row.updated_by_admin_id = admin_id
if commit:
db.commit()
db.refresh(row)
else:
db.flush()
out = _merge(row.value)
out["updated_at"] = row.updated_at.isoformat() if row.updated_at else None
return out
return get_config(db, scene)
def update_config(
@@ -98,193 +139,431 @@ def update_config(
enabled: bool | None = None,
max_plays: int | None = None,
reward_coin: int | None = None,
scene: str = "coupon",
admin_id: int,
commit: bool = True,
) -> tuple[dict[str, Any], dict[str, Any]]:
"""改开关 / 次数 / 金币(只改传了的字段;视频走 set_video)。返回 (before, after) 供审计。"""
row = db.get(AppConfig, _KEY)
before = _merge(row.value if row is not None else None)
new_value = {k: before[k] for k in _FIELDS}
before = get_config(db, scene)
raw = {key: before[key] for key in _FIELDS}
candidate_plays = int(max_plays if max_plays is not None else raw["max_plays"])
candidate_reward = int(reward_coin if reward_coin is not None else raw["reward_coin"])
_validate_settings(candidate_plays, candidate_reward)
if enabled is True and (
not raw.get("video_url")
or raw.get("analysis_status") != "valid"
or not raw.get("duration_ms")
):
raise ValueError("请先上传并通过分析的视频,再启用")
if enabled is not None:
new_value["enabled"] = enabled
if max_plays is not None:
new_value["max_plays"] = max(0, min(int(max_plays), MAX_PLAYS_LIMIT))
if reward_coin is not None:
new_value["reward_coin"] = max(0, min(int(reward_coin), REWARD_COIN_LIMIT))
after = _write(db, new_value, admin_id=admin_id, commit=commit)
raw["enabled"] = enabled
raw["max_plays"] = candidate_plays
raw["reward_coin"] = candidate_reward
raw["config_version"] = int(raw.get("config_version") or 0) + 1
after = _write(
db,
raw,
scene=scene,
admin_id=admin_id,
commit=False,
)
if scene == "coupon" and max_plays is not None:
# 兼容仍调用旧专用接口的客户端/脚本,并把统一策略全局值一并更新。
limit_policy.set_global_limits(
db,
{"guide.video.lifetime": candidate_plays},
admin_id=admin_id,
commit=False,
)
# set_global_limits 已 flush;即使由 admin 路由统一在外层提交,也要把
# 同一事务内的最新统一策略值写进审计 after 快照。
after = get_config(db, scene)
if commit:
db.commit()
after = get_config(db, scene)
return before, after
def set_video(
db: Session, video_url: str | None, *, admin_id: int, commit: bool = True
db: Session,
video_url: str | None,
*,
analysis: dict[str, Any] | None = None,
scene: str = "coupon",
admin_id: int,
commit: bool = True,
) -> tuple[dict[str, Any], dict[str, Any]]:
"""设置/清空引导视频地址。返回 (before, after);before['video_url'] 供调用方删旧文件。"""
row = db.get(AppConfig, _KEY)
before = _merge(row.value if row is not None else None)
new_value = {k: before[k] for k in _FIELDS}
new_value["video_url"] = video_url
after = _write(db, new_value, admin_id=admin_id, commit=commit)
before = get_config(db, scene)
raw = {key: before[key] for key in _FIELDS}
raw["video_url"] = video_url
if video_url:
if not analysis or analysis.get("analysis_status") != "valid":
raise ValueError("视频必须先完成服务端分析")
for key in (
"duration_ms", "video_codec", "audio_codec",
"analysis_status", "analysis_error",
):
raw[key] = analysis.get(key)
else:
raw.update(
duration_ms=None,
video_codec=None,
audio_codec=None,
analysis_status="missing",
analysis_error=None,
)
raw["enabled"] = False
raw["config_version"] = int(raw.get("config_version") or 0) + 1
after = _write(db, raw, scene=scene, admin_id=admin_id, commit=commit)
return before, after
# ===== 播放计次 =====
def used_plays(db: Session, user_id: int) -> int:
"""该账号已用掉的引导视频次数(开播即算,含未发币的)。"""
return int(
db.execute(
select(func.count()).select_from(GuideVideoPlay).where(
GuideVideoPlay.user_id == user_id
)
).scalar_one()
def used_plays(
db: Session,
user_id: int,
scene: str = "coupon",
*,
reset_at: datetime | None = None,
) -> int:
stmt = select(func.count()).select_from(GuideVideoPlay).where(
GuideVideoPlay.user_id == user_id,
GuideVideoPlay.scene == scene,
GuideVideoPlay.status != "prepared",
)
if reset_at is not None:
# guide_video_play 使用北京时间 naive 墙钟;白名单重置点是带时区时间。
reset_value = reset_at.astimezone(rewards.CN_TZ).replace(tzinfo=None)
stmt = stmt.where(GuideVideoPlay.started_at >= reset_value)
return int(db.execute(stmt).scalar_one())
def play_stats(db: Session) -> dict[str, int]:
"""全站播放统计(admin 页展示):总播放次数 / 其中已发币次数。"""
total = int(
db.execute(select(func.count()).select_from(GuideVideoPlay)).scalar_one()
def _effective_quota(
db: Session,
user_id: int,
scene: str,
cfg: dict[str, Any],
) -> tuple[int | None, datetime | None]:
if scene != "coupon":
return int(cfg.get("max_plays") or 0), None
policy = limit_policy.resolve_for_user(
db, "guide.video.lifetime", user_id
)
granted = int(
db.execute(
select(func.count()).select_from(GuideVideoPlay).where(
GuideVideoPlay.status == "granted"
)
).scalar_one()
)
return {"total_plays": total, "granted_plays": granted}
return policy.limit, policy.reset_at
def start_play(
db: Session, user_id: int, *, scene: str = "coupon", commit: bool = True
def _remaining(maximum: int | None, used: int) -> int:
# API 字段保持整数兼容;不限时使用足够大的展示值,不参与服务端判定。
exposed_maximum = maximum if maximum is not None else 1_000_000
return max(0, exposed_maximum - used)
def _prepare_miss(
scene: str,
reason: str,
cfg: dict[str, Any],
used: int,
maximum: int | None,
) -> dict[str, Any]:
"""决定这次浮层是否放引导视频;命中则**当场计次**并返回 play_token。
return {
"should_play": False,
"reason": reason,
"scene": scene,
"video_url": None,
"play_token": "",
"config_version": int(cfg.get("config_version") or 0),
"duration_ms": 0,
"circle_count": CIRCLE_COUNT,
"circle_duration_ms": 0,
"reward_coin": int(cfg.get("reward_coin") or 0),
"reward_per_circle": int(cfg.get("reward_coin") or 0) // CIRCLE_COUNT,
"seq": used,
"remaining": _remaining(maximum, used),
"expires_at": None,
}
返回 dict:
should_play 是否放引导视频(False 客户端照旧放广告)
video_url 相对地址(/media/...);客户端自行拼 BASE_URL
play_token 发币幂等键(should_play=False 时为空串)
reward_coin 播完/中途关闭都发的固定金币
seq / remaining 第几次 / 发完这次还剩几次(仅展示与排查用)
"""
cfg = get_config(db)
video_url = (cfg.get("video_url") or "").strip()
max_plays = int(cfg.get("max_plays") or 0)
reward_coin = int(cfg.get("reward_coin") or 0)
used = used_plays(db, user_id)
def _miss(used_now: int) -> dict[str, Any]:
return {
"should_play": False,
"video_url": None,
"play_token": "",
"reward_coin": reward_coin,
"seq": used_now,
"remaining": max(0, max_plays - used_now),
}
def prepare_play(db: Session, user_id: int, *, scene: str = "coupon") -> dict[str, Any]:
cfg = get_config(db, scene)
maximum, reset_at = _effective_quota(db, user_id, scene, cfg)
used = used_plays(db, user_id, scene, reset_at=reset_at)
video_url = str(cfg.get("video_url") or "").strip()
duration = int(cfg.get("duration_ms") or 0)
if not cfg.get("enabled"):
return _prepare_miss(scene, "disabled", cfg, used, maximum)
if not video_url or cfg.get("analysis_status") != "valid" or duration <= 0:
return _prepare_miss(
scene, "video_unavailable", cfg, used, maximum
)
if maximum is not None and used >= maximum:
return _prepare_miss(
scene, "play_limit_reached", cfg, used, maximum
)
if not cfg.get("enabled") or not video_url or max_plays <= 0 or used >= max_plays:
return _miss(used)
seq = used + 1
now = _now()
play = GuideVideoPlay(
user_id=user_id,
play_token=uuid.uuid4().hex,
scene=scene,
seq=seq,
seq=None,
video_url=video_url,
coin=0,
status="playing",
coin=int(cfg["reward_coin"]),
duration_ms=duration,
config_version=int(cfg["config_version"]),
settled_circles=0,
status="prepared",
completed=0,
started_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
prepared_at=now,
expires_at=now + PLAN_TTL,
started_at=None,
)
db.add(play)
# 上面的 COUNT 判定是无锁 check-then-insert:并发 /start 会都算出同一个 seq。
# (user_id, seq) 唯一键让只有一个能落库,其余撞键 → 回滚后按"这次不放视频"降级,
# 客户端照旧走广告链路。没有它,并发就能绕过 max_plays 无限刷金币。
try:
if commit:
db.commit()
else:
db.flush()
except IntegrityError:
db.rollback()
return _miss(used_plays(db, user_id))
db.commit()
return {
"should_play": True,
"reason": "ready",
"scene": scene,
"video_url": video_url,
"play_token": play.play_token,
"reward_coin": reward_coin,
"seq": seq,
"remaining": max(0, max_plays - seq),
"config_version": play.config_version,
"duration_ms": duration,
"circle_count": CIRCLE_COUNT,
"circle_duration_ms": duration / CIRCLE_COUNT,
"reward_coin": play.coin,
"reward_per_circle": play.coin // CIRCLE_COUNT,
"seq": used + 1,
"remaining": _remaining(maximum, used + 1),
"expires_at": play.expires_at.isoformat(),
}
def _find_play(db: Session, user_id: int, token: str) -> GuideVideoPlay | None:
"""按 (play_token, user_id) 取播放行 —— 带 user_id 是防拿别人的 token 来兑。"""
return db.execute(
select(GuideVideoPlay).where(
GuideVideoPlay.play_token == token,
GuideVideoPlay.play_token == token.strip(),
GuideVideoPlay.user_id == user_id,
)
).scalar_one_or_none()
def grant_play(
db: Session, user_id: int, *, play_token: str, completed: bool
def _start_out(
play: GuideVideoPlay,
maximum: int | None,
used: int,
status: str,
) -> dict[str, Any]:
"""按 play_token 发这次引导视频的金币(幂等)。播完 / 中途关闭都发。
assert play.started_at is not None and play.seq is not None and play.video_url
return {
"started": True,
"status": status,
"play_token": play.play_token,
"scene": play.scene,
"video_url": play.video_url,
"config_version": play.config_version,
"duration_ms": play.duration_ms,
"circle_count": CIRCLE_COUNT,
"circle_duration_ms": play.duration_ms / CIRCLE_COUNT,
"reward_coin": play.coin,
"reward_per_circle": play.coin // CIRCLE_COUNT,
"seq": play.seq,
"remaining": _remaining(maximum, used),
"started_at": play.started_at.isoformat(),
}
返回 {granted, coin, status}:granted=True 表示**本次调用真的入账了**;
重复上报返回 granted=False + 已发金币(客户端据此不重复累加 toast 金额)
"""
token = (play_token or "").strip()
# 金币额度以**服务端配置**为准,不信客户端(客户端只上报"播完/关闭")。
coin = int(get_config(db).get("reward_coin") or 0)
# 幂等核心:把 status 放进 WHERE 做条件更新(compare-and-set),而不是"先读再判再写"。
# 「播完」与「✕ 关闭」抢跑、或客户端超时重试时,两个请求会都读到 status='playing',
# 无锁的话就都往下发币、都 commit,金币入账两次(不用恶意,重试就会中招)。改成条件更新后
# 并发里只有一条 rowcount=1,另一条拿 0 → 按已发返回,不二次铸币。
# (PG READ COMMITTED 下后到的 UPDATE 阻塞到对手提交,再按新版本重判 status;SQLite 写串行。)
#
# 别指望 IntegrityError 兜底:这里只 UPDATE 不 INSERT,撞不到 uq_guide_video_play_token;
# 而 biz_type='guide_video' 的金币流水也不在 ux_coin_transaction_task_ref 的谓词
# (biz_type LIKE 'task%')覆盖范围内 —— 两个唯一键在这条路径上都是不生效的。
def start_play(db: Session, user_id: int, *, play_token: str) -> dict[str, Any]:
play = _find_play(db, user_id, play_token)
if play is None:
raise PlayStateError("play_not_found", "播放计划不存在")
cfg = get_config(db, play.scene)
maximum, reset_at = _effective_quota(
db, user_id, play.scene, cfg
)
used = used_plays(
db, user_id, play.scene, reset_at=reset_at
)
if play.status in {"started", "completed"}:
return _start_out(play, maximum, used, "already_started")
if play.status != "prepared":
raise PlayStateError("play_not_found", "播放计划不可用")
now = _now()
if play.expires_at is None or now > play.expires_at:
raise PlayStateError(
"plan_expired", "播放计划已过期,请重新获取",
reprepare_required=True,
)
if int(cfg["config_version"]) != play.config_version:
raise PlayStateError(
"config_changed", "视频配置已变化,请重新获取",
reprepare_required=True,
)
if maximum is not None and used >= maximum:
raise PlayStateError("play_limit_reached", "播放次数已用完")
# seq 在数据库中是账号+场景生命周期唯一值;白名单重置只重置额度,
# 不能从 1 重新编号,否则会与历史记录冲突。
play.seq = used_plays(db, user_id, play.scene) + 1
play.status = "started"
play.started_at = now
try:
db.commit()
db.refresh(play)
except IntegrityError as exc:
db.rollback()
raise PlayStateError("play_limit_reached", "并发起播冲突,请重新获取") from exc
return _start_out(play, maximum, used + 1, "started")
def _coin_balance(db: Session, user_id: int) -> int:
account = db.get(CoinAccount, user_id)
return int(account.coin_balance if account else 0)
def grant_circle(
db: Session, user_id: int, *, play_token: str, circle: int
) -> dict[str, Any]:
play = _find_play(db, user_id, play_token)
balance = _coin_balance(db, user_id)
base = {
"granted": False,
"retryable": False,
"retry_after_ms": 0,
"circle": circle,
"granted_coin": 0,
"settled_circles": int(play.settled_circles if play else 0),
"coin_balance": balance,
}
if play is None:
return {**base, "status": "not_found"}
if play.status not in {"started", "completed"} or play.started_at is None:
return {**base, "status": "not_started"}
settled = int(play.settled_circles)
if circle <= settled:
return {
**base,
"status": "already_granted",
"granted_coin": play.coin // CIRCLE_COUNT,
"settled_circles": settled,
}
if settled >= CIRCLE_COUNT:
return {**base, "status": "finished", "settled_circles": settled}
if circle != settled + 1:
return {
**base,
"status": "out_of_order",
"retryable": True,
"settled_circles": settled,
}
# 向上取整到毫秒,绝不让第 N 圈早于精确 N/10 边界发奖。
boundary_ms = (circle * play.duration_ms + CIRCLE_COUNT - 1) // CIRCLE_COUNT
elapsed_ms = max(0, int((_now() - play.started_at).total_seconds() * 1000))
if elapsed_ms < boundary_ms:
return {
**base,
"status": "too_early",
"retryable": True,
"retry_after_ms": boundary_ms - elapsed_ms,
"settled_circles": settled,
}
final = circle == CIRCLE_COUNT
won = db.execute(
update(GuideVideoPlay)
.where(
GuideVideoPlay.play_token == token,
GuideVideoPlay.user_id == user_id,
GuideVideoPlay.status == "playing",
GuideVideoPlay.id == play.id,
GuideVideoPlay.status == "started",
GuideVideoPlay.settled_circles == settled,
)
.values(
status="granted",
coin=coin,
completed=1 if completed else 0,
granted_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
settled_circles=circle,
status="completed" if final else "started",
completed=1 if final else 0,
granted_at=_now() if final else None,
)
.execution_options(synchronize_session=False)
).rowcount
if not won:
# 没抢到:token 不存在 / 不是本人的 / 已被另一次上报发过。回滚拿干净快照再区分两者
# (对手此时必然已提交 —— 我们就是被它挡下的,所以读得到它写的 coin)。
db.rollback()
play = _find_play(db, user_id, token)
if play is None:
return {"granted": False, "coin": 0, "status": "not_found"}
return {"granted": False, "coin": play.coin, "status": "already_granted"}
fresh = _find_play(db, user_id, play_token)
fresh_settled = int(fresh.settled_circles if fresh else 0)
return {
**base,
"status": "already_granted" if circle <= fresh_settled else "out_of_order",
"retryable": circle > fresh_settled,
"granted_coin": (
play.coin // CIRCLE_COUNT if circle <= fresh_settled else 0
),
"settled_circles": fresh_settled,
"coin_balance": _coin_balance(db, user_id),
}
if coin > 0:
crud_wallet.grant_coins(
db,
user_id,
coin,
biz_type=BIZ_TYPE,
ref_id=token,
remark="新手引导视频奖励",
)
db.commit()
return {"granted": True, "coin": coin, "status": "granted"}
per_circle = play.coin // CIRCLE_COUNT
account, _ = crud_wallet.grant_coins(
db,
user_id,
per_circle,
biz_type=BIZ_TYPE,
ref_id=f"{play.play_token}:{circle}",
remark=f"新手引导视频{circle}奖励",
)
try:
db.commit()
except IntegrityError:
# 流水唯一键是第二道幂等保险;若撞键,回滚后返回权威状态。
db.rollback()
fresh = _find_play(db, user_id, play_token)
return {
**base,
"status": "already_granted",
"granted_coin": play.coin // CIRCLE_COUNT,
"settled_circles": int(fresh.settled_circles if fresh else settled),
"coin_balance": _coin_balance(db, user_id),
}
return {
**base,
"granted": True,
"status": "granted",
"granted_coin": per_circle,
"settled_circles": circle,
"coin_balance": int(account.coin_balance),
}
def cleanup_old_videos(db: Session, *, protection: timedelta = PLAN_TTL) -> int:
"""清理超过保护期且未被当前配置/有效计划/近期起播引用的托管旧片。"""
now = _now()
protected_urls: set[str] = set()
for scene in SCENES:
url = get_config(db, scene).get("video_url")
if url:
protected_urls.add(str(url))
protected_urls.update(
str(url)
for url in db.execute(
select(GuideVideoPlay.video_url).where(
GuideVideoPlay.video_url.is_not(None),
(
(
(GuideVideoPlay.status == "prepared")
& (GuideVideoPlay.expires_at >= now)
)
| (
(GuideVideoPlay.status == "started")
& (GuideVideoPlay.started_at >= now - protection)
)
),
)
).scalars()
if url
)
directory = Path(settings.MEDIA_ROOT) / "guide_video"
if not directory.is_dir():
return 0
cutoff = now.timestamp() - protection.total_seconds()
removed = 0
for path in directory.glob("guide_*.mp4"):
url = f"{settings.MEDIA_URL_PREFIX}/guide_video/{path.name}"
try:
if url not in protected_urls and path.stat().st_mtime <= cutoff:
media.delete_guide_video(url)
removed += 1
except OSError:
continue
return removed
+29 -7
View File
@@ -2,7 +2,7 @@
from __future__ import annotations
import math
from datetime import datetime, timedelta, timezone
from datetime import UTC, datetime, timedelta
from sqlalchemy import func, select
from sqlalchemy.orm import Session
@@ -10,9 +10,22 @@ from sqlalchemy.orm import Session
from app.models.phone_rebind_log import PhoneRebindLog
def rebound_within_days(db: Session, phone: str, days: int) -> bool:
def rebound_within_days(
db: Session,
phone: str,
days: int,
*,
reset_at: datetime | None = None,
) -> bool:
"""该手机号在最近 days 天内是否换绑过(命中 → 禁止再次换绑)。"""
since = datetime.now(timezone.utc) - timedelta(days=days)
since = datetime.now(UTC) - timedelta(days=days)
if reset_at is not None:
reset_value = (
reset_at.replace(tzinfo=UTC)
if reset_at.tzinfo is None
else reset_at.astimezone(UTC)
)
since = max(since, reset_value)
stmt = (
select(PhoneRebindLog.id)
.where(PhoneRebindLog.phone == phone, PhoneRebindLog.rebound_at >= since)
@@ -21,16 +34,25 @@ def rebound_within_days(db: Session, phone: str, days: int) -> bool:
return db.execute(stmt).first() is not None
def remaining_block_days(db: Session, phone: str, days: int) -> int:
def remaining_block_days(
db: Session,
phone: str,
days: int,
*,
reset_at: datetime | None = None,
) -> int:
"""距离该手机号可再次换绑还剩几天(向上取整;无记录返回 0)。"""
conditions = [PhoneRebindLog.phone == phone]
if reset_at is not None:
conditions.append(PhoneRebindLog.rebound_at >= reset_at)
last = db.execute(
select(func.max(PhoneRebindLog.rebound_at)).where(PhoneRebindLog.phone == phone)
select(func.max(PhoneRebindLog.rebound_at)).where(*conditions)
).scalar_one_or_none()
if last is None:
return 0
if last.tzinfo is None: # SQLite 取回 naive datetime,按 UTC 归一
last = last.replace(tzinfo=timezone.utc)
remaining = (last + timedelta(days=days) - datetime.now(timezone.utc)).total_seconds()
last = last.replace(tzinfo=UTC)
remaining = (last + timedelta(days=days) - datetime.now(UTC)).total_seconds()
return max(0, math.ceil(remaining / 86400))
+92 -37
View File
@@ -9,15 +9,10 @@ from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from app.core.config_schema import (
RISK_COMPARE_DAILY_THRESHOLD_KEY,
RISK_ONECLICK_DAILY_THRESHOLD_KEY,
RISK_SMS_HOURLY_THRESHOLD_KEY,
)
from app.core import limit_policy
from app.models.app_config import AppConfig
from app.models.comparison import ComparisonRecord
from app.models.risk import BehaviorEvent, RiskIncident, SubjectRestriction
from app.repositories import app_config
CN_TZ = ZoneInfo("Asia/Shanghai")
@@ -41,7 +36,6 @@ class RuleSpec:
code: str
event_type: str
subject_type: str
threshold_key: str
window: str
count_outcomes: tuple[str, ...]
@@ -51,7 +45,6 @@ RULES: dict[str, RuleSpec] = {
code=RULE_SMS_HOURLY,
event_type=EVENT_SMS_SEND,
subject_type="device",
threshold_key=RISK_SMS_HOURLY_THRESHOLD_KEY,
window="hour",
count_outcomes=("success",),
),
@@ -59,22 +52,23 @@ RULES: dict[str, RuleSpec] = {
code=RULE_ONECLICK_DAILY,
event_type=EVENT_ONECLICK_LOGIN,
subject_type="device",
threshold_key=RISK_ONECLICK_DAILY_THRESHOLD_KEY,
window="day",
count_outcomes=("success", "failed"),
),
}
RULE_THRESHOLD_KEYS: dict[str, str] = {
RULE_SMS_HOURLY: RISK_SMS_HOURLY_THRESHOLD_KEY,
RULE_ONECLICK_DAILY: RISK_ONECLICK_DAILY_THRESHOLD_KEY,
RULE_COMPARE_DAILY: RISK_COMPARE_DAILY_THRESHOLD_KEY,
RISK_LIMIT_RULE_CODES: dict[str, str] = {
RULE_SMS_HOURLY: "risk.sms.hourly",
RULE_ONECLICK_DAILY: "risk.oneclick.daily",
RULE_COMPARE_DAILY: "risk.compare.daily",
}
def get_rule_threshold(db: Session, rule_code: str) -> int:
"""读取规则当前阈值;配置表为空时回退上线前的 5/20/100 默认值。"""
return int(app_config.get_value(db, RULE_THRESHOLD_KEYS[rule_code]))
return limit_policy.resolve(
db, RISK_LIMIT_RULE_CODES[rule_code]
).global_limit
def utcnow() -> datetime:
@@ -257,22 +251,34 @@ def _upsert_incident(
def evaluate_behavior_rule(
db: Session, *, rule_code: str, subject_id: str, at: datetime
db: Session,
*,
rule_code: str,
subject_id: str,
at: datetime,
threshold: int | None = None,
subject_reset_at: datetime | None = None,
) -> RiskIncident | None:
spec = RULES[rule_code]
threshold = get_rule_threshold(db, rule_code)
effective_threshold = threshold or get_rule_threshold(db, rule_code)
window_key, window_start, end = _window_bounds(at, spec.window)
reset_at = get_rule_reset_at(db, rule_code)
start = max(window_start, reset_at) if reset_at else window_start
baselines = [value for value in (reset_at, subject_reset_at) if value is not None]
start = max(window_start, *baselines) if baselines else window_start
count, first_at, last_at, triggered_at = _event_stats(
db,
spec,
subject_id=subject_id,
start=start,
end=end,
threshold=threshold,
threshold=effective_threshold,
)
if count < threshold or first_at is None or last_at is None or triggered_at is None:
if (
count < effective_threshold
or first_at is None
or last_at is None
or triggered_at is None
):
return None
return _upsert_incident(
db,
@@ -300,7 +306,6 @@ def reconcile_behavior_rule(
"""按当前阈值重算短信/一键登录当前窗口,并收起已不再命中的待处理告警。"""
spec = RULES[rule_code]
current = at or utcnow()
threshold = get_rule_threshold(db, rule_code)
window_key, window_start, end = _window_bounds(current, spec.window)
reset_at = get_rule_reset_at(db, rule_code)
start = max(window_start, reset_at) if reset_at else window_start
@@ -311,23 +316,39 @@ def reconcile_behavior_rule(
BehaviorEvent.occurred_at >= start,
BehaviorEvent.occurred_at < end,
)
qualifying_rows = db.execute(
subject_rows = db.execute(
select(
BehaviorEvent.subject_id,
func.max(BehaviorEvent.occurred_at),
func.max(BehaviorEvent.phone),
)
.where(*filters)
.group_by(BehaviorEvent.subject_id)
.having(func.count(BehaviorEvent.id) >= threshold)
).all()
qualifying = {str(subject_id) for subject_id, _ in qualifying_rows}
for subject_id, last_at in qualifying_rows:
evaluate_behavior_rule(
qualifying: set[str] = set()
policy_code = {
RULE_SMS_HOURLY: "risk.sms.hourly",
RULE_ONECLICK_DAILY: "risk.oneclick.daily",
}[rule_code]
for subject_id, last_at, phone in subject_rows:
policy = limit_policy.resolve(
db,
policy_code,
phone=phone,
device=str(subject_id),
)
if policy.suppressed or policy.limit is None:
continue
incident = evaluate_behavior_rule(
db,
rule_code=rule_code,
subject_id=str(subject_id),
at=last_at or current,
threshold=policy.limit,
subject_reset_at=policy.reset_at,
)
if incident is not None:
qualifying.add(str(subject_id))
open_incidents = db.scalars(
select(RiskIncident).where(
@@ -383,7 +404,29 @@ def record_behavior_event(
db.add(event)
db.flush()
if evaluate_rule:
evaluate_behavior_rule(db, rule_code=evaluate_rule, subject_id=subject_id, at=at)
policy_code = {
RULE_SMS_HOURLY: "risk.sms.hourly",
RULE_ONECLICK_DAILY: "risk.oneclick.daily",
}.get(evaluate_rule)
policy = (
limit_policy.resolve(
db,
policy_code,
phone=phone,
device=device_id or subject_id,
)
if policy_code
else None
)
if policy is None or not policy.suppressed:
evaluate_behavior_rule(
db,
rule_code=evaluate_rule,
subject_id=subject_id,
at=at,
threshold=policy.limit if policy else None,
subject_reset_at=policy.reset_at if policy else None,
)
if commit:
db.commit()
db.refresh(event)
@@ -396,20 +439,33 @@ def sync_compare_incident(
user_id: int,
at: datetime,
threshold: int | None = None,
device_id: str | None = None,
commit: bool = True,
) -> RiskIncident | None:
effective_threshold = threshold or get_rule_threshold(db, RULE_COMPARE_DAILY)
policy = limit_policy.resolve_for_user(
db,
"risk.compare.daily",
user_id,
device=device_id,
)
if policy.suppressed:
return None
effective_threshold = threshold or policy.limit or get_rule_threshold(
db, RULE_COMPARE_DAILY
)
# comparison_record 的既有写入口统一落“北京时间 naive”时间;这里必须沿用同一
# 口径,否则 SQLite/PG session timezone 不同时会把凌晨记录算到前一天。
local = at.astimezone(CN_TZ).replace(tzinfo=None) if at.tzinfo else at
window_start = local.replace(hour=0, minute=0, second=0, microsecond=0)
end = window_start + timedelta(days=1)
window_key = window_start.strftime("%Y-%m-%d")
reset_at = get_rule_reset_at(db, RULE_COMPARE_DAILY)
reset_local = (
reset_at.astimezone(CN_TZ).replace(tzinfo=None) if reset_at else None
)
start = max(window_start, reset_local) if reset_local else window_start
global_reset_at = get_rule_reset_at(db, RULE_COMPARE_DAILY)
baselines = [
value.astimezone(CN_TZ).replace(tzinfo=None)
for value in (global_reset_at, policy.reset_at)
if value is not None
]
start = max(window_start, *baselines) if baselines else window_start
filters = (
ComparisonRecord.user_id == user_id,
ComparisonRecord.created_at >= start,
@@ -464,7 +520,6 @@ def reconcile_compare_rule(
reset_at.astimezone(CN_TZ).replace(tzinfo=None) if reset_at else None
)
start = max(window_start, reset_local) if reset_local else window_start
threshold = get_rule_threshold(db, RULE_COMPARE_DAILY)
rows = db.execute(
select(
ComparisonRecord.user_id,
@@ -476,17 +531,17 @@ def reconcile_compare_rule(
ComparisonRecord.created_at < end,
)
.group_by(ComparisonRecord.user_id)
.having(func.count(ComparisonRecord.id) >= threshold)
).all()
qualifying = {str(user_id) for user_id, _ in rows}
qualifying: set[str] = set()
for user_id, last_at in rows:
sync_compare_incident(
incident = sync_compare_incident(
db,
user_id=int(user_id),
at=last_at or current,
threshold=threshold,
commit=False,
)
if incident is not None:
qualifying.add(str(user_id))
open_incidents = db.scalars(
select(RiskIncident).where(
+20 -7
View File
@@ -132,10 +132,20 @@ def get_or_create_account(
) -> CoinAccount:
"""取用户金币账户,不存在则建一个空账户。
lock=True 时对已存在的账户行加 SELECT FOR UPDATE(--写余额的调用方串行化,防并发
双写余额错位, admin set 模式连点);默认 False 不改 C 端发奖行为SQLite 下为 no-op
lock=True 时对已存在的账户行加 SELECT FOR UPDATE并强制刷新 identity map 中可能
缓存的旧余额所有金币写入口都会使用它SQLite FOR UPDATE no-op
"""
acc = db.get(CoinAccount, user_id, with_for_update=True) if lock else db.get(CoinAccount, user_id)
if lock:
# populate_existing 很关键:同一 Session 可能早已缓存旧余额。只加 FOR UPDATE 而
# 继续复用 identity map 里的旧对象,仍会在另一笔发奖提交后把余额覆盖回去。
acc = db.execute(
select(CoinAccount)
.where(CoinAccount.user_id == user_id)
.with_for_update()
.execution_options(populate_existing=True)
).scalar_one_or_none()
else:
acc = db.get(CoinAccount, user_id)
if acc is None:
acc = CoinAccount(
user_id=user_id,
@@ -165,7 +175,9 @@ def grant_coins(
返回 (account, transaction)调用方负责 commit
"""
acc = get_or_create_account(db, user_id, commit=False)
# 所有金币来源统一遵守账户行锁协议;否则 guide 锁了账户,广告/签到等未锁路径
# 仍可拿旧余额做 ORM 读改写,最终余额会小于流水累计。
acc = get_or_create_account(db, user_id, commit=False, lock=True)
acc.coin_balance += amount
if amount > 0:
acc.total_coin_earned += amount
@@ -198,7 +210,7 @@ def grant_cash(
[grant_coins] 同模式(运营手动调现金 / 测试发现金用)返回 (account, transaction),
调用方负责 commit不在此校验扣成负由调用方(admin router)按业务保护
"""
acc = get_or_create_account(db, user_id, commit=False)
acc = get_or_create_account(db, user_id, commit=False, lock=True)
acc.cash_balance_cents += amount_cents
txn = CashTransaction(
@@ -228,7 +240,7 @@ def grant_invite_cash(
invite_cash_transaction, commit与金币兑换的 cash_balance_cents **物理隔离**
(产品红线:邀请奖励金 金币现金,两本账不可累加)返回 (account, transaction),
调用方负责 commit不在此校验扣成负由调用方按业务保护"""
acc = get_or_create_account(db, user_id, commit=False)
acc = get_or_create_account(db, user_id, commit=False, lock=True)
acc.invite_cash_balance_cents += amount_cents
txn = InviteCashTransaction(
@@ -286,7 +298,8 @@ def exchange_coins_to_cash(
if coin_amount < floor_min or coin_amount % COIN_PER_CENT != 0:
raise InvalidExchangeAmountError
acc = get_or_create_account(db, user_id, commit=False)
# 余额充足校验必须和扣减遵守同一账户锁,否则并发消费都可能通过旧余额校验。
acc = get_or_create_account(db, user_id, commit=False, lock=True)
if acc.coin_balance < coin_amount:
raise InsufficientCoinError
+12 -2
View File
@@ -107,6 +107,13 @@ class ComparisonRecordIn(BaseModel):
# 明细
items: list[ComparisonItemIn] = Field(default_factory=list)
comparison_results: list[ComparisonResultIn] = Field(default_factory=list)
# 展示模型统一数组(pricebot done.params.platforms 原样透传): 每平台一行、自带
# status/is_best/display/display_order,记录页据此直渲染。宽松 list[dict] 存(结构由
# pricebot 定,server 只原样落库),前端读它、老记录空时回退 comparison_results。
platforms: list[dict] = Field(default_factory=list)
# 记录级结局(pricebot 下发): success/below_minimum/store_closed/failed。让"未满起送"不再
# 被塌缩成 failed。_derive 优先用它、其次客户端 status、再兜底二态派生。
record_status: str | None = None
# 逐平台结局摘要(含失败平台的细分原因 status: store_not_found/items_not_found/below_minimum/
# unsupported/...)。来自 done.params.platform_results,客户端透传;落 raw_payload(不单列),
# admin「卡在哪一步」从这里读。dict{platform_id: {...}} 宽松存(结构由 pricebot 定——是
@@ -177,6 +184,9 @@ 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。
@@ -212,9 +222,9 @@ class CompareStartReserveIn(BaseModel):
class CompareStartReserveOut(BaseModel):
limit: int
limit: int | None
used: int
remaining: int
remaining: int | None
class CompareStatsOut(BaseModel):
+56 -20
View File
@@ -1,36 +1,72 @@
"""新手引导视频(领券等候浮层前 N 次替代广告)的客户端请求/响应契约"""
from __future__ import annotations
"""客户端引导视频三阶段协议"""
from typing import Literal
from pydantic import BaseModel, Field
GuideScene = Literal["coupon", "comparison"]
class GuideVideoPrepareIn(BaseModel):
scene: GuideScene = "coupon"
class GuideVideoPrepareOut(BaseModel):
should_play: bool
reason: str
scene: GuideScene
video_url: str | None = None
play_token: str = ""
config_version: int = 0
duration_ms: int = 0
circle_count: int = 10
circle_duration_ms: float = 0
reward_coin: int = 0
reward_per_circle: int = 0
seq: int = 0
remaining: int = 0
expires_at: str | None = None
class GuideVideoStartIn(BaseModel):
"""开播询问。scene 目前只有 coupon(领券浮层);预留给日后比价等场景。"""
scene: str = Field(default="coupon", max_length=16)
play_token: str = Field(min_length=1, max_length=64)
class GuideVideoStartOut(BaseModel):
"""should_play=False 时客户端照旧走广告链路,其余字段无意义。"""
should_play: bool
video_url: str | None = None # 相对地址 /media/...;客户端自行拼 BASE_URL
play_token: str = "" # 发奖幂等键
reward_coin: int = 0 # 播完/中途关闭都发的固定金币
seq: int = 0 # 本账号第几次
remaining: int = 0 # 发完这次还剩几次
started: bool
status: Literal["started", "already_started"]
play_token: str
scene: GuideScene
video_url: str
config_version: int
duration_ms: int
circle_count: int = 10
circle_duration_ms: float
reward_coin: int
reward_per_circle: int
seq: int
remaining: int
started_at: str
class GuideVideoRewardIn(BaseModel):
"""播完或中途关闭都调这个;completed 只做留痕,两者都发币。"""
play_token: str = Field(min_length=1, max_length=64)
completed: bool = False
circle: int = Field(ge=1, le=10)
class GuideVideoRewardOut(BaseModel):
"""granted=True 表示本次调用真的入账(重复上报为 False,coin 是已发金额)。"""
granted: bool
coin: int
status: str
status: Literal[
"granted",
"already_granted",
"too_early",
"out_of_order",
"not_started",
"not_found",
"finished",
]
retryable: bool = False
retry_after_ms: int = 0
circle: int
granted_coin: int = 0
settled_circles: int = 0
coin_balance: int = 0
+85 -11
View File
@@ -85,6 +85,10 @@ def _dispatch(
push_vars: dict[str, str] | None = None,
) -> Notification | None:
"""落一条站内消息并向该用户设备直推。返回落库行;去重命中/失败返回 None。"""
logger.info(
"notification dispatch started user_id=%s type=%s dedup_key=%s",
user_id, type_key, dedup_key,
)
try:
row = notif_repo.create_notification(
db,
@@ -111,6 +115,10 @@ def _dispatch(
logger.exception("rollback after notification failure also failed")
return None
logger.info(
"notification created user_id=%s type=%s notification_id=%s dedup_key=%s",
user_id, type_key, row.id, dedup_key,
)
_push_to_user_devices(db, row, push_vars)
return row
@@ -124,29 +132,95 @@ def _push_to_user_devices(db: Session, row: Notification, push_vars: dict[str, s
extras.update({str(k): str(v) for k, v in (row.extra or {}).items()})
extras["notificationId"] = str(row.id)
for dev in device_repo.list_push_targets(db, user_id=row.user_id):
targets = device_repo.list_push_targets(db, user_id=row.user_id)
logger.info(
"push targets resolved user_id=%s type=%s notification_id=%s target_count=%s",
row.user_id, row.type, row.id, len(targets),
)
if not targets:
logger.warning(
"push skipped no targets user_id=%s type=%s notification_id=%s",
row.user_id, row.type, row.id,
)
return
sent = failed = skipped = data_sent = data_failed = 0
for dev in targets:
vendor = vendor_push.normalize_vendor(dev.push_vendor)
if not vendor or vendor not in vendor_push.SUPPORTED_VENDORS:
continue
if vendor_push.missing_settings(vendor):
# 本地/测试环境凭据不齐 → 只落站内消息,不发真推送(与 push/vendors 的报缺口径一致)
logger.info(
"skip push (vendor %s not configured) user_id=%s type=%s",
vendor, row.user_id, row.type,
skipped += 1
logger.warning(
"push target skipped unsupported vendor user_id=%s type=%s "
"notification_id=%s device_id=%s raw_vendor=%s normalized_vendor=%s",
row.user_id, row.type, row.id, dev.device_id, dev.push_vendor, vendor,
)
continue
missing = vendor_push.missing_settings(vendor)
if missing:
skipped += 1
# 本地/测试环境凭据不齐 → 只落站内消息,不发真推送(与 push/vendors 的报缺口径一致)
logger.warning(
"push target skipped vendor not configured user_id=%s type=%s "
"notification_id=%s device_id=%s vendor=%s missing_settings=%s",
row.user_id, row.type, row.id, dev.device_id, vendor, missing,
)
continue
logger.info(
"push send started user_id=%s type=%s notification_id=%s "
"device_id=%s vendor=%s",
row.user_id, row.type, row.id, dev.device_id, vendor,
)
try:
vendor_push.send_notification(
response = vendor_push.send_notification(
vendor, dev.push_token, title=title, body=body, extras=extras
)
if vendor == "huawei" and row.type in {"feedback_reply", "feedback_reward"}:
# 华为反馈推送联调日志:保留厂商返回码/requestId 等排障信息,
# 请求本身的 token、Authorization 和应用密钥不会进入 response。
logger.info(
"huawei feedback push response user_id=%s type=%s "
"notification_id=%s response=%s",
row.user_id, row.type, row.id, response,
)
logger.info(
"push sent user_id=%s type=%s vendor=%s notification_id=%s",
row.user_id, row.type, vendor, row.id,
"push sent user_id=%s type=%s vendor=%s notification_id=%s device_id=%s",
row.user_id, row.type, vendor, row.id, dev.device_id,
)
sent += 1
except vendor_push.VendorPushError as e:
failed += 1
logger.warning(
"push failed user_id=%s type=%s vendor=%s notification_id=%s "
"device_id=%s error=%s",
row.user_id, row.type, vendor, row.id, dev.device_id, e,
)
try:
data_response = vendor_push.send_data_event(
vendor,
dev.push_token,
event=vendor_push.DATA_EVENT_NOTIFICATION_CREATED,
notification_id=str(row.id),
)
data_sent += 1
logger.info(
"push data event completed user_id=%s type=%s vendor=%s "
"notification_id=%s device_id=%s response=%s",
row.user_id, row.type, vendor, row.id, dev.device_id, data_response,
)
except vendor_push.VendorPushError as e:
data_failed += 1
# 透传只负责前台铃铛实时刷新,失败不能影响通知栏消息或站内消息。
logger.warning(
"push failed user_id=%s type=%s vendor=%s: %s", row.user_id, row.type, vendor, e
"push data event failed user_id=%s type=%s vendor=%s notification_id=%s "
"device_id=%s error=%s",
row.user_id, row.type, vendor, row.id, dev.device_id, e,
)
logger.info(
"push dispatch completed user_id=%s type=%s notification_id=%s targets=%s "
"sent=%s failed=%s skipped=%s data_sent=%s data_failed=%s",
row.user_id, row.type, row.id, len(targets),
sent, failed, skipped, data_sent, data_failed,
)
except Exception: # noqa: BLE001 — 渲染/查设备等意外失败同样不外抛
logger.exception("push notification failed user_id=%s type=%s", row.user_id, row.type)
+48 -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_daily_limit" in items and "ad_cooldown_sec" in items
# hidden 项(任务/里程碑、首页轮播数据源、广告组的单次金币/每轮次数/信息流广告开关)不在配置页返回。
# 非 hidden 项照常返回;广告次数上限迁到「白名单」统一配置
assert "signin_rewards" in items and "ad_cooldown_sec" in items
# hidden 项(任务/里程碑、首页轮播数据源、广告次数/单次金币/每轮次数/信息流广告开关)不在配置页返回。
for hidden_key in (
"task_rewards", "record_milestones", "marquee_feed_mode",
"ad_reward_coin", "ad_round_count", "comparing_ad_enabled",
"ad_daily_limit", "ad_reward_coin", "ad_round_count", "comparing_ad_enabled",
):
assert hidden_key not in items, f"{hidden_key} 应被 hidden 过滤"
assert items["signin_rewards"]["value"] == [
@@ -124,6 +124,50 @@ def test_update_ad_limit_takes_effect(admin_client: TestClient, token: str) -> N
db.close()
def test_list_config_reads_limit_values_from_global_bundle(
admin_client: TestClient,
token: str,
) -> None:
changed = admin_client.patch(
"/admin/api/config/ad_cooldown_sec",
json={"value": 17},
headers=_auth(token),
)
assert changed.status_code == 200, changed.text
items = {
item["key"]: item
for item in admin_client.get(
"/admin/api/config",
headers=_auth(token),
).json()
}
assert items["ad_cooldown_sec"]["value"] == 17
assert items["ad_cooldown_sec"]["overridden"] is True
@pytest.mark.parametrize(
("key", "value"),
[
("ad_daily_limit", 0),
("ad_daily_limit", 100_001),
("ad_cooldown_sec", 86_401),
],
)
def test_update_limit_config_rejects_out_of_range_values(
admin_client: TestClient,
token: str,
key: str,
value: int,
) -> None:
response = admin_client.patch(
f"/admin/api/config/{key}",
json={"value": value},
headers=_auth(token),
)
assert response.status_code == 400, response.text
def test_update_bool_config(admin_client: TestClient, token: str) -> None:
# 提现自动对账开关默认 True
items = {
+109
View File
@@ -219,6 +219,115 @@ def test_user_reward_stats_can_scope_withdrawals_by_account(
assert invite.json()["cash_balance_cents"] == 456
def test_user_reward_stats_draw_ecpm_uses_all_filtered_impressions(
admin_client: TestClient, admin_token: str
) -> None:
"""Draw 平均 eCPM 应与收益报表一致,不能只平均成功发奖记录。"""
from app.models.ad_ecpm import AdEcpmRecord
from app.models.ad_feed_reward import AdFeedRewardRecord
uid = _seed_user_with_data("13800000024")
created_at = datetime(2038, 1, 15, 4, tzinfo=UTC)
db = SessionLocal()
try:
db.add_all(
[
AdFeedRewardRecord(
client_event_id="reward-stats-granted-high",
ad_session_id="reward-stats-granted-high",
user_id=uid,
reward_date="2038-01-15",
duration_seconds=10,
unit_count=1,
ecpm_raw="9000",
ad_type="draw",
feed_scene="coupon",
app_env="prod",
our_code_id="104098712",
coin=9,
status="granted",
created_at=created_at,
),
AdEcpmRecord(
user_id=uid,
ad_type="draw",
feed_scene="coupon",
ad_session_id="reward-stats-impression-low",
app_env="prod",
our_code_id="104098712",
ecpm_raw="1000",
report_date="2038-01-15",
created_at=created_at,
),
AdEcpmRecord(
user_id=uid,
ad_type="feed",
feed_scene="coupon",
ad_session_id="reward-stats-impression-mid",
app_env="prod",
our_code_id="104098712",
ecpm_raw="3000",
report_date="2038-01-15",
created_at=created_at,
),
# 同用户但不同场景/环境/非业务代码位,均不应进入本次详情筛选。
AdEcpmRecord(
user_id=uid,
ad_type="draw",
feed_scene="comparison",
ad_session_id="reward-stats-other-scene",
app_env="prod",
our_code_id="104098712",
ecpm_raw="7000",
report_date="2038-01-15",
created_at=created_at,
),
AdEcpmRecord(
user_id=uid,
ad_type="draw",
feed_scene="coupon",
ad_session_id="reward-stats-test-env",
app_env="test",
our_code_id="104127529",
ecpm_raw="8000",
report_date="2038-01-15",
created_at=created_at,
),
AdEcpmRecord(
user_id=uid,
ad_type="draw",
feed_scene="coupon",
ad_session_id="reward-stats-non-business",
app_env="prod",
our_code_id="demo-slot",
ecpm_raw="9000",
report_date="2038-01-15",
created_at=created_at,
),
]
)
db.commit()
finally:
db.close()
response = admin_client.get(
f"/admin/api/users/{uid}/reward-stats",
params={
"date_from": "2038-01-15T00:00:00Z",
"date_to": "2038-01-15T23:59:59Z",
"app_env": "prod",
"revenue_scope": "business",
"feed_scene": "coupon",
},
headers=_auth(admin_token),
)
assert response.status_code == 200, response.text
data = response.json()
assert data["feed_count"] == 1
# 全部真实展示 (1000 + 3000) / 2;不能返回成功发奖记录的 9000。
assert data["feed_avg_ecpm"] == 2000.0
def test_user_coin_record_sort_accepts_mixed_timezone_datetimes() -> None:
"""线上 PostgreSQL 返回 awareSQLite/历史转换可能返回 naive,二者必须可混排。"""
naive = datetime(2038, 1, 1, 8, 0)
+7 -2
View File
@@ -73,6 +73,7 @@ def test_monitoring_audit_catalog_and_api_permissions(
"analytics-health",
"event-logs",
"audit-logs",
"limit-whitelist",
]
# 运营默认可查风控和设备存活,不能绕过导航直调技术/审计接口。
@@ -82,6 +83,9 @@ def test_monitoring_audit_catalog_and_api_permissions(
assert admin_client.get(
"/admin/api/device-liveness/stats", headers=_auth(operator_token)
).status_code == 200
assert admin_client.get(
"/admin/api/limit-whitelist/rules", headers=_auth(operator_token)
).status_code == 200
for path in (
"/admin/api/analytics-health/overview?date_from=2026-07-01T00:00:00Z&date_to=2026-07-02T00:00:00Z",
"/admin/api/event-logs",
@@ -89,13 +93,14 @@ def test_monitoring_audit_catalog_and_api_permissions(
):
assert admin_client.get(path, headers=_auth(operator_token)).status_code == 403
# 技术角色默认拥有监控审计组全部项权限。
# 技术角色默认拥有监控审计组全部项权限。
for path in (
"/admin/api/risk-monitor/summary",
"/admin/api/device-liveness/stats",
"/admin/api/analytics-health/overview?date_from=2026-07-01T00:00:00Z&date_to=2026-07-02T00:00:00Z",
"/admin/api/event-logs",
"/admin/api/audit-logs",
"/admin/api/limit-whitelist/rules",
):
assert admin_client.get(path, headers=_auth(tech_token)).status_code == 200
@@ -199,7 +204,7 @@ def test_builtin_roles_labels_and_pages(admin_client, super_token) -> None:
}
assert set(roles["tech"]["pages"]) == {
"dashboard", "risk-monitor", "device-liveness", "analytics-health", "config",
"ad-revenue", "huawei-review", "event-logs", "audit-logs",
"ad-revenue", "huawei-review", "event-logs", "audit-logs", "limit-whitelist",
}
+69
View File
@@ -140,6 +140,75 @@ def test_harvest_done_failed_derives_fail_reason(client) -> None:
assert rec.information == "比价过程出错,请稍后重试" # 原文案仍留存
def _done_params_platforms_no_isbest() -> dict:
"""id 3304 型:done 帧 platforms 全平台「相似替换/仅供参考」(has_dish_diff),pricebot
一个 is_best 都没标,但有有价目标(美团 57.8 < 源淘宝闪购 60.8)"""
return {
"platforms": [
{"role": "source", "platform_id": "eleme", "platform_name": "淘宝闪购",
"package": "me.ele", "price": 60.8, "is_best": False, "has_dish_diff": False,
"store_name": "窑鸡王", "items": [{"name": "招牌窑鸡 整只-香辣", "qty": 1}]},
{"role": "target", "platform_id": "meituan_waimai", "platform_name": "美团外卖",
"package": "com.sankuai.meituan.takeoutnew", "price": 57.8, "is_best": False,
"has_dish_diff": True},
{"role": "target", "platform_id": "jd_waimai_standalone", "platform_name": "京东外卖",
"package": "com.jd.waimai", "price": 74.9, "is_best": False, "has_dish_diff": True},
],
"information": "美团更便宜(含相似商品替换)",
}
def test_harvest_done_platforms_no_isbest_falls_back_to_cheapest_target(client) -> None:
"""回归(id 3304):platforms 全无 is_best(全平台相似替换)但有有价目标 → best 应兜底取
有价目标里最低那家,不能让 best_*/saved 整条落 NULL(否则首页价 0.00 / 记录页无最低红框 /
省额丢失)"""
tid = _tid()
with SessionLocal() as db:
crud.harvest_running(db, trace_id=tid, user_id=None)
rec, _ = crud.harvest_done(db, trace_id=tid, user_id=None,
done_params=_done_params_platforms_no_isbest())
assert rec.status == "success"
assert rec.best_platform_id == "meituan_waimai" # 有价目标里最低
assert rec.best_price_cents == 5780 # 57.8 元
assert rec.source_price_cents == 6080 # 源淘宝闪购 60.8
assert rec.saved_amount_cents == 300 # 60.8 - 57.8
assert rec.is_source_best is False # 兜底选的是目标,非源
def _done_params_no_isbest_source_cheapest() -> dict:
"""无 is_best 且源本身最便宜:源淘宝闪购 50.0 < 全部 dish-diff 目标(美团 57.8 / 京东 74.9)。
此时不能强选更贵的目标当 best(否则 saved 变负污染累计发现可省),应回落源"""
return {
"platforms": [
{"role": "source", "platform_id": "eleme", "platform_name": "淘宝闪购",
"package": "me.ele", "price": 50.0, "is_best": False, "has_dish_diff": False,
"store_name": "窑鸡王", "items": [{"name": "招牌窑鸡 整只-香辣", "qty": 1}]},
{"role": "target", "platform_id": "meituan_waimai", "platform_name": "美团外卖",
"package": "com.sankuai.meituan.takeoutnew", "price": 57.8, "is_best": False,
"has_dish_diff": True},
{"role": "target", "platform_id": "jd_waimai_standalone", "platform_name": "京东外卖",
"package": "com.jd.waimai", "price": 74.9, "is_best": False, "has_dish_diff": True},
],
"information": "源平台已是最低(其余为相似替换)",
}
def test_harvest_done_no_isbest_source_cheapest_falls_back_to_source(client) -> None:
"""回归:无 is_best 且源最便宜 → best 回落源(与 _derive「全目标缺菜回落源、不虚报省」同语义),
saved=0is_source_best=True,绝不因强选更贵目标而让 saved 变负倒扣累计发现可省"""
tid = _tid()
with SessionLocal() as db:
crud.harvest_running(db, trace_id=tid, user_id=None)
rec, _ = crud.harvest_done(db, trace_id=tid, user_id=None,
done_params=_done_params_no_isbest_source_cheapest())
assert rec.status == "success"
assert rec.best_platform_id == "eleme" # 回落到源(源最便宜)
assert rec.best_price_cents == 5000 # 源 50.0
assert rec.source_price_cents == 5000
assert rec.saved_amount_cents == 0 # 没省到,绝不为负
assert rec.is_source_best is True # 源就是最便宜
def test_harvest_abort_cancels_running(client) -> None:
tid = _tid()
with SessionLocal() as db:
+1 -1
View File
@@ -201,7 +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: 0)
monkeypatch.setattr(crud_guide, "used_plays", lambda db, user_id, scene="coupon": 0)
with SessionLocal() as db:
result = crud_guide.start_play(db, uid)
File diff suppressed because it is too large Load Diff
+95
View File
@@ -7,7 +7,9 @@ mock/真发/变量覆盖/站内联动/设备反查/参数校验。厂商 HTTP
from __future__ import annotations
import json
import logging
import httpx
import pytest
from fastapi.testclient import TestClient
@@ -110,6 +112,99 @@ def test_huawei_non_success_code_raises(monkeypatch) -> None:
vendor_push.send_notification("huawei", "bad-token", title="t", body="b")
def test_vendor_http_logs_device_token_raw_response_and_elapsed(monkeypatch, caplog) -> None:
vendor_push._token_cache.clear()
def _fake_request(method, url, **kwargs): # noqa: ANN001
if url == vendor_push.settings.HUAWEI_PUSH_TOKEN_ENDPOINT:
return _Resp({"access_token": "raw-access-token", "expires_in": 3600})
return _Resp({"code": "80000000", "msg": "Success", "requestId": "vendor-request-1"})
monkeypatch.setattr(vendor_push.settings, "HUAWEI_PUSH_APP_ID", "10086001")
monkeypatch.setattr(vendor_push.settings, "HUAWEI_PUSH_APP_SECRET", "raw-app-secret")
monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
with caplog.at_level(logging.INFO, logger=vendor_push.logger.name):
vendor_push.send_notification(
"huawei",
"raw-device-token",
title="需要进入日志的标题",
body="需要进入日志的正文",
extras={"type": "withdraw_success", "notificationId": "90001"},
)
messages = "\n".join(record.getMessage() for record in caplog.records)
assert "operation=authenticate" in messages
assert "operation=send_notification" in messages
assert "request=" in messages
assert "response=" in messages
assert "elapsed_ms=" in messages
assert "vendor-request-1" in messages
assert "withdraw_success" in messages
assert "raw-access-token" in messages
assert "raw-device-token" in messages
assert "需要进入日志的标题" in messages
assert "需要进入日志的正文" in messages
assert "<redacted" in messages
assert "raw-app-secret" not in messages
send_http_log = next(
record.getMessage()
for record in caplog.records
if "operation=send_notification" in record.getMessage()
and "vendor push http completed" in record.getMessage()
)
assert "raw-device-token" in send_http_log
assert "raw-access-token" not in send_http_log # 请求头中的厂商鉴权 token 仍脱敏
def test_vendor_http_network_error_logs_request_context(monkeypatch, caplog) -> None:
def _fake_request(*args, **kwargs): # noqa: ANN002, ANN003
raise httpx.ConnectError("connection refused")
monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_APP_SECRET", "raw-xiaomi-secret")
monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
with (
caplog.at_level(logging.ERROR, logger=vendor_push.logger.name),
pytest.raises(vendor_push.VendorPushError, match="push http error"),
):
vendor_push.send_notification(
"xiaomi",
"raw-xiaomi-token",
title="title",
body="body",
extras={"type": "push_test"},
)
messages = "\n".join(record.getMessage() for record in caplog.records)
assert "outcome=network_error" in messages
assert "response=<no_response>" in messages
assert "operation=send_notification" in messages
assert "elapsed_ms=" in messages
assert "raw-xiaomi-secret" not in messages
assert "raw-xiaomi-token" in messages
@pytest.mark.parametrize(
("request_kwargs", "device_token"),
[
({"json": {"token": ["honor-device-token"]}}, "honor-device-token"),
({"json": {"message": {"token": ["huawei-device-token"]}}}, "huawei-device-token"),
({"json": {"regId": "vivo-device-token"}}, "vivo-device-token"),
({"data": {"registration_id": "xiaomi-device-token"}}, "xiaomi-device-token"),
(
{"data": {"message": '{"target_value":"oppo-device-token"}'}},
"oppo-device-token",
),
],
)
def test_all_vendor_device_token_fields_remain_visible_in_request_summary(
request_kwargs, device_token
) -> None:
summary = vendor_push._request_summary(request_kwargs)
assert device_token in summary
def test_vendor_aliases_normalize() -> None:
assert vendor_push.normalize_vendor("华为") == "huawei"
assert vendor_push.normalize_vendor("HMS") == "huawei"
+3 -9
View File
@@ -10,11 +10,7 @@ from fastapi.testclient import TestClient
from app.admin.main import admin_app
from app.admin.repositories import admin_user as admin_repo
from app.api.v1 import compare as compare_api
from app.core.config_schema import (
RISK_COMPARE_DAILY_THRESHOLD_KEY,
RISK_ONECLICK_DAILY_THRESHOLD_KEY,
RISK_SMS_HOURLY_THRESHOLD_KEY,
)
from app.core.config_schema import LIMIT_POLICY_GLOBAL_KEY
from app.core.security import issue_token_pair
from app.db.session import SessionLocal
from app.models.app_config import AppConfig
@@ -28,9 +24,7 @@ from app.repositories import user as user_repo
@pytest.fixture(autouse=True)
def _reset_risk_rule_config():
keys = (
RISK_SMS_HOURLY_THRESHOLD_KEY,
RISK_ONECLICK_DAILY_THRESHOLD_KEY,
RISK_COMPARE_DAILY_THRESHOLD_KEY,
LIMIT_POLICY_GLOBAL_KEY,
risk_repo.RISK_RESET_BASELINES_KEY,
)
with SessionLocal() as db:
@@ -533,7 +527,7 @@ def test_admin_can_edit_rules_and_current_window_is_reconciled() -> None:
"/admin/api/risk-monitor/rules",
headers=headers,
json={
"sms_hourly_threshold": 21,
"sms_hourly_threshold": 100_001,
"oneclick_daily_threshold": 20,
"compare_daily_threshold": 100,
},