From 667cda566f8ba74ec98e564d930b5984384b2b5e Mon Sep 17 00:00:00 2001 From: zhuzihao Date: Fri, 26 Jun 2026 23:33:31 +0800 Subject: [PATCH] =?UTF-8?q?feat(analytics):=20=E5=9F=8B=E7=82=B9=E4=BA=8B?= =?UTF-8?q?=E4=BB=B6=E6=8E=A5=E6=94=B6=E6=8E=A5=E5=8F=A3=20+=20admin=20?= =?UTF-8?q?=E6=9F=A5=E8=AF=A2=E6=8E=A5=E5=8F=A3=20+=20=E5=9F=8B=E7=82=B9?= =?UTF-8?q?=E8=A1=A8=20(#81)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新手引导埋点的服务端: - 表 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 --------- Co-authored-by: zzhyyyyy <2685922758@qq.com> Reviewed-on: https://gitea.shaguabijia.com/WonderableAI/shaguabijia-app-server/pulls/81 Co-authored-by: zhuzihao Co-committed-by: zhuzihao --- .../1699fc2c069f_add_analytics_event_table.py | 65 +++++++++++++++++++ app/admin/main.py | 2 + app/admin/repositories/queries.py | 39 +++++++++++ app/admin/routers/event_logs.py | 52 +++++++++++++++ app/admin/schemas/analytics.py | 33 ++++++++++ app/api/v1/analytics.py | 31 +++++++++ app/main.py | 2 + app/models/__init__.py | 1 + app/models/analytics_event.py | 60 +++++++++++++++++ app/repositories/analytics.py | 34 ++++++++++ app/schemas/analytics.py | 42 ++++++++++++ 11 files changed, 361 insertions(+) create mode 100644 alembic/versions/1699fc2c069f_add_analytics_event_table.py create mode 100644 app/admin/routers/event_logs.py create mode 100644 app/admin/schemas/analytics.py create mode 100644 app/api/v1/analytics.py create mode 100644 app/models/analytics_event.py create mode 100644 app/repositories/analytics.py create mode 100644 app/schemas/analytics.py diff --git a/alembic/versions/1699fc2c069f_add_analytics_event_table.py b/alembic/versions/1699fc2c069f_add_analytics_event_table.py new file mode 100644 index 0000000..3e51ca5 --- /dev/null +++ b/alembic/versions/1699fc2c069f_add_analytics_event_table.py @@ -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 ### diff --git a/app/admin/main.py b/app/admin/main.py index e4027fb..57d044a 100644 --- a/app/admin/main.py +++ b/app/admin/main.py @@ -25,6 +25,7 @@ from app.admin.routers.cps import router as cps_router from app.admin.routers.dashboard import router as dashboard_router from app.admin.routers.device_liveness import router as device_liveness_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 @@ -93,6 +94,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) diff --git a/app/admin/repositories/queries.py b/app/admin/repositories/queries.py index 5afe09a..93a2d80 100644 --- a/app/admin/repositories/queries.py +++ b/app/admin/repositories/queries.py @@ -16,6 +16,7 @@ from app.core.config import settings 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.device import DeviceLiveness from app.models.feedback import Feedback @@ -564,6 +565,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( diff --git a/app/admin/routers/event_logs.py b/app/admin/routers/event_logs.py new file mode 100644 index 0000000..eceb155 --- /dev/null +++ b/app/admin/routers/event_logs.py @@ -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, + ) diff --git a/app/admin/schemas/analytics.py b/app/admin/schemas/analytics.py new file mode 100644 index 0000000..2aef672 --- /dev/null +++ b/app/admin/schemas/analytics.py @@ -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 diff --git a/app/api/v1/analytics.py b/app/api/v1/analytics.py new file mode 100644 index 0000000..265954d --- /dev/null +++ b/app/api/v1/analytics.py @@ -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) diff --git a/app/main.py b/app/main.py index 5a4a6bd..dd7ebbd 100644 --- a/app/main.py +++ b/app/main.py @@ -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) diff --git a/app/models/__init__.py b/app/models/__init__.py index 05c6c57..35662b3 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -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 diff --git a/app/models/analytics_event.py b/app/models/analytics_event.py new file mode 100644 index 0000000..63110c3 --- /dev/null +++ b/app/models/analytics_event.py @@ -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"" diff --git a/app/repositories/analytics.py b/app/repositories/analytics.py new file mode 100644 index 0000000..a43dcb3 --- /dev/null +++ b/app/repositories/analytics.py @@ -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) diff --git a/app/schemas/analytics.py b/app/schemas/analytics.py new file mode 100644 index 0000000..f0347c0 --- /dev/null +++ b/app/schemas/analytics.py @@ -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