merge: T3 埋点上报（POST /events/batch，FR-SYS-06）
This commit is contained in:
@@ -1,10 +1,25 @@
|
||||
"""``POST /events/batch`` — 客户端埋点批量上报（骨架，T3 填充实现）。
|
||||
"""``POST /events/batch`` — 客户端埋点批量上报（FR-SYS-06 服务端入口）。
|
||||
|
||||
FR-SYS-06 服务端入口；实现契约见 docs/开发任务包/T3-埋点上报.md。
|
||||
埋点链路对客户端静默降级：非法批返回 40020（无 hint），客户端丢弃本地
|
||||
缓存即可，不打扰用户。校验规则见 packages/analytics/service.py。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from fastapi import APIRouter
|
||||
|
||||
from apps.api.deps import AnalyticsServiceDep, CurrentUserId
|
||||
from packages.analytics.schemas import EventBatchRequest, EventBatchResponse
|
||||
from packages.shared.response import ok
|
||||
|
||||
router = APIRouter(prefix="/events", tags=["events"])
|
||||
|
||||
|
||||
@router.post("/batch", summary="批量上报埋点事件（整批校验，任一非法整批拒绝）")
|
||||
async def post_event_batch(
|
||||
user_id: CurrentUserId,
|
||||
analytics: AnalyticsServiceDep,
|
||||
payload: EventBatchRequest,
|
||||
) -> dict[str, object]:
|
||||
accepted = await analytics.record(user_id=user_id, batch=payload.events)
|
||||
return ok(EventBatchResponse(accepted=accepted).model_dump())
|
||||
|
||||
@@ -0,0 +1,43 @@
|
||||
"""埋点上报的请求/响应模型（架构文档 §9.3 events，FR-SYS-06 服务端部分）。
|
||||
|
||||
统一信封（§9.2）之外，这里只描述 ``data`` 载荷。事件字典（event dictionary）
|
||||
V1 不做服务端枚举校验 —— 客户端事件名集合由 TD-23 埋点字典维护、尚未定稿，
|
||||
先只用形状正则兜底，字典定稿后再收紧。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
#: 事件名形状：小写字母开头，仅 [a-z0-9_.]，总长 2~64（架构 §6.2 event_name）。
|
||||
#:
|
||||
#: 注意它刻意不写成 ``Field(pattern=...)`` —— 请求体校验失败会落进通用的
|
||||
#: 40001/422 信封，而埋点链路要求 fail-close 地返回域错误 40020（客户端据此
|
||||
#: 丢弃本地缓存、静默降级）。因此正则在 service 层 ``validate_batch`` 里强制。
|
||||
EVENT_NAME_PATTERN = re.compile(r"^[a-z][a-z0-9_.]{1,63}$")
|
||||
|
||||
|
||||
class EventIn(BaseModel):
|
||||
"""单条埋点事件（客户端时钟不可信，client_ts 仅作参考）。"""
|
||||
|
||||
event_name: str = Field(description="事件名；形状见 EVENT_NAME_PATTERN，服务层 fail-close 校验")
|
||||
props: dict[str, Any] = Field(
|
||||
default_factory=dict, description="事件属性；PII 键会被服务端剔除"
|
||||
)
|
||||
client_ts: datetime | None = Field(
|
||||
default=None, description="客户端事件时间（ISO 8601）；缺省时仅用服务端时间"
|
||||
)
|
||||
|
||||
|
||||
class EventBatchRequest(BaseModel):
|
||||
"""整批上报：任一条非法则整批拒绝，不部分落库。"""
|
||||
|
||||
events: list[EventIn] = Field(min_length=1, description="本批事件，上限见 EVENTS_MAX_BATCH")
|
||||
|
||||
|
||||
class EventBatchResponse(BaseModel):
|
||||
accepted: int = Field(description="本批成功落库的事件条数；失败走 40020，不会返回部分值")
|
||||
@@ -1,10 +1,122 @@
|
||||
"""analytics 域 service — 骨架（T3 任务填充实现）。"""
|
||||
"""analytics 域 service —— ``ob_event`` 落库通道（FR-SYS-06 服务端部分）。
|
||||
|
||||
校验拆成模块级纯函数 :func:`validate_batch`（不碰 DB，单测直接覆盖），
|
||||
``AnalyticsService.record`` 只负责取配置、组装与落库。整批 fail-close：
|
||||
任一条不合法 → ``EVENTS_BATCH_REJECTED``（40020），绝不部分落库 ——
|
||||
半批数据会污染看板口径（TD-24 三张看板都吃这张表），宁可让客户端整批重试。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from collections.abc import Sequence
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from packages.analytics.schemas import EVENT_NAME_PATTERN, EventIn
|
||||
from packages.domain.ids import next_id
|
||||
from packages.shared.config import get_settings
|
||||
from packages.shared.errors import AppError, ErrorCode
|
||||
from packages.shared.logging import get_logger
|
||||
from packages.shared.models import ObEvent
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
#: 客户端时钟最多比服务端快 5 分钟（超出视为异常时间戳，防未来事件污染当日看板）。
|
||||
_CLIENT_TS_MAX_SKEW = timedelta(minutes=5)
|
||||
#: 离线缓存上报最多回溯 7 天（再久的旧事件口径已失真，直接拒收）。
|
||||
_CLIENT_TS_MAX_AGE = timedelta(days=7)
|
||||
|
||||
#: props 顶层出现即剔除的 PII 键（NFR-SEC-03：埋点与日志同族，都不落 PII）。
|
||||
_PII_PROPS_KEYS = frozenset({"phone", "token", "authorization"})
|
||||
|
||||
|
||||
def _drop_pii_keys(props: dict[str, Any], *, event_name: str) -> dict[str, Any]:
|
||||
"""剔除 props 顶层的 PII 键（大小写不敏感，``Token`` 这类变体也拦）。
|
||||
|
||||
日志只记被剔除的**键名**，不记值 —— 值本身就是 PII，不能二次泄漏。
|
||||
"""
|
||||
if not any(key.lower() in _PII_PROPS_KEYS for key in props):
|
||||
return props
|
||||
dropped = [key for key in props if key.lower() in _PII_PROPS_KEYS]
|
||||
logger.warning("events_pii_dropped", event_name=event_name, dropped_keys=dropped)
|
||||
return {key: value for key, value in props.items() if key.lower() not in _PII_PROPS_KEYS}
|
||||
|
||||
|
||||
def validate_batch(
|
||||
events: Sequence[EventIn],
|
||||
*,
|
||||
user_id: int,
|
||||
max_batch: int,
|
||||
max_bytes: int,
|
||||
now: datetime,
|
||||
) -> list[ObEvent]:
|
||||
"""整批 fail-close 校验并组装 ``ObEvent`` 行（纯函数，不触碰 DB）。
|
||||
|
||||
返回的行尚未进 session，由 ``record`` 统一 ``add_all`` + ``flush``，
|
||||
保证「要么全落、要么全拒」在服务边界内成立。``now`` 由调用方注入，
|
||||
既让单测可以钉死时间，也保证 server_ts 与 client_ts 窗口用的是同一时刻。
|
||||
"""
|
||||
if not 1 <= len(events) <= max_batch:
|
||||
raise AppError(ErrorCode.EVENTS_BATCH_REJECTED)
|
||||
|
||||
rows: list[ObEvent] = []
|
||||
for event in events:
|
||||
# 事件字典枚举校验等 TD-23 定稿后再收紧，V1 只约束形状。
|
||||
if EVENT_NAME_PATTERN.fullmatch(event.event_name) is None:
|
||||
raise AppError(ErrorCode.EVENTS_BATCH_REJECTED)
|
||||
|
||||
props = _drop_pii_keys(event.props, event_name=event.event_name)
|
||||
|
||||
# 字节上限以 UTF-8 序列化后计（与客户端计算口径一致），不可序列化同样拒收。
|
||||
try:
|
||||
encoded = json.dumps(props, ensure_ascii=False).encode()
|
||||
except (TypeError, ValueError) as exc:
|
||||
raise AppError(ErrorCode.EVENTS_BATCH_REJECTED) from exc
|
||||
if len(encoded) > max_bytes:
|
||||
raise AppError(ErrorCode.EVENTS_BATCH_REJECTED)
|
||||
|
||||
client_ts = event.client_ts
|
||||
if client_ts is not None:
|
||||
# 客户端漏发时区时按 UTC 解释（与服务端口径一致），仍落在窗口外则拒收。
|
||||
if client_ts.tzinfo is None:
|
||||
client_ts = client_ts.replace(tzinfo=UTC)
|
||||
if client_ts < now - _CLIENT_TS_MAX_AGE or client_ts > now + _CLIENT_TS_MAX_SKEW:
|
||||
raise AppError(ErrorCode.EVENTS_BATCH_REJECTED)
|
||||
|
||||
rows.append(
|
||||
ObEvent(
|
||||
event_id=next_id(),
|
||||
user_id=user_id,
|
||||
event_name=event.event_name,
|
||||
props=props,
|
||||
client_ts=client_ts,
|
||||
server_ts=now,
|
||||
)
|
||||
)
|
||||
return rows
|
||||
|
||||
|
||||
class AnalyticsService:
|
||||
def __init__(self, session: AsyncSession) -> None:
|
||||
self._session = session
|
||||
|
||||
async def record(self, *, user_id: int, batch: list[EventIn]) -> int:
|
||||
"""校验并落库一批埋点事件，返回落库条数。
|
||||
|
||||
只 flush 不 commit —— 批量原子性交给 ``session_dep`` 的统一提交/回滚，
|
||||
校验失败时一条都不会写入。
|
||||
"""
|
||||
settings = get_settings()
|
||||
rows = validate_batch(
|
||||
batch,
|
||||
user_id=user_id,
|
||||
max_batch=settings.events_max_batch,
|
||||
max_bytes=settings.events_props_max_bytes,
|
||||
now=datetime.now(UTC),
|
||||
)
|
||||
self._session.add_all(rows)
|
||||
await self._session.flush()
|
||||
return len(rows)
|
||||
|
||||
@@ -0,0 +1,137 @@
|
||||
"""埋点上报集成用例（真实 PostgreSQL，FR-SYS-06）。
|
||||
|
||||
覆盖：/v1 与 /api/v1 两个前缀、落库行内容（含 PII 剔除）、
|
||||
非法批 40020 且整批原子（库里没有半条）、无 token 40100。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import uuid
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
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
|
||||
async def client() -> Any:
|
||||
import httpx
|
||||
|
||||
from apps.api.main import create_app
|
||||
from packages.shared.config import get_settings
|
||||
|
||||
# 埋点链路不碰 LLM，无需替换 provider；lifespan 照常起网关与连接池
|
||||
app = create_app(get_settings())
|
||||
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 auth(client: Any) -> dict[str, Any]:
|
||||
"""dev-token 换一个专属测试用户，避免与其它用例的 ob_event 行串号。"""
|
||||
resp = await client.post(
|
||||
"/api/v1/auth/dev-token", json={"device_key": f"t3-{uuid.uuid4().hex[:8]}"}
|
||||
)
|
||||
assert resp.json()["code"] == 0
|
||||
data = resp.json()["data"]
|
||||
headers = {"Authorization": f"Bearer {data['access_token']}"}
|
||||
return {"headers": headers, "user_id": int(data["user_id"])}
|
||||
|
||||
|
||||
async def _count_events(user_id: int) -> int:
|
||||
from sqlalchemy import func, select
|
||||
|
||||
from packages.shared.db import get_session_factory
|
||||
from packages.shared.models import ObEvent
|
||||
|
||||
async with get_session_factory()() as session:
|
||||
stmt = select(func.count()).select_from(ObEvent).where(ObEvent.user_id == user_id)
|
||||
return int(await session.scalar(stmt))
|
||||
|
||||
|
||||
async def _fetch_events(user_id: int) -> list[Any]:
|
||||
from sqlalchemy import select
|
||||
|
||||
from packages.shared.db import get_session_factory
|
||||
from packages.shared.models import ObEvent
|
||||
|
||||
async with get_session_factory()() as session:
|
||||
result = await session.execute(
|
||||
select(ObEvent).where(ObEvent.user_id == user_id).order_by(ObEvent.event_id)
|
||||
)
|
||||
return list(result.scalars().all())
|
||||
|
||||
|
||||
async def test_batch_lands_two_rows_with_pii_stripped(client: Any, auth: dict[str, Any]) -> None:
|
||||
resp = await client.post(
|
||||
"/v1/events/batch",
|
||||
json={
|
||||
"events": [
|
||||
{
|
||||
"event_name": "page_view",
|
||||
"props": {"page": "home", "phone": "13800001234"},
|
||||
"client_ts": datetime.now(UTC).isoformat(),
|
||||
},
|
||||
{"event_name": "outfit_card.tap", "props": {"card_id": 42}},
|
||||
]
|
||||
},
|
||||
headers=auth["headers"],
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert body["code"] == 0
|
||||
assert body["data"] == {"accepted": 2}
|
||||
|
||||
rows = await _fetch_events(auth["user_id"])
|
||||
assert len(rows) == 2
|
||||
assert rows[0].event_name == "page_view"
|
||||
assert rows[0].props == {"page": "home"} # phone 已在服务端剔除
|
||||
assert rows[0].client_ts is not None
|
||||
assert rows[0].server_ts is not None
|
||||
assert rows[1].event_name == "outfit_card.tap"
|
||||
assert rows[1].props == {"card_id": 42}
|
||||
assert rows[1].client_ts is None
|
||||
|
||||
|
||||
async def test_invalid_batch_is_rejected_atomically(client: Any, auth: dict[str, Any]) -> None:
|
||||
before = await _count_events(auth["user_id"])
|
||||
|
||||
resp = await client.post(
|
||||
"/api/v1/events/batch",
|
||||
json={
|
||||
"events": [
|
||||
{"event_name": "page_view"},
|
||||
{"event_name": "PageView"}, # 大写 → 形状非法
|
||||
]
|
||||
},
|
||||
headers=auth["headers"],
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
assert resp.json()["code"] == 40020
|
||||
|
||||
# 整批原子：合法的那条也不能半落库
|
||||
assert await _count_events(auth["user_id"]) == before
|
||||
|
||||
|
||||
async def test_events_require_authentication(client: Any) -> None:
|
||||
resp = await client.post("/api/v1/events/batch", json={"events": [{"event_name": "page_view"}]})
|
||||
assert resp.status_code == 401
|
||||
assert resp.json()["code"] == 40100
|
||||
@@ -0,0 +1,136 @@
|
||||
"""埋点整批校验的纯逻辑用例（FR-SYS-06，不连库）。
|
||||
|
||||
校验入口是 packages.analytics.service.validate_batch —— 模块级纯函数，
|
||||
时间与上限全部注入，便于钉死边界值逐条断言。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from packages.analytics.schemas import EventIn
|
||||
from packages.analytics.service import validate_batch
|
||||
from packages.shared.errors import AppError, ErrorCode
|
||||
|
||||
#: 钉死的"服务端当前时间"，所有 client_ts 边界都相对它构造。
|
||||
NOW = datetime(2026, 8, 14, 12, 0, 0, tzinfo=UTC)
|
||||
USER_ID = 7_100_000_000_000_000_001
|
||||
|
||||
MAX_BATCH = 50
|
||||
MAX_BYTES = 8192
|
||||
|
||||
|
||||
def _ev(
|
||||
name: str = "page_view",
|
||||
props: dict[str, Any] | None = None,
|
||||
client_ts: datetime | None = None,
|
||||
) -> EventIn:
|
||||
return EventIn(event_name=name, props=props if props is not None else {}, client_ts=client_ts)
|
||||
|
||||
|
||||
def _run(events: list[EventIn]) -> list[Any]:
|
||||
return validate_batch(
|
||||
events, user_id=USER_ID, max_batch=MAX_BATCH, max_bytes=MAX_BYTES, now=NOW
|
||||
)
|
||||
|
||||
|
||||
def _assert_rejected(events: list[EventIn]) -> None:
|
||||
"""fail-close：任一规则不过 → 整批 40020。"""
|
||||
with pytest.raises(AppError) as excinfo:
|
||||
_run(events)
|
||||
assert excinfo.value.code == ErrorCode.EVENTS_BATCH_REJECTED
|
||||
assert int(excinfo.value.code) == 40020
|
||||
|
||||
|
||||
def test_valid_batch_passes_and_assembles_rows() -> None:
|
||||
rows = _run(
|
||||
[
|
||||
_ev("outfit_card.tap", {"card_id": 42}, NOW - timedelta(minutes=1)),
|
||||
_ev("page_view"),
|
||||
]
|
||||
)
|
||||
assert len(rows) == 2
|
||||
assert all(row.user_id == USER_ID for row in rows)
|
||||
# server_ts 用注入的 now，与 client_ts 窗口判定同一时刻
|
||||
assert all(row.server_ts == NOW for row in rows)
|
||||
assert rows[0].event_name == "outfit_card.tap"
|
||||
assert rows[0].props == {"card_id": 42}
|
||||
assert rows[0].client_ts == NOW - timedelta(minutes=1)
|
||||
assert rows[1].client_ts is None # 缺省 client_ts 允许
|
||||
# 雪花 id 逐条独立分配
|
||||
assert len({row.event_id for row in rows}) == 2
|
||||
|
||||
|
||||
def test_empty_and_oversized_batch_are_rejected() -> None:
|
||||
_assert_rejected([]) # schema 的 min_length=1 之外，服务层同样兜底
|
||||
_assert_rejected([_ev() for _ in range(MAX_BATCH + 1)])
|
||||
# 恰好等于上限是合法的
|
||||
assert len(_run([_ev() for _ in range(MAX_BATCH)])) == MAX_BATCH
|
||||
|
||||
|
||||
def test_oversize_props_are_rejected() -> None:
|
||||
# 默认上限 8192 字节；8KB 的值序列化后必然超限
|
||||
big = {"blob": "x" * 8192}
|
||||
assert len(json.dumps(big, ensure_ascii=False).encode()) > MAX_BYTES
|
||||
_assert_rejected([_ev(props=big)])
|
||||
# 与合法事件混在同一批也要整批拒绝（fail-close，不部分落库）
|
||||
_assert_rejected([_ev(), _ev(props=big)])
|
||||
|
||||
|
||||
def test_unserializable_props_are_rejected() -> None:
|
||||
# set 不可 JSON 序列化 → fail-close 拒收，而不是落库时炸事务
|
||||
_assert_rejected([_ev(props={"bad": {1, 2}})]) # type: ignore[dict-item]
|
||||
|
||||
|
||||
def test_event_name_shape_is_enforced() -> None:
|
||||
for bad_name in (
|
||||
"PageView", # 大写
|
||||
"1page", # 数字开头
|
||||
"p", # 太短（正则要求 >= 2 位）
|
||||
"page view", # 空格
|
||||
"page_view!", # 非法字符
|
||||
"a" * 65, # 超长（正则上限 64）
|
||||
"page_view\n", # $ 可被尾随换行骗过，fullmatch 不行
|
||||
):
|
||||
_assert_rejected([_ev(bad_name)])
|
||||
# 64 位以内、含点与下划线、数字开头之外的位置都是合法的
|
||||
assert len(_run([_ev("a" + "b" * 63)])) == 1
|
||||
assert len(_run([_ev("wardrobe.item_2.sync_ok")])) == 1
|
||||
|
||||
|
||||
def test_client_ts_window_boundaries() -> None:
|
||||
# 不晚于 now+5min：恰好 +5min 合法，再多 1 秒拒收
|
||||
assert len(_run([_ev(client_ts=NOW + timedelta(minutes=5))])) == 1
|
||||
_assert_rejected([_ev(client_ts=NOW + timedelta(minutes=5, seconds=1))])
|
||||
# 不早于 7 天前：恰好 7 天前合法，再早 1 秒拒收
|
||||
assert len(_run([_ev(client_ts=NOW - timedelta(days=7))])) == 1
|
||||
_assert_rejected([_ev(client_ts=NOW - timedelta(days=7, seconds=1))])
|
||||
# 同批一条合法 + 一条越界 → 整批拒
|
||||
_assert_rejected([_ev(client_ts=NOW), _ev(client_ts=NOW + timedelta(hours=1))])
|
||||
|
||||
|
||||
def test_naive_client_ts_is_interpreted_as_utc() -> None:
|
||||
# 客户端漏发时区按 UTC 解释，窗口内仍可落库
|
||||
rows = _run([_ev(client_ts=datetime(2026, 8, 14, 11, 0, 0))])
|
||||
assert rows[0].client_ts == datetime(2026, 8, 14, 11, 0, 0, tzinfo=UTC)
|
||||
|
||||
|
||||
def test_pii_keys_are_stripped_from_props() -> None:
|
||||
rows = _run(
|
||||
[
|
||||
_ev(
|
||||
"report.share",
|
||||
{
|
||||
"phone": "13800001234",
|
||||
"Token": "Bearer abc", # 大小写变体也拦
|
||||
"authorization": "Basic xyz",
|
||||
"scene": "home",
|
||||
},
|
||||
)
|
||||
]
|
||||
)
|
||||
assert rows[0].props == {"scene": "home"}
|
||||
Reference in New Issue
Block a user