learning.py

# core/learning.py
import re
import json
import threading
from datetime import datetime
from memory import ConversationMemory
import sqlite3


class LearningEngine:
    """
    Self-learning engine that continuously observes user conversations,
    extracts communication patterns, interests, and preferences,
    and consolidates them into an adaptive user model.

    The more a user interacts, the better Invicta understands them.
    """

    def __init__(self, memory: ConversationMemory):
        self.memory = memory
        self._lock = threading.Lock()

    # ═══════════════════════════════════════════════════════════════════════
    # REAL-TIME MESSAGE ANALYSIS
    # ═══════════════════════════════════════════════════════════════════════

    def analyze_message(self, user_id, message):
        if not message or not user_id:
            return
        try:
            self.memory.increment_interaction_count(user_id)
            self._detect_interests(user_id, message)
            self._detect_communication_style(user_id, message)
            self._detect_emotional_state(user_id, message)
            self._detect_preferences(user_id, message)
            self._detect_language(user_id, message)  # ← ADD THIS
        except Exception as e:
            print(f"⚠️ Learning analysis error: {e}")

    def _detect_interests(self, user_id, message):
        """Detect topics the user is interested in based on what they ask about."""
        interest_patterns = [
            # Technology
            (r'\b(?:python|javascript|typescript|rust|golang|programming|coding|developer|software|algorithm|api|database|web\s*dev|machine\s*learning|deep\s*learning|ai|artificial\s*intelligence|neural\s*network|data\s*science|cybersecurity|blockchain|crypto|cloud\s*computing|devops)\b', 'technology'),
            # Science
            (r'\b(?:physics|chemistry|biology|quantum|astronomy|space|nasa|evolution|genetics|neuroscience|psychology|math|calculus|statistics|research)\b', 'science'),
            # Business & Finance
            (r'\b(?:startup|business|entrepreneur|invest|stock|market|finance|crypto|trading|marketing|sales|revenue|profit|startup|company|venture)\b', 'business'),
            # Creative
            (r'\b(?:design|art|music|writing|creative|paint|draw|compose|fiction|novel|poetry|photography|filmmaking|video|animation)\b', 'creative'),
            # Health & Fitness
            (r'\b(?:fitness|workout|exercise|diet|nutrition|health|medical|yoga|meditation|mental\s*health|therapy|wellness|gym|running|weight)\b', 'health'),
            # Gaming
            (r'\b(?:game|gaming|gamer|xbox|playstation|nintendo|steam|esports|rpg|fps|mmorpg|minecraft|fortnite)\b', 'gaming'),
            # Travel
            (r'\b(?:travel|trip|vacation|flight|hotel|country|city|backpack|adventure|explore|destination|tourist|visa)\b', 'travel'),
            # Food & Cooking
            (r'\b(?:recipe|cooking|food|cuisine|bake|chef|restaurant|meal|ingredient|kitchen|dish)\b', 'food'),
            # Education
            (r'\b(?:study|university|college|school|learn|course|degree|exam|academic|education|teach|student|homework|tutorial)\b', 'education'),
            # Philosophy & Religion
            (r'\b(?:philosophy|existential|meaning\s*of\s*life|moral|ethics|religion|spiritual|meditation|mindfulness|consciousness)\b', 'philosophy'),
        ]

        msg_lower = message.lower()
        for pattern, category in interest_patterns:
            matches = re.findall(pattern, msg_lower)
            if matches:
                # Store each matched keyword as an interest observation
                for match in matches:
                    keyword = match.strip().lower()
                    self.memory.store_learning(
                        user_id=user_id,
                        category="interest",
                        key=keyword,
                        value=f"Interested in {category}: {keyword}",
                        confidence=0.4
                    )

    def _detect_language(self, user_id, message):
        """Detect if user is writing in a non-English language and store preference."""
        # Simple detection for common scripts
        import re
        if re.search(r'[\u0900-\u097F]', message):  # Devanagari (Hindi)
            self.memory.store_learning(user_id, "preference", "language", "hindi", 0.6)
        elif re.search(r'[\u0600-\u06FF]', message):  # Arabic
            self.memory.store_learning(user_id, "preference", "language", "arabic", 0.6)
        elif re.search(r'[\u3040-\u30FF]', message):  # Japanese
            self.memory.store_learning(user_id, "preference", "language", "japanese", 0.6)
        elif re.search(r'[\u4E00-\u9FFF]', message):  # Chinese
            self.memory.store_learning(user_id, "preference", "language", "chinese", 0.6)
        elif re.search(r'[\uAC00-\uD7AF]', message):  # Korean
            self.memory.store_learning(user_id, "preference", "language", "korean", 0.6)
        elif re.search(r'[áéíóúñ¿¡]', message, re.IGNORECASE):  # Spanish
            self.memory.store_learning(user_id, "preference", "language", "spanish", 0.3)

    def _detect_communication_style(self, user_id, message):
        """Detect how the user communicates — formality, length, tone."""
        # ── FIX: msg_lower was missing here, causing the crash ──
        msg_lower = message.lower()
        words = message.split()
        word_count = len(words)

        # ── Formality detection ──
        formal_indicators = sum(1 for w in words if w.lower() in {
            "therefore", "however", "furthermore", "consequently", "nevertheless",
            "accordingly", "moreover", "henceforth", "would", "shall", "may",
            "perhaps", "kindly", "regards", "sincerely", "please", "respectfully",
        })
        casual_indicators = sum(1 for w in words if w.lower() in {
            "lol", "haha", "yeah", "nah", "yep", "nope", "gonna", "wanna",
            "gotta", "dunno", "kinda", "sorta", "btw", "omg", "wtf", "lmao",
            "bruh", "dude", "hey", "yo", "sup", "ngl", "fr", "tbh", "imo",
        })
        contractions = sum(1 for w in words if re.match(r"\w+'\w+", w))

        if formal_indicators > 2 or (formal_indicators > 0 and casual_indicators == 0 and contractions == 0):
            self.memory.store_learning(user_id, "style", "formality_signal", "formal", 0.3)
        elif casual_indicators > 0 or contractions > 2:
            self.memory.store_learning(user_id, "style", "formality_signal", "casual", 0.3)

        # ── Response length preference ──
        if word_count <= 5:
            self.memory.store_learning(user_id, "style", "length_signal", "short", 0.25)
        elif word_count <= 20:
            self.memory.store_learning(user_id, "style", "length_signal", "medium", 0.2)
        else:
            self.memory.store_learning(user_id, "style", "length_signal", "detailed", 0.25)

        # ── Tone detection ──
        # Question-heavy = curious
        questions = message.count('?')
        exclamations = message.count('!')
        emojis = len(re.findall(r'[🔥💡❤️👍😊😂🤔👏🙌✨🎯💪🚀🧠⭐🌟💫]', message))

        if questions >= 2:
            self.memory.store_learning(user_id, "style", "tone_signal", "curious", 0.3)
        if exclamations >= 2 or emojis >= 2:
            self.memory.store_learning(user_id, "style", "tone_signal", "enthusiastic", 0.3)
        if any(phrase in msg_lower for phrase in ['help me', 'can you', 'please', 'how do i', 'how to']):
            self.memory.store_learning(user_id, "style", "tone_signal", "practical", 0.25)

        # Detect humor/wit
        humor_words = ['joke', 'funny', 'lol', 'haha', 'lmao', 'pun', 'humor']
        if any(w in msg_lower for w in humor_words):
            self.memory.store_learning(user_id, "style", "tone_signal", "witty", 0.35)

    def _detect_emotional_state(self, user_id, message):
        """Detect the user's emotional state from their messages."""
        msg_lower = message.lower()

        # Stress/frustration
        stress_words = ['stressed', 'frustrated', 'overwhelmed', 'anxious', 'worried',
                        'tired', 'exhausted', 'burnt out', 'burnout', 'struggling',
                        'can\'t handle', 'too much', 'losing hope']
        if any(w in msg_lower for w in stress_words):
            self.memory.store_learning(user_id, "emotion", "recent_state", "stressed", 0.5)
            self.memory.store_learning(user_id, "emotion", "stress_prone", "yes", 0.2)

        # Happiness/excitement
        happy_words = ['excited', 'happy', 'great', 'awesome', 'amazing', 'wonderful',
                       'fantastic', 'love it', 'so good', 'thrilled', 'pumped', 'stoked']
        if any(w in msg_lower for w in happy_words):
            self.memory.store_learning(user_id, "emotion", "recent_state", "happy", 0.5)

        # Sadness
        sad_words = ['sad', 'depressed', 'lonely', 'miss', 'lost', 'heartbroken',
                     'crying', 'hurt', 'pain', 'grief', 'mourn']
        if any(w in msg_lower for w in sad_words):
            self.memory.store_learning(user_id, "emotion", "recent_state", "sad", 0.5)

        # Curiosity
        curious_words = ['wonder', 'curious', 'interesting', 'fascinating', 'tell me more',
                         'how come', 'why is', 'what if', 'explain']
        if any(w in msg_lower for w in curious_words):
            self.memory.store_learning(user_id, "emotion", "recent_state", "curious", 0.4)

    def _detect_preferences(self, user_id, message):
        """Detect explicit and implicit preferences."""
        msg_lower = message.lower()

        # Explicit preferences
        pref_patterns = [
            (r"i (?:prefer|like|want|need) (?:my (?:answers?|responses?) )?(?:to be )?(short|brief|concise)", "short"),
            (r"i (?:prefer|like|want|need) (?:my (?:answers?|responses?) )?(?:to be )?(long|detailed|thorough|in.depth|comprehensive)", "detailed"),
            (r"i (?:prefer|like|want|need) (?:my (?:answers?|responses?) )?(?:to be )?(simple|easy|basic|beginner)", "simple"),
            (r"i (?:prefer|like|want|need) (?:my (?:answers?|responses?) )?(?:to be )?(technical|advanced|complex)", "technical"),
            (r"(?:don't|do not) (?:use|give me) (?:code|programming)", "no_code"),
            (r"(?:give|show) me (?:the )?code", "wants_code"),
            (r"(?:give|show) me (?:an )?example", "wants_examples"),
            (r"(?:step.by.step|step by step)", "step_by_step"),
        ]

        for pattern, pref in pref_patterns:
            if re.search(pattern, msg_lower):
                self.memory.store_learning(user_id, "preference", pref, "true", 0.6)

    # ═══════════════════════════════════════════════════════════════════════
    # CONSOLIDATION (runs periodically)
    # ═══════════════════════════════════════════════════════════════════════

    def run_daily_consolidation(self):
        """
        Consolidate raw learning observations into user insights.
        Called periodically by the background thread.
        """
        try:
            # Get all users who have learning data
            all_users = self._get_active_users()
            for user_id in all_users:
                try:
                    self._consolidate_user(user_id)
                except Exception as e:
                    print(f"⚠️ Consolidation error for user {user_id}: {e}")
        except Exception as e:
            print(f"⚠️ Daily consolidation error: {e}")

    def _get_active_users(self):
        """Get list of user IDs that have learning data."""
        db_path = self.memory.db_path
        conn = sqlite3.connect(db_path)
        conn.row_factory = sqlite3.Row
        cur = conn.cursor()
        cur.execute("SELECT DISTINCT user_id FROM user_learning")
        rows = cur.fetchall()
        conn.close()
        return [row["user_id"] for row in rows]

    def _consolidate_user(self, user_id):
        """Consolidate all learning data for one user into insights."""
        insights = self.memory.get_user_insights(user_id) or {}

        # ── Formality ──
        style_data = self.memory.get_learning_by_category(user_id, "style")
        formality_signals = []
        for key, val in style_data.items():
            if key == "formality_signal":
                formality_signals.append(val)

        if formality_signals:
            # Most frequent signal wins
            from collections import Counter
            most_common = Counter(formality_signals).most_common(1)[0][0]
            insights["preferred_formality"] = most_common

        # ── Response length ──
        length_signals = [v for k, v in style_data.items() if k == "length_signal"]
        if length_signals:
            from collections import Counter
            most_common = Counter(length_signals).most_common(1)[0][0]
            insights["preferred_response_length"] = most_common

        # ── Tone ──
        tone_signals = [v for k, v in style_data.items() if k == "tone_signal"]
        if tone_signals:
            from collections import Counter
            most_common = Counter(tone_signals).most_common(1)[0][0]
            insights["preferred_tone"] = most_common

        # ── Topics of interest ──
        top_topics = self.memory.get_top_learning_topics(user_id, limit=15)
        interest_list = [t["key"] for t in top_topics if t.get("times_observed", 0) >= 2]
        if interest_list:
            insights["topics_of_interest"] = interest_list[:10]

        # ── Communication patterns ──
        patterns = {}
        pref_data = self.memory.get_learning_by_category(user_id, "preference")
        for k, v in pref_data.items():
            patterns[k] = v
        if patterns:
            insights["communication_patterns"] = patterns

        # Save consolidated insights
        self.memory.update_user_insights(user_id, insights)

        # Prune old learning data
        self.memory.prune_old_learning(user_id, max_entries=500)

        print(f"🧠 Consolidated learning for user {user_id}")

    # ═══════════════════════════════════════════════════════════════════════
    # USER MODEL (used by Brain to adapt responses)
    # ═══════════════════════════════════════════════════════════════════════

    def get_user_model(self, user_id):
        """
        Build the current user model from consolidated insights
        + real-time learning data. Used by Brain._build_adaptation_prompt().
        """
        if not user_id:
            return None

        insights = self.memory.get_user_insights(user_id)
        model = {}

        # Formality
        model["formality"] = insights.get("preferred_formality", "casual") if insights else "casual"

        # Response length
        model["response_length"] = insights.get("preferred_response_length", "medium") if insights else "medium"

        # Tone
        model["tone"] = insights.get("preferred_tone", "friendly") if insights else "friendly"

        # Interests
        model["interests"] = insights.get("topics_of_interest", []) if insights else []

        # Preferences
        if insights and insights.get("communication_patterns"):
            model["preferences"] = insights["communication_patterns"]
        else:
            model["preferences"] = {}

        # Emotional state (from real-time learning)
        emotion_data = self.memory.get_learning_by_category(user_id, "emotion")
        recent_state = emotion_data.get("recent_state", "")
        if recent_state:
            model["recent_emotional_state"] = recent_state

        # Interaction count
        if insights:
            model["interaction_count"] = insights.get("total_interactions", 0)
        else:
            model["interaction_count"] = 0

        return model