memory.py

from collections import Counter
from dataclasses import asdict, dataclass
from datetime import datetime

try:
    from langchain_core.chat_history import InMemoryChatMessageHistory
except Exception:
    InMemoryChatMessageHistory = None


@dataclass
class MemoryEvent:
    iteration: int
    trainer: str
    query: str
    category: str
    priority: str
    timestamp: str


class SimpleMemory:
    """Memoria simple para la simulación PokéSupport Town."""

    def __init__(self):
        self.iteration = 0
        self.conversation_history = []
        self.event_history = []
        self.detected_patterns = []
        self.tickets = []
        self.chat_history = (
            InMemoryChatMessageHistory()
            if InMemoryChatMessageHistory is not None
            else None
        )

    def reset(self):
        self.__init__()

    def add_conversation_turn(self, user_message, assistant_message):
        self.conversation_history.append({
            "user": user_message,
            "assistant": assistant_message,
        })

        if self.chat_history is not None:
            self.chat_history.add_user_message(user_message)
            self.chat_history.add_ai_message(assistant_message)

    def add_event(self, trainer, query, category, priority):
        self.iteration += 1
        event = MemoryEvent(
            iteration=self.iteration,
            trainer=trainer,
            query=query,
            category=category,
            priority=priority,
            timestamp=datetime.now().isoformat(timespec="seconds"),
        )
        self.event_history.append(asdict(event))
        return asdict(event)

    def count_by_category(self):
        return Counter(event["category"] for event in self.event_history)

    def recent_events(self, limit=4):
        return self.event_history[-limit:]

    def recent_conversation(self, limit=6):
        if self.chat_history is None:
            return self.conversation_history[-limit:]

        return [
            {"type": message.type, "content": message.content}
            for message in self.chat_history.messages[-limit:]
        ]

    def detect_pattern(self, category, threshold=3):
        category_count = self.count_by_category()[category]

        if category_count < threshold:
            return None

        existing_pattern = next(
            (
                pattern for pattern in self.detected_patterns
                if pattern["category"] == category
            ),
            None,
        )

        if existing_pattern:
            return existing_pattern

        pattern = {
            "category": category,
            "count": category_count,
            "message": (
                f"Se han detectado {category_count} incidencias de tipo "
                f"'{category}'. Puede ser un problema general de la ciudad."
            ),
        }
        self.detected_patterns.append(pattern)
        return pattern

    def create_ticket(self, trainer, category, query, priority):
        ticket = {
            "id": f"T-{len(self.tickets) + 1:03d}",
            "trainer": trainer,
            "category": category,
            "query": query,
            "priority": priority,
            "status": "abierto",
            "iteration": self.iteration,
        }
        self.tickets.append(ticket)
        return ticket

    def summary(self):
        return {
            "iterations": self.iteration,
            "events": self.event_history,
            "category_counts": dict(self.count_by_category()),
            "patterns": self.detected_patterns,
            "tickets": self.tickets,
        }