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
11 changed files with 361 additions and 0 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 ###
@@ -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)
@@ -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(
@@ -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,
+    )
@@ -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
@@ -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)
@@ -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)
@@ -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
@@ -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}>"
@@ -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)
@@ -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