taolib.testing.analytics.services.analytics_service

分析业务逻辑服务。

提供事件摄入和分析查询的业务逻辑。

Classes

AnalyticsService

分析服务。

Module Contents

class taolib.testing.analytics.services.analytics_service.AnalyticsService(event_repo: taolib.testing.analytics.repository.event_repo.EventRepository, session_repo: taolib.testing.analytics.repository.session_repo.SessionRepository)

分析服务。

_event_repo

_session_repo

async ingest_events(events: list[taolib.testing.analytics.models.event.EventCreate]) → dict[str, Any]

摄入事件数据。

参数:

events -- 事件列表

返回:

int, "rejected": int}

返回类型:

摄入结果 {"accepted"

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

获取概览统计。

参数:

• app_id -- 应用标识

• start -- 开始时间

• end -- 结束时间

返回:

概览统计数据

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

获取转化漏斗分析。

参数:

• app_id -- 应用标识

• steps -- 漏斗步骤列表

• start -- 开始时间

• end -- 结束时间

返回:

漏斗分析数据

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

获取功能使用排名。

参数:

• app_id -- 应用标识

• start -- 开始时间

• end -- 结束时间

• limit -- 返回数量限制

返回:

功能排名数据

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

获取用户导航路径。

参数:

• app_id -- 应用标识

• start -- 开始时间

• end -- 结束时间

• limit -- 返回数量限制

返回:

导航路径数据 (Sankey 格式)

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

获取区域停留时间分析。

参数:

• app_id -- 应用标识

• start -- 开始时间

• end -- 结束时间

返回:

区域停留时间数据

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

获取流失点分析。

参数:

• app_id -- 应用标识

• flow_steps -- 流程步骤列表

• start -- 开始时间

• end -- 结束时间

返回:

流失分析数据

static default_time_range() → tuple[datetime.datetime, datetime.datetime]

获取默认时间范围（最近 7 天）。