From 37fd51a498a908d4b5a72d4538c6f1077021b6ba Mon Sep 17 00:00:00 2001 From: guke Date: Thu, 9 Jul 2026 17:31:24 +0800 Subject: [PATCH] =?UTF-8?q?feat(analytics):=20=E5=9F=8B=E7=82=B9/=E4=B8=8A?= =?UTF-8?q?=E6=8A=A5=E5=81=A5=E5=BA=B7=E5=BA=A6=E8=87=AA=E6=8A=A5=E8=AE=A1?= =?UTF-8?q?=E6=95=B0(selfstat)+=20admin=20=E5=81=A5=E5=BA=B7=E5=BA=A6?= =?UTF-8?q?=E8=81=9A=E5=90=88=E7=AB=AF=E7=82=B9=20(#127)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 背景 / 目标 客户端埋点从「采集」到「上报落库」全链路存在丢失(采集端丢帧、网络丢包、上报失败),现有 analytics_event(行为事件)无法度量这条链路的健康度。本 MR 引入一条与行为埋点完全独立的自报计数(selfstat)链路:客户端周期上报「自 epoch 起算的累计计数」,服务端只 append 存原始快照,admin 查询时在 Python 侧差分聚合出两段成功率(埋点成功率 / 上报成功率),支持总览 / 按天趋势 / 按维度下钻。 改动概览 ① App 侧 — 上报接入(不强制登录) 新增 POST /api/v1/analytics/selfstat:一份快照 = 设备头 + N 条 event 累计计数,单事务落库,返回 snapshot_id。 稳态兜底:落库异常不裸奔 500,logger.exception 后回 503(计数链路要稳,不阻塞端上主流程)。 events 允许空列表(某次只报设备级计数也合法),上限 200 条/次;与行为埋点 min_length=1 有意不同。 ② Admin 侧 — 健康度聚合(只读,需 admin 鉴权) 新增 GET /admin/api/analytics-health/{overview,trend,breakdown}。 差分聚合:原始累计快照按 (device_id, epoch_id, event) 分区、created_at 升序做相邻差分;分区首行增量=累计值,负值(epoch 重置 / 乱序)夹 0;每条增量按其快照 created_at 归入北京天桶。 基线行:取 date_from 左侧每分区最后一条快照,保证区间第一条增量正确(参与差分后丢弃)。 两段率(分母为 0 → null): 埋点成功率 track_success_rate = (attempted − drop_capture) / attempted 上报成功率 report_success_rate = delivered / (delivered + drop_undelivered) breakdown 维度限 event | app_ver | oem(路由正则校验);结果按上报成功率升序(最差在前,None 垫底)。 ③ 数据模型 / 基建 新表 analytics_selfstat(快照头 + 设备维度 app_ver/oem/os + 设备级诊断量 batches_attempted/ok/fail、retries、queue_depth、端 sent_at + 服务端权威 created_at)与 analytics_selfstat_event(四类累计计数,FK → 快照头)。 Alembic 迁移 11c44afbea58(down_revision admin_user_plain_password):建两表 + 索引。 模型登记进 app/models/__init__.py(供 Alembic 发现);路由挂进 app/admin/main.py。 关键设计取舍 只存原始累计、查询时 Python 差分:admin 低频、量级小,跨 PG/SQLite 无方言坑(与 cps.py/coupon_data.py 同款约定)。 选「最新一条」用 max(id) 而非 max(created_at):id 严格单调,规避 SQLite 秒级时间戳撞车的歧义。 tz 口径统一:SQLite 返回 naive UTC、PG 返回 aware UTC,过滤前统一转 naive UTC 再比较。 基线子查询无下界:扫 date_from 左侧全量(spec §7 已接受的取舍;量级变大再上物化 rollup)。 测试(+14,全绿) tests/test_analytics_selfstat.py(3):正常落库 / 空 events / 落库异常回 503。 tests/test_analytics_health.py(11):差分逻辑(首行=累计值、相邻差、epoch 重置换分区、乱序夹 0、同时间戳按 id 定序)、两段率公式、零分母→None、北京天边界、端点(overview 鉴权 / overview / breakdown)。 迁移 / 部署注意 部署需执行 alembic upgrade head(建两张新表)。纯新增表,无回填、无破坏性改动,向后兼容。 客户端需按约定 payload(snake_case、累计语义)对接 /api/v1/analytics/selfstat;admin 前端(独立仓 shaguabijia-admin-web)消费三个 analytics-health 端点,不在本 MR 范围。 --------- Co-authored-by: guke Reviewed-on: https://gitea.shaguabijia.com/WonderableAI/shaguabijia-app-server/pulls/127 --- .../11c44afbea58_analytics_selfstat_tables.py | 72 ++++++++ app/admin/main.py | 2 + app/admin/repositories/analytics_health.py | 146 ++++++++++++++++ app/admin/routers/analytics_health.py | 49 ++++++ app/admin/schemas/analytics_health.py | 21 +++ app/api/v1/analytics.py | 17 +- app/models/__init__.py | 4 + app/models/analytics_selfstat.py | 59 +++++++ app/repositories/analytics_selfstat.py | 38 +++++ app/schemas/analytics_selfstat.py | 34 ++++ tests/test_analytics_health.py | 160 ++++++++++++++++++ tests/test_analytics_selfstat.py | 43 +++++ 12 files changed, 644 insertions(+), 1 deletion(-) create mode 100644 alembic/versions/11c44afbea58_analytics_selfstat_tables.py create mode 100644 app/admin/repositories/analytics_health.py create mode 100644 app/admin/routers/analytics_health.py create mode 100644 app/admin/schemas/analytics_health.py create mode 100644 app/models/analytics_selfstat.py create mode 100644 app/repositories/analytics_selfstat.py create mode 100644 app/schemas/analytics_selfstat.py create mode 100644 tests/test_analytics_health.py create mode 100644 tests/test_analytics_selfstat.py diff --git a/alembic/versions/11c44afbea58_analytics_selfstat_tables.py b/alembic/versions/11c44afbea58_analytics_selfstat_tables.py new file mode 100644 index 0000000..1184a31 --- /dev/null +++ b/alembic/versions/11c44afbea58_analytics_selfstat_tables.py @@ -0,0 +1,72 @@ +"""analytics_selfstat tables + +Revision ID: 11c44afbea58 +Revises: admin_user_plain_password +Create Date: 2026-07-08 16:32:49.351817 + +""" +from collections.abc import Sequence + +import sqlalchemy as sa + +from alembic import op + + +# revision identifiers, used by Alembic. +revision: str = '11c44afbea58' +down_revision: str | Sequence[str] | None = 'admin_user_plain_password' +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.create_table( + 'analytics_selfstat', + sa.Column('id', sa.Integer(), autoincrement=True, nullable=False), + sa.Column('device_id', sa.String(length=64), nullable=False), + sa.Column('epoch_id', sa.String(length=64), nullable=False), + sa.Column('app_ver', sa.String(length=32), nullable=True), + sa.Column('oem', sa.String(length=32), nullable=True), + sa.Column('os', sa.String(length=32), nullable=True), + sa.Column('batches_attempted', sa.BigInteger(), nullable=False, server_default='0'), + sa.Column('batches_ok', sa.BigInteger(), nullable=False, server_default='0'), + sa.Column('batches_fail', sa.BigInteger(), nullable=False, server_default='0'), + sa.Column('retries', sa.BigInteger(), nullable=False, server_default='0'), + sa.Column('queue_depth', sa.Integer(), nullable=False, server_default='0'), + sa.Column('sent_at', sa.BigInteger(), 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_selfstat', schema=None) as batch_op: + batch_op.create_index(batch_op.f('ix_analytics_selfstat_created_at'), ['created_at'], unique=False) + batch_op.create_index(batch_op.f('ix_analytics_selfstat_device_id'), ['device_id'], unique=False) + batch_op.create_index(batch_op.f('ix_analytics_selfstat_epoch_id'), ['epoch_id'], unique=False) + + op.create_table( + 'analytics_selfstat_event', + sa.Column('id', sa.Integer(), autoincrement=True, nullable=False), + sa.Column('snapshot_id', sa.Integer(), nullable=False), + sa.Column('event', sa.String(length=64), nullable=False), + sa.Column('attempted', sa.BigInteger(), nullable=False, server_default='0'), + sa.Column('drop_capture', sa.BigInteger(), nullable=False, server_default='0'), + sa.Column('delivered', sa.BigInteger(), nullable=False, server_default='0'), + sa.Column('drop_undelivered', sa.BigInteger(), nullable=False, server_default='0'), + sa.ForeignKeyConstraint(['snapshot_id'], ['analytics_selfstat.id'], ), + sa.PrimaryKeyConstraint('id'), + ) + with op.batch_alter_table('analytics_selfstat_event', schema=None) as batch_op: + batch_op.create_index(batch_op.f('ix_analytics_selfstat_event_event'), ['event'], unique=False) + batch_op.create_index(batch_op.f('ix_analytics_selfstat_event_snapshot_id'), ['snapshot_id'], unique=False) + + +def downgrade() -> None: + with op.batch_alter_table('analytics_selfstat_event', schema=None) as batch_op: + batch_op.drop_index(batch_op.f('ix_analytics_selfstat_event_snapshot_id')) + batch_op.drop_index(batch_op.f('ix_analytics_selfstat_event_event')) + op.drop_table('analytics_selfstat_event') + with op.batch_alter_table('analytics_selfstat', schema=None) as batch_op: + batch_op.drop_index(batch_op.f('ix_analytics_selfstat_epoch_id')) + batch_op.drop_index(batch_op.f('ix_analytics_selfstat_device_id')) + batch_op.drop_index(batch_op.f('ix_analytics_selfstat_created_at')) + op.drop_table('analytics_selfstat') diff --git a/app/admin/main.py b/app/admin/main.py index 89669fc..7345803 100644 --- a/app/admin/main.py +++ b/app/admin/main.py @@ -26,6 +26,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.analytics_health import router as analytics_health_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 @@ -97,6 +98,7 @@ 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(analytics_health_router) admin_app.include_router(feedback_qr_router) admin_app.include_router(admins_router) admin_app.include_router(roles_router) diff --git a/app/admin/repositories/analytics_health.py b/app/admin/repositories/analytics_health.py new file mode 100644 index 0000000..6098963 --- /dev/null +++ b/app/admin/repositories/analytics_health.py @@ -0,0 +1,146 @@ +"""埋点健康度聚合(埋点成功率 / 上报成功率)。 + +只存原始累计快照,查询时在 Python 侧差分聚合(admin 低频、量级小,跨 PG/SQLite 无方言坑; +与 cps.py / coupon_data.py 同款约定)。差分按 (device_id, epoch_id, event) 分区、created_at +升序,相邻做差、负值夹 0;每增量按其快照 created_at 归入北京天桶。 +""" +from __future__ import annotations + +from collections import defaultdict +from datetime import UTC, datetime + +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from app.core import rewards +from app.models.analytics_selfstat import AnalyticsSelfStat as H +from app.models.analytics_selfstat import AnalyticsSelfStatEvent as E + +_COUNTS = ("attempted", "drop_capture", "delivered", "drop_undelivered") + + +def diff_snapshots(rows: list[dict]) -> list[dict]: + """累计快照行 → 每快照增量行(纯逻辑)。 + + rows 每行含 device_id/epoch_id/event/created_at/app_ver/oem/os + 四个累计计数。 + 返回每行含 dims + created_at + 四个增量 d_*(分区首行增量=累计值;负值夹 0)。 + """ + parts: dict[tuple, list[dict]] = defaultdict(list) + for r in rows: + parts[(r["device_id"], r["epoch_id"], r["event"])].append(r) + + out: list[dict] = [] + for group in parts.values(): + group.sort(key=lambda r: (r["created_at"], r.get("id", 0))) + prev = {k: 0 for k in _COUNTS} + for r in group: + deltas = {f"d_{k}": max(0, int(r[k]) - prev[k]) for k in _COUNTS} + out.append({ + "device_id": r["device_id"], "epoch_id": r["epoch_id"], "event": r["event"], + "created_at": r["created_at"], "app_ver": r["app_ver"], + "oem": r["oem"], "os": r["os"], **deltas, + }) + prev = {k: int(r[k]) for k in _COUNTS} + return out + + +def _cn_day(dt: datetime) -> str: + """created_at(UTC 口径)→ 北京日期字符串 YYYY-MM-DD。naive 当 UTC,tz-aware 直接换算。""" + if dt.tzinfo is None: + dt = dt.replace(tzinfo=UTC) + return dt.astimezone(rewards.CN_TZ).date().isoformat() + + +def _rates(sums: dict) -> dict: + """由四个增量和派生两段率(分母 0 → None)。""" + persisted_denom = sums["attempted"] + report_denom = sums["delivered"] + sums["drop_undelivered"] + return { + **sums, + "track_success_rate": ( + (sums["attempted"] - sums["drop_capture"]) / persisted_denom + if persisted_denom else None + ), + "report_success_rate": ( + sums["delivered"] / report_denom if report_denom else None + ), + } + + +def _sum_deltas(deltas: list[dict]) -> dict: + return {k: sum(d[f"d_{k}"] for d in deltas) for k in _COUNTS} + + +def _fetch_rows(db: Session, date_from: datetime, date_to: datetime) -> list[dict]: + """取 [from, to) 区间行 + 每分区在 from 左侧的最后一条基线行(供第一条区间增量做差)。""" + cols = ( + H.id, H.device_id, H.epoch_id, E.event, H.created_at, + H.app_ver, H.oem, H.os, + E.attempted, E.drop_capture, E.delivered, E.drop_undelivered, + ) + in_range = db.execute( + select(*cols).join(E, E.snapshot_id == H.id) + .where(H.created_at >= date_from, H.created_at < date_to) + ).mappings().all() + + # 注:基线子查询无下界扫 from 左侧全量(spec §7 已接受的取舍;量级变大再上物化 rollup)。 + # 用 max(id) 而非 max(created_at) 选"最新一条":id 严格单调,避免 SQLite 秒级时间戳撞车时选歧义。 + sub = ( + select(H.device_id, H.epoch_id, E.event, func.max(H.id).label("max_id")) + .join(E, E.snapshot_id == H.id) + .where(H.created_at < date_from) + .group_by(H.device_id, H.epoch_id, E.event) + .subquery() + ) + baseline = db.execute( + select(*cols).join(E, E.snapshot_id == H.id).join( + sub, sub.c.max_id == H.id + ) + ).mappings().all() + + return [dict(r) for r in list(baseline) + list(in_range)] + + +def _in_range_deltas(db: Session, date_from: datetime, date_to: datetime) -> list[dict]: + """差分后只保留 created_at ∈ [from, to) 的增量(基线行被差分用后丢弃)。 + + Python 侧过滤需对齐 tz 口径:SQLite 返回 naive UTC,PG 返回 aware UTC。 + 统一转成 naive UTC 再比较,兼容两种后端。 + """ + def _to_naive_utc(dt: datetime) -> datetime: + if dt.tzinfo is not None: + return dt.astimezone(UTC).replace(tzinfo=None) + return dt + + from_naive = _to_naive_utc(date_from) + to_naive = _to_naive_utc(date_to) + deltas = diff_snapshots(_fetch_rows(db, date_from, date_to)) + return [d for d in deltas if from_naive <= _to_naive_utc(d["created_at"]) < to_naive] + + +def overview(db: Session, date_from: datetime, date_to: datetime) -> dict: + deltas = _in_range_deltas(db, date_from, date_to) + return _rates(_sum_deltas(deltas)) + + +def trend(db: Session, date_from: datetime, date_to: datetime) -> list[dict]: + deltas = _in_range_deltas(db, date_from, date_to) + by_day: dict[str, list[dict]] = defaultdict(list) + for d in deltas: + by_day[_cn_day(d["created_at"])].append(d) + return [ + {"day": day, **_rates(_sum_deltas(items))} + for day, items in sorted(by_day.items()) + ] + + +def breakdown(db: Session, date_from: datetime, date_to: datetime, dim: str) -> list[dict]: + if dim not in ("event", "app_ver", "oem"): + raise ValueError(f"invalid dim: {dim!r}") + deltas = _in_range_deltas(db, date_from, date_to) + by_key: dict[str, list[dict]] = defaultdict(list) + for d in deltas: + by_key[d.get(dim) or "(unknown)"].append(d) + rows = [{"key": key, **_rates(_sum_deltas(items))} for key, items in by_key.items()] + rows.sort(key=lambda r: (r["report_success_rate"] is None, r["report_success_rate"] or 0.0)) + return rows diff --git a/app/admin/routers/analytics_health.py b/app/admin/routers/analytics_health.py new file mode 100644 index 0000000..517feb3 --- /dev/null +++ b/app/admin/routers/analytics_health.py @@ -0,0 +1,49 @@ +"""admin 埋点健康度:埋点成功率 / 上报成功率 总览 + 趋势 + 下钻(只读)。""" +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 analytics_health as repo +from app.admin.schemas.analytics_health import ( + HealthBreakdownRow, + HealthMetrics, + HealthTrendPoint, +) + +router = APIRouter( + prefix="/admin/api/analytics-health", + tags=["admin-analytics-health"], + dependencies=[Depends(get_current_admin)], +) + + +@router.get("/overview", response_model=HealthMetrics, summary="两段成功率总览") +def overview( + db: AdminDb, + date_from: Annotated[datetime, Query()], + date_to: Annotated[datetime, Query()], +) -> HealthMetrics: + return HealthMetrics(**repo.overview(db, date_from, date_to)) + + +@router.get("/trend", response_model=list[HealthTrendPoint], summary="按北京天趋势") +def trend( + db: AdminDb, + date_from: Annotated[datetime, Query()], + date_to: Annotated[datetime, Query()], +) -> list[HealthTrendPoint]: + return [HealthTrendPoint(**p) for p in repo.trend(db, date_from, date_to)] + + +@router.get("/breakdown", response_model=list[HealthBreakdownRow], summary="按维度下钻") +def breakdown( + db: AdminDb, + date_from: Annotated[datetime, Query()], + date_to: Annotated[datetime, Query()], + dim: Annotated[str, Query(pattern="^(event|app_ver|oem)$")] = "event", +) -> list[HealthBreakdownRow]: + return [HealthBreakdownRow(**r) for r in repo.breakdown(db, date_from, date_to, dim)] diff --git a/app/admin/schemas/analytics_health.py b/app/admin/schemas/analytics_health.py new file mode 100644 index 0000000..4d3328d --- /dev/null +++ b/app/admin/schemas/analytics_health.py @@ -0,0 +1,21 @@ +"""埋点健康度 admin 响应 schema。""" +from __future__ import annotations + +from pydantic import BaseModel + + +class HealthMetrics(BaseModel): + attempted: int + drop_capture: int + delivered: int + drop_undelivered: int + track_success_rate: float | None + report_success_rate: float | None + + +class HealthTrendPoint(HealthMetrics): + day: str + + +class HealthBreakdownRow(HealthMetrics): + key: str diff --git a/app/api/v1/analytics.py b/app/api/v1/analytics.py index 265954d..23d2f82 100644 --- a/app/api/v1/analytics.py +++ b/app/api/v1/analytics.py @@ -6,13 +6,18 @@ POST /api/v1/analytics/events — 批量接收新手引导(及后续)埋点,appe """ from __future__ import annotations -from fastapi import APIRouter, Request +import logging + +from fastapi import APIRouter, HTTPException, Request from app.api.deps import DbSession from app.repositories import analytics as analytics_repo +from app.repositories import analytics_selfstat as selfstat_repo from app.schemas.analytics import AnalyticsBatchIn, AnalyticsIngestOut +from app.schemas.analytics_selfstat import SelfStatBatchIn, SelfStatIngestOut router = APIRouter(prefix="/api/v1/analytics", tags=["analytics"]) +logger = logging.getLogger("shagua.analytics") def _client_ip(request: Request) -> str: @@ -29,3 +34,13 @@ def ingest_events( ) -> AnalyticsIngestOut: n = analytics_repo.record_batch(db, batch, client_ip=_client_ip(request)) return AnalyticsIngestOut(received=n) + + +@router.post("/selfstat", response_model=SelfStatIngestOut, summary="上报自报计数快照") +def ingest_selfstat(batch: SelfStatBatchIn, db: DbSession) -> SelfStatIngestOut: + try: + snap_id = selfstat_repo.record_selfstat(db, batch) + except Exception: # noqa: BLE001 — 计数链路要稳,落库失败不裸奔 500,记日志回明确错误 + logger.exception("selfstat ingest failed device=%s epoch=%s", batch.device_id, batch.epoch_id) + raise HTTPException(status_code=503, detail="selfstat ingest failed") from None + return SelfStatIngestOut(snapshot_id=snap_id) diff --git a/app/models/__init__.py b/app/models/__init__.py index 1f9a0d8..694aa00 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -7,6 +7,10 @@ from app.models.ad_watch_log import AdWatchLog # noqa: F401 from app.models.admin import AdminAuditLog, AdminUser # noqa: F401 from app.models.admin_role import AdminRole # noqa: F401 from app.models.analytics_event import AnalyticsEvent # noqa: F401 +from app.models.analytics_selfstat import ( # noqa: F401 + AnalyticsSelfStat, + AnalyticsSelfStatEvent, +) 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_selfstat.py b/app/models/analytics_selfstat.py new file mode 100644 index 0000000..03bb79c --- /dev/null +++ b/app/models/analytics_selfstat.py @@ -0,0 +1,59 @@ +"""埋点/上报成功率自报计数快照表(append-only)。 + +客户端周期上报「自 epoch 起算的累计计数」;服务端只存原始快照,查询时在 Python 侧差分聚合 +(见 app/admin/repositories/analytics_health.py)。与既有 analytics_event 表完全独立。 + +- analytics_selfstat :一快照一行(快照头 + 设备维度 + 设备级诊断量) +- analytics_selfstat_event :一 event 一行(四类累计计数),外键指向快照头 +""" +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy import BigInteger, DateTime, ForeignKey, Integer, String, func +from sqlalchemy.orm import Mapped, mapped_column + +from app.db.base import Base + + +class AnalyticsSelfStat(Base): + __tablename__ = "analytics_selfstat" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + device_id: Mapped[str] = mapped_column(String(64), index=True, nullable=False) + epoch_id: Mapped[str] = mapped_column(String(64), index=True, nullable=False) + # 设备维度(每设备固定,下钻用) + app_ver: Mapped[str | None] = mapped_column(String(32), nullable=True) + oem: Mapped[str | None] = mapped_column(String(32), nullable=True) + os: Mapped[str | None] = mapped_column(String(32), nullable=True) + # 设备级诊断量(累计;queue_depth 是瞬时 gauge) + batches_attempted: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0) + batches_ok: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0) + batches_fail: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0) + retries: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0) + queue_depth: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + sent_at: Mapped[int | None] = mapped_column(BigInteger, nullable=True) # 端上报时刻 epoch ms + # 服务端接收时间(权威,用于时间分桶与分区排序) + 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"" + + +class AnalyticsSelfStatEvent(Base): + __tablename__ = "analytics_selfstat_event" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + snapshot_id: Mapped[int] = mapped_column( + ForeignKey("analytics_selfstat.id"), index=True, nullable=False + ) + event: Mapped[str] = mapped_column(String(64), index=True, nullable=False) + attempted: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0) + drop_capture: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0) + delivered: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0) + drop_undelivered: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0) + + def __repr__(self) -> str: # pragma: no cover + return f"" diff --git a/app/repositories/analytics_selfstat.py b/app/repositories/analytics_selfstat.py new file mode 100644 index 0000000..ee2a1e4 --- /dev/null +++ b/app/repositories/analytics_selfstat.py @@ -0,0 +1,38 @@ +"""自报计数快照落库。一次事务:插 1 条快照头 + N 条 event 行,返回快照 id。""" +from __future__ import annotations + +from sqlalchemy.orm import Session + +from app.models.analytics_selfstat import AnalyticsSelfStat, AnalyticsSelfStatEvent +from app.schemas.analytics_selfstat import SelfStatBatchIn + + +def record_selfstat(db: Session, batch: SelfStatBatchIn) -> int: + snap = AnalyticsSelfStat( + device_id=batch.device_id, + epoch_id=batch.epoch_id, + app_ver=batch.app_ver, + oem=batch.oem, + os=batch.os, + batches_attempted=batch.batches_attempted, + batches_ok=batch.batches_ok, + batches_fail=batch.batches_fail, + retries=batch.retries, + queue_depth=batch.queue_depth, + sent_at=batch.sent_at, + ) + db.add(snap) + db.flush() # 拿到 snap.id + db.add_all([ + AnalyticsSelfStatEvent( + snapshot_id=snap.id, + event=e.event, + attempted=e.attempted, + drop_capture=e.drop_capture, + delivered=e.delivered, + drop_undelivered=e.drop_undelivered, + ) + for e in batch.events + ]) + db.commit() + return snap.id diff --git a/app/schemas/analytics_selfstat.py b/app/schemas/analytics_selfstat.py new file mode 100644 index 0000000..6edf545 --- /dev/null +++ b/app/schemas/analytics_selfstat.py @@ -0,0 +1,34 @@ +"""自报计数上报 schema。字段名对齐客户端 payload(snake_case),累计值语义。""" +from __future__ import annotations + +from pydantic import BaseModel, Field + + +class SelfStatEventIn(BaseModel): + event: str = Field(max_length=64) + attempted: int = 0 + drop_capture: int = 0 + delivered: int = 0 + drop_undelivered: int = 0 + + +class SelfStatBatchIn(BaseModel): + device_id: str = Field(max_length=64) + epoch_id: str = Field(max_length=64) + sent_at: int | None = None + app_ver: str | None = Field(default=None, max_length=32) + oem: str | None = Field(default=None, max_length=32) + os: str | None = Field(default=None, max_length=32) + batches_attempted: int = 0 + batches_ok: int = 0 + batches_fail: int = 0 + retries: int = 0 + queue_depth: int = 0 + # 允许空列表:某次快照只上报设备级计数(batches_*/retries/queue_depth)、无 event 细分时也合法 + # (与 AnalyticsBatchIn 的 min_length=1 有意不同——那是行为事件、必须至少一条)。 + events: list[SelfStatEventIn] = Field(default_factory=list, max_length=200) + + +class SelfStatIngestOut(BaseModel): + ok: bool = True + snapshot_id: int diff --git a/tests/test_analytics_health.py b/tests/test_analytics_health.py new file mode 100644 index 0000000..39fe093 --- /dev/null +++ b/tests/test_analytics_health.py @@ -0,0 +1,160 @@ +"""埋点健康度聚合:纯差分函数 + admin 端点。""" +from __future__ import annotations + +from datetime import datetime, timezone + +from app.admin.repositories.analytics_health import _cn_day, _rates, diff_snapshots + + +def _row(device, epoch, event, ts, **cum) -> dict: + base = {"attempted": 0, "drop_capture": 0, "delivered": 0, "drop_undelivered": 0} + base.update(cum) + return { + "device_id": device, "epoch_id": epoch, "event": event, + "created_at": datetime(2026, 7, 1, ts, 0, tzinfo=timezone.utc), + "app_ver": "0.2.12(62)", "oem": "ColorOS", "os": "Android 14", + **base, + } + + +def test_diff_first_row_is_full_cumulative() -> None: + rows = [_row("d", "e", "video_play", 1, attempted=100, delivered=90)] + out = diff_snapshots(rows) + assert len(out) == 1 + assert out[0]["d_attempted"] == 100 + assert out[0]["d_delivered"] == 90 + + +def test_diff_consecutive_delta() -> None: + rows = [ + _row("d", "e", "video_play", 1, attempted=100, delivered=90), + _row("d", "e", "video_play", 2, attempted=150, delivered=140), + ] + out = sorted(diff_snapshots(rows), key=lambda r: r["created_at"]) + assert out[1]["d_attempted"] == 50 + assert out[1]["d_delivered"] == 50 + + +def test_diff_epoch_reset_new_partition() -> None: + rows = [ + _row("d", "e1", "video_play", 1, attempted=100), + _row("d", "e2", "video_play", 2, attempted=5), + ] + out = {(r["epoch_id"]): r["d_attempted"] for r in diff_snapshots(rows)} + assert out["e1"] == 100 + assert out["e2"] == 5 + + +def test_diff_clamps_negative_on_reorder() -> None: + rows = [ + _row("d", "e", "video_play", 1, attempted=100), + _row("d", "e", "video_play", 2, attempted=80), + ] + out = sorted(diff_snapshots(rows), key=lambda r: r["created_at"]) + assert out[1]["d_attempted"] == 0 + + +def test_rates_computes_both_formulas() -> None: + out = _rates({"attempted": 150, "drop_capture": 0, "delivered": 140, "drop_undelivered": 10}) + assert out["track_success_rate"] == 1.0 + assert out["report_success_rate"] == 140 / 150 + + +def test_rates_zero_denominator_yields_none() -> None: + out = _rates({"attempted": 0, "drop_capture": 0, "delivered": 0, "drop_undelivered": 0}) + assert out["track_success_rate"] is None + assert out["report_success_rate"] is None + + +def test_cn_day_beijing_boundary() -> None: + # UTC 15:59 → 北京 23:59 同日;UTC 16:00 → 北京 次日 00:00 + assert _cn_day(datetime(2026, 7, 1, 15, 59, tzinfo=timezone.utc)) == "2026-07-01" + assert _cn_day(datetime(2026, 7, 1, 16, 0, tzinfo=timezone.utc)) == "2026-07-02" + + +def test_diff_same_timestamp_ordered_by_id() -> None: + ts = datetime(2026, 7, 1, 1, 0, tzinfo=timezone.utc) + # 相同 created_at,乱序传入(id=2 在前);应按 id 排序 → id=1(cum100) 在前 + rows = [ + {"id": 2, "device_id": "d", "epoch_id": "e", "event": "vp", "created_at": ts, + "app_ver": "v", "oem": "o", "os": "s", + "attempted": 150, "drop_capture": 0, "delivered": 0, "drop_undelivered": 0}, + {"id": 1, "device_id": "d", "epoch_id": "e", "event": "vp", "created_at": ts, + "app_ver": "v", "oem": "o", "os": "s", + "attempted": 100, "drop_capture": 0, "delivered": 0, "drop_undelivered": 0}, + ] + out = diff_snapshots(rows) + assert out[0]["d_attempted"] == 100 # id=1 sorts first → its cumulative + assert out[1]["d_attempted"] == 50 # id=2 second → 150-100 + + +import pytest +from fastapi.testclient import TestClient + +from app.admin.main import admin_app +from app.admin.repositories import admin_user as admin_repo +from app.db.session import SessionLocal + + +@pytest.fixture() +def admin_client() -> TestClient: + return TestClient(admin_app) + + +@pytest.fixture() +def admin_token() -> str: + db = SessionLocal() + try: + if admin_repo.get_by_username(db, "health_admin") is None: + admin_repo.create_admin(db, username="health_admin", password="pw", role="super_admin") + finally: + db.close() + c = TestClient(admin_app) + r = c.post("/admin/api/auth/login", json={"username": "health_admin", "password": "pw"}) + return r.json()["access_token"] + + +def _auth(t: str) -> dict: + return {"Authorization": f"Bearer {t}"} + + +def _seed_two_snapshots(client: TestClient) -> None: + for attempted, delivered in ((100, 90), (150, 140)): + client.post("/api/v1/analytics/selfstat", json={ + "device_id": "d-health", "epoch_id": "e-health", + "app_ver": "0.2.12(62)", "oem": "ColorOS", "os": "Android 14", + "events": [{"event": "video_play", "attempted": attempted, + "drop_capture": 0, "delivered": delivered, "drop_undelivered": 0}], + }) + + +def test_health_overview_requires_auth(admin_client: TestClient) -> None: + r = admin_client.get("/admin/api/analytics-health/overview", + params={"date_from": "2026-07-01T00:00:00Z", "date_to": "2030-01-01T00:00:00Z"}) + assert r.status_code == 401 + + +def test_health_overview(client: TestClient, admin_client: TestClient, admin_token: str) -> None: + _seed_two_snapshots(client) # 播种走公开 app(ingest 端点在 app,不在 admin_app) + r = admin_client.get( + "/admin/api/analytics-health/overview", + params={"date_from": "2026-07-01T00:00:00Z", "date_to": "2030-01-01T00:00:00Z"}, + headers=_auth(admin_token), + ) + assert r.status_code == 200, r.text + data = r.json() + # 两快照累计 100→150,差分后 attempted 总 150(首快照 100 + 增量 50) + assert data["attempted"] == 150 + assert data["track_success_rate"] == 1.0 + + +def test_health_breakdown(client: TestClient, admin_client: TestClient, admin_token: str) -> None: + _seed_two_snapshots(client) # 播种走公开 app + r = admin_client.get( + "/admin/api/analytics-health/breakdown", + params={"date_from": "2026-07-01T00:00:00Z", "date_to": "2030-01-01T00:00:00Z", "dim": "event"}, + headers=_auth(admin_token), + ) + assert r.status_code == 200, r.text + rows = r.json() + assert any(row["key"] == "video_play" for row in rows) diff --git a/tests/test_analytics_selfstat.py b/tests/test_analytics_selfstat.py new file mode 100644 index 0000000..a096806 --- /dev/null +++ b/tests/test_analytics_selfstat.py @@ -0,0 +1,43 @@ +"""自报计数上报端点测试。""" +from __future__ import annotations + +from fastapi.testclient import TestClient + + +def _payload(**over) -> dict: + base = { + "device_id": "dev-1", "epoch_id": "ep-1", "sent_at": 1700000000000, + "app_ver": "0.2.12(62)", "oem": "ColorOS", "os": "Android 14", + "batches_attempted": 10, "batches_ok": 9, "batches_fail": 1, + "retries": 1, "queue_depth": 2, + "events": [ + {"event": "video_play", "attempted": 100, "drop_capture": 1, + "delivered": 95, "drop_undelivered": 2}, + ], + } + base.update(over) + return base + + +def test_selfstat_ingest_ok(client: TestClient) -> None: + r = client.post("/api/v1/analytics/selfstat", json=_payload()) + assert r.status_code == 200, r.text + body = r.json() + assert body["ok"] is True + assert isinstance(body["snapshot_id"], int) + + +def test_selfstat_ingest_empty_events(client: TestClient) -> None: + r = client.post("/api/v1/analytics/selfstat", json=_payload(events=[])) + assert r.status_code == 200, r.text + assert r.json()["ok"] is True + + +def test_selfstat_ingest_db_error_returns_503(client: TestClient, monkeypatch) -> None: + """落库异常时端点回 503(计数链路稳定性守卫),不裸奔 500。""" + def _boom(*_args, **_kwargs): + raise RuntimeError("db gone") + + monkeypatch.setattr("app.repositories.analytics_selfstat.record_selfstat", _boom) + r = client.post("/api/v1/analytics/selfstat", json=_payload()) + assert r.status_code == 503, r.text