#!/usr/bin/env python """ Analytics Event Tracking System Tracks player actions and events for analytics and insights. """ import time import json import uuid from typing import Dict, Any, Optional from datetime import datetime import logging logger = logging.getLogger(__name__) # Database Schema for analytics_events table """ CREATE TABLE IF NOT EXISTS analytics_events ( id VARCHAR(36) PRIMARY KEY, player_id VARCHAR(36) NOT NULL, session_id VARCHAR(36), event_name VARCHAR(100) NOT NULL, properties JSON, timestamp BIGINT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, INDEX idx_player_id (player_id), INDEX idx_event_name (event_name), INDEX idx_timestamp (timestamp), INDEX idx_session_id (session_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; CREATE TABLE IF NOT EXISTS analytics_sessions ( session_id VARCHAR(36) PRIMARY KEY, player_id VARCHAR(36) NOT NULL, started_at BIGINT NOT NULL, ended_at BIGINT, duration_seconds INT, events_count INT DEFAULT 0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, INDEX idx_player_id (player_id), INDEX idx_started_at (started_at) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ class AnalyticsTracker: """Tracks analytics events for player actions.""" def __init__(self, db_connection=None): """ Initialize the analytics tracker. Args: db_connection: MySQL database connection """ self.db = db_connection self._sessions = {} # player_id -> session_id self._event_buffer = [] # Buffer for batch inserts self._buffer_size = 100 def start_session(self, player_id: str) -> str: """ Start a new analytics session for a player. Args: player_id: Player's unique identifier Returns: Session ID """ session_id = str(uuid.uuid4()) self._sessions[player_id] = session_id if self.db: try: cursor = self.db.cursor() cursor.execute(""" INSERT INTO analytics_sessions (session_id, player_id, started_at) VALUES (%s, %s, %s) """, (session_id, player_id, int(time.time()))) self.db.commit() cursor.close() except Exception as e: logger.error(f"Failed to start session: {e}") logger.info(f"Started analytics session {session_id} for player {player_id}") return session_id def end_session(self, player_id: str): """ End the current analytics session for a player. Args: player_id: Player's unique identifier """ session_id = self._sessions.get(player_id) if not session_id: return if self.db: try: cursor = self.db.cursor() ended_at = int(time.time()) # Calculate duration cursor.execute(""" UPDATE analytics_sessions SET ended_at = %s, duration_seconds = %s - started_at WHERE session_id = %s """, (ended_at, ended_at, session_id)) self.db.commit() cursor.close() except Exception as e: logger.error(f"Failed to end session: {e}") del self._sessions[player_id] logger.info(f"Ended analytics session {session_id} for player {player_id}") def track_event(self, player_id: str, event_name: str, properties: Optional[Dict[str, Any]] = None): """ Track an analytics event. Args: player_id: Player's unique identifier event_name: Name of the event (e.g., 'purchase_completed') properties: Optional event properties/metadata """ event_id = str(uuid.uuid4()) session_id = self._sessions.get(player_id) timestamp = int(time.time()) event = { 'id': event_id, 'player_id': player_id, 'session_id': session_id, 'event_name': event_name, 'properties': json.dumps(properties) if properties else None, 'timestamp': timestamp } self._event_buffer.append(event) # Flush buffer if it's full if len(self._event_buffer) >= self._buffer_size: self._flush_events() logger.debug(f"Tracked event '{event_name}' for player {player_id}") def _flush_events(self): """Flush buffered events to database atomically.""" if not self._event_buffer or not self.db: return from database.transactions import transaction try: with transaction() as (conn, cursor): # Insert all events atomically cursor.executemany(""" INSERT INTO analytics_events (id, player_id, session_id, event_name, properties, timestamp) VALUES (%s, %s, %s, %s, %s, %s) """, [(e['id'], e['player_id'], e['session_id'], e['event_name'], e['properties'], e['timestamp']) for e in self._event_buffer]) # Update session event counts for event in self._event_buffer: if event['session_id']: cursor.execute(""" UPDATE analytics_sessions SET events_count = events_count + 1 WHERE session_id = %s """, (event['session_id'],)) # Transaction commits automatically on success logger.info(f"Flushed {len(self._event_buffer)} analytics events to database") self._event_buffer = [] except Exception as e: logger.error(f"Failed to flush events: {e}") # Keep batch for retry def get_session_id(self, player_id: str) -> Optional[str]: """ Get the current session ID for a player. Args: player_id: Player's unique identifier Returns: Session ID or None """ return self._sessions.get(player_id) # Global analytics tracker _tracker = None def get_tracker(db_connection=None) -> AnalyticsTracker: """ Get or create the global analytics tracker instance. Args: db_connection: Optional database connection Returns: AnalyticsTracker instance """ global _tracker if _tracker is None: _tracker = AnalyticsTracker(db_connection) return _tracker def track_event(player_id: str, event_name: str, properties: Optional[Dict[str, Any]] = None): """ Track an analytics event (convenience function). Args: player_id: Player's unique identifier event_name: Name of the event properties: Optional event properties """ tracker = get_tracker() tracker.track_event(player_id, event_name, properties) # Convenience functions for common events def track_purchase_initiated(player_id: str, item_id: str, price: float, currency: str = 'USD'): """Track when a player initiates a purchase.""" track_event(player_id, 'purchase_initiated', { 'item_id': item_id, 'price': price, 'currency': currency }) def track_purchase_completed(player_id: str, item_id: str, price: float, currency: str = 'USD', transaction_id: str = None): """Track when a purchase is completed.""" track_event(player_id, 'purchase_completed', { 'item_id': item_id, 'price': price, 'currency': currency, 'transaction_id': transaction_id }) def track_conversation_sent(player_id: str, npc_id: str, message_type: str): """Track when a player sends a conversation message.""" track_event(player_id, 'conversation_sent', { 'npc_id': npc_id, 'message_type': message_type }) def track_level_up(player_id: str, new_level: int, category: str = 'general'): """Track when a player levels up.""" track_event(player_id, 'level_up', { 'new_level': new_level, 'category': category }) def track_tutorial_step(player_id: str, step_name: str, completed: bool = True): """Track tutorial progress.""" track_event(player_id, 'tutorial_step', { 'step_name': step_name, 'completed': completed }) def track_relationship_milestone(player_id: str, npc_id: str, relationship_type: str, affinity_level: int): """Track relationship milestones.""" track_event(player_id, 'relationship_milestone', { 'npc_id': npc_id, 'relationship_type': relationship_type, 'affinity_level': affinity_level }) def track_death(player_id: str, age: int, cause: str): """Track when a player's character dies.""" track_event(player_id, 'character_death', { 'age': age, 'cause': cause })