feat(analytics): admin analytics-health endpoints (overview/trend/breakdown)

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)

app/admin/repositories/analytics_health.py

@@ -102,9 +102,20 @@ def _fetch_rows(db: Session, date_from: datetime, date_to: datetime) -> list[dic
 
 
 def _in_range_deltas(db: Session, date_from: datetime, date_to: datetime) -> list[dict]:
-    """差分后只保留 created_at ∈ [from, to) 的增量(基线行被差分用后丢弃)。"""
+    """差分后只保留 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 date_from <= d["created_at"] < 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:

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)]

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

tests/test_analytics_health.py

@@ -86,3 +86,75 @@ def test_diff_same_timestamp_ordered_by_id() -> None:
     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)