03_System_Tools/workspace_monitor_service.py

#!/usr/bin/env python3
"""
Luuna Workspace Monitor Service - Automatic document processing and skill extraction
Monitors new workspace folders and processes documents automatically
"""

import os
import json
import time
import threading
from pathlib import Path
from typing import Dict, List, Optional, Any
import logging
from datetime import datetime
import sys

# Add current directory to path for imports
sys.path.append(str(Path(__file__).parent))

from document_processor import DocumentAnalysisPipeline, DocumentConverter

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class WorkspaceMonitorService:
    """Service that monitors workspace for new folders and processes documents"""

    def __init__(self, base_workspace_path: Path):
        self.base_workspace_path = base_workspace_path
        self.monitored_workspaces = {}
        self.running = False
        self.check_interval = 60  # Check every 60 seconds
        self.converter = DocumentConverter()

    def start_monitoring(self):
        """Start the monitoring service"""
        self.running = True
        logger.info("Starting Workspace Monitor Service")

        # Initial scan
        self._scan_for_new_workspaces()

        # Start monitoring thread
        monitor_thread = threading.Thread(target=self._monitoring_loop, daemon=True)
        monitor_thread.start()

        logger.info("Workspace Monitor Service started successfully")

    def stop_monitoring(self):
        """Stop the monitoring service"""
        self.running = False
        logger.info("Workspace Monitor Service stopped")

    def _monitoring_loop(self):
        """Main monitoring loop"""
        while self.running:
            try:
                self._scan_for_new_workspaces()
                self._process_existing_workspaces()
                time.sleep(self.check_interval)
            except Exception as e:
                logger.error(f"Error in monitoring loop: {e}")
                time.sleep(self.check_interval)

    def _scan_for_new_workspaces(self):
        """Scan for new workspace folders"""
        if not self.base_workspace_path.exists():
            return

        for item in self.base_workspace_path.iterdir():
            if item.is_dir() and not item.name.startswith('.') and item.name != 'analysis_results':
                workspace_name = item.name

                if workspace_name not in self.monitored_workspaces:
                    logger.info(f"Found new workspace: {workspace_name}")
                    self._initialize_workspace(item)

    def _initialize_workspace(self, workspace_path: Path):
        """Initialize monitoring for a new workspace"""
        workspace_name = workspace_path.name

        # Create pipeline for this workspace
        pipeline = DocumentAnalysisPipeline(workspace_path)

        self.monitored_workspaces[workspace_name] = {
            'path': workspace_path,
            'pipeline': pipeline,
            'last_processed': 0,
            'status': 'active'
        }

        # Run initial processing
        self._process_workspace(workspace_name)

    def _process_existing_workspaces(self):
        """Process existing workspaces for updates"""
        for workspace_name, workspace_info in self.monitored_workspaces.items():
            if workspace_info['status'] == 'active':
                # Check if there are new files (basic check)
                workspace_path = workspace_info['path']
                if self._has_new_files(workspace_path, workspace_info['last_processed']):
                    logger.info(f"Processing updates for workspace: {workspace_name}")
                    self._process_workspace(workspace_name)

    def _has_new_files(self, workspace_path: Path, last_processed: float) -> bool:
        """Check if workspace has new files since last processing"""
        supported_extensions = {'.pdf', '.csv', '.docx', '.doc', '.txt'}

        for file_path in workspace_path.rglob('*'):
            if (file_path.is_file() and
                file_path.suffix.lower() in supported_extensions and
                file_path.stat().st_mtime > last_processed):
                return True

        return False

    def _process_workspace(self, workspace_name: str):
        """Process a specific workspace"""
        workspace_info = self.monitored_workspaces.get(workspace_name)
        if not workspace_info:
            return

        try:
            pipeline = workspace_info['pipeline']
            results = pipeline.run_full_pipeline()

            # Update last processed time
            workspace_info['last_processed'] = time.time()

            logger.info(f"Processed workspace {workspace_name}: {results}")

            # Log results
            self._log_processing_results(workspace_name, results)

        except Exception as e:
            logger.error(f"Error processing workspace {workspace_name}: {e}")

    def _log_processing_results(self, workspace_name: str, results: Dict):
        """Log processing results"""
        log_file = self.base_workspace_path / "workspace_monitor.log"

        log_entry = {
            'timestamp': datetime.now().isoformat(),
            'workspace': workspace_name,
            'results': results
        }

        try:
            # Read existing log
            existing_logs = []
            if log_file.exists():
                with open(log_file, 'r', encoding='utf-8') as f:
                    for line in f:
                        if line.strip():
                            try:
                                existing_logs.append(json.loads(line))
                            except:
                                continue

            # Add new entry
            existing_logs.append(log_entry)

            # Keep only last 1000 entries
            if len(existing_logs) > 1000:
                existing_logs = existing_logs[-1000:]

            # Write back
            with open(log_file, 'w', encoding='utf-8') as f:
                for entry in existing_logs:
                    f.write(json.dumps(entry, ensure_ascii=False) + '\n')

        except Exception as e:
            logger.error(f"Error logging results: {e}")

