Trixy7/trixy_core/conversation/context.py

# -*- coding: utf-8 -*-
"""
Conversation Context - Kontext-Verwaltung für Sessions.
"""

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any


class ContextScope(Enum):
    """Gültigkeitsbereich von Kontext-Variablen."""
    TURN = "turn"           # Nur für aktuellen Turn
    SESSION = "session"     # Für gesamte Session
    USER = "user"           # Persistent für User
    GLOBAL = "global"       # Global über alle Sessions


@dataclass
class ContextVariable:
    """Eine Kontext-Variable."""

    name: str
    value: Any
    scope: ContextScope = ContextScope.SESSION
    created_at: datetime = field(default_factory=datetime.now)
    updated_at: datetime = field(default_factory=datetime.now)
    metadata: dict[str, Any] = field(default_factory=dict)

    def update(self, value: Any) -> None:
        """Aktualisiert den Wert."""
        self.value = value
        self.updated_at = datetime.now()

    def to_dict(self) -> dict:
        """Konvertiert zu Dictionary."""
        return {
            "name": self.name,
            "value": self.value,
            "scope": self.scope.value,
            "created_at": self.created_at.isoformat(),
            "updated_at": self.updated_at.isoformat(),
        }


class ConversationContext:
    """
    Kontext-Container für Conversation-Sessions.

    Verwaltet Variablen verschiedener Gültigkeitsbereiche:
    - TURN: Nur für aktuellen Gesprächsbeitrag
    - SESSION: Für gesamte Conversation-Session
    - USER: Persistent für einen User (über Sessions hinweg)
    - GLOBAL: Global verfügbar
    """

    def __init__(self):
        """Initialisiert den Kontext."""
        self._variables: dict[str, ContextVariable] = {}
        self._turn_variables: dict[str, ContextVariable] = {}

    def set(
        self,
        name: str,
        value: Any,
        scope: ContextScope = ContextScope.SESSION,
        metadata: dict | None = None,
    ) -> None:
        """
        Setzt eine Kontext-Variable.

        Args:
            name: Variablen-Name
            value: Wert
            scope: Gültigkeitsbereich
            metadata: Zusätzliche Metadaten
        """
        var = ContextVariable(
            name=name,
            value=value,
            scope=scope,
            metadata=metadata or {},
        )

        if scope == ContextScope.TURN:
            self._turn_variables[name] = var
        else:
            self._variables[name] = var

    def get(self, name: str, default: Any = None) -> Any:
        """
        Gibt Kontext-Variable zurück.

        Args:
            name: Variablen-Name
            default: Standard-Wert wenn nicht gefunden

        Returns:
            Wert oder Default
        """
        # Erst Turn-Variables prüfen
        if name in self._turn_variables:
            return self._turn_variables[name].value

        if name in self._variables:
            return self._variables[name].value

        return default

    def get_variable(self, name: str) -> ContextVariable | None:
        """
        Gibt Kontext-Variable mit Metadaten zurück.

        Args:
            name: Variablen-Name

        Returns:
            Variable oder None
        """
        if name in self._turn_variables:
            return self._turn_variables[name]

        return self._variables.get(name)

    def has(self, name: str) -> bool:
        """
        Prüft ob Variable existiert.

        Args:
            name: Variablen-Name

        Returns:
            True wenn vorhanden
        """
        return name in self._turn_variables or name in self._variables

    def remove(self, name: str) -> bool:
        """
        Entfernt Variable.

        Args:
            name: Variablen-Name

        Returns:
            True wenn entfernt
        """
        if name in self._turn_variables:
            del self._turn_variables[name]
            return True

        if name in self._variables:
            del self._variables[name]
            return True

        return False

    def clear_turn(self) -> None:
        """Löscht alle Turn-Variablen."""
        self._turn_variables.clear()

    def clear_scope(self, scope: ContextScope) -> None:
        """
        Löscht alle Variablen eines Scopes.

        Args:
            scope: Zu löschender Scope
        """
        if scope == ContextScope.TURN:
            self._turn_variables.clear()
        else:
            to_remove = [
                name for name, var in self._variables.items()
                if var.scope == scope
            ]
            for name in to_remove:
                del self._variables[name]

    def clear(self) -> None:
        """Löscht alle Variablen."""
        self._variables.clear()
        self._turn_variables.clear()

    def get_by_scope(self, scope: ContextScope) -> dict[str, Any]:
        """
        Gibt alle Variablen eines Scopes zurück.

        Args:
            scope: Scope

        Returns:
            Dictionary mit Variablen
        """
        if scope == ContextScope.TURN:
            return {name: var.value for name, var in self._turn_variables.items()}

        return {
            name: var.value
            for name, var in self._variables.items()
            if var.scope == scope
        }

    def get_all(self) -> dict[str, Any]:
        """
        Gibt alle Variablen zurück.

        Returns:
            Dictionary mit allen Variablen
        """
        result = {}

        # Session/User/Global zuerst
        for name, var in self._variables.items():
            result[name] = var.value

        # Turn überschreibt
        for name, var in self._turn_variables.items():
            result[name] = var.value

        return result

    def merge(self, other: "ConversationContext") -> None:
        """
        Merged anderen Kontext in diesen.

        Args:
            other: Anderer Kontext
        """
        for name, var in other._variables.items():
            self._variables[name] = var

        for name, var in other._turn_variables.items():
            self._turn_variables[name] = var

    def to_dict(self) -> dict:
        """Konvertiert zu Dictionary."""
        return {
            "variables": {
                name: var.to_dict()
                for name, var in self._variables.items()
            },
            "turn_variables": {
                name: var.to_dict()
                for name, var in self._turn_variables.items()
            },
        }

    # === Spezielle Kontext-Methoden für Conversations ===

    def set_intent(self, intent: str, confidence: float = 1.0) -> None:
        """Setzt erkannten Intent."""
        self.set("_intent", intent, ContextScope.TURN)
        self.set("_intent_confidence", confidence, ContextScope.TURN)

    def get_intent(self) -> str | None:
        """Gibt aktuellen Intent zurück."""
        return self.get("_intent")

    def set_entities(self, entities: dict[str, Any]) -> None:
        """Setzt extrahierte Entities."""
        self.set("_entities", entities, ContextScope.TURN)

    def get_entities(self) -> dict[str, Any]:
        """Gibt Entities zurück."""
        return self.get("_entities", {})

    def set_slot(self, name: str, value: Any) -> None:
        """Setzt Slot-Wert (für Dialoge)."""
        self.set(f"slot_{name}", value, ContextScope.SESSION)

    def get_slot(self, name: str, default: Any = None) -> Any:
        """Gibt Slot-Wert zurück."""
        return self.get(f"slot_{name}", default)

    def clear_slots(self) -> None:
        """Löscht alle Slots."""
        to_remove = [
            name for name in self._variables
            if name.startswith("slot_")
        ]
        for name in to_remove:
            del self._variables[name]

    @property
    def slots(self) -> dict[str, Any]:
        """Gibt alle Slots zurück."""
        return {
            name[5:]: var.value  # Entferne "slot_" Prefix
            for name, var in self._variables.items()
            if name.startswith("slot_")
        }