user_profile.py

"""
User Profile System for Voice Agent
Implements personalized user preferences, settings, and consent management
Following specifications from rules.mdc
"""

import os
import json
import uuid
import logging
from datetime import datetime
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, asdict, field
from enum import Enum
import sqlite3
from pathlib import Path

from openai import OpenAI


class ProactivityLevel(Enum):
    """Proactivity levels for user control"""
    OFF = 0
    LOW = 1
    NORMAL = 2
    HIGH = 3


@dataclass
class ConversationStyle:
    """User's preferred conversation style"""
    formality: str = "casual"  # casual, formal, friendly
    concise: bool = True       # Brief vs detailed responses
    emoji_use: bool = False    # Use emojis in responses
    humor_level: str = "light" # none, light, moderate, high
    
    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)
    
    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'ConversationStyle':
        return cls(**data)


@dataclass
class DomainConsents:
    """User consent settings for different memory domains"""
    preferences: bool = True      # Food, music, general preferences
    calendar: bool = False        # Schedule and appointments
    health: bool = False          # Health-related information
    finance: bool = False         # Financial information
    work: bool = True             # Work and career related
    personal: bool = True         # Personal experiences and stories
    interests: bool = True        # Hobbies and interests
    location: bool = False        # Location and travel information
    
    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)
    
    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'DomainConsents':
        return cls(**data)
    
    def get_allowed_domains(self) -> List[str]:
        """Get list of domains user has consented to"""
        allowed = []
        for domain, consented in self.to_dict().items():
            if consented:
                allowed.append(domain)
        return allowed


@dataclass
class SafetyOverrides:
    """Safety settings and topic restrictions"""
    avoid_topics: List[str] = field(default_factory=lambda: ["medical", "legal"])
    content_filter_level: str = "normal"  # strict, normal, relaxed
    age_appropriate: bool = True
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "avoid_topics": self.avoid_topics,
            "content_filter_level": self.content_filter_level,
            "age_appropriate": self.age_appropriate
        }
    
    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'SafetyOverrides':
        return cls(
            avoid_topics=data.get("avoid_topics", ["medical", "legal"]),
            content_filter_level=data.get("content_filter_level", "normal"),
            age_appropriate=data.get("age_appropriate", True)
        )


@dataclass
class UserProfile:
    """Complete user profile following rules.mdc schema"""
    user_id: str
    name: str = "User"
    locales: List[str] = field(default_factory=lambda: ["en"])
    timezone: str = "UTC"
    style: ConversationStyle = field(default_factory=ConversationStyle)
    interests: List[str] = field(default_factory=list)
    proactivity_level: ProactivityLevel = ProactivityLevel.NORMAL
    domain_consents: DomainConsents = field(default_factory=DomainConsents)
    safety_overrides: SafetyOverrides = field(default_factory=SafetyOverrides)
    created_at: str = field(default_factory=lambda: datetime.now().isoformat())
    updated_at: str = field(default_factory=lambda: datetime.now().isoformat())
    version: int = 1
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert to dictionary for storage"""
        return {
            "user_id": self.user_id,
            "name": self.name,
            "locales": self.locales,
            "timezone": self.timezone,
            "style": self.style.to_dict(),
            "interests": self.interests,
            "proactivity_level": self.proactivity_level.value,
            "domain_consents": self.domain_consents.to_dict(),
            "safety_overrides": self.safety_overrides.to_dict(),
            "created_at": self.created_at,
            "updated_at": self.updated_at,
            "version": self.version
        }
    
    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'UserProfile':
        """Create from dictionary"""
        return cls(
            user_id=data["user_id"],
            name=data.get("name", "User"),
            locales=data.get("locales", ["en"]),
            timezone=data.get("timezone", "UTC"),
            style=ConversationStyle.from_dict(data.get("style", {})),
            interests=data.get("interests", []),
            proactivity_level=ProactivityLevel(data.get("proactivity_level", 2)),
            domain_consents=DomainConsents.from_dict(data.get("domain_consents", {})),
            safety_overrides=SafetyOverrides.from_dict(data.get("safety_overrides", {})),
            created_at=data.get("created_at", datetime.now().isoformat()),
            updated_at=data.get("updated_at", datetime.now().isoformat()),
            version=data.get("version", 1)
        )
    
    def update_timestamp(self):
        """Update the last modified timestamp"""
        self.updated_at = datetime.now().isoformat()
        self.version += 1


class ProfileExtractor:
    """Extracts profile information from conversations using LLM"""
    
    def __init__(self, openai_client: OpenAI):
        self.client = openai_client
        self.extraction_prompt = """You are a profile extraction system. Analyze the conversation to identify user preferences, interests, and characteristics that should be stored in their profile.

