Files
whale-town/scripts/UserBehaviorAnalytics.gd
2025-12-05 19:00:14 +08:00

669 lines
20 KiB
GDScript
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
extends Node
class_name UserBehaviorAnalytics
## 用户行为分析系统
## 收集、分析和报告用户在游戏中的行为数据
# 行为事件类型枚举
enum EventType {
LOGIN, # 登录
LOGOUT, # 登出
CHARACTER_CREATED, # 角色创建
MOVEMENT, # 移动
DIALOGUE_STARTED, # 开始对话
DIALOGUE_ENDED, # 结束对话
MESSAGE_SENT, # 发送消息
FRIEND_REQUEST, # 好友请求
EVENT_JOINED, # 参加活动
INTERACTION, # 交互行为
UI_ACTION, # UI操作
ERROR_OCCURRED, # 错误发生
PERFORMANCE_METRIC # 性能指标
}
# 行为数据结构
class BehaviorEvent:
var event_id: String
var event_type: EventType
var user_id: String
var timestamp: float
var session_id: String
var data: Dictionary = {}
var context: Dictionary = {}
func _init(type: EventType, uid: String, sid: String, event_data: Dictionary = {}):
event_id = generate_event_id()
event_type = type
user_id = uid
session_id = sid
timestamp = Time.get_unix_time_from_system()
data = event_data.duplicate()
context = _get_context_data()
func generate_event_id() -> String:
return "evt_%d_%d" % [Time.get_unix_time_from_system(), randi()]
func _get_context_data() -> Dictionary:
return {
"platform": OS.get_name(),
"viewport_size": get_viewport().get_visible_rect().size if get_viewport() else Vector2.ZERO,
"fps": Engine.get_frames_per_second()
}
# 数据存储和管理
var behavior_events: Array[BehaviorEvent] = []
var current_session_id: String = ""
var current_user_id: String = "player"
var session_start_time: float = 0.0
var max_events_in_memory: int = 1000
var analytics_enabled: bool = true
# 数据持久化
var analytics_file_path: String = "user://user_analytics.json"
var session_file_path: String = "user://session_data.json"
# 统计数据缓存
var cached_statistics: Dictionary = {}
var statistics_cache_time: float = 0.0
var cache_duration: float = 300.0 # 5分钟缓存
# 信号
signal behavior_recorded(event_type: EventType, data: Dictionary)
signal session_started(session_id: String)
signal session_ended(session_id: String, duration: float)
signal analytics_report_generated(report: Dictionary)
func _ready():
"""初始化用户行为分析系统"""
_start_new_session()
_load_analytics_data()
# 设置定时保存
var save_timer = Timer.new()
save_timer.wait_time = 60.0 # 每分钟保存一次
save_timer.timeout.connect(_save_analytics_data)
save_timer.autostart = true
add_child(save_timer)
print("UserBehaviorAnalytics initialized")
## 开始新会话
func _start_new_session() -> void:
"""开始新的用户会话"""
current_session_id = _generate_session_id()
session_start_time = Time.get_unix_time_from_system()
# 记录会话开始事件
record_event(EventType.LOGIN, {"session_start": true})
session_started.emit(current_session_id)
print("New session started: ", current_session_id)
## 结束当前会话
func end_current_session() -> void:
"""结束当前用户会话"""
if current_session_id.is_empty():
return
var session_duration = Time.get_unix_time_from_system() - session_start_time
# 记录会话结束事件
record_event(EventType.LOGOUT, {
"session_duration": session_duration,
"events_count": behavior_events.size()
})
session_ended.emit(current_session_id, session_duration)
# 保存数据
_save_analytics_data()
print("Session ended: ", current_session_id, " Duration: ", session_duration, "s")
## 记录行为事件
func record_event(event_type: EventType, event_data: Dictionary = {}) -> void:
"""
记录用户行为事件
@param event_type: 事件类型
@param event_data: 事件数据
"""
if not analytics_enabled:
return
var event = BehaviorEvent.new(event_type, current_user_id, current_session_id, event_data)
behavior_events.append(event)
# 限制内存中的事件数量
if behavior_events.size() > max_events_in_memory:
behavior_events.pop_front()
# 清除统计缓存
_clear_statistics_cache()
# 发射信号
behavior_recorded.emit(event_type, event_data)
# 调试输出(仅在开发模式)
if OS.is_debug_build():
print("Behavior recorded: ", EventType.keys()[event_type], " - ", event_data)
## 记录移动行为
func record_movement(from_position: Vector2, to_position: Vector2, duration: float) -> void:
"""
记录角色移动行为
@param from_position: 起始位置
@param to_position: 目标位置
@param duration: 移动时长
"""
var distance = from_position.distance_to(to_position)
record_event(EventType.MOVEMENT, {
"from": {"x": from_position.x, "y": from_position.y},
"to": {"x": to_position.x, "y": to_position.y},
"distance": distance,
"duration": duration,
"speed": distance / max(duration, 0.1)
})
## 记录对话行为
func record_dialogue_interaction(target_character: String, message_count: int, duration: float) -> void:
"""
记录对话交互行为
@param target_character: 对话目标角色
@param message_count: 消息数量
@param duration: 对话时长
"""
record_event(EventType.DIALOGUE_STARTED, {
"target": target_character,
"message_count": message_count,
"duration": duration
})
## 记录UI操作
func record_ui_action(action_type: String, ui_element: String, additional_data: Dictionary = {}) -> void:
"""
记录UI操作行为
@param action_type: 操作类型click, hover, input等
@param ui_element: UI元素标识
@param additional_data: 额外数据
"""
var ui_data = {
"action": action_type,
"element": ui_element
}
ui_data.merge(additional_data)
record_event(EventType.UI_ACTION, ui_data)
## 记录性能指标
func record_performance_metric(metric_name: String, value: float, unit: String = "") -> void:
"""
记录性能指标
@param metric_name: 指标名称
@param value: 指标值
@param unit: 单位
"""
record_event(EventType.PERFORMANCE_METRIC, {
"metric": metric_name,
"value": value,
"unit": unit,
"fps": Engine.get_frames_per_second(),
"memory_usage": OS.get_static_memory_usage_by_type()
})
## 记录错误事件
func record_error(error_type: String, error_message: String, context_data: Dictionary = {}) -> void:
"""
记录错误事件
@param error_type: 错误类型
@param error_message: 错误消息
@param context_data: 上下文数据
"""
var error_data = {
"error_type": error_type,
"message": error_message,
"stack_trace": get_stack() if OS.is_debug_build() else []
}
error_data.merge(context_data)
record_event(EventType.ERROR_OCCURRED, error_data)
## 生成会话ID
func _generate_session_id() -> String:
"""生成唯一的会话ID"""
var timestamp = Time.get_unix_time_from_system()
var random = randi()
return "session_%d_%d" % [timestamp, random]
## 清除统计缓存
func _clear_statistics_cache() -> void:
"""清除统计数据缓存"""
cached_statistics.clear()
statistics_cache_time = 0.0## 获
func get_basic_statistics() -> Dictionary:
"""
获取基础统计信息
@return: 统计数据字典
"""
# 检查缓存
var current_time = Time.get_unix_time_from_system()
if not cached_statistics.is_empty() and (current_time - statistics_cache_time) < cache_duration:
return cached_statistics
var stats = {}
# 事件总数统计
stats["total_events"] = behavior_events.size()
stats["current_session_events"] = _count_session_events(current_session_id)
stats["session_duration"] = current_time - session_start_time
# 事件类型分布
var event_type_counts = {}
for event in behavior_events:
var type_name = EventType.keys()[event.event_type]
event_type_counts[type_name] = event_type_counts.get(type_name, 0) + 1
stats["event_types"] = event_type_counts
# 活跃度分析
stats["activity_metrics"] = _calculate_activity_metrics()
# 性能统计
stats["performance_metrics"] = _calculate_performance_metrics()
# 错误统计
stats["error_metrics"] = _calculate_error_metrics()
# 缓存结果
cached_statistics = stats
statistics_cache_time = current_time
return stats
## 计算活跃度指标
func _calculate_activity_metrics() -> Dictionary:
"""计算用户活跃度指标"""
var metrics = {}
var current_time = Time.get_unix_time_from_system()
# 最近1小时的活动
var recent_events = _get_events_in_timeframe(current_time - 3600)
metrics["events_last_hour"] = recent_events.size()
# 移动活跃度
var movement_events = _get_events_by_type(EventType.MOVEMENT)
metrics["total_movements"] = movement_events.size()
if movement_events.size() > 0:
var total_distance = 0.0
for event in movement_events:
total_distance += event.data.get("distance", 0.0)
metrics["total_distance_moved"] = total_distance
metrics["average_movement_distance"] = total_distance / movement_events.size()
# 社交活跃度
var dialogue_events = _get_events_by_type(EventType.DIALOGUE_STARTED)
metrics["dialogue_sessions"] = dialogue_events.size()
var message_events = _get_events_by_type(EventType.MESSAGE_SENT)
metrics["messages_sent"] = message_events.size()
# UI交互频率
var ui_events = _get_events_by_type(EventType.UI_ACTION)
metrics["ui_interactions"] = ui_events.size()
return metrics
## 计算性能指标
func _calculate_performance_metrics() -> Dictionary:
"""计算性能相关指标"""
var metrics = {}
var performance_events = _get_events_by_type(EventType.PERFORMANCE_METRIC)
if performance_events.size() > 0:
var fps_values = []
var memory_values = []
for event in performance_events:
if event.context.has("fps"):
fps_values.append(event.context.fps)
if event.data.has("memory_usage"):
memory_values.append(event.data.memory_usage)
if fps_values.size() > 0:
metrics["average_fps"] = _calculate_average(fps_values)
metrics["min_fps"] = fps_values.min()
metrics["max_fps"] = fps_values.max()
if memory_values.size() > 0:
metrics["average_memory"] = _calculate_average(memory_values)
metrics["peak_memory"] = memory_values.max()
# 当前性能状态
metrics["current_fps"] = Engine.get_frames_per_second()
metrics["current_memory"] = OS.get_static_memory_usage_by_type()
return metrics
## 计算错误指标
func _calculate_error_metrics() -> Dictionary:
"""计算错误相关指标"""
var metrics = {}
var error_events = _get_events_by_type(EventType.ERROR_OCCURRED)
metrics["total_errors"] = error_events.size()
# 错误类型分布
var error_types = {}
for event in error_events:
var error_type = event.data.get("error_type", "unknown")
error_types[error_type] = error_types.get(error_type, 0) + 1
metrics["error_types"] = error_types
# 最近错误
if error_events.size() > 0:
var latest_error = error_events[-1]
metrics["latest_error"] = {
"type": latest_error.data.get("error_type", "unknown"),
"message": latest_error.data.get("message", ""),
"timestamp": latest_error.timestamp
}
return metrics
## 生成用户行为报告
func generate_behavior_report(timeframe_hours: float = 24.0) -> Dictionary:
"""
生成用户行为分析报告
@param timeframe_hours: 分析时间范围(小时)
@return: 行为报告
"""
var current_time = Time.get_unix_time_from_system()
var start_time = current_time - (timeframe_hours * 3600)
var report = {}
report["report_generated_at"] = current_time
report["timeframe_hours"] = timeframe_hours
report["user_id"] = current_user_id
report["session_id"] = current_session_id
# 获取时间范围内的事件
var timeframe_events = _get_events_in_timeframe(start_time)
report["events_in_timeframe"] = timeframe_events.size()
# 活动模式分析
report["activity_patterns"] = _analyze_activity_patterns(timeframe_events)
# 用户偏好分析
report["user_preferences"] = _analyze_user_preferences(timeframe_events)
# 会话质量分析
report["session_quality"] = _analyze_session_quality(timeframe_events)
# 发射报告生成信号
analytics_report_generated.emit(report)
return report
## 分析活动模式
func _analyze_activity_patterns(events: Array[BehaviorEvent]) -> Dictionary:
"""分析用户活动模式"""
var patterns = {}
# 按小时分组活动
var hourly_activity = {}
for event in events:
var hour = Time.get_datetime_dict_from_unix_time(event.timestamp).hour
hourly_activity[hour] = hourly_activity.get(hour, 0) + 1
patterns["hourly_distribution"] = hourly_activity
# 最活跃时段
var max_activity = 0
var peak_hour = 0
for hour in hourly_activity:
if hourly_activity[hour] > max_activity:
max_activity = hourly_activity[hour]
peak_hour = hour
patterns["peak_activity_hour"] = peak_hour
patterns["peak_activity_count"] = max_activity
# 活动类型偏好
var type_preferences = {}
for event in events:
var type_name = EventType.keys()[event.event_type]
type_preferences[type_name] = type_preferences.get(type_name, 0) + 1
patterns["activity_type_preferences"] = type_preferences
return patterns
## 分析用户偏好
func _analyze_user_preferences(events: Array[BehaviorEvent]) -> Dictionary:
"""分析用户偏好和兴趣"""
var preferences = {}
# 对话偏好
var dialogue_targets = {}
var dialogue_events = events.filter(func(e): return e.event_type == EventType.DIALOGUE_STARTED)
for event in dialogue_events:
var target = event.data.get("target", "unknown")
dialogue_targets[target] = dialogue_targets.get(target, 0) + 1
preferences["preferred_dialogue_targets"] = dialogue_targets
# UI使用偏好
var ui_preferences = {}
var ui_events = events.filter(func(e): return e.event_type == EventType.UI_ACTION)
for event in ui_events:
var element = event.data.get("element", "unknown")
ui_preferences[element] = ui_preferences.get(element, 0) + 1
preferences["ui_usage_patterns"] = ui_preferences
# 移动模式偏好
var movement_events = events.filter(func(e): return e.event_type == EventType.MOVEMENT)
if movement_events.size() > 0:
var distances = movement_events.map(func(e): return e.data.get("distance", 0.0))
preferences["average_movement_distance"] = _calculate_average(distances)
preferences["movement_frequency"] = movement_events.size()
return preferences
## 分析会话质量
func _analyze_session_quality(events: Array[BehaviorEvent]) -> Dictionary:
"""分析会话质量指标"""
var quality = {}
# 错误率
var error_events = events.filter(func(e): return e.event_type == EventType.ERROR_OCCURRED)
quality["error_rate"] = float(error_events.size()) / max(events.size(), 1)
# 参与度(基于事件多样性)
var unique_event_types = {}
for event in events:
unique_event_types[event.event_type] = true
quality["engagement_score"] = float(unique_event_types.size()) / EventType.size()
# 会话连续性(事件间隔分析)
if events.size() > 1:
var intervals = []
for i in range(1, events.size()):
intervals.append(events[i].timestamp - events[i-1].timestamp)
quality["average_event_interval"] = _calculate_average(intervals)
quality["session_continuity"] = 1.0 / (1.0 + _calculate_average(intervals))
return quality
## 辅助函数:计算平均值
func _calculate_average(values: Array) -> float:
"""计算数组平均值"""
if values.is_empty():
return 0.0
var sum = 0.0
for value in values:
sum += float(value)
return sum / values.size()
## 辅助函数:按类型获取事件
func _get_events_by_type(event_type: EventType) -> Array[BehaviorEvent]:
"""获取指定类型的事件"""
return behavior_events.filter(func(event): return event.event_type == event_type)
## 辅助函数:获取时间范围内的事件
func _get_events_in_timeframe(start_time: float) -> Array[BehaviorEvent]:
"""获取指定时间范围内的事件"""
return behavior_events.filter(func(event): return event.timestamp >= start_time)
## 辅助函数:统计会话事件数
func _count_session_events(session_id: String) -> int:
"""统计指定会话的事件数量"""
return behavior_events.filter(func(event): return event.session_id == session_id).size()
## 保存分析数据
func _save_analytics_data() -> void:
"""保存分析数据到本地文件"""
var data = {
"events": [],
"session_id": current_session_id,
"session_start_time": session_start_time,
"user_id": current_user_id,
"saved_at": Time.get_unix_time_from_system()
}
# 序列化事件数据(只保存最近的事件)
var events_to_save = behavior_events.slice(max(0, behavior_events.size() - 500), behavior_events.size())
for event in events_to_save:
data.events.append({
"event_id": event.event_id,
"event_type": event.event_type,
"user_id": event.user_id,
"session_id": event.session_id,
"timestamp": event.timestamp,
"data": event.data,
"context": event.context
})
var file = FileAccess.open(analytics_file_path, FileAccess.WRITE)
if file:
var json_string = JSON.stringify(data)
file.store_string(json_string)
file.close()
print("Analytics data saved: ", events_to_save.size(), " events")
else:
print("Failed to save analytics data")
## 加载分析数据
func _load_analytics_data() -> void:
"""从本地文件加载分析数据"""
if not FileAccess.file_exists(analytics_file_path):
print("No analytics data file found, starting fresh")
return
var file = FileAccess.open(analytics_file_path, FileAccess.READ)
if file:
var json_string = file.get_as_text()
file.close()
var json = JSON.new()
var parse_result = json.parse(json_string)
if parse_result == OK:
var data = json.data
# 加载事件数据
if data.has("events"):
for event_data in data.events:
var event = BehaviorEvent.new(
event_data.get("event_type", EventType.UI_ACTION),
event_data.get("user_id", ""),
event_data.get("session_id", "")
)
event.event_id = event_data.get("event_id", "")
event.timestamp = event_data.get("timestamp", 0.0)
event.data = event_data.get("data", {})
event.context = event_data.get("context", {})
behavior_events.append(event)
print("Analytics data loaded: ", behavior_events.size(), " events")
else:
print("Failed to parse analytics data JSON")
else:
print("Failed to open analytics data file")
## 导出分析报告
func export_analytics_report(export_path: String, timeframe_hours: float = 24.0) -> bool:
"""
导出分析报告到文件
@param export_path: 导出文件路径
@param timeframe_hours: 分析时间范围
@return: 是否成功导出
"""
var report = generate_behavior_report(timeframe_hours)
var file = FileAccess.open(export_path, FileAccess.WRITE)
if file:
var json_string = JSON.stringify(report)
file.store_string(json_string)
file.close()
print("Analytics report exported to: ", export_path)
return true
else:
print("Failed to export analytics report to: ", export_path)
return false
## 清理旧数据
func cleanup_old_data(days_to_keep: int = 7) -> void:
"""
清理旧的分析数据
@param days_to_keep: 保留天数
"""
var cutoff_time = Time.get_unix_time_from_system() - (days_to_keep * 86400)
var original_count = behavior_events.size()
behavior_events = behavior_events.filter(func(event): return event.timestamp >= cutoff_time)
var removed_count = original_count - behavior_events.size()
if removed_count > 0:
print("Cleaned up ", removed_count, " old analytics events")
_save_analytics_data()
## 获取实时统计
func get_realtime_statistics() -> Dictionary:
"""获取实时统计信息(不使用缓存)"""
var stats = {}
stats["current_session_duration"] = Time.get_unix_time_from_system() - session_start_time
stats["events_this_session"] = _count_session_events(current_session_id)
stats["current_fps"] = Engine.get_frames_per_second()
stats["memory_usage"] = OS.get_static_memory_usage_by_type()
# 最近5分钟的活动
var recent_time = Time.get_unix_time_from_system() - 300
var recent_events = _get_events_in_timeframe(recent_time)
stats["recent_activity"] = recent_events.size()
return stats
## 设置分析配置
func set_analytics_config(enabled: bool, max_events: int = 1000) -> void:
"""
设置分析系统配置
@param enabled: 是否启用分析
@param max_events: 最大事件数量
"""
analytics_enabled = enabled
max_events_in_memory = max_events
print("Analytics config updated - Enabled: ", enabled, " Max events: ", max_events)
func _exit_tree():
"""节点退出时保存数据"""
end_current_session()