class AgentTaskDistributor:
    """Distributes extracted skills and tasks to appropriate agents"""

    def __init__(self, base_workspace_path: Path):
        self.base_workspace_path = base_workspace_path
        self.agent_categories = self._define_agent_categories()

    def _define_agent_categories(self) -> Dict[str, Dict]:
        """Define agent categories and their capabilities"""
        return {
            'crypto_trading_agent': {
                'skills': ['crypto_trading', 'market_analysis', 'trading_strategy'],
                'file_patterns': ['*crypto*', '*trading*', '*market*'],
                'priority': 'high'
            },
            'data_processing_agent': {
                'skills': ['data_analysis', 'csv_processing', 'statistics'],
                'file_patterns': ['*data*', '*csv*', '*analysis*'],
                'priority': 'medium'
            },
            'ai_agent_developer': {
                'skills': ['ai_agent', 'automation', 'prompt_engineering'],
                'file_patterns': ['*agent*', '*ai*', '*automation*'],
                'priority': 'high'
            },
            'business_intelligence_agent': {
                'skills': ['business_intelligence', 'reporting', 'strategy'],
                'file_patterns': ['*business*', '*strategy*', '*intelligence*'],
                'priority': 'medium'
            },
            'financial_analysis_agent': {
                'skills': ['financial_analysis', 'loan_analysis', 'credit_scoring'],
                'file_patterns': ['*finance*', '*loan*', '*credit*', '*bank*'],
                'priority': 'high'
            }
        }

    def distribute_tasks_from_skills(self) -> Dict[str, Any]:
        """Analyze extracted skills and distribute to appropriate agents"""
        results_file = self.base_workspace_path / "analysis_results" / "extracted_skills.json"

        if not results_file.exists():
            logger.warning("No extracted skills found")
            return {'error': 'no_skills_found'}

        # Load extracted skills
        try:
            with open(results_file, 'r', encoding='utf-8') as f:
                skills_data = json.load(f)
        except Exception as e:
            logger.error(f"Error loading skills: {e}")
            return {'error': 'load_failed'}

        # Distribute skills to agents
        agent_assignments = {}
        unassigned_skills = []

        for skill_data in skills_data:
            assigned = False

            for agent_name, agent_config in self.agent_categories.items():
                if self._skill_matches_agent(skill_data, agent_config):
                    if agent_name not in agent_assignments:
                        agent_assignments[agent_name] = {
                            'agent_config': agent_config,
                            'assigned_skills': [],
                            'total_confidence': 0,
                            'skill_count': 0
                        }

                    agent_assignments[agent_name]['assigned_skills'].append(skill_data)
                    agent_assignments[agent_name]['total_confidence'] += skill_data.get('confidence_score', 0)
                    agent_assignments[agent_name]['skill_count'] += 1
                    assigned = True
                    break

            if not assigned:
                unassigned_skills.append(skill_data)

        # Calculate agent priorities and create task assignments
        task_assignments = self._create_task_assignments(agent_assignments, unassigned_skills)

        # Save assignments
        self._save_task_assignments(task_assignments)

        return {
            'agent_assignments': agent_assignments,
            'unassigned_skills': unassigned_skills,
            'task_assignments': task_assignments,
            'total_skills': len(skills_data),
            'assigned_skills': sum(len(agent['assigned_skills']) for agent in agent_assignments.values())
        }

    def _skill_matches_agent(self, skill_data: Dict, agent_config: Dict) -> bool:
        """Check if a skill matches an agent"""
        skill_category = skill_data.get('category', '').lower()
        skill_name = skill_data.get('name', '').lower()
        source_document = skill_data.get('source_document', '').lower()

        # Check skill categories
        for agent_skill in agent_config['skills']:
            if agent_skill.lower() in skill_category:
                return True

        # Check file patterns
        for pattern in agent_config['file_patterns']:
            if pattern.replace('*', '') in source_document:
                return True

        # Check skill name keywords
        for pattern in agent_config['file_patterns']:
            clean_pattern = pattern.replace('*', '')
            if clean_pattern in skill_name:
                return True

        return False

    def _create_task_assignments(self, agent_assignments: Dict, unassigned_skills: List) -> Dict:
        """Create task assignments for agents"""
        task_assignments = {}

        for agent_name, agent_data in agent_assignments.items():
            skills = agent_data['assigned_skills']
            avg_confidence = agent_data['total_confidence'] / len(skills) if skills else 0

            # Create tasks based on skills
            tasks = []
            for skill in skills:
                task = {
                    'task_id': f"task_{skill['skill_id']}",
                    'skill_id': skill['skill_id'],
                    'task_type': self._determine_task_type(skill),
                    'description': f"Implement and test skill: {skill['name']}",
                    'priority': 'high' if avg_confidence > 0.8 else 'medium',
                    'estimated_complexity': skill.get('implementation_complexity', 'medium'),
                    'dependencies': skill.get('dependencies', []),
                    'deliverables': [
                        f"Implement {skill['name']} skill",
                        f"Create test cases for {skill['name']}",
                        f"Document usage of {skill['name']}"
                    ]
                }
                tasks.append(task)

            task_assignments[agent_name] = {
                'agent_name': agent_name,
                'total_skills': len(skills),
                'average_confidence': round(avg_confidence, 2),
                'tasks': tasks,
                'status': 'pending',
                'created_at': datetime.now().isoformat()
            }

        return task_assignments

    def _determine_task_type(self, skill_data: Dict) -> str:
        """Determine task type based on skill"""
        category = skill_data.get('category', '').lower()

        if 'analysis' in category:
            return 'data_analysis'
        elif 'trading' in category:
            return 'trading_automation'
        elif 'ai' in category or 'agent' in category:
            return 'ai_development'
        elif 'business' in category:
            return 'business_intelligence'
        else:
            return 'general_implementation'

    def _save_task_assignments(self, task_assignments: Dict):
        """Save task assignments to file"""
        output_dir = self.base_workspace_path / "task_assignments"
        output_dir.mkdir(exist_ok=True)

        assignments_file = output_dir / "agent_task_assignments.json"

        try:
            with open(assignments_file, 'w', encoding='utf-8') as f:
                json.dump(task_assignments, f, indent=2, ensure_ascii=False)

            logger.info(f"Saved task assignments for {len(task_assignments)} agents")

        except Exception as e:
            logger.error(f"Error saving task assignments: {e}")

