chore: CPS 数据表字典 + 按群对账统计 mock 脚本 + CLAUDE.md

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

alembic/versions/1699fc2c069f_add_analytics_event_table.py

@@ -1,65 +0,0 @@
-"""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 ###

app/admin/main.py

@@ -24,7 +24,6 @@ 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
@@ -92,7 +91,6 @@ 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)

app/admin/repositories/queries.py

@@ -15,7 +15,6 @@ 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
@@ -413,44 +412,6 @@ 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(

app/admin/routers/event_logs.py

@@ -1,52 +0,0 @@
-"""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,
-    )

app/admin/schemas/analytics.py

@@ -1,33 +0,0 @@
-"""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

app/api/v1/ad.py

@@ -120,21 +120,6 @@ 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(

app/api/v1/analytics.py

@@ -1,31 +0,0 @@
-"""客户端埋点上报接口。
-
-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)

app/api/v1/auth.py

@@ -12,11 +12,11 @@ from __future__ import annotations
 
 import logging
 
-from fastapi import APIRouter, Depends, HTTPException, Request, status
+from fastapi import APIRouter, Depends, HTTPException, status
 
 from app.api.deps import CurrentUser, DbSession
 from app.core import test_account
-from app.core.ratelimit import enforce_rate_limit, rate_limit
+from app.core.ratelimit import 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,12 +38,6 @@ 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 = 10
-
 
 def _login_response(
     user, *, onboarding_completed: bool, force_onboarding: bool = False
@@ -92,26 +86,13 @@ def jverify_login(req: JverifyLoginRequest, db: DbSession) -> TokenWithUser:
     summary="发送短信验证码",
     dependencies=[Depends(rate_limit(10, 60, "sms-send"))],  # 同 IP 每分钟≤10 次(防一 IP 刷不同号)
 )
-def sms_send(req: SmsSendRequest, request: Request) -> SmsSendResponse:
+def sms_send(req: SmsSendRequest) -> 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:
@@ -128,10 +109,9 @@ def sms_send(req: SmsSendRequest, request: Request) -> SmsSendResponse:
     summary="手机号+验证码登录",
     dependencies=[Depends(rate_limit(20, 60, "sms-login"))],  # 防撞库爆破(另有单码失败次数上限)
 )
-def sms_login(req: SmsLoginRequest, request: Request, db: DbSession) -> TokenWithUser:
+def sms_login(req: SmsLoginRequest, 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="测试账号今日使用次数已达上限,请明天再试")
@@ -143,19 +123,6 @@ def sms_login(req: SmsLoginRequest, request: Request, db: DbSession) -> TokenWit
         # 不读/不写 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")
 

app/core/ratelimit.py

@@ -57,29 +57,3 @@ 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,
-        )

app/main.py

@@ -15,7 +15,6 @@ 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
@@ -106,7 +105,6 @@ 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)

app/models/__init__.py

@@ -4,7 +4,6 @@ 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

app/models/analytics_event.py

@@ -1,60 +0,0 @@
-"""新手引导(及后续)埋点事件表。
-
-每行 = 客户端上报的一条行为埋点,按「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}>"

app/repositories/analytics.py

@@ -1,34 +0,0 @@
-"""埋点事件批量落库。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)

app/schemas/analytics.py

@@ -1,42 +0,0 @@
-"""客户端埋点上报 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

app/schemas/auth.py

@@ -71,10 +71,6 @@ 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):

scripts/sim_pangle_callback.py

@@ -12,7 +12,6 @@ FastAPI TestClient 在进程内打**真实** endpoint(`/api/v1/ad/pangle-callbac
 from __future__ import annotations
 
 import argparse
-import json
 import os
 import sys
 import uuid
@@ -57,11 +56,6 @@ 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:
@@ -81,9 +75,6 @@ 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)

tests/test_ad_reward.py

@@ -5,8 +5,6 @@
 """
 from __future__ import annotations
 
-import json
-
 from sqlalchemy import select
 
 from app.core.config import settings
@@ -181,38 +179,6 @@ 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")

tests/test_auth.py

@@ -74,73 +74,6 @@ 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 默认关限流;本用例临时打开 + 调小阈值(避开同 IP 10/分钟那道)+ 清计数隔离。"""
-    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"

tests/test_test_account.py

@@ -142,54 +142,6 @@ 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: