Compare commits

...

5 Commits

Author SHA1 Message Date
zzhyyyyy e1e390a406 feat(auth): 发码防刷统一走设备维度 — 删 IP 那道 + 阈值 10→5
- /sms/send 去掉路由上的 IP 维度限流(rate_limit 依赖):统一由设备(device_id)那道兜
  (产品决议:IP 那道会风控同 IP 的测试号,而测试号已豁免设备那道)。
- 设备发码限流阈值 10 → 5(与登录一致)。
- 同步 sms.py 防刷分层注释。
⚠️ 取舍:device_id 客户端可伪造/轮换,脚本轮换 id 可绕过发码限流;挡脚本狂发改主要靠极光控制台侧。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-26 23:29:12 +08:00
zzhyyyyy b59145c784 feat(ad): 信息流广告切换为 Draw 信息流并区分比价/领券收益
后端支持 Draw 信息流广告:
- ad_feed_reward_record 新增 ad_type(feed/draw),ad_ecpm_record 新增 feed_scene(comparison/coupon/welfare)+ Alembic 迁移
- ecpm-report / feed-reward 接口接收并落 ad_type=draw 与 feed_scene
- 广告配置字段 compare/coupon_feed_code_id 改名为 compare/coupon_draw_code_id(默认 Draw 位 104098712)
- 收益报表与对账(ad_revenue / ad_audit)支持 draw 类型,按 feed_scene 区分比价/领券收益

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-26 22:58:23 +08:00
zzhyyyyy c01cc15bb0 feat(analytics): 埋点事件接收接口 + admin 查询接口 + 埋点表
新手引导埋点的服务端:
- 表 analytics_event(五维硬性列 + props JSON 扩展字段;event/device_id/user_id/session_id/created_at 带索引)
- POST /api/v1/analytics/events:客户端批量上报(不鉴权、body 读可选 user_id、补 client_ip + server_at)
- admin GET /admin/api/event-logs:列表 + 按 事件/设备/用户/会话/时间 筛选(offset 分页,照 list_feedbacks)
- alembic migration 建表(autogenerate 顺带检出的 ad/cps 历史索引漂移已手动剔除)