EXTRACTION CRITERIA:
- Extract clear preferences (food, music, activities, etc.)
- Identify interests and hobbies mentioned
- Note communication style preferences
- Detect timezone/location if mentioned
- Identify any safety or consent-related preferences

IMPORTANT: Only extract information that is explicitly stated or clearly implied. Do not make assumptions.

Return ONLY a valid JSON object. If nothing to extract, return: {{"updates": []}}

Format:
{{
  "updates": [
    {{
      "field": "interests",
      "action": "add",
      "value": "specific interest mentioned",
      "confidence": 0.85
    }},
    {{
      "field": "style.formality",
      "action": "set",
      "value": "casual",
      "confidence": 0.90
    }}
  ]
}}

Available fields:
- name, locales, timezone
- interests (array)
- style.formality, style.concise, style.emoji_use, style.humor_level
- proactivity_level (0-3)
- domain_consents.preferences, domain_consents.work, etc.

Conversation context:
{conversation_context}

JSON object:"""

    def extract_profile_updates(self, conversation_context: str) -> List[Dict[str, Any]]:
        """Extract profile updates from conversation"""
        try:
            # Generate extraction prompt
            prompt = self.extraction_prompt.format(
                conversation_context=conversation_context
            )
            
            # Call LLM for extraction
            response = self.client.chat.completions.create(
                model="gpt-4",
                messages=[{"role": "user", "content": prompt}],
                max_tokens=300,
                temperature=0.1
            )
            
            # Parse response
            content = response.choices[0].message.content.strip()
            
            # Clean up response
            if content.startswith('```json'):
                content = content[7:]
            if content.endswith('```'):
                content = content[:-3]
            content = content.strip()
            
            try:
                result = json.loads(content)
                return result.get('updates', [])
            except json.JSONDecodeError:
                # Try to extract JSON from response
                import re
                json_match = re.search(r'\{.*\}', content, re.DOTALL)
                if json_match:
                    try:
                        result = json.loads(json_match.group(0))
                        return result.get('updates', [])
                    except json.JSONDecodeError:
                        logging.warning(f"Failed to parse extracted JSON: {json_match.group(0)}")
                        return []
                else:
                    logging.warning(f"Failed to parse profile extraction: {content[:100]}...")
                    return []
            
        except Exception as e:
            logging.warning(f"Profile extraction failed: {e}")
            return []


class UserProfileManager:
    """Manages user profiles with persistence and updates"""
    
    def __init__(self, db_path: str = "./user_profiles.db"):
        self.db_path = db_path
        self.profile_extractor = None
        self._init_database()
        logging.info(f"UserProfileManager initialized with database: {db_path}")
    
    def _init_database(self):
        """Initialize SQLite database for profile storage"""
        os.makedirs(os.path.dirname(self.db_path) if os.path.dirname(self.db_path) else ".", exist_ok=True)
        
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS user_profiles (
                    user_id TEXT PRIMARY KEY,
                    profile_data TEXT NOT NULL,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    version INTEGER DEFAULT 1
                )
            """)
            
            # Create audit table for tracking changes
            conn.execute("""
                CREATE TABLE IF NOT EXISTS profile_changes (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    user_id TEXT NOT NULL,
                    field_changed TEXT NOT NULL,
                    old_value TEXT,
                    new_value TEXT,
                    changed_at TEXT NOT NULL,
                    change_reason TEXT
                )
            """)
            conn.commit()
    
    def set_profile_extractor(self, openai_client: OpenAI):
        """Set up profile extraction capabilities"""
        self.profile_extractor = ProfileExtractor(openai_client)
    
    def get_profile(self, user_id: str) -> UserProfile:
        """Get user profile, creating default if not exists"""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute(
                "SELECT profile_data FROM user_profiles WHERE user_id = ?",
                (user_id,)
            )
            row = cursor.fetchone()
            
            if row:
                profile_data = json.loads(row[0])
                return UserProfile.from_dict(profile_data)
            else:
                # Create default profile
                profile = UserProfile(user_id=user_id)
                self.save_profile(profile)
                return profile
    
    def save_profile(self, profile: UserProfile, change_reason: str = "manual_update"):
        """Save user profile with audit trail"""
        profile.update_timestamp()
        
        with sqlite3.connect(self.db_path) as conn:
            # Check if profile exists for audit trail
            cursor = conn.execute(
                "SELECT profile_data FROM user_profiles WHERE user_id = ?",
                (profile.user_id,)
            )
            existing_row = cursor.fetchone()
            
            profile_json = json.dumps(profile.to_dict())
            
            if existing_row:
                # Update existing profile
                conn.execute("""
                    UPDATE user_profiles 
                    SET profile_data = ?, updated_at = ?, version = ?
                    WHERE user_id = ?
                """, (profile_json, profile.updated_at, profile.version, profile.user_id))
                
                # Track changes in audit table
                if existing_row[0] != profile_json:
                    conn.execute("""
                        INSERT INTO profile_changes 
                        (user_id, field_changed, old_value, new_value, changed_at, change_reason)
                        VALUES (?, ?, ?, ?, ?, ?)
                    """, (profile.user_id, "full_profile", existing_row[0], 
                          profile_json, profile.updated_at, change_reason))
            else:
                # Insert new profile
                conn.execute("""
                    INSERT INTO user_profiles (user_id, profile_data, created_at, updated_at, version)
                    VALUES (?, ?, ?, ?, ?)
                """, (profile.user_id, profile_json, profile.created_at, 
                      profile.updated_at, profile.version))
                
                # Track creation in audit table
                conn.execute("""
                    INSERT INTO profile_changes 
                    (user_id, field_changed, old_value, new_value, changed_at, change_reason)
                    VALUES (?, ?, ?, ?, ?, ?)
                """, (profile.user_id, "profile_created", None, 
                      profile_json, profile.created_at, "profile_creation"))
            
            conn.commit()
    
    def update_profile_from_conversation(self, user_id: str, conversation_context: str) -> List[str]:
        """Update profile based on conversation analysis"""
        if not self.profile_extractor:
            return []
        
        try:
            # Extract profile updates
            updates = self.profile_extractor.extract_profile_updates(conversation_context)
            
            if not updates:
                return []
            
            # Get current profile
            profile = self.get_profile(user_id)
            changes_made = []
            
            # Apply updates
            for update in updates:
                if update.get('confidence', 0) < 0.7:  # Only apply high-confidence updates
                    continue
                
                field = update.get('field')
                action = update.get('action')
                value = update.get('value')
                
                if self._apply_profile_update(profile, field, action, value):
                    changes_made.append(f"{action} {field}: {value}")
            
            # Save updated profile
            if changes_made:
                self.save_profile(profile, "conversation_analysis")
                logging.info(f"Profile updated for {user_id}: {changes_made}")
            
            return changes_made
            
        except Exception as e:
            logging.error(f"Failed to update profile from conversation: {e}")
            return []
    
    def _apply_profile_update(self, profile: UserProfile, field: str, action: str, value: Any) -> bool:
        """Apply a single profile update"""
        try:
            if field == "name":
                profile.name = value
            elif field == "interests":
                if action == "add" and value not in profile.interests:
                    profile.interests.append(value)
                elif action == "remove" and value in profile.interests:
                    profile.interests.remove(value)
                else:
                    return False
            elif field == "timezone":
                profile.timezone = value
            elif field.startswith("style."):
                style_field = field.split(".", 1)[1]
                if hasattr(profile.style, style_field):
                    setattr(profile.style, style_field, value)
                else:
                    return False
            elif field == "proactivity_level":
                if isinstance(value, int) and 0 <= value <= 3:
                    profile.proactivity_level = ProactivityLevel(value)
                else:
                    return False
            elif field.startswith("domain_consents."):
                domain = field.split(".", 1)[1]
                if hasattr(profile.domain_consents, domain):
                    setattr(profile.domain_consents, domain, bool(value))
                else:
                    return False
            else:
                return False
            
            return True
            
        except Exception as e:
            logging.warning(f"Failed to apply profile update {field}={value}: {e}")
            return False
    
    def get_personalized_context(self, user_id: str) -> str:
        """Get personalized context string for LLM prompts"""
        profile = self.get_profile(user_id)
        
        context_parts = []
        
        # Basic info
        if profile.name != "User":
            context_parts.append(f"User's name: {profile.name}")
        
        # Interests
        if profile.interests:
            context_parts.append(f"User's interests: {', '.join(profile.interests)}")
        
        # Communication style
        style_desc = []
        if profile.style.formality != "casual":
            style_desc.append(f"prefers {profile.style.formality} communication")
        if profile.style.concise:
            style_desc.append("prefers brief responses")
        if profile.style.emoji_use:
            style_desc.append("enjoys emojis")
        if profile.style.humor_level != "light":
            style_desc.append(f"appreciates {profile.style.humor_level} humor")
        
        if style_desc:
            context_parts.append(f"Communication style: {', '.join(style_desc)}")
        
        # Proactivity preference
        context_parts.append(f"Proactivity preference: {profile.proactivity_level.name.lower()}")
        
        # Safety considerations
        if profile.safety_overrides.avoid_topics:
            context_parts.append(f"Avoid topics: {', '.join(profile.safety_overrides.avoid_topics)}")
        
        return "\n".join(context_parts) if context_parts else "No specific user preferences recorded."
    
    def get_profile_stats(self, user_id: str) -> Dict[str, Any]:
        """Get profile statistics and metadata"""
        profile = self.get_profile(user_id)
        
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute(
                "SELECT COUNT(*) FROM profile_changes WHERE user_id = ?",
                (user_id,)
            )
            change_count = cursor.fetchone()[0]
        
        return {
            "user_id": user_id,
            "name": profile.name,
            "created_at": profile.created_at,
            "updated_at": profile.updated_at,
            "version": profile.version,
            "interests_count": len(profile.interests),
            "allowed_domains": profile.domain_consents.get_allowed_domains(),
            "proactivity_level": profile.proactivity_level.name,
            "total_changes": change_count
        }


# Example usage and testing
if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    
    # Test profile system
    manager = UserProfileManager()
    
    # Create test profile
    profile = UserProfile(
        user_id="test_user",
        name="Alice",
        interests=["cooking", "photography"],
        timezone="America/New_York"
    )
    
    # Save and retrieve
    manager.save_profile(profile)
    retrieved = manager.get_profile("test_user")
    
    print(f"Profile created: {retrieved.name}")
    print(f"Interests: {retrieved.interests}")
    print(f"Personalized context: {manager.get_personalized_context('test_user')}")
    print(f"Profile stats: {manager.get_profile_stats('test_user')}")