core/session.py

"""
Kariana Unified UI - Session Management
Manages user sessions and persists state
"""
import json
import logging
import os
import tempfile
from dataclasses import dataclass, field, asdict
from datetime import datetime
from typing import Dict, List, Optional
from pathlib import Path

logger = logging.getLogger(__name__)


@dataclass
class Session:
    """User session data"""
    session_id: str
    created_at: str
    last_activity: str
    connected_instance_port: Optional[int] = None
    connected_instance_host: Optional[str] = None
    command_history: List[Dict] = field(default_factory=list)
    workflows: List[Dict] = field(default_factory=list)
    preferences: Dict = field(default_factory=dict)

    def to_dict(self) -> dict:
        return asdict(self)

    @classmethod
    def from_dict(cls, data: dict) -> "Session":
        return cls(**data)


class SessionManager:
    """Manages session persistence and state"""

    def __init__(self, storage_dir: str = None):
        if storage_dir is None:
            storage_dir = os.path.join(tempfile.gettempdir(), "kariana_ui_sessions")
        self.storage_dir = Path(storage_dir)
        self.storage_dir.mkdir(parents=True, exist_ok=True)
        self._current_session: Optional[Session] = None

    def _generate_session_id(self) -> str:
        """Generate unique session ID"""
        import uuid
        return str(uuid.uuid4())[:8]

    def _get_session_path(self, session_id: str) -> Path:
        """Get path to session file"""
        return self.storage_dir / f"session_{session_id}.json"

    def create_session(self) -> Session:
        """Create a new session"""
        now = datetime.utcnow().isoformat()
        session = Session(
            session_id=self._generate_session_id(),
            created_at=now,
            last_activity=now,
        )
        self._current_session = session
        self._save_session(session)
        logger.info(f"Created session {session.session_id}")
        return session

    def get_or_create_session(self) -> Session:
        """Get current session or create new one"""
        if self._current_session is None:
            # Try to load most recent session
            sessions = self._list_sessions()
            if sessions:
                self._current_session = self._load_session(sessions[-1])
            else:
                self._current_session = self.create_session()
        return self._current_session

    def _save_session(self, session: Session):
        """Save session to disk"""
        try:
            path = self._get_session_path(session.session_id)
            with open(path, 'w') as f:
                json.dump(session.to_dict(), f, indent=2)
        except Exception as e:
            logger.error(f"Failed to save session: {e}")

    def _load_session(self, session_id: str) -> Optional[Session]:
        """Load session from disk"""
        try:
            path = self._get_session_path(session_id)
            if path.exists():
                with open(path, 'r') as f:
                    data = json.load(f)
                return Session.from_dict(data)
        except Exception as e:
            logger.error(f"Failed to load session {session_id}: {e}")
        return None

    def _list_sessions(self) -> List[str]:
        """List all session IDs"""
        sessions = []
        for path in self.storage_dir.glob("session_*.json"):
            session_id = path.stem.replace("session_", "")
            sessions.append(session_id)
        return sorted(sessions)

    def update_activity(self):
        """Update last activity timestamp"""
        if self._current_session:
            self._current_session.last_activity = datetime.utcnow().isoformat()
            self._save_session(self._current_session)

    def set_connected_instance(self, host: str, port: int):
        """Record connected instance"""
        session = self.get_or_create_session()
        session.connected_instance_host = host
        session.connected_instance_port = port
        self._save_session(session)

    def add_command_to_history(self, command: Dict, response: Dict = None):
        """Add command to history"""
        session = self.get_or_create_session()
        entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "command": command,
            "response": response,
        }
        session.command_history.append(entry)
        # Keep last 100 commands
        session.command_history = session.command_history[-100:]
        self._save_session(session)

    def get_command_history(self, limit: int = 50) -> List[Dict]:
        """Get recent command history"""
        session = self.get_or_create_session()
        return session.command_history[-limit:]

    def save_workflow(self, workflow: Dict):
        """Save a workflow"""
        session = self.get_or_create_session()
        # Update if exists, otherwise append
        for i, w in enumerate(session.workflows):
            if w.get("name") == workflow.get("name"):
                session.workflows[i] = workflow
                self._save_session(session)
                return
        session.workflows.append(workflow)
        self._save_session(session)

    def get_workflows(self) -> List[Dict]:
        """Get saved workflows"""
        session = self.get_or_create_session()
        return session.workflows

    def delete_workflow(self, name: str) -> bool:
        """Delete a workflow by name"""
        session = self.get_or_create_session()
        for i, w in enumerate(session.workflows):
            if w.get("name") == name:
                session.workflows.pop(i)
                self._save_session(session)
                return True
        return False

    def set_preference(self, key: str, value):
        """Set a preference"""
        session = self.get_or_create_session()
        session.preferences[key] = value
        self._save_session(session)

    def get_preference(self, key: str, default=None):
        """Get a preference"""
        session = self.get_or_create_session()
        return session.preferences.get(key, default)


# Global session manager
_session_manager: Optional[SessionManager] = None


def get_session_manager() -> SessionManager:
    """Get or create global session manager"""
    global _session_manager
    if _session_manager is None:
        _session_manager = SessionManager()
    return _session_manager