backend/app/services/patterns.py

"""
Project Jarvis — Autonomous Pattern Engine (ML Core)
Discovers user habits and routines by analyzing interaction history.
"""
import logging
from typing import List, Dict, Any, Optional
from datetime import datetime
from sqlalchemy import func
from app.models.entities import InteractionLog
from app.core.database import SessionLocal

logger = logging.getLogger("friday.patterns")

def discover_routines() -> List[Dict[str, Any]]:
    """
    Analyzes historical InteractionLogs to identify recurring patterns.
    Uses simple frequency-time clustering as a baseline for ML discovery.
    """
    db = SessionLocal()
    try:
        # 1. Fetch all logs
        logs = db.query(InteractionLog).all()
        if not logs:
            return []

        # 2. Daily Pulse Analysis: Identify frequent intents by hour
        # (Using a simple dict-based clustering for speed on local system)
        intent_hour_map = {}
        for log in logs:
            hour = log.timestamp.hour
            key = (hour, log.intent)
            intent_hour_map[key] = intent_hour_map.get(key, 0) + 1

        # 3. Filter for significant patterns (Intent occurs > 2 times in the same hour window)
        routines = []
        for (hour, intent), count in intent_hour_map.items():
            if count >= 2:
                # Basic Routine Mapping
                time_str = f"{hour:02d}:00"
                label = f"Recurring {intent.capitalize()}"
                
                # Contextual Enhancements
                if intent == "automation":
                    label = "System Maintenance Protocol"
                elif intent == "search":
                    label = "Deep Intelligence Routine"
                elif intent == "expense":
                    label = "Sovereign Audit"
                
                routines.append({
                    "time": time_str,
                    "intent": intent,
                    "label": label,
                    "confidence": min(count / 5.0, 1.0) # Scaled confidence
                })

        # Sort by time
        routines.sort(key=lambda x: x["time"])
        logger.info(f"PatternEngine: Discovered {len(routines)} autonomous routines.")
        return routines
    except Exception as e:
        logger.error(f"Pattern discovery failed: {e}")
        return []
    finally:
        db.close()

def get_predicted_action(session_id: str) -> Optional[Dict[str, Any]]:
    """
    Predicts the user's next likely action based on the current hour.
    """
    now_hour = datetime.now().hour
    routines = discover_routines()
    
    for r in routines:
        # If routine is in the same hour or next hour
        r_hour = int(r["time"].split(":")[0])
        if r_hour == now_hour:
            return r
    return None