xsl
64175a1ea1
ob_event 落库通道：形状校验 + props 字节上限 + client_ts 合理性 + PII 键剔除，整批原子（FR-SYS-06 服务端部分）。 实现：packages/analytics/schemas.py（EventIn/EventBatchRequest/ EventBatchResponse + 事件名形状正则常量）；service.py 拆出纯函数 validate_batch（批量/props 字节/事件名形状/client_ts 窗口/PII 剔除， 任一不过整批 40020），AnalyticsService.record 组装 ObEvent 逐条 add + 一次 flush；routers/events.py 提供 POST /events/batch （/api/v1 与 /v1 别名，认证必需）。 测试：单测 tests/unit/test_events_validation.py 8 例（合法批、超批量、 props 超字节/不可序列化、事件名形状含尾随换行、client_ts 双边界、 naive 时区按 UTC、PII 键含大小写变体剔除）；集成 tests/integration/test_events.py 3 例（两前缀可用+落库断言、 非法批 40020 且库无半条、无 token 40100）。门禁全绿：ruff check/format、mypy（T2 列表 + domain strict）、lint-imports 4 契约、 单测 177 通过、集成 6 通过。 偏差说明：event_name 正则未写成 Field(pattern=...)，而是作为 EVENT_NAME_PATTERN 常量放 schemas.py、由服务层 validate_batch 强制—— 若挂在请求体校验上，非法事件名会先被 RequestValidationError 拦截、 返回通用 40001/422 信封，无法满足任务契约 §4「大写 event_name → 40020」的集成行为。正则本体与文档一致。
123 lines
4.9 KiB
Python
123 lines
4.9 KiB
Python
"""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)
|