331 lines
16 KiB
Python
331 lines
16 KiB
Python
# 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'
|
|
} |