# FIXED DYNAMIC TECHNOLOGY ROUTER - Correctly extracts from YOUR tech-stack-selector # Fixed paths and structure to match your actual tech-stack-selector response from typing import Dict, Any from loguru import logger # Import dynamic technology specialists from designers.frontend.react_designer import ReactDesigner from designers.backend.nodejs_designer import NodejsDesigner from designers.database.postgresql_designer import PostgreSQLDesigner from designers.database.mongodb_designer import DynamicMongoDBDesigner from designers.frontend.angular_designer_18 import Angular18Designer from designers.backend.aspnet_designer_8 import AspNetCore8Designer from designers.database.mssql_designer_2022 import MSSQLServer2022Designer class TechnologyStack: """Technology stack specification extracted from tech-stack-selector""" def __init__(self, tech_recommendations: Dict[str, Any]): # Extract from YOUR exact tech-stack-selector structure frontend_config = tech_recommendations.get('frontend', {}) backend_config = tech_recommendations.get('backend', {}) database_config = tech_recommendations.get('database', {}) security_config = tech_recommendations.get('security', {}) infrastructure_config = tech_recommendations.get('infrastructure', {}) # Extract exact values from your response self.frontend_framework = frontend_config.get('framework', 'React').lower() self.backend_language = backend_config.get('language', 'Node.js').lower() self.database_system = database_config.get('primary', 'PostgreSQL').lower() # Extract additional tech stack details self.ui_library = self._extract_ui_library(frontend_config.get('libraries', [])) self.state_management = self._extract_state_management(frontend_config.get('libraries', [])) self.authentication = security_config.get('authentication', 'JWT') self.cloud_provider = infrastructure_config.get('cloud_provider', 'AWS') logger.info(f"✅ Technology Stack extracted:") logger.info(f" Frontend: {self.frontend_framework}") logger.info(f" Backend: {self.backend_language}") logger.info(f" Database: {self.database_system}") logger.info(f" UI Library: {self.ui_library}") logger.info(f" State Management: {self.state_management}") logger.info(f" Authentication: {self.authentication}") def _extract_ui_library(self, libraries: list) -> str: """Extract UI library from frontend libraries""" ui_libraries = ['tailwind css', 'material-ui', 'chakra ui', 'ant design', 'bootstrap'] for lib in libraries: if any(ui_lib in lib.lower() for ui_lib in ui_libraries): return lib return 'Tailwind CSS' # Default def _extract_state_management(self, libraries: list) -> str: """Extract state management from frontend libraries""" state_libs = ['redux toolkit', 'zustand', 'context api', 'recoil', 'jotai'] for lib in libraries: if any(state_lib in lib.lower() for state_lib in state_libs): return lib return 'Redux Toolkit' # Default for complex apps class TechnologyRouter: """FIXED router that correctly handles YOUR tech-stack-selector output""" def __init__(self): # Initialize available specialists self.frontend_specialists = { "react": ReactDesigner(), "angular": Angular18Designer(), # NEW "angular 18": Angular18Designer(), # NEW "angular18": Angular18Designer(), # NEW "vue": None, # Will add later } self.backend_specialists = { "node.js": NodejsDesigner(), "asp.net": AspNetCore8Designer(), # NEW "asp.net core": AspNetCore8Designer(), # NEW "aspnet": AspNetCore8Designer(), # NEW "c#": AspNetCore8Designer(), # NEW "nodejs": NodejsDesigner(), "python": None, # Will add later "java": None, # Will add later } self.database_specialists = { "postgresql": PostgreSQLDesigner(), "postgres": PostgreSQLDesigner(), "mysql": None, # Will add later "ms sql server": MSSQLServer2022Designer(), # NEW "ms sql server 2022": MSSQLServer2022Designer(), # NEW "mssql": MSSQLServer2022Designer(), # NEW "sqlserver": MSSQLServer2022Designer(), # NEW "sql server": MSSQLServer2022Designer(), # NEW "mongodb": DynamicMongoDBDesigner(), # Will add later "postgresql": PostgreSQLDesigner(), "postgres": PostgreSQLDesigner(), "PostgreSQL": PostgreSQLDesigner(), # Add uppercase "Postgres": PostgreSQLDesigner(), # Add mixed case "mongodb": DynamicMongoDBDesigner(), "MongoDB": DynamicMongoDBDesigner(), } logger.info("🔄 FIXED Technology Router initialized") logger.info(f" Available Frontend: {[k for k, v in self.frontend_specialists.items() if v]}") logger.info(f" Available Backend: {[k for k, v in self.backend_specialists.items() if v]}") logger.info(f" Available Database: {[k for k, v in self.database_specialists.items() if v]}") def extract_technology_stack(self, tech_stack_selector_output: Dict[str, Any]) -> TechnologyStack: """Extract technology stack from YOUR tech-stack-selector response - WITH USER EDITS SUPPORT""" try: logger.info("🎯 Extracting technology stack from tech-stack-selector...") # CHECK FOR USER'S EDITED TECH STACK FIRST user_tech_choices = tech_stack_selector_output.get('user_technology_choices') if user_tech_choices: logger.info("✅ Found user's edited technology choices - using them!") logger.info(f" User Frontend: {user_tech_choices.get('frontend', {}).get('framework', 'Unknown')}") logger.info(f" User Backend: {user_tech_choices.get('backend', {}).get('language', 'Unknown')}") logger.info(f" User Database: {user_tech_choices.get('database', {}).get('primary', 'Unknown')}") # Use user's edited tech stack tech_stack = TechnologyStack(user_tech_choices) else: logger.info("⚠️ No user edits found, using original AI recommendations...") # Fallback to original logic claude_recommendations = tech_stack_selector_output.get('claude_recommendations', {}) technology_recommendations = claude_recommendations.get('technology_recommendations', {}) if not technology_recommendations: logger.warning("⚠️ No technology_recommendations found, checking alternative paths...") if 'frontend' in claude_recommendations: technology_recommendations = claude_recommendations else: logger.error("❌ Cannot find technology recommendations in response") raise ValueError("No technology recommendations found in tech-stack-selector output") tech_stack = TechnologyStack(technology_recommendations) return tech_stack except Exception as e: logger.error(f"❌ Technology stack extraction failed: {e}") logger.error(f" Available keys in response: {list(tech_stack_selector_output.keys())}") raise async def route_to_specialists(self, tech_stack: TechnologyStack, functional_requirements: Dict[str, Any], business_context: Dict[str, Any]) -> Dict[str, Any]: """Route to appropriate specialists based on extracted technology stack""" try: logger.info("🤖 Routing to technology specialists...") # Get appropriate specialists frontend_specialist = self._get_specialist( tech_stack.frontend_framework, self.frontend_specialists, 'frontend' ) backend_specialist = self._get_specialist( tech_stack.backend_language, self.backend_specialists, 'backend' ) database_specialist = self._get_specialist( tech_stack.database_system, self.database_specialists, 'database' ) # Prepare context for specialists (using YOUR data structure) design_context = self._prepare_design_context( tech_stack, functional_requirements, business_context ) # Call specialists in parallel logger.info("🎨 Calling Frontend specialist...") frontend_result = await frontend_specialist.design_architecture(design_context) logger.info("⚙️ Calling Backend specialist...") backend_result = await backend_specialist.design_architecture(design_context) logger.info("🗄️ Calling Database specialist...") database_result = await database_specialist.design_architecture(design_context) logger.info("✅ All specialists completed successfully") return { 'frontend': frontend_result, 'backend': backend_result, 'database': database_result } except Exception as e: logger.error(f"❌ Specialist routing failed: {e}") raise def _get_specialist(self, technology: str, specialists_dict: Dict, specialist_type: str): """Get appropriate specialist for the technology""" technology_key = technology.lower().replace('.', '').replace(' ', '') specialist = specialists_dict.get(technology_key) if specialist is None: logger.warning(f"⚠️ No {specialist_type} specialist found for '{technology}', using fallback") # Use first available specialist as fallback for key, spec in specialists_dict.items(): if spec is not None: logger.info(f" Using {key} specialist as fallback") return spec raise Exception(f"No {specialist_type} specialists available") logger.info(f"✅ Using {technology_key} specialist for {specialist_type}") return specialist def _prepare_design_context(self, tech_stack: TechnologyStack, functional_requirements: Dict[str, Any], business_context: Dict[str, Any]) -> Dict[str, Any]: """Prepare context using YOUR tech-stack-selector data structure""" return { 'project_id': 'generated_project', # Technology stack information 'technology_stack': { 'frontend': { 'framework': tech_stack.frontend_framework, 'libraries': [tech_stack.ui_library, tech_stack.state_management] }, 'backend': { 'language': tech_stack.backend_language, 'framework': 'Express.js' if 'node' in tech_stack.backend_language else tech_stack.backend_language }, 'database': { 'primary': tech_stack.database_system }, 'security': { 'authentication': tech_stack.authentication }, 'infrastructure': { 'cloud_provider': tech_stack.cloud_provider } }, # Functional requirements from YOUR structure 'functional_requirements': { 'feature_name': functional_requirements.get('feature_name', 'Unknown'), 'description': functional_requirements.get('description', ''), 'technical_requirements': functional_requirements.get('technical_requirements', []), 'business_logic_rules': functional_requirements.get('business_logic_rules', []), 'complexity_level': functional_requirements.get('complexity_level', 'medium'), 'all_features': functional_requirements.get('all_features', []) }, # Business context from YOUR structure 'business_context': business_context } async def route_and_design(self, tech_stack_output: Dict[str, Any], project_id: str) -> Dict[str, Any]: """MAIN ENTRY POINT - takes YOUR tech-stack-selector output and routes to specialists""" try: logger.info("🏗️ Starting architecture design with tech-stack-selector output...") # Extract technology stack from YOUR response tech_stack = self.extract_technology_stack(tech_stack_output) # Extract functional requirements from YOUR response functional_requirements = tech_stack_output.get('functional_requirements', {}) business_context = tech_stack_output.get('claude_recommendations', {}) # Route to specialists specialist_results = await self.route_to_specialists( tech_stack, functional_requirements, business_context ) # Combine results combined_result = { 'technologies_used': { 'frontend': tech_stack.frontend_framework, 'backend': tech_stack.backend_language, 'database': tech_stack.database_system }, 'technology_specifications': tech_stack.__dict__, 'architecture_design': { 'frontend_architecture': specialist_results['frontend'].get('architecture', {}), 'backend_architecture': specialist_results['backend'].get('architecture', {}), 'database_architecture': specialist_results['database'].get('architecture', {}) }, 'specialist_results': specialist_results, 'integration_ready': True, 'source': 'tech_stack_selector_v11' } logger.info("✅ Architecture design completed successfully") return combined_result except Exception as e: logger.error(f"❌ Architecture design failed: {e}") return self._create_fallback_result(tech_stack_output) def _create_fallback_result(self, tech_stack_output: Dict[str, Any]) -> Dict[str, Any]: """Create fallback result when routing fails""" logger.warning("⚠️ Creating fallback architecture result") return { 'technologies_used': {'frontend': 'react', 'backend': 'node.js', 'database': 'postgresql'}, 'technology_specifications': { 'frontend_framework': 'react', 'backend_language': 'node.js', 'database_system': 'postgresql' }, 'architecture_design': { 'frontend_architecture': {'framework': 'React', 'note': 'Fallback architecture'}, 'backend_architecture': {'framework': 'Node.js/Express', 'note': 'Fallback architecture'}, 'database_architecture': {'system': 'PostgreSQL', 'note': 'Fallback architecture'} }, 'fallback': True, 'source': 'fallback_due_to_error' }