diff --git a/apps/admin/app.py b/apps/admin/app.py index 150b011..2cc1cdc 100644 --- a/apps/admin/app.py +++ b/apps/admin/app.py @@ -12,7 +12,7 @@ from contextlib import asynccontextmanager from fastapi import FastAPI -from apps.admin.routers import assets, gaps, kb, rules +from apps.admin.routers import assets, gaps, kb, rules, stats, users from packages.shared.config import Settings, get_settings from packages.shared.db import dispose_engine, get_session_factory from packages.shared.handlers import install_exception_handlers @@ -71,7 +71,8 @@ def create_admin_app(settings: Settings | None = None) -> FastAPI: setup_tracing(app, cfg) if deployed: - for router in (kb.router, rules.router, assets.router, gaps.router): + routers = (kb.router, rules.router, assets.router, gaps.router, users.router, stats.router) + for router in routers: app.include_router(router, prefix=ADMIN_API_PREFIX) else: logger.warning( diff --git a/apps/admin/routers/stats.py b/apps/admin/routers/stats.py new file mode 100644 index 0000000..0efbc5a --- /dev/null +++ b/apps/admin/routers/stats.py @@ -0,0 +1,158 @@ +"""行为统计:原始事件流 + 按天 × event_name 的 PV 透视(FR-ADM 行为分析能力)。 + +只读透视,直接走 ``ob_event``(ix_ob_event_user_ts / ix_ob_event_name_ts 两个 +索引支撑这里的过滤与排序)。日桶统一 UTC+8 自然日(与 billing 的日预算桶、 +sms/email 频控桶同一时区纪律),在 SQL 里 ``+ interval '8 hours'`` 后 +``date_trunc``——不把整表捞回来再按 Python 侧分桶。 + +ob_event 数据量会先于其他表膨胀:所有端点强制分页(page_size 上限 200), +时间范围过滤是推荐用法而非可选项。 +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from typing import Any + +from fastapi import APIRouter, Depends, Query +from sqlalchemy import Date, cast, desc, func, select, text + +from apps.admin.deps import SessionDep, admin_auth +from packages.shared.errors import AppError, ErrorCode +from packages.shared.models import ObEvent, User +from packages.shared.response import ok + +router = APIRouter(tags=["admin:stats"], dependencies=[Depends(admin_auth)]) + +#: 原始事件流分页上限(契约固定值:ob_event 量大,不许一次拉爆)。 +MAX_PAGE_SIZE = 200 + +#: 日桶时区:与 billing/频控的 UTC+8 自然窗口一致(FR-BILL-01 纪律)。 +_DAY_BUCKET_OFFSET = text("INTERVAL '8 hours'") + +#: 总览回溯窗口上限:90 天,防 ``days=100000`` 把整表做 group by。 +MAX_OVERVIEW_DAYS = 90 + + +def _day_bucket() -> Any: + """server_ts 的 UTC+8 自然日桶(SQL 表达式,group by / select 两用)。""" + return cast(func.date_trunc("day", ObEvent.server_ts + _DAY_BUCKET_OFFSET), Date).label("day") + + +def _serialize_event(event: ObEvent) -> dict[str, Any]: + return { + "event_id": str(event.event_id), + "user_id": str(event.user_id), + "event_name": event.event_name, + "props": event.props, + "client_ts": event.client_ts.isoformat() if event.client_ts else None, + "server_ts": event.server_ts.isoformat() if event.server_ts else None, + } + + +def _parse_ts(raw: str, *, param: str) -> datetime: + """时间参数解析:ISO 8601 或 epoch 秒(int/float)都接受。 + + 无时区的 ISO 按 UTC 解释(与服务端落库口径一致)。 + """ + try: + epoch = float(raw) + except ValueError: + try: + parsed = datetime.fromisoformat(raw) + except ValueError as exc: + raise AppError( + ErrorCode.BAD_REQUEST, + message=f"{param} 时间格式不合法", + hint="用 ISO 时间或 epoch 秒", + ) from exc + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=UTC) + return parsed + return datetime.fromtimestamp(epoch, tz=UTC) + + +@router.get("/events", summary="原始事件流:user/event_name/时间范围过滤 + 分页") +async def list_events( + session: SessionDep, + user_id: int | None = Query(default=None), + event_name: str | None = Query(default=None, max_length=64), + start: str | None = Query(default=None, description="ISO 时间或 epoch 秒(含)"), + end: str | None = Query(default=None, description="ISO 时间或 epoch 秒(含)"), + page: int = Query(default=1, ge=1), + page_size: int = Query(default=50, ge=1, le=MAX_PAGE_SIZE), +) -> dict[str, Any]: + conditions = [] + if user_id is not None: + conditions.append(ObEvent.user_id == user_id) + if event_name: + conditions.append(ObEvent.event_name == event_name) + if start: + conditions.append(ObEvent.server_ts >= _parse_ts(start, param="start")) + if end: + conditions.append(ObEvent.server_ts <= _parse_ts(end, param="end")) + + where = select(ObEvent).where(*conditions) + total = await session.scalar(select(func.count()).select_from(where.subquery())) + rows = ( + ( + await session.execute( + where.order_by(desc(ObEvent.server_ts)) + .offset((page - 1) * page_size) + .limit(page_size) + ) + ) + .scalars() + .all() + ) + return ok( + { + "items": [_serialize_event(row) for row in rows], + "total": total or 0, + "page": page, + "page_size": page_size, + } + ) + + +@router.get("/stats/overview", summary="行为总览:按天 × event_name PV 透视 + 注册/活跃序列") +async def stats_overview( + session: SessionDep, + days: int = Query(default=7, ge=1, le=MAX_OVERVIEW_DAYS, description="回溯天数"), +) -> dict[str, Any]: + window_start = datetime.now(UTC) - timedelta(days=days) + + day_bucket = _day_bucket() + pv_stmt = ( + select(day_bucket, ObEvent.event_name, func.count().label("pv")) + .where(ObEvent.server_ts >= window_start) + .group_by("day", ObEvent.event_name) + .order_by("day", desc("pv")) + ) + pv_rows = (await session.execute(pv_stmt)).all() + + active_stmt = ( + select(day_bucket, func.count(func.distinct(ObEvent.user_id)).label("active_users")) + .where(ObEvent.server_ts >= window_start) + .group_by("day") + .order_by("day") + ) + active_rows = (await session.execute(active_stmt)).all() + + registered_total = await session.scalar(select(func.count()).select_from(User)) + + return ok( + { + "days": days, + "window_start": window_start.isoformat(), + "registered_total": registered_total or 0, + "daily_active_users": [ + {"day": row.day.isoformat(), "active_users": row.active_users} + for row in active_rows + ], + "daily_event_pv": [ + {"day": row.day.isoformat(), "event_name": row.event_name, "pv": row.pv} + for row in pv_rows + ], + } + ) diff --git a/apps/admin/routers/users.py b/apps/admin/routers/users.py new file mode 100644 index 0000000..dca6ad4 --- /dev/null +++ b/apps/admin/routers/users.py @@ -0,0 +1,257 @@ +"""用户行为看板:用户列表 + 用户详情(FR-ADM 行为分析能力)。 + +与 kb.py 的关键差异:这里直接读 ``packages.shared.models`` 的观测/计费/ +报告表做只读透视,不经任何域 service——后台看板要的是「跨域拼宽表」, +而域 service 的接口是为客户端单用户视角设计的,复用反而拧。 + +PII 纪律:``union_id`` 只留前 4 后 4、中间打码后出(防 PII 直出后台), +掩码逻辑写在响应组装处(本模块),不许下沉到查询层。 + +鉴权走 ``X-Admin-Token``(共享令牌,见 apps.admin.deps)。 +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from typing import Any + +from fastapi import APIRouter, Depends, Query +from sqlalchemy import Select, desc, func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from apps.admin.deps import SessionDep, admin_auth +from packages.domain.billing.quota import QuotaState, period_key +from packages.shared.errors import AppError, ErrorCode +from packages.shared.models import ( + ObEvent, + Quota, + QuotaRule, + RpReport, + User, + UserIdentity, +) +from packages.shared.response import ok + +router = APIRouter(prefix="/users", tags=["admin:users"], dependencies=[Depends(admin_auth)]) + +#: 列表/详情分页与「最近事件」窗口的共用上限(ob_event 量大,不许一次拉爆)。 +MAX_PAGE_SIZE = 200 +RECENT_EVENTS_LIMIT = 50 +RECENT_REPORTS_LIMIT = 10 +_EVENT_COUNT_WINDOW = timedelta(days=7) + + +def _mask_union_id(union_id: str) -> str: + """PII 打码:只留前 4 后 4,中间一律 ``*``;不足 9 位的整体打码。""" + if len(union_id) <= 8: + return union_id[:2] + "***" + return f"{union_id[:4]}{'*' * (len(union_id) - 8)}{union_id[-4:]}" + + +def _user_list_stmt(query: str | None) -> Select[Any]: + """用户列表主查询:三个标量关联子查询拼出 last_active/event_count/report_count。 + + 都是 ``(user_id)`` 上的索引查找,不走 join 放大行数——ob_event/rp_report + 与 id_user 一对多,join 后 count 需要 distinct, correlated subquery 更直白。 + """ + window_start = datetime.now(UTC) - _EVENT_COUNT_WINDOW + last_active = ( + select(func.max(ObEvent.server_ts)).where(ObEvent.user_id == User.user_id).scalar_subquery() + ) + event_count_7d = ( + select(func.count()) + .where(ObEvent.user_id == User.user_id, ObEvent.server_ts >= window_start) + .scalar_subquery() + ) + report_count = ( + select(func.count()) + .select_from(RpReport) + .where(RpReport.user_id == User.user_id) + .scalar_subquery() + ) + stmt = select( + User.user_id, + User.register_source, + User.created_at, + last_active.label("last_active_at"), + event_count_7d.label("event_count_7d"), + report_count.label("report_count"), + ) + if query: + # query 兼容两种过滤:能解析成 int → user_id 精确;否则按注册来源等值 + if query.isdigit(): + stmt = stmt.where(User.user_id == int(query)) + else: + stmt = stmt.where(User.register_source == query) + return stmt + + +def _serialize_user_row(row: Any) -> dict[str, Any]: + return { + "user_id": str(row.user_id), + "register_source": row.register_source, + "created_at": row.created_at.isoformat() if row.created_at else None, + "last_active_at": row.last_active_at.isoformat() if row.last_active_at else None, + "event_count_7d": row.event_count_7d, + "report_count": row.report_count, + } + + +@router.get("", summary="用户列表:query 按 user_id 精确 / register_source 过滤 + 分页") +async def list_users( + session: SessionDep, + query: str | None = Query(default=None, max_length=64, description="user_id 或注册来源"), + page: int = Query(default=1, ge=1), + page_size: int = Query(default=20, ge=1, le=MAX_PAGE_SIZE), +) -> dict[str, Any]: + stmt = _user_list_stmt(query) + total = await session.scalar(select(func.count()).select_from(stmt.subquery())) + rows = ( + await session.execute( + stmt.order_by(desc(User.created_at)).offset((page - 1) * page_size).limit(page_size) + ) + ).all() + return ok( + { + "items": [_serialize_user_row(row) for row in rows], + "total": total or 0, + "page": page, + "page_size": page_size, + } + ) + + +async def _quota_snapshot(session: AsyncSession, user_id: int) -> list[dict[str, Any]]: + """各 quota_key 的本周期用量快照(used/limit/remaining)。 + + 只读:不 ``lock_or_create_quota``(那会为用户造行,看板不该有写副作用)。 + 规则表全量读出(V1 规则个位数),计量行按本周期 period_key 一次性查回。 + """ + rules = (await session.execute(select(QuotaRule).order_by(QuotaRule.quota_key))).scalars().all() + now = datetime.now(UTC) + keys = { + rule.quota_key: period_key(now, rule.period_type) # type: ignore[arg-type] + for rule in rules + } + quota_rows = ( + ( + await session.execute( + select(Quota).where( + Quota.user_id == user_id, + Quota.quota_key.in_(keys.keys()), + Quota.period_key.in_(set(keys.values())), + ) + ) + ) + .scalars() + .all() + ) + by_key = {(row.quota_key, row.period_key): row for row in quota_rows} + snapshot = [] + for rule in rules: + row = by_key.get((rule.quota_key, keys[rule.quota_key])) + used = row.used if row is not None else 0 + bonus = row.bonus_granted if row is not None else 0 + state = QuotaState( + used=used, base_limit=rule.base_limit, bonus=bonus, hard_cap=rule.hard_cap + ) + snapshot.append( + { + "quota_key": rule.quota_key, + "period_key": keys[rule.quota_key], + "used": used, + "limit": state.effective_limit, + "remaining": state.remaining, + } + ) + return snapshot + + +def _serialize_event(event: ObEvent) -> dict[str, Any]: + return { + "event_id": str(event.event_id), + "event_name": event.event_name, + "props": event.props, + "client_ts": event.client_ts.isoformat() if event.client_ts else None, + "server_ts": event.server_ts.isoformat() if event.server_ts else None, + } + + +@router.get("/{user_id}", summary="用户详情:identities(掩码)+ 配额快照 + 最近事件/报告") +async def get_user_detail(user_id: int, session: SessionDep) -> dict[str, Any]: + user = await session.get(User, user_id) + if user is None: + raise AppError(ErrorCode.NOT_FOUND, message="用户不存在") + + identities = ( + ( + await session.execute( + select(UserIdentity) + .where(UserIdentity.user_id == user_id) + .order_by(UserIdentity.created_at) + ) + ) + .scalars() + .all() + ) + events = ( + ( + await session.execute( + select(ObEvent) + .where(ObEvent.user_id == user_id) + .order_by(desc(ObEvent.server_ts)) + .limit(RECENT_EVENTS_LIMIT) + ) + ) + .scalars() + .all() + ) + reports = ( + ( + await session.execute( + select(RpReport) + .where(RpReport.user_id == user_id) + .order_by(desc(RpReport.created_at)) + .limit(RECENT_REPORTS_LIMIT) + ) + ) + .scalars() + .all() + ) + quotas = await _quota_snapshot(session, user_id) + + return ok( + { + "user": { + "user_id": str(user.user_id), + "status": user.status, + "nickname": user.nickname, + "register_source": user.register_source, + "created_at": user.created_at.isoformat() if user.created_at else None, + "last_active_at": ( + user.last_active_at.isoformat() if user.last_active_at else None + ), + "deleted_at": user.deleted_at.isoformat() if user.deleted_at else None, + }, + "identities": [ + { + "provider": ident.provider, + "union_id": _mask_union_id(ident.union_id), + "created_at": ident.created_at.isoformat() if ident.created_at else None, + } + for ident in identities + ], + "quotas": quotas, + "recent_events": [_serialize_event(event) for event in events], + "recent_reports": [ + { + "report_id": str(report.report_id), + "tier": report.tier, + "version": report.version, + "status": report.status, + "created_at": report.created_at.isoformat() if report.created_at else None, + } + for report in reports + ], + } + ) diff --git a/tests/integration/test_admin_behavior.py b/tests/integration/test_admin_behavior.py new file mode 100644 index 0000000..6cfa911 --- /dev/null +++ b/tests/integration/test_admin_behavior.py @@ -0,0 +1,380 @@ +"""admin 用户行为看板集成用例(真实 PostgreSQL,FR-ADM 行为分析能力)。 + +覆盖四个端点的字段与分页、X-Admin-Token 鉴权口径与现有 admin 一致 +(错误令牌 → 40100)、union_id 掩码(前 4 后 4,防 PII 直出)、配额快照 +只读不造行。种子数据用专属随机 user_id,断言全部相对种子而非全表。 +""" + +from __future__ import annotations + +import os +import uuid +from collections.abc import AsyncIterator +from datetime import UTC, datetime, timedelta +from typing import Any + +import pytest + +ADMIN_TOKEN = "it-admin-token-behavior" +AUTH = {"X-Admin-Token": ADMIN_TOKEN} +BAD_AUTH = {"X-Admin-Token": "wrong-token"} + +_SEED_EVENT_NAMES = ("page_view", "outfit_card.tap") + + +def _infra_available() -> bool: + return bool(os.getenv("MEIDA_RUN_INTEGRATION")) + + +pytestmark = [ + pytest.mark.integration, + pytest.mark.skipif( + not _infra_available(), + reason="set MEIDA_RUN_INTEGRATION=1 with PostgreSQL and Redis running", + ), +] + + +@pytest.fixture(autouse=True) +async def _dispose_engine_after_test() -> AsyncIterator[None]: + """用例结束即释放全局引擎(理由同 test_admin_console.py)。""" + yield + from packages.shared.db import dispose_engine + + await dispose_engine() + + +@pytest.fixture +async def client() -> Any: + import httpx + + from apps.admin.app import create_admin_app + from packages.shared.config import Settings + + app = create_admin_app(Settings(admin_token=ADMIN_TOKEN)) + transport = httpx.ASGITransport(app=app) + async with ( + httpx.AsyncClient(transport=transport, base_url="http://test") as http, + app.router.lifespan_context(app), + ): + yield http + + +@pytest.fixture +async def seed() -> dict[str, Any]: + """造 2 个用户 + 身份 + 事件(7 天内/外各一)+ 报告 + 配额规则与用量。""" + from sqlalchemy import select + + from packages.domain.billing.quota import period_key + from packages.domain.ids import next_id + from packages.shared.db import get_session_factory + from packages.shared.models import ( + ObEvent, + Quota, + QuotaRule, + RpReport, + User, + UserIdentity, + ) + + suffix = uuid.uuid4().hex[:8] + user_a, user_b = next_id(), next_id() + union_id_a = f"email:it-user-{suffix}@example.com" + now = datetime.now(UTC) + quota_key = f"it-quota-{suffix}" + day_key = period_key(now, "day") + + async with get_session_factory()() as session: + session.add_all( + [ + User( + user_id=user_a, + status="active", + nickname=f"it-a-{suffix}", + register_source="email", + created_at=now - timedelta(days=3), + ), + User( + user_id=user_b, + status="active", + nickname=f"it-b-{suffix}", + register_source="dev", + created_at=now - timedelta(days=1), + ), + UserIdentity( + identity_id=next_id(), + user_id=user_a, + provider="email", + union_id=union_id_a, + ), + ObEvent( + event_id=next_id(), + user_id=user_a, + event_name=_SEED_EVENT_NAMES[0], + props={"page": "home"}, + server_ts=now - timedelta(hours=2), + ), + ObEvent( + event_id=next_id(), + user_id=user_a, + event_name=_SEED_EVENT_NAMES[1], + props={"card_id": 1}, + server_ts=now - timedelta(days=1), + ), + # 8 天前:在 7 天窗口外——event_count_7d 不该算它 + ObEvent( + event_id=next_id(), + user_id=user_a, + event_name=_SEED_EVENT_NAMES[0], + props={"page": "old"}, + server_ts=now - timedelta(days=8), + ), + ObEvent( + event_id=next_id(), + user_id=user_b, + event_name=_SEED_EVENT_NAMES[0], + props={"page": "home"}, + server_ts=now - timedelta(hours=1), + ), + RpReport( + report_id=next_id(), + user_id=user_a, + tier="basic", + status="ready", + version=1, + ), + QuotaRule( + quota_key=quota_key, + period_type="day", + base_limit=10, + task_bonus_max=0, + hard_cap=20, + ), + Quota( + user_id=user_a, + quota_key=quota_key, + period_key=day_key, + used=3, + limit_value=10, + bonus_granted=0, + ), + ] + ) + await session.commit() + rules_before = len( + (await session.execute(select(QuotaRule).where(QuotaRule.quota_key == quota_key))) + .scalars() + .all() + ) + return { + "user_a": user_a, + "user_b": user_b, + "quota_key": quota_key, + "day_key": day_key, + "rules_before": rules_before, + "union_id_a": union_id_a, + } + + +async def test_wrong_token_rejected_like_existing_admin(client: Any) -> None: + """鉴权口径与现有 admin 一致:错误令牌 → 40100。""" + resp = await client.get("/admin/api/v1/users", headers=BAD_AUTH) + assert resp.status_code == 401 + assert resp.json()["code"] == 40100 + + resp = await client.get("/admin/api/v1/events", headers=BAD_AUTH) + assert resp.status_code == 401 + assert resp.json()["code"] == 40100 + + resp = await client.get("/admin/api/v1/stats/overview", headers=BAD_AUTH) + assert resp.status_code == 401 + assert resp.json()["code"] == 40100 + + # 无令牌头:同样 40100(compare_digest 对空串,不炸 500) + resp = await client.get("/admin/api/v1/users") + assert resp.status_code == 401 + assert resp.json()["code"] == 40100 + + +async def test_users_list_fields_and_filters(client: Any, seed: dict[str, Any]) -> None: + resp = await client.get("/admin/api/v1/users", headers=AUTH, params={"query": "email"}) + assert resp.status_code == 200 + body = resp.json()["data"] + assert body["page"] == 1 + assert body["page_size"] == 20 + assert body["total"] >= 1 + assert all(item["register_source"] == "email" for item in body["items"]) + + # query 按 user_id 精确:只命中种子用户 A——字段契约全在它身上断言, + # 不受库里其它用户数据/分页位置影响 + resp = await client.get( + "/admin/api/v1/users", headers=AUTH, params={"query": str(seed["user_a"])} + ) + body = resp.json()["data"] + assert [item["user_id"] for item in body["items"]] == [str(seed["user_a"])] + row_a = body["items"][0] + assert row_a["register_source"] == "email" + assert row_a["created_at"] is not None + assert row_a["last_active_at"] is not None + assert row_a["event_count_7d"] == 2 # 8 天外那条不算 + assert row_a["report_count"] == 1 + + +async def test_users_list_pagination(client: Any, seed: dict[str, Any]) -> None: + resp = await client.get( + "/admin/api/v1/users", + headers=AUTH, + params={"query": "dev", "page": 1, "page_size": 1}, + ) + body = resp.json()["data"] + assert body["page_size"] == 1 + assert len(body["items"]) == 1 + assert body["total"] >= 1 + + +async def test_user_detail_masks_identity_and_snapshots_quota( + client: Any, seed: dict[str, Any] +) -> None: + from sqlalchemy import select + + from packages.shared.db import get_session_factory + from packages.shared.models import Quota, UserIdentity + + user_id = seed["user_a"] + resp = await client.get(f"/admin/api/v1/users/{user_id}", headers=AUTH) + assert resp.status_code == 200 + data = resp.json()["data"] + + # 基本信息 + assert data["user"]["user_id"] == str(user_id) + assert data["user"]["register_source"] == "email" + + # 身份:provider 明文、union_id 只留前 4 后 4(PII 直出防护,逐字符断言) + assert len(data["identities"]) == 1 + ident = data["identities"][0] + assert ident["provider"] == "email" + raw = seed["union_id_a"] + expected = f"{raw[:4]}{'*' * (len(raw) - 8)}{raw[-4:]}" + assert ident["union_id"] == expected + assert raw not in str(data["identities"]) + + # 配额快照:种子用量 used=3,limit=base 10,remaining=7 + quota_rows = [q for q in data["quotas"] if q["quota_key"] == seed["quota_key"]] + assert len(quota_rows) == 1 + assert quota_rows[0]["used"] == 3 + assert quota_rows[0]["limit"] == 10 + assert quota_rows[0]["remaining"] == 7 + assert quota_rows[0]["period_key"] == seed["day_key"] + + # 只读断言:看板不产生写副作用(身份与计量行数不变) + async with get_session_factory()() as session: + idents = ( + (await session.execute(select(UserIdentity).where(UserIdentity.user_id == user_id))) + .scalars() + .all() + ) + quotas = ( + ( + await session.execute( + select(Quota).where( + Quota.user_id == user_id, Quota.quota_key == seed["quota_key"] + ) + ) + ) + .scalars() + .all() + ) + assert len(idents) == 1 + assert len(quotas) == 1 + assert quotas[0].used == 3 + + # 最近事件/报告:各落一条种子数据 + assert len(data["recent_events"]) == 3 # 2 条 7 天内 + 1 条 8 天前(窗口不按详情截) + assert data["recent_events"][0]["server_ts"] >= data["recent_events"][-1]["server_ts"] + assert len(data["recent_reports"]) == 1 + report = data["recent_reports"][0] + assert report["tier"] == "basic" + assert report["status"] == "ready" + assert report["version"] == 1 + assert report["report_id"] + + +async def test_user_detail_not_found(client: Any) -> None: + resp = await client.get("/admin/api/v1/users/0", headers=AUTH) + assert resp.status_code == 404 + assert resp.json()["code"] == 40400 + + +async def test_events_stream_filters_and_pagination(client: Any, seed: dict[str, Any]) -> None: + user_id = seed["user_a"] + # user_id 过滤:只出种子用户的事件,server_ts 倒序 + resp = await client.get( + "/admin/api/v1/events", + headers=AUTH, + params={"user_id": user_id, "page_size": 2, "page": 1}, + ) + assert resp.status_code == 200 + body = resp.json()["data"] + assert body["total"] == 3 + assert len(body["items"]) == 2 + assert all(item["user_id"] == str(user_id) for item in body["items"]) + first, second = body["items"][0]["server_ts"], body["items"][1]["server_ts"] + assert first >= second + + # 翻第二页拿到最早那条(8 天前),证明分页生效 + resp = await client.get( + "/admin/api/v1/events", + headers=AUTH, + params={"user_id": user_id, "page_size": 2, "page": 2}, + ) + page2 = resp.json()["data"] + assert len(page2["items"]) == 1 + + # event_name 过滤 + ISO 时间范围:8 天前那条被 start 挡在窗外 + start = (datetime.now(UTC) - timedelta(days=7)).isoformat() + resp = await client.get( + "/admin/api/v1/events", + headers=AUTH, + params={"user_id": user_id, "event_name": _SEED_EVENT_NAMES[0], "start": start}, + ) + body = resp.json()["data"] + assert body["total"] == 1 + assert body["items"][0]["props"] == {"page": "home"} + + # epoch 秒也接受(与 ISO 等价窗口) + epoch_start = str(int((datetime.now(UTC) - timedelta(days=7)).timestamp())) + resp = await client.get( + "/admin/api/v1/events", + headers=AUTH, + params={"user_id": user_id, "event_name": _SEED_EVENT_NAMES[0], "start": epoch_start}, + ) + assert resp.json()["data"]["total"] == 1 + + # page_size 超上限 → 422(契约上限 200) + resp = await client.get("/admin/api/v1/events", headers=AUTH, params={"page_size": 201}) + assert resp.status_code == 422 + + +async def test_stats_overview_pivot_and_series(client: Any, seed: dict[str, Any]) -> None: + resp = await client.get("/admin/api/v1/stats/overview", headers=AUTH, params={"days": 7}) + assert resp.status_code == 200 + data = resp.json()["data"] + assert data["days"] == 7 + assert data["registered_total"] >= 2 + + # 透视表:种子用户的两个事件名都出现,PV 至少为种子条数 + pv = {(row["event_name"]) for row in data["daily_event_pv"]} + assert _SEED_EVENT_NAMES[0] in pv + assert _SEED_EVENT_NAMES[1] in pv + page_view_pv = sum( + row["pv"] for row in data["daily_event_pv"] if row["event_name"] == _SEED_EVENT_NAMES[0] + ) + assert page_view_pv >= 2 # 种子 A、B 各至少一条(库内其它数据只会更多) + + # 活跃序列:至少一天有活跃用户 + assert data["daily_active_users"] + assert max(row["active_users"] for row in data["daily_active_users"]) >= 1 + + # days 超上限 → 422 + resp = await client.get("/admin/api/v1/stats/overview", headers=AUTH, params={"days": 91}) + assert resp.status_code == 422