taolib.testing.analytics.repository.event_repo

taolib.testing.analytics.repository.event_repo#

事件 Repository 模块。

提供事件的数据访问和 MongoDB 聚合管道。

Classes#

EventRepository

事件数据访问层。

Module Contents#

class taolib.testing.analytics.repository.event_repo.EventRepository(collection: motor.motor_asyncio.AsyncIOMotorCollection)#

Bases: taolib.testing._base.repository.AsyncRepository[taolib.testing.analytics.models.event.EventDocument]

事件数据访问层。

async bulk_create(events: list[dict[str, Any]]) → int#

批量创建事件。

参数:

events -- 事件数据字典列表

返回:

成功插入的事件数量

async find_by_session(session_id: str) → list[taolib.testing.analytics.models.event.EventDocument]#

根据会话 ID 查询事件（按时间排序）。

参数:

session_id -- 会话 ID

返回:

事件文档列表

async find_by_app(app_id: str, start: datetime.datetime, end: datetime.datetime, event_type: taolib.testing.analytics.models.enums.EventType | None = None, skip: int = 0, limit: int = 100) → list[taolib.testing.analytics.models.event.EventDocument]#

按应用和时间范围查询事件。

参数:

  • app_id -- 应用标识

  • start -- 开始时间

  • end -- 结束时间

  • event_type -- 事件类型（可选）

  • skip -- 跳过记录数

  • limit -- 限制记录数

返回:

事件文档列表

async aggregate_funnel(app_id: str, steps: list[str], start: datetime.datetime, end: datetime.datetime) → list[dict[str, Any]]#

转化漏斗聚合分析。

统计每个步骤的完成人数（基于 session_id 去重）。

参数:

  • app_id -- 应用标识

  • steps -- 漏斗步骤名称列表（对应 page_url 或 metadata.feature_name）

  • start -- 开始时间

  • end -- 结束时间

返回:

每个步骤的统计数据列表

async aggregate_feature_usage(app_id: str, start: datetime.datetime, end: datetime.datetime, limit: int = 20) → list[dict[str, Any]]#

功能使用排名聚合。

参数:

  • app_id -- 应用标识

  • start -- 开始时间

  • end -- 结束时间

  • limit -- 返回数量限制

返回:

功能使用排名列表

async aggregate_navigation_paths(app_id: str, start: datetime.datetime, end: datetime.datetime, limit: int = 50) → list[dict[str, Any]]#

用户导航路径聚合（页面流向）。

参数:

  • app_id -- 应用标识

  • start -- 开始时间

  • end -- 结束时间

  • limit -- 返回数量限制

返回:

页面导航路径列表 [{source, target, value}]

async aggregate_time_on_section(app_id: str, start: datetime.datetime, end: datetime.datetime) → list[dict[str, Any]]#

区域停留时间聚合。

参数:

  • app_id -- 应用标识

  • start -- 开始时间

  • end -- 结束时间

返回:

每个区域的平均停留时间

async aggregate_drop_off(app_id: str, flow_steps: list[str], start: datetime.datetime, end: datetime.datetime) → list[dict[str, Any]]#

流失点聚合分析。

计算多步流程中每一步的完成人数和流失率。

参数:

  • app_id -- 应用标识

  • flow_steps -- 流程步骤列表

  • start -- 开始时间

  • end -- 结束时间

返回:

每步的进入/完成/流失数据

async get_overview_stats(app_id: str, start: datetime.datetime, end: datetime.datetime) → dict[str, Any]#

获取概览统计数据。

参数:

  • app_id -- 应用标识

  • start -- 开始时间

  • end -- 结束时间

返回:

概览统计字典

async create_indexes() → None#

创建所有必要的索引。