class AutoWorkspaceProcessor:
    """Main processor that combines monitoring and task distribution"""

    def __init__(self, base_workspace_path: Path):
        self.base_workspace_path = base_workspace_path
        self.monitor_service = WorkspaceMonitorService(base_workspace_path)
        self.task_distributor = AgentTaskDistributor(base_workspace_path)

    def process_all_workspaces(self) -> Dict[str, Any]:
        """Process all workspaces and distribute tasks"""
        logger.info("Starting auto workspace processing")

        # Start monitoring service briefly to scan
        self.monitor_service.start_monitoring()
        time.sleep(2)  # Give it time to scan
        self.monitor_service.stop_monitoring()

        # Distribute tasks from extracted skills
        task_results = self.task_distributor.distribute_tasks_from_skills()

        # Generate summary report
        summary = self._generate_processing_summary(task_results)

        logger.info("Auto workspace processing completed")
        return summary

    def _generate_processing_summary(self, task_results: Dict) -> Dict:
        """Generate comprehensive processing summary"""
        summary = {
            'processing_timestamp': datetime.now().isoformat(),
            'total_workspaces': len(self.monitor_service.monitored_workspaces),
            'workspace_names': list(self.monitor_service.monitored_workspaces.keys()),
            'task_distribution': task_results,
            'agent_workload': {}
        }

        # Calculate agent workload
        if 'agent_assignments' in task_results:
            for agent_name, agent_data in task_results['agent_assignments'].items():
                summary['agent_workload'][agent_name] = {
                    'skill_count': agent_data['skill_count'],
                    'total_confidence': round(agent_data['total_confidence'], 2),
                    'average_confidence': round(agent_data['total_confidence'] / agent_data['skill_count'], 2) if agent_data['skill_count'] > 0 else 0,
                    'priority': self.task_distributor.agent_categories.get(agent_name, {}).get('priority', 'unknown')
                }

        # Save summary
        summary_file = self.base_workspace_path / "processing_summary.json"
        try:
            with open(summary_file, 'w', encoding='utf-8') as f:
                json.dump(summary, f, indent=2, ensure_ascii=False)
        except Exception as e:
            logger.error(f"Error saving summary: {e}")

        return summary

# Main execution
if __name__ == "__main__":
    # Process all workspaces automatically
    base_path = Path("c:/Luuna/SKILL_BASE")
    processor = AutoWorkspaceProcessor(base_path)

    results = processor.process_all_workspaces()

    print("=== AUTO WORKSPACE PROCESSING RESULTS ===")
    print(json.dumps(results, indent=2, ensure_ascii=False))

    print("\n=== AGENT WORKLOAD SUMMARY ===")
    if 'agent_workload' in results:
        for agent, workload in results['agent_workload'].items():
            print(f"{agent}: {workload['skill_count']} skills (avg confidence: {workload['average_confidence']})")