analytics_tracker.py

# analytics_tracker.py

import json
import datetime
import sqlite3
import logging
from collections import defaultdict

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("analytics.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger("ecommerce_tracker")

# Database setup
DB_PATH = "analytics_data.db"

class EcommerceTracker:
    def __init__(self):
        self.session_data = defaultdict(dict)
        self.init_database()

    def init_database(self):
        """Initialize database with ecommerce tables"""
        try:
            conn = sqlite3.connect(DB_PATH)
            cursor = conn.cursor()
            
            cursor.execute('''
            CREATE TABLE IF NOT EXISTS conversion_events (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                session_id TEXT,
                conversation_id TEXT,
                event_type TEXT CHECK(event_type IN ('product_view', 'cart_add', 'purchase')),
                product_id TEXT,
                timestamp TEXT,
                metadata TEXT
            )''')
            
            cursor.execute('''
            CREATE TABLE IF NOT EXISTS conversation_metrics (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                conversation_id TEXT UNIQUE,
                session_id TEXT,
                intent_category TEXT,
                product_mentions TEXT,
                satisfaction INTEGER,
                conversion_count INTEGER,
                timestamp TEXT
            )''')
            
            conn.commit()
            conn.close()
            
        except Exception as e:
            logger.error(f"Database initialization failed: {str(e)}")
            raise

    def log_conversation(self, user_id, message, response, metadata=None):
        """Log conversation with ecommerce tracking"""
        try:
            conversation_id = metadata.get('conversation_id') if metadata else None
            session_id = metadata.get('session_id') if metadata else None
            
            # Store conversation in database
            conn = sqlite3.connect(DB_PATH)
            cursor = conn.cursor()
            
            cursor.execute('''
            INSERT INTO conversations 
            (user_id, conversation_id, session_id, timestamp, message, response, 
             response_time, sentiment_score, completed, satisfaction_rating, 
             intent_category, product_mentions, conversion_events, metadata)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            ''', (
                user_id,
                conversation_id,
                session_id,
                datetime.datetime.now().isoformat(),
                message,
                response,
                metadata.get('response_time', 0) if metadata else 0,
                metadata.get('sentiment', 0) if metadata else 0,
                metadata.get('completed', 0) if metadata else 0,
                metadata.get('satisfaction', 0) if metadata else 0,
                metadata.get('intent') if metadata else None,
                json.dumps(metadata.get('products', [])) if metadata else '[]',
                json.dumps(metadata.get('conversions', [])) if metadata else '[]',
                json.dumps(metadata) if metadata else '{}'
            ))
            
            conn.commit()
            conn.close()
            return True
            
        except Exception as e:
            logger.error(f"Error logging conversation: {str(e)}")
            return False

    def track_conversion(self, event_type, product_id, metadata=None):
        """Track ecommerce conversion events"""
        try:
            session_id = metadata.get('session_id') if metadata else None
            conversation_id = metadata.get('conversation_id') if metadata else None
            
            conn = sqlite3.connect(DB_PATH)
            cursor = conn.cursor()
            
            cursor.execute('''
            INSERT INTO conversion_events 
            (session_id, conversation_id, event_type, product_id, timestamp, metadata)
            VALUES (?, ?, ?, ?, ?, ?)
            ''', (
                session_id,
                conversation_id,
                event_type,
                product_id,
                datetime.datetime.now().isoformat(),
                json.dumps(metadata) if metadata else '{}'
            ))
            
            conn.commit()
            conn.close()
            return True
            
        except Exception as e:
            logger.error(f"Error tracking conversion: {str(e)}")
            return False

    def generate_ecommerce_report(self, period='day'):
        """Generate ecommerce performance report"""
        try:
            conn = sqlite3.connect(DB_PATH)
            cursor = conn.cursor()
            
            # Get conversion metrics
            cursor.execute('''
            SELECT 
                COUNT(DISTINCT conversation_id) as total_conversations,
                SUM(CASE WHEN event_type = 'product_view' THEN 1 ELSE 0 END) as product_views,
                SUM(CASE WHEN event_type = 'cart_add' THEN 1 ELSE 0 END) as cart_adds,
                SUM(CASE WHEN event_type = 'purchase' THEN 1 ELSE 0 END) as purchases
            FROM conversion_events
            WHERE timestamp >= ?
            ''', (self._get_start_time(period),))
            
            conversion_data = cursor.fetchone()
            
            # Get top products
            cursor.execute('''
            SELECT product_id, COUNT(*) as count 
            FROM conversion_events 
            WHERE event_type = 'product_view'
            GROUP BY product_id 
            ORDER BY count DESC 
            LIMIT 10
            ''')
            top_products = cursor.fetchall()
            
            # Get intent analysis
            cursor.execute('''
            SELECT intent_category, COUNT(*) as count
            FROM conversation_metrics
            GROUP BY intent_category
            ORDER BY count DESC
            ''')
            intent_analysis = cursor.fetchall()
            
            conn.close()
            
            return {
                "period": period,
                "total_conversations": conversion_data[0],
                "product_views": conversion_data[1],
                "cart_additions": conversion_data[2],
                "purchases": conversion_data[3],
                "conversion_rate": conversion_data[3]/conversion_data[0] if conversion_data[0] > 0 else 0,
                "top_products": [{"product_id": p[0], "views": p[1]} for p in top_products],
                "intent_distribution": [{"intent": i[0], "count": i[1]} for i in intent_analysis]
            }
            
        except Exception as e:
            logger.error(f"Error generating report: {str(e)}")
            return {"error": str(e)}

    def _get_start_time(self, period):
        """Calculate start time for reporting period"""
        now = datetime.datetime.now()
        if period == 'day':
            return (now - datetime.timedelta(days=1)).isoformat()
        elif period == 'week':
            return (now - datetime.timedelta(weeks=1)).isoformat()
        elif period == 'month':
            return (now - datetime.timedelta(days=30)).isoformat()
        return "1970-01-01"

    def anonymize_data(self):
        """Anonymize PII data for compliance"""
        try:
            conn = sqlite3.connect(DB_PATH)
            cursor = conn.cursor()
            
            # Anonymize user IDs and IP addresses
            cursor.execute('''
            UPDATE conversations 
            SET user_id = 'anon_' || substr(hex(randomblob(16)), 1, 32),
                metadata = json_set(metadata, '$.ip_address', 'redacted')
            ''')
            
            conn.commit()
            conn.close()
            return True
            
        except Exception as e:
            logger.error(f"Error anonymizing data: {str(e)}")
            return False

# Initialize database when module is imported
EcommerceTracker().init_database()