app 主后端 :8770 与 admin :8771 共用同一 SQLite,admin 同库直接查、无需跨库。
配套客户端五维上报 + admin 日志页(另两仓库 PR)。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-26 19:11:23 +08:00
zhuzihao ea22a11ef9 fix(ad): 激励视频发奖按环境隔离,堵测试包看广告给生产同 id 用户串号发币 (#77)
## 问题
激励视频发奖走穿山甲 S2S 回调 /api/v1/ad/pangle-callback,发币时只认回调里的裸数字
user_id(grant_ad_reward 仅校验「本库存在该 id」),无任何环境归属校验。

同一 user_id 在测试库(本地 localhost)与正式库(线上)是两个不同的人;穿山甲 S2S 回调
地址只配一个(指向生产),测试包(dev)看广告的回调也会打到生产 → 生产把奖发给「本库里
同 user_id 的另一个真人」,造成跨库串号:测试看一次广告,正式库同号用户也被发币(线上
广告收益页因此多出一条本不该有的发奖记录)。

## 修复(服务端,按环境隔离)
pangle_callback 解析激励视频 mediaExtra 透传的 srv_env(dev/prod),与本服 settings.APP_ENV
不符则受理但不发币、不写记录(is_verify=true 防穿山甲重试)。不带 srv_env 的旧客户端按本
环境放行(兼容,不误伤存量正式用户)。

- 新增回归测试 test_callback_foreign_env_skipped(srv_env=prod 打到 APP_ENV=dev 的服务 → 不发)
- sim_pangle_callback 增加 --srv-env 便于本地手验

客户端配套(mediaExtra 写入 srv_env)见 app-android 对应 PR。

## 生效前提(三者缺一仍漏)
1. 生产 app-server 部署本代码并重启
2. 生产 .env 为 APP_ENV=prod
3. app 重新构建(旧包不带 srv_env 仍走兼容放行)

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

---------

Co-authored-by: zzhyyyyy <2685922758@qq.com>
Reviewed-on: #77
Co-authored-by: zhuzihao <zhuzihao@wonderable.ai>
Co-committed-by: zhuzihao <zhuzihao@wonderable.ai>
2026-06-26 13:17:27 +08:00
zhuzihao 2edebfad83 feat(auth): 短信登录防刷 — 同一设备id同IP 每小时限 5 次 (#78)
Co-authored-by: zzhyyyyy <2685922758@qq.com>
Reviewed-on: #78
Co-authored-by: zhuzihao <zhuzihao@wonderable.ai>
Co-committed-by: zhuzihao <zhuzihao@wonderable.ai>
2026-06-26 13:17:07 +08:00
34 changed files with 766 additions and 40 deletions
@@ -0,0 +1,65 @@
"""add analytics_event table
Revision ID: 1699fc2c069f
Revises: bcfcaf07152b
Create Date: 2026-06-26 16:35:16.133975
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision: str = '1699fc2c069f'
down_revision: str | Sequence[str] | None = 'bcfcaf07152b'
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('analytics_event',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('event', sa.String(length=64), nullable=False),
sa.Column('props', sa.JSON(), nullable=True),
sa.Column('device_id', sa.String(length=64), nullable=False),
sa.Column('user_id', sa.Integer(), nullable=True),
sa.Column('session_id', sa.String(length=64), nullable=True),
sa.Column('client_ts', sa.BigInteger(), nullable=False),
sa.Column('sent_at', sa.BigInteger(), nullable=True),
sa.Column('page', sa.String(length=64), nullable=True),
sa.Column('client_ip', sa.String(length=64), nullable=True),
sa.Column('oem', sa.String(length=32), nullable=True),
sa.Column('os', sa.String(length=32), nullable=True),
sa.Column('model', sa.String(length=64), nullable=True),
sa.Column('app_ver', sa.String(length=32), nullable=True),
sa.Column('network', sa.String(length=16), nullable=True),
sa.Column('channel', sa.String(length=32), nullable=True),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.PrimaryKeyConstraint('id')
)
with op.batch_alter_table('analytics_event', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_analytics_event_created_at'), ['created_at'], unique=False)
batch_op.create_index(batch_op.f('ix_analytics_event_device_id'), ['device_id'], unique=False)
batch_op.create_index(batch_op.f('ix_analytics_event_event'), ['event'], unique=False)
batch_op.create_index(batch_op.f('ix_analytics_event_session_id'), ['session_id'], unique=False)
batch_op.create_index(batch_op.f('ix_analytics_event_user_id'), ['user_id'], unique=False)
# 注:autogenerate 顺带检出 ad_ecpm/cps_*/invite_fingerprint 的历史索引漂移,
# 与本次「新增埋点表」无关,已手动移除,避免本迁移误改他人表。
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
with op.batch_alter_table('analytics_event', schema=None) as batch_op:
batch_op.drop_index(batch_op.f('ix_analytics_event_user_id'))
batch_op.drop_index(batch_op.f('ix_analytics_event_session_id'))
batch_op.drop_index(batch_op.f('ix_analytics_event_event'))
batch_op.drop_index(batch_op.f('ix_analytics_event_device_id'))
batch_op.drop_index(batch_op.f('ix_analytics_event_created_at'))
op.drop_table('analytics_event')
# ### end Alembic commands ###
@@ -0,0 +1,46 @@
"""add ad_type to ad_feed_reward and feed_scene to ad_ecpm
把信息流广告全面改造成 Draw 信息流(draw):
- ad_feed_reward_record.ad_type:广告形态 feed(信息流) / draw(Draw 信息流)。可空,旧数据 NULL
一律视为 feed(向后兼容);每日上限与因子2(LT)仍按本表全表 unit 累计,不按 ad_type 拆。
- ad_ecpm_record.feed_scene:点位场景 comparison(比价) / coupon(领券) / welfare(福利),供广告
收益报表区分比价/领券 Draw 收益;仅信息流/Draw 上报,激励视频为 NULL。
注:autogenerate 会顺带探测到 analytics_event / cps_* / invite_fingerprint 等无关索引差异(本地库与
metadata 漂移、analytics 模型尚未并入 __init__),与本次改动无关,已手工剔除——本迁移只 add 两列。
SQLite 经 env.py 的 render_as_batch 自动走 batch_alter_table 重建表。
Revision ID: c2874d2bf705
Revises: 1699fc2c069f
Create Date: 2026-06-26 16:48:52.158609
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision: str = "c2874d2bf705"
down_revision: str | Sequence[str] | None = "1699fc2c069f"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
# 信息流发奖记录:新增广告形态 feed/draw(可空,旧数据 NULL=feed)
op.add_column(
"ad_feed_reward_record",
sa.Column("ad_type", sa.String(length=16), nullable=True),
)
# eCPM 展示上报:新增点位场景 comparison/coupon/welfare(可空,激励视频/旧数据 NULL)
op.add_column(
"ad_ecpm_record",
sa.Column("feed_scene", sa.String(length=16), nullable=True),
)
def downgrade() -> None:
op.drop_column("ad_ecpm_record", "feed_scene")
op.drop_column("ad_feed_reward_record", "ad_type")
+2
View File
@@ -24,6 +24,7 @@ from app.admin.routers.config import router as config_router
from app.admin.routers.cps import router as cps_router
from app.admin.routers.dashboard import router as dashboard_router
from app.admin.routers.ops_stat_config import router as ops_stat_config_router
from app.admin.routers.event_logs import router as event_logs_router
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.onboarding import router as onboarding_router
@@ -91,6 +92,7 @@ admin_app.include_router(wallet_router)
admin_app.include_router(withdraw_router)
admin_app.include_router(price_report_router)
admin_app.include_router(feedback_router)
admin_app.include_router(event_logs_router)
admin_app.include_router(feedback_qr_router)
admin_app.include_router(admins_router)
admin_app.include_router(audit_router)
+38 -6
View File
@@ -128,8 +128,28 @@ def _feed_prior_granted_units(
return {uid: int(n) for uid, n in db.execute(stmt).all()}
def _feed_rows(db: Session, *, date: str, user_id: int | None) -> list[dict]:
"""信息流记录复算。granted 记录逐份累加,LT 序号沿用账号累计份数(含本日之前)。"""
def _feed_scene_matches(rec: AdFeedRewardRecord, scene: str | None) -> bool:
"""信息流记录是否落入请求的展示筛选 scene。
- scene=="feed":ad_type in ("feed", NULL)(旧数据 NULL 视为 feed,向后兼容)
- scene=="draw":ad_type=="draw"
- scene 为 None:不筛(两类都要)。
"""
if scene == "feed":
return rec.ad_type in (None, "feed")
if scene == "draw":
return rec.ad_type == "draw"
return True
def _feed_rows(
db: Session, *, date: str, user_id: int | None, scene: str | None = None
) -> list[dict]:
"""信息流记录复算。granted 记录逐份累加,LT 序号沿用账号累计份数(含本日之前)。
**关键:LT 因子账号累计按全表 unit 累计(feed+draw 共享同一发奖池/上限),不按 ad_type 拆分**——
故无论 scene 怎么筛展示,这里都遍历当日**全部**信息流记录维持 granted_units 累加;scene 只决定
哪些行被**留下展示**(由 _feed_scene_matches 判断),不影响累计基线,保证复算序号与正式发奖一致。
"""
stmt = (
select(AdFeedRewardRecord)
.where(AdFeedRewardRecord.reward_date == date)
@@ -142,18 +162,23 @@ def _feed_rows(db: Session, *, date: str, user_id: int | None) -> list[dict]:
granted_units: dict[int, int] = _feed_prior_granted_units(db, date=date, user_id=user_id)
rows: list[dict] = []
for rec in db.execute(stmt).scalars():
keep = _feed_scene_matches(rec, scene) # 累计照常推进,这里只决定是否展示本行
if rec.status == "granted":
existing = granted_units.get(rec.user_id, 0)
units = rec.unit_count
granted_units[rec.user_id] = existing + units
if not keep:
continue
expected = sum(
rewards.calculate_ad_reward_coin(rec.ecpm_raw, existing + offset)
for offset in range(1, units + 1)
)
granted_units[rec.user_id] = existing + units
start = existing + 1 if units > 0 else None
end = existing + units if units > 0 else None
rows.append({
"scene": "feed",
"ad_type": rec.ad_type or "feed",
"feed_scene": rec.feed_scene,
"record_id": rec.id,
"user_id": rec.user_id,
"ad_session_id": rec.ad_session_id,
@@ -173,8 +198,12 @@ def _feed_rows(db: Session, *, date: str, user_id: int | None) -> list[dict]:
"matched": expected == rec.coin,
})
else:
if not keep:
continue
rows.append({
"scene": "feed",
"ad_type": rec.ad_type or "feed",
"feed_scene": rec.feed_scene,
"record_id": rec.id,
"user_id": rec.user_id,
"ad_session_id": rec.ad_session_id,
@@ -199,16 +228,19 @@ def _feed_rows(db: Session, *, date: str, user_id: int | None) -> list[dict]:
def audit_rows(
db: Session, *, date: str, user_id: int | None, scene: str | None = None
) -> list[dict]:
"""当日逐条发奖复算行(未排序)。scene: None=两类 / "reward_video" / "feed"
"""当日逐条发奖复算行(未排序)。scene: None=两类 / "reward_video" / "feed" / "draw"
"feed""draw" 都查 ad_feed_reward_record(同一发奖表),按 ad_type 区分:feed 含历史 NULL,
draw 仅 ad_type=="draw"。信息流行额外带 `ad_type`/`feed_scene`,供收益报表区分比价/领券 Draw 收益。
每行含 `app_env`/`our_code_id`/`expected_coin`/`actual_coin` 等,供金币审计逐条对账,
也供广告收益报表把「应发/实发」按 用户×类型×应用×代码位 聚合(见 ad_revenue,复用同一复算口径)。
**LT 因子账号累计仍按全表 unit 累计(feed+draw 共享),scene 只筛展示,不拆累计。**
"""
rows: list[dict] = []
if scene in (None, "reward_video"):
rows.extend(_reward_video_rows(db, date=date, user_id=user_id))
if scene in (None, "feed"):
rows.extend(_feed_rows(db, date=date, user_id=user_id))
if scene in (None, "feed", "draw"):
rows.extend(_feed_rows(db, date=date, user_id=user_id, scene=scene))
return rows
+27 -8
View File
@@ -11,12 +11,17 @@
(ad_audit.audit_rows,与正式发奖同一公式口径,不另写公式)。合计与对账在全量上统计,
不受 limit(只截断 items)影响。
⚠️ 局限:① 历史 Draw 发奖混在 ad_feed_reward_record 无类型标记,金币侧统一记 feed。
② 跨天 S2S 回调:同一次广告的展示与发奖偶尔落相邻日,各自按 report_date / reward_date 归日
每行带 ad_type(reward_video/feed/draw)与 feed_scene(comparison/coupon/welfare),供前端区分
「比价 Draw 收益」与「领券 Draw 收益」(比价/领券共用同一代码位,只能靠 feed_scene 分)
⚠️ 局限:① 历史信息流/Draw 发奖 ad_type 为 NULL 的旧记录统一视为 feed(向后兼容);Draw 仅
ad_type=="draw" 的新记录单独成类。② 跨天 S2S 回调:同一次广告的展示与发奖偶尔落相邻日,各自按
report_date / reward_date 归日。
"""
from __future__ import annotations
from datetime import date as _date, datetime, timedelta, timezone
from datetime import UTC, datetime, timedelta
from datetime import date as _date
from sqlalchemy import select
from sqlalchemy.orm import Session
@@ -30,7 +35,7 @@ from app.models.user import User
def _cn_hour(dt: datetime) -> int:
"""created_at(UTC 口径)→ 北京时间小时(023)。naive 当 UTC 处理(sqlite),tz-aware 直接换算(pg)。"""
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
dt = dt.replace(tzinfo=UTC)
return dt.astimezone(rewards.CN_TZ).hour
@@ -46,8 +51,18 @@ def _date_range(date_from: str, date_to: str) -> list[str]:
return out
# 审计行的 scene 与报表 ad_type 一一对应
_SCENE_TO_AD_TYPE = {"reward_video": "reward_video", "feed": "feed"}
# 报表 ad_type 与审计 scene 取值一致(reward_video / feed / draw):feed 与 draw 同查发奖表
# ad_feed_reward_record,由 audit 内部按 ad_type 区分(feed 含历史 NULL,draw 仅 ad_type=="draw")。
_AUDIT_SCENES = {"reward_video", "feed", "draw"}
def _event_ad_type(row: dict) -> str:
"""纯发奖事件行的 ad_type:信息流行用 audit 带回的真实 ad_type(feed/draw),回退 feed;
激励视频行恒 reward_video。不再用 scene 硬映射,避免把 draw 丢成 feed。"""
if row["scene"] == "reward_video":
return "reward_video"
return row.get("ad_type") or "feed"
# 发奖复算明细字段(展开下钻看「金币怎么算出来的」)——从 audit 行原样取这些 key。
_REWARD_DETAIL_KEYS = (
@@ -84,7 +99,9 @@ def ad_revenue_report(
# 同时保留全量列表,未被展示合并的成「纯发奖」事件。
reward_by_session: dict[tuple[int, str], list[dict]] = {}
all_reward_rows: list[dict] = []
audit_scene = _SCENE_TO_AD_TYPE.get(ad_type) if ad_type is not None else None
# 报表 ad_type 直接当 audit scene 用(取值一致);未知/无效 ad_type 不取发奖行。draw 在此被
# 正确传成 scene="draw",audit 会按 ad_type 筛出 Draw 发奖,不再丢成 feed。
audit_scene = ad_type if ad_type in _AUDIT_SCENES else None
if ad_type is None or audit_scene is not None:
for d in _date_range(date_from, date_to):
for row in ad_audit.audit_rows(db, date=d, user_id=user_id, scene=audit_scene):
@@ -123,6 +140,7 @@ def ad_revenue_report(
"report_date": rec.report_date,
"user_id": rec.user_id,
"ad_type": rec.ad_type,
"feed_scene": rec.feed_scene,
"app_env": rec.app_env,
"our_code_id": rec.our_code_id,
"created_at": rec.created_at,
@@ -162,7 +180,8 @@ def ad_revenue_report(
"event_key": f"rwd-{row['record_id']}",
"report_date": row["_report_date"],
"user_id": row["user_id"],
"ad_type": _SCENE_TO_AD_TYPE.get(row["scene"], row["scene"]),
"ad_type": _event_ad_type(row),
"feed_scene": row.get("feed_scene"),
"app_env": row.get("app_env"),
"our_code_id": row.get("our_code_id"),
"created_at": row["created_at"],
+39
View File
@@ -15,6 +15,7 @@ from app.core import rewards
from app.models.ad_feed_reward import AdFeedRewardRecord
from app.models.ad_reward import AdRewardRecord
from app.models.admin import AdminAuditLog
from app.models.analytics_event import AnalyticsEvent
from app.models.comparison import ComparisonRecord
from app.models.feedback import Feedback
from app.models.onboarding import OnboardingCompletion
@@ -412,6 +413,44 @@ def list_feedbacks(
return offset_paginate(db, stmt, (order_fn(sort_col), id_order), limit=limit, cursor=cursor)
def list_analytics_events(
db: Session,
*,
event: str | None = None,
device_id: str | None = None,
user_id: int | None = None,
session_id: str | None = None,
created_from: datetime | None = None,
created_to: datetime | None = None,
sort_by: str = "id",
sort_order: str = "desc",
limit: int = 20,
cursor: int | None = None,
) -> tuple[list[AnalyticsEvent], int | None, int]:
"""埋点日志列表(admin 全量)。按 事件名 / 设备ID前缀 / 用户ID / 会话ID / 接收时间范围 筛选,
按 id·接收时间排序。offset 分页(同 [list_feedbacks])。created_at 为 timestamptz,
日期入参统一转 tz-aware UTC 比较。"""
stmt = select(AnalyticsEvent)
if event:
stmt = stmt.where(AnalyticsEvent.event == event)
if device_id and device_id.strip():
stmt = stmt.where(AnalyticsEvent.device_id.like(f"{device_id.strip()}%"))
if user_id is not None:
stmt = stmt.where(AnalyticsEvent.user_id == user_id)
if session_id and session_id.strip():
stmt = stmt.where(AnalyticsEvent.session_id == session_id.strip())
if created_from is not None:
stmt = stmt.where(AnalyticsEvent.created_at >= _as_utc(created_from))
if created_to is not None:
stmt = stmt.where(AnalyticsEvent.created_at <= _as_utc(created_to))
sort_cols = {"id": AnalyticsEvent.id, "created_at": AnalyticsEvent.created_at}
sort_col = sort_cols.get(sort_by, AnalyticsEvent.id)
order_fn = asc if sort_order == "asc" else desc
id_order = asc(AnalyticsEvent.id) if sort_order == "asc" else desc(AnalyticsEvent.id)
return offset_paginate(db, stmt, (order_fn(sort_col), id_order), limit=limit, cursor=cursor)
def get_withdraw_by_out_bill_no(db: Session, out_bill_no: str) -> WithdrawOrder | None:
"""按商户单号查提现单(admin 重试打款先拿 user_id 用,M3)。"""
return db.execute(
+1 -1
View File
@@ -26,7 +26,7 @@ def get_ad_coin_audit(
date: Annotated[str | None, Query(description="北京时间 YYYY-MM-DD,默认今天")] = None,
user_id: Annotated[int | None, Query(description="只看某用户;不传=全部用户")] = None,
scene: Annotated[
str | None, Query(description="reward_video / feed;不传=两类都要")
str | None, Query(description="reward_video / feed / draw;不传=全部")
] = None,
limit: Annotated[int, Query(ge=1, le=500)] = 100,
only_mismatch: Annotated[
+52
View File
@@ -0,0 +1,52 @@
"""admin 埋点日志:列表 + 按事件 / 设备 / 用户 / 会话 / 时间筛选(只读,同库直接查 analytics_event)。"""
from __future__ import annotations
from datetime import datetime
from typing import Annotated
from fastapi import APIRouter, Depends, Query
from app.admin.deps import AdminDb, get_current_admin
from app.admin.repositories import queries
from app.admin.schemas.analytics import AnalyticsEventOut
from app.admin.schemas.common import CursorPage
router = APIRouter(
prefix="/admin/api/event-logs",
tags=["admin-event-logs"],
dependencies=[Depends(get_current_admin)],
)
@router.get("", response_model=CursorPage[AnalyticsEventOut], summary="埋点日志列表")
def list_event_logs(
db: AdminDb,
event: Annotated[str | None, Query(max_length=64)] = None,
device_id: Annotated[str | None, Query(max_length=64)] = None,
user_id: Annotated[int | None, Query()] = None,
session_id: Annotated[str | None, Query(max_length=64)] = None,
created_from: Annotated[datetime | None, Query()] = None,
created_to: Annotated[datetime | None, Query()] = None,
sort_by: Annotated[str, Query(pattern="^(id|created_at)$")] = "id",
sort_order: Annotated[str, Query(pattern="^(asc|desc)$")] = "desc",
limit: Annotated[int, Query(ge=1, le=100)] = 20,
cursor: Annotated[int | None, Query()] = None,
) -> CursorPage[AnalyticsEventOut]:
items, next_cursor, total = queries.list_analytics_events(
db,
event=event,
device_id=device_id,
user_id=user_id,
session_id=session_id,
created_from=created_from,
created_to=created_to,
sort_by=sort_by,
sort_order=sort_order,
limit=limit,
cursor=cursor,
)
return CursorPage(
items=[AnalyticsEventOut.model_validate(e) for e in items],
next_cursor=next_cursor,
total=total,
)
+4 -4
View File
@@ -9,8 +9,8 @@ class AdConfigOut(BaseModel):
app_id: str
reward_code_id: str
compare_feed_code_id: str
coupon_feed_code_id: str
compare_draw_code_id: str
coupon_draw_code_id: str
reward_mkey: str
reward_enabled: bool
compare_ad_enabled: bool
@@ -23,8 +23,8 @@ class AdConfigUpdate(BaseModel):
app_id: str | None = None
reward_code_id: str | None = None
compare_feed_code_id: str | None = None
coupon_feed_code_id: str | None = None
compare_draw_code_id: str | None = None
coupon_draw_code_id: str | None = None
reward_mkey: str | None = None
reward_enabled: bool | None = None
compare_ad_enabled: bool | None = None
+6 -1
View File
@@ -56,7 +56,12 @@ class AdRevenueRow(BaseModel):
report_date: str = Field(..., description="该事件所属日期(北京时间 YYYY-MM-DD)")
user_id: int
user_phone: str | None = Field(None, description="用户手机号(admin 展示用,完整;用户已删 / 查不到为空)")
ad_type: str = Field(..., description="reward_video(激励视频) / feed(信息流) / draw(历史 Draw 信息流)")
ad_type: str = Field(..., description="reward_video(激励视频) / feed(信息流) / draw(Draw 信息流);历史 NULL 视为 feed")
feed_scene: str | None = Field(
None,
description="点位场景:comparison(比价) / coupon(领券) / welfare(福利);供区分比价/领券 Draw 收益;"
"激励视频与旧数据为空",
)
app_env: str | None = Field(None, description="我们的应用:prod(傻瓜比价正式) / test(测试应用);旧数据为空")
our_code_id: str | None = Field(None, description="我们后台配置的代码位 ID(104xxx);旧数据为空")
hour: int | None = Field(None, description="北京时间小时 023(granularity=hour 时有值;按天为 null)")
+33
View File
@@ -0,0 +1,33 @@
"""admin 埋点日志列表响应。"""
from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, ConfigDict
class AnalyticsEventOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
event: str
# Who
device_id: str
user_id: int | None
# When
session_id: str | None
client_ts: int
sent_at: int | None
created_at: datetime # 服务端接收时间(server_at)
# Where
page: str | None
client_ip: str | None
# How
oem: str | None
os: str | None
model: str | None
app_ver: str | None
network: str | None
channel: str | None
# What 专属
props: dict | None
+22 -5
View File
@@ -9,8 +9,8 @@
"""
from __future__ import annotations
import logging
import json
import logging
import uuid
from fastapi import APIRouter, Depends, HTTPException, Request, status
@@ -18,8 +18,8 @@ from fastapi import APIRouter, Depends, HTTPException, Request, status
from app.api.deps import CurrentUser, DbSession
from app.core import rewards
from app.core.config import settings
from app.integrations import pangle
from app.core.ratelimit import rate_limit
from app.integrations import pangle
from app.repositories import ad_ecpm as crud_ecpm
from app.repositories import ad_feed_reward as crud_feed
from app.repositories import ad_reward as crud_ad
@@ -120,6 +120,21 @@ def pangle_callback(request: Request, db: DbSession) -> PangleCallbackOut:
ad_session_id = extra.get("ad_session_id")
ecpm = params.get("ecpm")
# 环境隔离:激励视频 mediaExtra 里带「这次观看属于哪个后端环境」(srv_env=dev/prod,客户端按
# BuildConfig.DEBUG 决定),穿山甲 S2S 回调原样带回。回调 URL 在穿山甲后台只配一个(指向生产),
# 所以测试包(dev)看广告的 S2S 也会打到生产——若不拦,生产会把奖发给「本库里同 user_id 的另一个
# 真人」(user_id 是跨库不隔离的裸数字 → 跨库串号)。这里只处理「属于本服环境」的回调:环境不符
# 直接受理但不发币、不写任何记录,保证各环境后台广告收益页只含本环境用户。is_verify=true 让穿山甲
# 不再重试(测试用户的币由 localhost 的 test-grant 单独发,不依赖这条 S2S)。
# 兼容:旧客户端不带 srv_env(取不到)→ 视为本环境,照常处理,不误伤存量正式用户。
callback_env = extra.get("srv_env")
if callback_env and callback_env != settings.APP_ENV:
logger.info(
"pangle callback foreign env skip: callback_env=%s self_env=%s user_id=%d trans_id=%s",
callback_env, settings.APP_ENV, user_id, trans_id,
)
return PangleCallbackOut(is_verify=True, reason=REASON_OK)
existing = crud_ad.find_by_trans(db, trans_id)
if existing is not None:
logger.info(
@@ -270,12 +285,13 @@ def ecpm_report(payload: EcpmReportIn, user: CurrentUser, db: DbSession) -> Ecpm
ad_type=payload.ad_type, ecpm_raw=payload.ecpm,
ad_session_id=payload.ad_session_id,
adn=payload.adn, slot_id=payload.slot_id,
feed_scene=payload.feed_scene,
app_env=payload.app_env, our_code_id=payload.our_code_id,
)
logger.info(
"ad ecpm report user_id=%d type=%s session=%s ecpm=%s adn=%s slot=%s app=%s code=%s",
user.id, payload.ad_type, payload.ad_session_id, payload.ecpm, payload.adn, payload.slot_id,
payload.app_env, payload.our_code_id,
"ad ecpm report user_id=%d type=%s scene=%s session=%s ecpm=%s adn=%s slot=%s app=%s code=%s",
user.id, payload.ad_type, payload.feed_scene, payload.ad_session_id, payload.ecpm,
payload.adn, payload.slot_id, payload.app_env, payload.our_code_id,
)
return EcpmReportOut(ok=True)
@@ -386,6 +402,7 @@ def feed_reward(payload: FeedRewardIn, user: CurrentUser, db: DbSession) -> Feed
client_event_id=payload.client_event_id,
ecpm=payload.ecpm,
duration_seconds=payload.duration_seconds,
ad_type=payload.ad_type,
ad_session_id=payload.ad_session_id,
adn=payload.adn,
slot_id=payload.slot_id,
+31
View File
@@ -0,0 +1,31 @@
"""客户端埋点上报接口。
POST /api/v1/analytics/events 批量接收新手引导(及后续)埋点,append analytics_event
**不强制登录**(未登录态也要采集行为):user_id 由客户端在 body 里可选带上,不靠 Bearer
服务端补 client_ip(X-Forwarded-For) created_at(接收时间 = server_at)
"""
from __future__ import annotations
from fastapi import APIRouter, Request
from app.api.deps import DbSession
from app.repositories import analytics as analytics_repo
from app.schemas.analytics import AnalyticsBatchIn, AnalyticsIngestOut
router = APIRouter(prefix="/api/v1/analytics", tags=["analytics"])
def _client_ip(request: Request) -> str:
"""取客户端 IP:生产经 nginx 反代优先 X-Forwarded-For 第一段,否则直连 IP(同 admin get_client_ip)。"""
xff = request.headers.get("x-forwarded-for")
if xff:
return xff.split(",")[0].strip()
return request.client.host if request.client else ""
@router.post("/events", response_model=AnalyticsIngestOut, summary="批量上报埋点事件")
def ingest_events(
batch: AnalyticsBatchIn, request: Request, db: DbSession
) -> AnalyticsIngestOut:
n = analytics_repo.record_batch(db, batch, client_ip=_client_ip(request))
return AnalyticsIngestOut(received=n)
+37 -5
View File
@@ -12,11 +12,11 @@ from __future__ import annotations
import logging
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi import APIRouter, Depends, HTTPException, Request, status
from app.api.deps import CurrentUser, DbSession
from app.core import test_account
from app.core.ratelimit import rate_limit
from app.core.ratelimit import enforce_rate_limit, rate_limit
from app.core.security import TokenError, decode_token, issue_token_pair
from app.integrations.jiguang import JiguangError, mask_phone, verify_and_get_phone
from app.integrations.sms import SmsError, send_code, verify_code
@@ -38,6 +38,12 @@ logger = logging.getLogger("shagua.auth")
router = APIRouter(prefix="/api/v1/auth", tags=["auth"])
# 手机号登录防刷:同一设备(device_id) + 同一 IP 每小时最多的登录尝试次数(成功/失败都计)。
SMS_LOGIN_MAX_PER_HOUR = 5
# 发码防刷:同一设备(device_id) + 同一 IP 每小时最多的发码次数。
# 堵「换手机号绕开单号 60s 冷却 / 单号每日上限」的洞 —— 那两道是单号维度,一机换号能绕开。
SMS_SEND_MAX_PER_HOUR_PER_DEVICE = 5
def _login_response(
user, *, onboarding_completed: bool, force_onboarding: bool = False
@@ -84,15 +90,27 @@ def jverify_login(req: JverifyLoginRequest, db: DbSession) -> TokenWithUser:
"/sms/send",
response_model=SmsSendResponse,
summary="发送短信验证码",
dependencies=[Depends(rate_limit(10, 60, "sms-send"))], # 同 IP 每分钟≤10 次(防一 IP 刷不同号)
)
def sms_send(req: SmsSendRequest) -> SmsSendResponse:
def sms_send(req: SmsSendRequest, request: Request) -> SmsSendResponse:
# 测试账号:不真发短信(号码非真实手机号,真发会失败/浪费),直接放行让客户端进入填码界面。
# 真正的"免验证码"在 /sms/login 跳过校验;每日上限也在 login 处算(send 不耗额度)。
# (也不受下面设备发码限流约束:QA 联调要反复发码。)
if test_account.is_test_account(req.phone):
logger.info("test_account sms_send short-circuit (不真发)")
return SmsSendResponse(sent=True, mock=True, cooldown_sec=0)
# 防刷:同一设备(device_id) + 同一 IP 每小时最多 SMS_SEND_MAX_PER_HOUR_PER_DEVICE 次发码。
# 补「换手机号绕开单号 60s 冷却 / 单号每日上限」的洞(那两道是单号维度,一机换号能绕);设备维度按机器封顶,
# 挡短信轰炸/烧钱。放在真发(send_code)之前 → 超限直接拦下、不真发短信。与路由上 IP 维度(10次/分钟)互补。
enforce_rate_limit(
request,
scope="sms-send-device",
subject=req.device_id,
limit=SMS_SEND_MAX_PER_HOUR_PER_DEVICE,
window_sec=3600,
detail="操作过于频繁,请稍后再试",
)
try:
cooldown = send_code(req.phone)
except SmsError as e:
@@ -109,9 +127,10 @@ def sms_send(req: SmsSendRequest) -> SmsSendResponse:
summary="手机号+验证码登录",
dependencies=[Depends(rate_limit(20, 60, "sms-login"))], # 防撞库爆破(另有单码失败次数上限)
)
def sms_login(req: SmsLoginRequest, db: DbSession) -> TokenWithUser:
def sms_login(req: SmsLoginRequest, request: Request, db: DbSession) -> TokenWithUser:
# 测试账号:免验证码登录 + 每日上限 + 每次都走新手引导(详见 app/core/test_account.py)。
# 放在最前面:命中即不校验验证码;先扣当日额度,超限直接拒,挡住有人猜到号后脚本刷。
# 测试账号走自己的每日额度、不受下面 (设备+IP) 每小时限流约束(QA 需在一小时内反复登录联调)。
if test_account.is_test_account(req.phone):
if not test_account.try_consume_quota():
raise HTTPException(status_code=429, detail="测试账号今日使用次数已达上限,请明天再试")
@@ -123,6 +142,19 @@ def sms_login(req: SmsLoginRequest, db: DbSession) -> TokenWithUser:
# 不读/不写 onboarding_completion 表。
return _login_response(user, onboarding_completed=False, force_onboarding=True)
# 防刷:同一设备(device_id) + 同一 IP 每小时最多 SMS_LOGIN_MAX_PER_HOUR 次登录尝试。放在验证码校验
# **之前** → 输错验证码的失败尝试也计数,才挡得住撞库/爆破。与路由上 IP 维度的 sms-login 限流(同 IP)互补。
# ⚠️ 按设备而非手机号 → 一台机器换不同手机号刷登录也受限(防一机狂登多号);device_id 空(老客户端)时
# 退化为该 IP 下所有空设备聚一桶,仍受限。
enforce_rate_limit(
request,
scope="sms-login-device",
subject=req.device_id,
limit=SMS_LOGIN_MAX_PER_HOUR,
window_sec=3600,
detail="登录尝试过于频繁,请稍后再试",
)
if not verify_code(req.phone, req.code):
raise HTTPException(status_code=400, detail="invalid sms code")
+2 -2
View File
@@ -63,8 +63,8 @@ def ad_config(db: DbSession) -> AdConfigPublicOut:
return AdConfigPublicOut(
app_id=c["app_id"],
reward_code_id=c["reward_code_id"],
compare_feed_code_id=c["compare_feed_code_id"],
coupon_feed_code_id=c["coupon_feed_code_id"],
compare_draw_code_id=c["compare_draw_code_id"],
coupon_draw_code_id=c["coupon_draw_code_id"],
reward_enabled=c["reward_enabled"],
compare_ad_enabled=c["compare_ad_enabled"],
coupon_ad_enabled=c["coupon_ad_enabled"],
+26
View File
@@ -57,3 +57,29 @@ def rate_limit(limit: int, window_sec: float, scope: str):
)
return _dep
def enforce_rate_limit(
request: Request,
scope: str,
subject: str,
limit: int,
window_sec: float,
*,
detail: str = "操作过于频繁,请稍后再试",
) -> None:
"""在路由内部手动限流,按 (subject, 客户端 IP) 计数。
用于限流 key 需要请求体字段(如手机号)Depends 阶段还拿不到 body
此时无法用 [rate_limit] 依赖,改在 handler 解析完 body 后调用本函数
key = `scope:subject:client_ip`;同一 (subject, IP) window_sec 内超过 limit 429
[settings.RATE_LIMIT_ENABLED] 总开关控制( [rate_limit] 一致)
"""
if not settings.RATE_LIMIT_ENABLED:
return
key = f"{scope}:{subject}:{_client_ip(request)}"
if not _hit(key, limit, window_sec):
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail=detail,
)
+3 -1
View File
@@ -14,7 +14,9 @@ worker / 多机时内存不共享 → 冷却、每日上限、校验都会失效
防刷三层(短信花钱 + `/sms/send` 在登录前无法 JWT 鉴权):
1. 单号 `SMS_SEND_INTERVAL_SEC` 冷却(本文件)
2. 单号每日 `SMS_DAILY_LIMIT_PER_PHONE` 条上限(本文件)
3. IP 频控(api rate_limit 依赖)+ 极光控制台 IP 白名单/防轰炸(运维侧)
3. 设备(device_id)每小时频控(api auth.sms_send enforce_rate_limit)+ 极光控制台 IP 白名单/防轰炸(运维侧)
IP 频控(rate_limit 依赖)2026-06-26 按产品要求删除改设备维度; device_id 客户端可伪造/轮换,
脚本轮换 id 能绕过本层 挡脚本狂发主要靠极光控制台侧(+ 可选 nginx 限流)
:单码校验失败 `SMS_MAX_VERIFY_ATTEMPTS` 次即作废(防爆破),验过即作废(一次性)
"""
from __future__ import annotations
+2
View File
@@ -15,6 +15,7 @@ from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from app.api.v1.ad import router as ad_router
from app.api.v1.analytics import router as analytics_router
from app.api.v1.auth import router as auth_router
from app.api.v1.compare import router as compare_router
from app.api.v1.compare_milestone import router as compare_milestone_router
@@ -105,6 +106,7 @@ def health() -> dict[str, str]:
app.include_router(auth_router)
app.include_router(user_router)
app.include_router(feedback_router)
app.include_router(analytics_router)
app.include_router(invite_router)
app.include_router(coupon_router)
app.include_router(device_router)
+1
View File
@@ -4,6 +4,7 @@ from app.models.ad_feed_reward import AdFeedRewardRecord # noqa: F401
from app.models.ad_reward import AdRewardRecord # noqa: F401
from app.models.ad_watch_log import AdWatchLog # noqa: F401
from app.models.admin import AdminAuditLog, AdminUser # noqa: F401
from app.models.analytics_event import AnalyticsEvent # noqa: F401
from app.models.app_config import AppConfig # noqa: F401
from app.models.comparison import ComparisonRecord # noqa: F401
from app.models.cps_activity import CpsActivity # noqa: F401
+3
View File
@@ -29,6 +29,9 @@ class AdEcpmRecord(Base):
)
# 广告类型:reward_video(激励视频) / draw(Draw 信息流) 等;不强行统一代码位,各类型各自上报
ad_type: Mapped[str] = mapped_column(String(32), nullable=False)
# 点位场景:comparison(比价) / coupon(领券) / welfare(福利),供收益报表区分比价/领券 Draw 收益;
# 仅信息流/Draw 上报(比价与领券共用同一代码位,只能客户端各调用点显式打标),激励视频为 NULL。
feed_scene: Mapped[str | None] = mapped_column(String(16), nullable=True)
# 客户端生成的一次广告会话 id;激励视频 S2S 回调 extra 会透传同值
ad_session_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
# 实际投放的 ADN(穿山甲 getShowEcpm().getSdkName(),如 pangle / gdt)
+3
View File
@@ -30,6 +30,9 @@ class AdFeedRewardRecord(Base):
ecpm_raw: Mapped[str] = mapped_column(String(32), nullable=False)
adn: Mapped[str | None] = mapped_column(String(32), nullable=True)
slot_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
# 广告类型:feed(信息流) / draw(Draw 信息流)。比价与领券共用同一 Draw 代码位,靠 feed_scene
# 区分收益;ad_type 区分广告形态。旧数据(未升级客户端)为 NULL,一律视为 feed,保持向后兼容。
ad_type: Mapped[str | None] = mapped_column(String(16), nullable=True, default="feed")
# 点位场景:comparison(比价等待) / coupon(领券) / welfare(福利页)。比价与领券共用同一信息流
# 代码位,slot_id/our_code_id 分不出,只能客户端各调用点显式打标;NULL=历史/未升级客户端=未分类。
feed_scene: Mapped[str | None] = mapped_column(String(16), nullable=True)
+60
View File
@@ -0,0 +1,60 @@
"""新手引导(及后续)埋点事件表。
每行 = 客户端上报的一条行为埋点,Who / When / Where / What / How五维组织:
- What :event(事件名, video_play)+ props(事件专属属性,JSON)
- Who :device_id(硬件级设备标识)+ user_id(登录后才有,可空)
- When :client_ts(端事件时间 epoch ms)+ session_id(本次引导会话)+ sent_at(端上报时间)
+ created_at(服务端接收时间 = server_at)
- Where:page(引导步/页面)+ client_ip(服务端从 X-Forwarded-For )
- How :oem / os / model / app_ver / network / channel(设备与环境)
append-only,不更新;客户端批量上报( app/api/v1/analytics.py),admin 同库直接查(
app/admin/routers/event_logs.py)未登录态也允许上报(user_id 为空), user_id 不设外键只索引
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import JSON, BigInteger, DateTime, Integer, String, func
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
class AnalyticsEvent(Base):
__tablename__ = "analytics_event"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
# ---- What:做了什么 ----
event: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
props: Mapped[dict | None] = mapped_column(JSON, nullable=True)
# ---- Who:谁 ----
device_id: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
user_id: Mapped[int | None] = mapped_column(Integer, index=True, nullable=True)
# ---- When:何时 ----
session_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
client_ts: Mapped[int] = mapped_column(BigInteger, nullable=False) # 端事件时间 epoch ms
sent_at: Mapped[int | None] = mapped_column(BigInteger, nullable=True) # 端上报时间 epoch ms
# ---- Where:何地 ----
page: Mapped[str | None] = mapped_column(String(64), nullable=True)
client_ip: Mapped[str | None] = mapped_column(String(64), nullable=True)
# ---- How:用什么环境 ----
oem: Mapped[str | None] = mapped_column(String(32), nullable=True)
os: Mapped[str | None] = mapped_column(String(32), nullable=True)
model: Mapped[str | None] = mapped_column(String(64), nullable=True)
app_ver: Mapped[str | None] = mapped_column(String(32), nullable=True)
network: Mapped[str | None] = mapped_column(String(16), nullable=True)
channel: Mapped[str | None] = mapped_column(String(32), nullable=True)
# 服务端接收时间(= When.server_at);客户端时间不可信,以此为权威落库时刻。
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
)
def __repr__(self) -> str: # pragma: no cover
return f"<AnalyticsEvent id={self.id} event={self.event} device={self.device_id}>"
+2
View File
@@ -23,6 +23,7 @@ def create_ecpm_record(
ad_session_id: str | None = None,
adn: str | None = None,
slot_id: str | None = None,
feed_scene: str | None = None,
app_env: str | None = None,
our_code_id: str | None = None,
) -> AdEcpmRecord:
@@ -41,6 +42,7 @@ def create_ecpm_record(
ad_session_id=ad_session_id,
adn=adn,
slot_id=slot_id,
feed_scene=feed_scene,
app_env=app_env,
our_code_id=our_code_id,
ecpm_raw=ecpm_raw,
+6 -1
View File
@@ -14,7 +14,6 @@ from app.core.rewards import cn_today
from app.models.ad_feed_reward import AdFeedRewardRecord
from app.repositories import wallet as crud_wallet
FEED_REWARD_UNIT_SECONDS = 10
# 单个 feed 事件的时长上限(秒):一期 duration_seconds 由客户端上报,伪造超长时长会刷份数
# (每 10 秒 1 份)。真实单条信息流视频远小于此;取 120s=12 份封顶,挡刷量、不影响正规单。
@@ -65,6 +64,7 @@ def grant_feed_reward(
client_event_id: str,
ecpm: str,
duration_seconds: int,
ad_type: str = "feed",
ad_session_id: str | None = None,
adn: str | None = None,
slot_id: str | None = None,
@@ -84,6 +84,8 @@ def grant_feed_reward(
duration_seconds **这一条**的观看秒数服务端两道硬闸防刷:时长钳到 FEED_MAX_DURATION_SECONDS
eCPM calculate_ad_reward_coin 内钳到 AD_ECPM_MAX_FEN;叠加每日 get_ad_daily_limit 条数上限
feed_scene:点位场景(comparison/coupon/welfare),仅归类落库,不参与计算
ad_type:广告形态(feed 信息流 / draw Draw 信息流),仅归类落库;**每日上限与因子2(LT)仍按本表
全表 unit 累计(feed+draw 共享同一发奖池/上限),不按 ad_type 拆分**
"""
existing = _find_by_event(db, client_event_id)
if existing is not None:
@@ -106,6 +108,7 @@ def grant_feed_reward(
ecpm_raw=ecpm,
adn=adn,
slot_id=slot_id,
ad_type=ad_type,
feed_scene=feed_scene,
trace_id=trace_id,
app_env=app_env,
@@ -126,6 +129,7 @@ def grant_feed_reward(
ecpm_raw=ecpm,
adn=adn,
slot_id=slot_id,
ad_type=ad_type,
feed_scene=feed_scene,
trace_id=trace_id,
app_env=app_env,
@@ -147,6 +151,7 @@ def grant_feed_reward(
ecpm_raw=ecpm,
adn=adn,
slot_id=slot_id,
ad_type=ad_type,
feed_scene=feed_scene,
trace_id=trace_id,
app_env=app_env,
+34
View File
@@ -0,0 +1,34 @@
"""埋点事件批量落库。append-only,一次 commit 提交整批。"""
from __future__ import annotations
from sqlalchemy.orm import Session
from app.models.analytics_event import AnalyticsEvent
from app.schemas.analytics import AnalyticsBatchIn
def record_batch(db: Session, batch: AnalyticsBatchIn, *, client_ip: str | None) -> int:
"""把一批上报事件展开成多行落库(公共维度复制到每行),返回写入条数。"""
rows = [
AnalyticsEvent(
event=e.event,
props=e.props or None,
device_id=batch.device_id,
user_id=batch.user_id,
session_id=e.session_id,
client_ts=e.client_ts,
sent_at=batch.sent_at,
page=e.page,
client_ip=client_ip or None,
oem=batch.oem,
os=batch.os,
model=batch.model,
app_ver=batch.app_ver,
network=e.network,
channel=batch.channel,
)
for e in batch.events
]
db.add_all(rows)
db.commit()
return len(rows)
+4 -4
View File
@@ -101,10 +101,10 @@ AD_CONFIG_KEY = "ad_config"
_AD_CONFIG_DEFAULTS: dict[str, Any] = {
"app_id": "5830519", # 穿山甲应用ID(正式)
"reward_code_id": "104099389", # 福利页激励视频位
# ⚠️ 2026-06-21 真机核对穿山甲后台:5830519 名下信息流真实位是 104142227「信息流 1」;
# 旧值 104090333 不在该应用名下(请求会报 44406/配置 null、出不了广告)。客户端接入下发后以本值为准
"compare_feed_code_id": "104142227", # 比价信息流位
"coupon_feed_code_id": "104142227", # 领券信息流位(初始同比价,运营可拆)
# 比价与领券共用同一穿山甲 Draw 代码位(默认 104098712 = 后台「Draw信息流」位),靠 feed_scene
# (comparison/coupon)区分收益;运营可在后台把两者拆成不同代码位。字段从旧 *_feed_code_id 改名为 *_draw_code_id
"compare_draw_code_id": "104098712", # 比价 Draw 代码位(104098712 = Draw 信息流位)
"coupon_draw_code_id": "104098712", # 领券 Draw 代码位(初始同比价,运营可拆)
"reward_mkey": "", # 激励位 GroMore 验签密钥(空则回退 .env PANGLE_REWARD_SECRET*)
"reward_enabled": True, # 福利激励视频开关
"compare_ad_enabled": True, # 比价广告开关
+11
View File
@@ -57,6 +57,12 @@ class EcpmReportIn(BaseModel):
)
adn: str | None = Field(None, description="实际投放 ADN(getSdkName),如 pangle")
slot_id: str | None = Field(None, description="实际展示代码位(底层 mediation rit)")
feed_scene: str | None = Field(
None,
max_length=16,
description="点位场景:comparison(比价等待) / coupon(领券) / welfare(福利页);"
"比价与领券共用同一 Draw 代码位,需客户端在各调用点显式标注,供收益报表区分比价/领券;激励视频为空",
)
app_env: str | None = Field(
None, max_length=16, description="我们的穿山甲应用环境:prod(傻瓜比价正式) / test(测试应用)"
)
@@ -128,6 +134,11 @@ class FeedRewardIn(BaseModel):
"""
client_event_id: str = Field(..., min_length=8, max_length=64, description="客户端生成的幂等事件 id")
ad_type: str = Field(
"feed",
max_length=16,
description="广告类型:feed(信息流) / draw(Draw 信息流);默认 feed 兼容旧客户端",
)
ad_session_id: str | None = Field(
None, min_length=8, max_length=64, description="客户端生成的一次信息流广告会话 id"
)
+42
View File
@@ -0,0 +1,42 @@
"""客户端埋点上报 schema(批量)。
客户端把设备固定维度(device_id / user_id / oem / os / model / app_ver / channel / sent_at)
放批次外层只传一次,events 列表里每条只带事件维度(event / client_ts / session_id / page /
network / props);服务端展开成多行 AnalyticsEvent 落库( app/repositories/analytics.py)
"""
from __future__ import annotations
from pydantic import BaseModel, Field
class AnalyticsEventIn(BaseModel):
"""单条事件维度。"""
event: str = Field(max_length=64)
client_ts: int = Field(description="端事件发生时间 epoch ms")
session_id: str | None = Field(default=None, max_length=64)
page: str | None = Field(default=None, max_length=64)
network: str | None = Field(default=None, max_length=16)
props: dict[str, str] = Field(default_factory=dict)
class AnalyticsBatchIn(BaseModel):
"""一批上报:公共维度 + 事件列表。"""
# Who(整批共享)
device_id: str = Field(max_length=64)
user_id: int | None = None
# When(本批上报时刻 epoch ms)
sent_at: int | None = None
# How(设备固定维度,整批共享)
oem: str | None = Field(default=None, max_length=32)
os: str | None = Field(default=None, max_length=32)
model: str | None = Field(default=None, max_length=64)
app_ver: str | None = Field(default=None, max_length=32)
channel: str | None = Field(default=None, max_length=32)
events: list[AnalyticsEventIn] = Field(min_length=1, max_length=200)
class AnalyticsIngestOut(BaseModel):
ok: bool = True
received: int
+4
View File
@@ -71,6 +71,10 @@ class JverifyLoginRequest(BaseModel):
class SmsSendRequest(BaseModel):
phone: str = Field(..., min_length=11, max_length=11, pattern=r"^1\d{10}$")
device_id: str = Field(
"", max_length=64,
description="硬件级设备标识(Android ANDROID_ID),用于发码防刷按 设备+IP 限流;空=按 IP 聚一桶",
)
class SmsSendResponse(BaseModel):
+2 -2
View File
@@ -38,8 +38,8 @@ class AdConfigPublicOut(BaseModel):
app_id: str # 穿山甲应用ID(改了客户端需冷启才生效,SDK init 一次性读)
reward_code_id: str # 福利页激励视频位
compare_feed_code_id: str # 比价信息流
coupon_feed_code_id: str # 领券信息流位
compare_draw_code_id: str # 比价 Draw 代码
coupon_draw_code_id: str # 领券 Draw 代码位(比价/领券共用同一位,靠 feed_scene 区分收益)
reward_enabled: bool # 福利激励视频开关
compare_ad_enabled: bool # 比价广告开关
coupon_ad_enabled: bool # 领券广告开关
+9
View File
@@ -12,6 +12,7 @@ FastAPI TestClient 在进程内打**真实** endpoint(`/api/v1/ad/pangle-callbac
from __future__ import annotations
import argparse
import json
import os
import sys
import uuid
@@ -56,6 +57,11 @@ def main() -> None:
p = argparse.ArgumentParser(description="模拟穿山甲发奖回调")
p.add_argument("user_id", nargs="?", type=int, default=None, help="目标 user_id(默认最近注册)")
p.add_argument("--trans", default=None, help="交易号(默认随机);同号重跑验幂等")
p.add_argument(
"--srv-env", default=None,
help="模拟客户端 mediaExtra 里的后端环境标记(dev/prod)。与本服 APP_ENV 不符时"
"应被环境隔离拦下(Δ0、不串号);留空=不带该字段(旧客户端,照常发)",
)
args = p.parse_args()
if not settings.pangle_callback_configured:
@@ -75,6 +81,9 @@ def main() -> None:
# 镜像生产:后台代码位"奖励数量"应配成 AD_REWARD_COIN,回调据此发金币
"reward_amount": str(AD_REWARD_COIN),
}
# 镜像穿山甲透传 mediaExtra:把后端环境标记放进 gromoreExtra(服务端据此做环境隔离)
if args.srv_env:
params["gromoreExtra"] = json.dumps({"srv_env": args.srv_env})
params["sign"] = pangle.build_sign(trans_id, settings.PANGLE_REWARD_SECRET)
resp = TestClient(app).get("/api/v1/ad/pangle-callback", params=params)
+34
View File
@@ -5,6 +5,8 @@
"""
from __future__ import annotations
import json
from sqlalchemy import select
from app.core.config import settings
@@ -179,6 +181,38 @@ def test_callback_unknown_user(client) -> None:
assert r.json() == {"is_verify": False, "reason": 2}
def test_callback_foreign_env_skipped(client) -> None:
"""回调 srv_env 与本服 APP_ENV 不符(测试包的 S2S 打到生产)→ 受理但不发币、不串号。
防的是跨库串号:同一 user_id 在测试库 / 正式库是两个人,回调地址只配生产,测试包(dev)看广告
S2S 也会打到生产测试环境 APP_ENV=dev:srv_env=prod 视为别的环境 跳过,余额不变
不写记录;srv_env=dev(本环境)照常发srv_env gromoreExtra 透传(穿山甲带回 mediaExtra 的字段)
兼容:不带 srv_env 的旧客户端按本环境处理(其余用例已覆盖)
"""
phone = "13800003501"
token = _login(client, phone)
uid = _user_id(phone)
# 外环境(prod)回调:受理(is_verify=true 防穿山甲重试)但不发币、不留记录
foreign = _signed(uid, "trans_foreign_env", ecpm="200",
gromoreExtra=json.dumps({"srv_env": "prod"}))
r = _callback(client, foreign)
assert r.status_code == 200, r.text
assert r.json() == {"is_verify": True, "reason": 0}
assert _coin_balance(client, token) == 0
st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
assert st["used_today"] == 0
# 本环境(dev)回调:照常发币
same = _signed(uid, "trans_same_env", ecpm="200",
gromoreExtra=json.dumps({"srv_env": "dev"}))
r = _callback(client, same)
assert r.status_code == 200, r.text
assert _coin_balance(client, token) == calculate_ad_reward_coin("200", 1)
st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
assert st["used_today"] == 1
def test_reward_status_requires_auth(client) -> None:
"""客户端进度接口需登录。"""
r = client.get("/api/v1/ad/reward-status")
+67
View File
@@ -74,6 +74,73 @@ def test_sms_send_too_frequent(client) -> None:
assert client.post("/api/v1/auth/sms/send", json={"phone": phone}).status_code == 429
def test_sms_send_device_ip_rate_limit(client, monkeypatch) -> None:
"""发码防刷:同一设备(device_id) + 同一 IP 每小时最多 N 次发码,超出 429。
用不同手机号(绕开单号 60s 冷却)证明限流按设备封顶 换号绕开单号冷却/每日上限的洞
conftest 默认关限流;本用例临时打开 + 调小阈值便于测 + 清计数隔离"""
from app.api.v1 import auth
from app.core import ratelimit
monkeypatch.setattr(ratelimit.settings, "RATE_LIMIT_ENABLED", True)
monkeypatch.setattr(auth, "SMS_SEND_MAX_PER_HOUR_PER_DEVICE", 3)
ratelimit._buckets.clear()
device = "dev-send-A"
# 同设备、每次换不同手机号(绕开单号冷却)发码,前 3 次 200
for i in range(3):
r = client.post(
"/api/v1/auth/sms/send",
json={"phone": f"137001370{i:02d}", "device_id": device},
)
assert r.status_code == 200, f"{i + 1} 次应放行: {r.text}"
# 第 4 次:同设备超限 → 429(即便又换了手机号)
r = client.post(
"/api/v1/auth/sms/send",
json={"phone": "13700137003", "device_id": device},
)
assert r.status_code == 429, r.text
# 换个设备 → 独立计数(限流键含 device_id),放行
r = client.post(
"/api/v1/auth/sms/send",
json={"phone": "13700137004", "device_id": "dev-send-B"},
)
assert r.status_code == 200, r.text
def test_sms_login_device_ip_rate_limit(client, monkeypatch) -> None:
"""防刷:同一设备(device_id) + 同一 IP 每小时最多 SMS_LOGIN_MAX_PER_HOUR 次登录尝试,超出 429。
conftest 默认 RATE_LIMIT_ENABLED=false(内存计数跨用例累加),本用例临时打开并清空计数隔离"""
from app.api.v1 import auth
from app.core import ratelimit
monkeypatch.setattr(ratelimit.settings, "RATE_LIMIT_ENABLED", True)
ratelimit._buckets.clear() # 隔离:清掉跨用例累加的内存计数
device = "dev-rl-A"
# 前 N 次故意每次换手机号、固定设备:mock 校验放行 → 200。证明限流按设备而非手机号,同设备共用一个桶
for i in range(auth.SMS_LOGIN_MAX_PER_HOUR):
r = client.post(
"/api/v1/auth/sms/login",
json={"phone": f"135331533{i:02d}", "code": "123456", "device_id": device},
)
assert r.status_code == 200, f"{i + 1} 次应放行: {r.text}"
# 第 N+1 次:同设备同 IP 超限 → 429(即便又换了个手机号)
r = client.post(
"/api/v1/auth/sms/login",
json={"phone": "13533153399", "code": "123456", "device_id": device},
)
assert r.status_code == 429, r.text
assert "频繁" in r.json()["detail"]
# 同 IP 换个设备 → 独立计数(限流键含 device_id、非纯 IP/手机号),仍放行
r = client.post(
"/api/v1/auth/sms/login",
json={"phone": "13533153300", "code": "123456", "device_id": "dev-rl-B"},
)
assert r.status_code == 200, r.text
def test_sms_login_bad_code(client) -> None:
"""mock 下校验位数:5 位过 schema 但业务要 6 位 → 400。"""
phone = "13700137000"
+48
View File
@@ -142,6 +142,54 @@ def test_send_does_not_consume_quota(client, enabled, monkeypatch) -> None:
assert client.post("/api/v1/auth/sms/login", json={"phone": TEST_PHONE, "code": "1234"}).status_code == 429
# ============================ 登录限流豁免 ============================
def test_exempt_from_device_ip_login_rate_limit(client, enabled, monkeypatch) -> None:
"""测试号豁免 (设备+IP) 每小时登录限流。
普通设备同设备同 IP SMS_LOGIN_MAX_PER_HOUR 次即被拦( test_auth.test_sms_login_device_ip_rate_limit);
测试号走自己的每日额度不受这道限流这里临时打开 RATE_LIMIT_ENABLED,同一设备连登远超该上限仍全 200
回归用:防有人把 enforce_rate_limit 挪到测试账号 early-return 之前而破坏豁免"""
from app.api.v1 import auth
from app.core import ratelimit
monkeypatch.setattr(ratelimit.settings, "RATE_LIMIT_ENABLED", True)
ratelimit._buckets.clear() # 隔离:清掉跨用例累加的内存计数
# 同一设备连登 (每小时上限 + 3) 次,远超普通设备会被拦的阈值;测试号豁免 → 全部放行
for i in range(auth.SMS_LOGIN_MAX_PER_HOUR + 3):
r = client.post(
"/api/v1/auth/sms/login",
json={"phone": TEST_PHONE, "code": "000000", "device_id": "dev-test-exempt"},
)
assert r.status_code == 200, (
f"测试号第 {i + 1} 次登录应放行(豁免每小时限流),实际 {r.status_code}: {r.text}"
)
def test_exempt_from_device_ip_send_rate_limit(client, enabled, monkeypatch) -> None:
"""测试号豁免 (设备+IP) 每小时**发码**限流(与登录那道同样靠早返回豁免)。
普通设备同设备到 SMS_SEND_MAX_PER_HOUR_PER_DEVICE 次发码即被拦( test_auth.test_sms_send_device_ip_rate_limit);
测试号发码本就短路(不真发)且在限流之前早返回 同一设备连发远超上限仍全 200
回归用:防有人把发码限流挪到测试账号 early-return 之前而把 QA 联调也风控了"""
from app.api.v1 import auth
from app.core import ratelimit
monkeypatch.setattr(ratelimit.settings, "RATE_LIMIT_ENABLED", True)
monkeypatch.setattr(auth, "SMS_SEND_MAX_PER_HOUR_PER_DEVICE", 2) # 调小:不豁免的话第 3 次就该被拦
ratelimit._buckets.clear()
for i in range(auth.SMS_SEND_MAX_PER_HOUR_PER_DEVICE + 2):
r = client.post(
"/api/v1/auth/sms/send",
json={"phone": TEST_PHONE, "device_id": "dev-test-send"},
)
assert r.status_code == 200, (
f"测试号第 {i + 1} 次发码应放行(豁免每小时限流),实际 {r.status_code}: {r.text}"
)
# ============================ 计数单元逻辑 ============================
def test_quota_resets_across_day(enabled, monkeypatch) -> None: