memory.py

"""
memory.py — Gestion de la mémoire conversationnelle
Implémente le stockage et la récupération du contexte de conversation.
"""

import json
import os
from datetime import datetime
from typing import List, Dict, Optional
from config import MAX_HISTORY_TURNS, BASE_JSON_FILE


class ConversationMemory:
    """
    Gère l'historique de conversation et la persistance des échanges.
    
    Cette classe maintient :
    - L'historique en cours de session (messages récents)
    - La persistance sur disque (apprentissage cumulatif)
    - Un résumé du contexte pour limiter la taille des prompts
    """

    def __init__(self, max_turns: int = MAX_HISTORY_TURNS):
        """
        Initialise la mémoire de conversation.
        
        Args:
            max_turns: Nombre maximum de tours (paires user/assistant) conservés
        """
        self.max_turns = max_turns
        self.history: List[Dict] = []          # Historique de la session courante
        self.metadata: Dict = {                # Métadonnées de la conversation
            "session_start": datetime.now().isoformat(),
            "message_count": 0,
            "detected_domains": [],
        }

    def add_user_message(self, content: str) -> None:
        """Ajoute un message utilisateur à l'historique."""
        self.history.append({
            "role": "user",
            "content": content,
            "timestamp": datetime.now().isoformat()
        })
        self.metadata["message_count"] += 1
        self._trim_history()

    def add_assistant_message(self, content: str) -> None:
        """Ajoute un message assistant à l'historique."""
        self.history.append({
            "role": "assistant",
            "content": content,
            "timestamp": datetime.now().isoformat()
        })
        self._trim_history()

    def get_recent_history(self, n_turns: int = 6) -> List[Dict]:
        """
        Retourne les N derniers tours de conversation (sans timestamps).
        
        Args:
            n_turns: Nombre de tours récents à retourner
        
        Returns:
            Liste de dicts {'role': str, 'content': str}
        """
        # Convertir les turns en messages (1 tour = 1 user + 1 assistant)
        recent = self.history[-(n_turns * 2):]
        return [{"role": m["role"], "content": m["content"]} for m in recent]

    def get_gradio_format(self) -> List[Dict]:
        """
        Retourne l'historique au format Gradio Chatbot (type="messages").
        
        Returns:
            Liste de dicts {'role': 'user'|'assistant', 'content': str}
        """
        return [{"role": m["role"], "content": m["content"]} for m in self.history]

    def set_domain(self, domain: str) -> None:
        """Enregistre le domaine détecté pour la session."""
        if domain and domain not in self.metadata["detected_domains"]:
            self.metadata["detected_domains"].append(domain)

    def clear(self) -> None:
        """Remet l'historique à zéro (nouvelle conversation)."""
        self.history = []
        self.metadata["message_count"] = 0
        self.metadata["detected_domains"] = []
        self.metadata["session_start"] = datetime.now().isoformat()

    def _trim_history(self) -> None:
        """Tronque l'historique si il dépasse la limite."""
        max_messages = self.max_turns * 2  # 2 messages par tour (user + assistant)
        if len(self.history) > max_messages:
            self.history = self.history[-max_messages:]

    def get_summary(self) -> str:
        """Génère un résumé court de la conversation pour le contexte."""
        if not self.history:
            return ""
        
        topics = set()
        for msg in self.history:
            # Extraction naïve des sujets (peut être amélioré avec NLP)
            words = msg["content"].lower().split()
            for word in words:
                if len(word) > 5:
                    topics.add(word)
        
        return f"[Sujets abordés : {', '.join(list(topics)[:5])}]"


# ─────────────────────────────────────────────
# GESTION DE LA BASE DE CONNAISSANCES PERSISTANTE
# ─────────────────────────────────────────────

class KnowledgeBase:
    """
    Gère la base de connaissances statique et apprise dynamiquement.
    
    Deux sources :
    - base_connaissances.txt  : entrées manuelles (question/réponse par ligne)
    - base_connaissances_auto.json : entrées apprises automatiquement
    """

    def __init__(self, txt_path: str = "./data/base_connaissances.txt",
                 json_path: str = BASE_JSON_FILE):
        self.txt_path = txt_path
        self.json_path = json_path
        self._ensure_data_dir()
        self._ensure_json_exists()

    def _ensure_data_dir(self):
        """Crée le répertoire data/ si absent."""
        os.makedirs(os.path.dirname(self.txt_path), exist_ok=True)
        os.makedirs(os.path.dirname(self.json_path), exist_ok=True)

    def _ensure_json_exists(self):
        """Crée le fichier JSON vide si absent."""
        if not os.path.exists(self.json_path):
            with open(self.json_path, "w", encoding="utf-8") as f:
                json.dump([], f, ensure_ascii=False, indent=2)

    def load_txt(self) -> Dict[str, str]:
        """
        Charge la base TXT au format question/réponse alternés.
        
        Format attendu :
            Ligne 1: question
            Ligne 2: réponse
            Ligne 3: question
            ...
        
        Returns:
            Dict {question: réponse}
        """
        if not os.path.exists(self.txt_path):
            return {}
        
        base = {}
        try:
            with open(self.txt_path, "r", encoding="utf-8") as f:
                lines = [l.strip() for l in f.readlines() if l.strip()]
            
            for i in range(0, len(lines) - 1, 2):
                question = lines[i]
                reponse = lines[i + 1]
                if question and reponse:
                    base[question] = reponse
        except Exception as e:
            print(f"[KnowledgeBase] Erreur chargement TXT : {e}")
        
        return base

    def load_json(self) -> List[Dict]:
        """
        Charge la base JSON dynamique.
        
        Returns:
            Liste de dicts {"question": str, "réponse": str}
        """
        try:
            with open(self.json_path, "r", encoding="utf-8") as f:
                return json.load(f)
        except Exception as e:
            print(f"[KnowledgeBase] Erreur chargement JSON : {e}")
            return []

    def save_qa(self, question: str, reponse: str) -> bool:
        """
        Enregistre une nouvelle paire Q/R dans la base JSON.
        
        Args:
            question: La question posée
            reponse: La réponse générée
        
        Returns:
            True si succès, False sinon
        """
        try:
            base = self.load_json()
            base.append({
                "question": question,
                "réponse": reponse,
                "timestamp": datetime.now().isoformat()
            })
            with open(self.json_path, "w", encoding="utf-8") as f:
                json.dump(base, f, ensure_ascii=False, indent=2)
            return True
        except Exception as e:
            print(f"[KnowledgeBase] Erreur sauvegarde : {e}")
            return False

    def save_manual_qa(self, question: str, reponse: str) -> str:
        """
        Enregistre manuellement une paire Q/R (depuis l'onglet Apprentissage).
        
        Returns:
            Message de confirmation
        """
        if not question.strip() or not reponse.strip():
            return "❌ Question et réponse ne peuvent pas être vides."
        
        success = self.save_qa(question.strip(), reponse.strip())
        if success:
            return f"✅ Enregistré avec succès : « {question[:50]}... »"
        return "❌ Erreur lors de l'enregistrement."

    def get_all_as_dict(self) -> Dict[str, str]:
        """Retourne toutes les entrées (TXT + JSON) sous forme de dict."""
        base = self.load_txt()
        for item in self.load_json():
            q = item.get("question", "")
            r = item.get("réponse", "")
            if q and r:
                base[q] = r
        return base

    def get_stats(self) -> Dict:
        """Retourne les statistiques de la base de connaissances."""
        txt_count = len(self.load_txt())
        json_count = len(self.load_json())
        return {
            "txt_entries": txt_count,
            "json_entries": json_count,
            "total": txt_count + json_count
        }