From afb5f478c9e3a846ee9aa51cf533189efdb27588 Mon Sep 17 00:00:00 2001 From: guke Date: Wed, 8 Jul 2026 17:32:34 +0800 Subject: [PATCH] feat(analytics): admin health aggregation (python-side diff) --- app/admin/repositories/analytics_health.py | 138 +++++++++++++++++++++ tests/test_analytics_health.py | 54 ++++++++ 2 files changed, 192 insertions(+) create mode 100644 app/admin/repositories/analytics_health.py create mode 100644 tests/test_analytics_health.py diff --git a/app/admin/repositories/analytics_health.py b/app/admin/repositories/analytics_health.py new file mode 100644 index 0000000..0a0f44c --- /dev/null +++ b/app/admin/repositories/analytics_health.py @@ -0,0 +1,138 @@ +"""埋点健康度聚合(埋点成功率 / 上报成功率)。 + +只存原始累计快照,查询时在 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 and_, 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"]) + 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.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() + + sub = ( + select(H.device_id, H.epoch_id, E.event, func.max(H.created_at).label("mx")) + .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, + and_( + sub.c.device_id == H.device_id, + sub.c.epoch_id == H.epoch_id, + sub.c.event == E.event, + sub.c.mx == H.created_at, + ), + ) + ).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) 的增量(基线行被差分用后丢弃)。""" + deltas = diff_snapshots(_fetch_rows(db, date_from, date_to)) + return [d for d in deltas if date_from <= d["created_at"] < date_to] + + +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]: + assert dim in ("event", "app_ver", "oem") + 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/tests/test_analytics_health.py b/tests/test_analytics_health.py new file mode 100644 index 0000000..02333c7 --- /dev/null +++ b/tests/test_analytics_health.py @@ -0,0 +1,54 @@ +"""埋点健康度聚合:纯差分函数 + admin 端点。""" +from __future__ import annotations + +from datetime import datetime, timezone + +from app.admin.repositories.analytics_health import 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