# ================================================================================================ # ENHANCED TECH STACK SELECTOR - MIGRATED VERSION # Uses PostgreSQL data migrated to Neo4j with proper price-based relationships # ================================================================================================ import os import sys import json from datetime import datetime from typing import Dict, Any, Optional, List from pydantic import BaseModel from fastapi import FastAPI, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware from loguru import logger import atexit import anthropic from neo4j import GraphDatabase import psycopg2 from psycopg2.extras import RealDictCursor # ================================================================================================ # NEO4J SERVICE FOR MIGRATED DATA # ================================================================================================ class MigratedNeo4jService: def __init__(self, uri, user, password): self.driver = GraphDatabase.driver( uri, auth=(user, password), connection_timeout=5 ) try: self.driver.verify_connectivity() logger.info("✅ Migrated Neo4j Service connected successfully") except Exception as e: logger.error(f"❌ Neo4j connection failed: {e}") def close(self): self.driver.close() def run_query(self, query: str, parameters: Optional[Dict[str, Any]] = None): with self.driver.session() as session: result = session.run(query, parameters or {}) return [record.data() for record in result] def get_recommendations_by_budget(self, budget: float, domain: Optional[str] = None, preferred_techs: Optional[List[str]] = None): """Get recommendations based on budget using migrated data""" # Normalize domain for better matching normalized_domain = domain.lower().strip() if domain else None # Create domain mapping for better matching domain_mapping = { 'web development': ['portfolio', 'blog', 'website', 'landing', 'documentation', 'personal', 'small', 'learning', 'prototype', 'startup', 'mvp', 'api', 'e-commerce', 'online', 'marketplace', 'retail'], 'ecommerce': ['e-commerce', 'online', 'marketplace', 'retail', 'store', 'shop'], 'portfolio': ['portfolio', 'personal', 'blog', 'website'], 'blog': ['blog', 'content', 'writing', 'documentation'], 'startup': ['startup', 'mvp', 'prototype', 'small', 'business'], 'api': ['api', 'backend', 'service', 'microservice'], 'mobile': ['mobile', 'app', 'ios', 'android', 'react native', 'flutter'], 'ai': ['ai', 'ml', 'machine learning', 'artificial intelligence', 'data', 'analytics'], 'gaming': ['game', 'gaming', 'unity', 'unreal'], 'healthcare': ['healthcare', 'medical', 'health', 'patient', 'clinic'], 'finance': ['finance', 'fintech', 'banking', 'payment', 'financial'], 'education': ['education', 'learning', 'course', 'training', 'elearning'] } # Get related domain keywords related_keywords = [] if normalized_domain: for key, keywords in domain_mapping.items(): if any(keyword in normalized_domain for keyword in [key] + keywords): related_keywords.extend(keywords) break # If no mapping found, use the original domain if not related_keywords: related_keywords = [normalized_domain] # First try to get existing tech stacks with domain filtering existing_stacks = self.run_query(""" MATCH (s:TechStack)-[:BELONGS_TO_TIER]->(p:PriceTier) WHERE s.monthly_cost <= $budget AND ($domain IS NULL OR toLower(s.name) CONTAINS $normalized_domain OR toLower(s.description) CONTAINS $normalized_domain OR EXISTS { MATCH (d:Domain)-[:RECOMMENDS]->(s) WHERE toLower(d.name) = $normalized_domain } OR EXISTS { MATCH (d:Domain)-[:RECOMMENDS]->(s) WHERE toLower(d.name) CONTAINS $normalized_domain } OR (s.recommended_domains IS NOT NULL AND ANY(rd IN s.recommended_domains WHERE ANY(keyword IN $related_keywords WHERE toLower(rd) CONTAINS keyword)))) OPTIONAL MATCH (s)-[:USES_FRONTEND]->(frontend:Technology) OPTIONAL MATCH (s)-[:USES_BACKEND]->(backend:Technology) OPTIONAL MATCH (s)-[:USES_DATABASE]->(database:Technology) OPTIONAL MATCH (s)-[:USES_CLOUD]->(cloud:Technology) OPTIONAL MATCH (s)-[:USES_TESTING]->(testing:Technology) OPTIONAL MATCH (s)-[:USES_MOBILE]->(mobile:Technology) OPTIONAL MATCH (s)-[:USES_DEVOPS]->(devops:Technology) OPTIONAL MATCH (s)-[:USES_AI_ML]->(ai_ml:Technology) WITH s, frontend, backend, database, cloud, testing, mobile, devops, ai_ml, p, (s.satisfaction_score * 0.4 + s.success_rate * 0.3 + CASE WHEN $budget IS NOT NULL THEN (100 - (s.monthly_cost / $budget * 100)) * 0.3 ELSE 30 END) AS base_score WITH s, frontend, backend, database, cloud, testing, mobile, devops, ai_ml, base_score, p, CASE WHEN $preferred_techs IS NOT NULL THEN size([x IN $preferred_techs WHERE toLower(x) IN [toLower(frontend.name), toLower(backend.name), toLower(database.name), toLower(cloud.name), toLower(testing.name), toLower(mobile.name), toLower(devops.name), toLower(ai_ml.name)]]) * 5 ELSE 0 END AS preference_bonus RETURN s.name AS stack_name, s.monthly_cost AS monthly_cost, s.setup_cost AS setup_cost, s.team_size_range AS team_size, s.development_time_months AS development_time, s.satisfaction_score AS satisfaction, s.success_rate AS success_rate, s.price_tier AS price_tier, s.recommended_domains AS recommended_domains, s.description AS description, s.pros AS pros, s.cons AS cons, COALESCE(frontend.name, s.frontend_tech, 'Not specified') AS frontend, COALESCE(backend.name, s.backend_tech, 'Not specified') AS backend, COALESCE(database.name, s.database_tech, 'Not specified') AS database, COALESCE(cloud.name, s.cloud_tech, 'Not specified') AS cloud, COALESCE(testing.name, s.testing_tech, 'Not specified') AS testing, COALESCE(mobile.name, s.mobile_tech, 'Not specified') AS mobile, COALESCE(devops.name, s.devops_tech, 'Not specified') AS devops, COALESCE(ai_ml.name, s.ai_ml_tech, 'Not specified') AS ai_ml, base_score + preference_bonus AS recommendation_score ORDER BY recommendation_score DESC, s.monthly_cost ASC LIMIT 10 """, { "budget": budget, "domain": domain, "normalized_domain": normalized_domain, "related_keywords": related_keywords, "preferred_techs": preferred_techs or [] }) logger.info(f"🔍 Found {len(existing_stacks)} existing stacks from Neo4j with domain filtering") if existing_stacks: logger.info("✅ Using existing Neo4j stacks") return existing_stacks # If no domain-specific stacks found, try without domain filtering logger.info("🔍 No domain-specific stacks found, trying without domain filter...") existing_stacks_no_domain = self.run_query(""" MATCH (s:TechStack)-[:BELONGS_TO_TIER]->(p:PriceTier) WHERE s.monthly_cost <= $budget OPTIONAL MATCH (s)-[:USES_FRONTEND]->(frontend:Technology) OPTIONAL MATCH (s)-[:USES_BACKEND]->(backend:Technology) OPTIONAL MATCH (s)-[:USES_DATABASE]->(database:Technology) OPTIONAL MATCH (s)-[:USES_CLOUD]->(cloud:Technology) OPTIONAL MATCH (s)-[:USES_TESTING]->(testing:Technology) OPTIONAL MATCH (s)-[:USES_MOBILE]->(mobile:Technology) OPTIONAL MATCH (s)-[:USES_DEVOPS]->(devops:Technology) OPTIONAL MATCH (s)-[:USES_AI_ML]->(ai_ml:Technology) WITH s, frontend, backend, database, cloud, testing, mobile, devops, ai_ml, p, (s.satisfaction_score * 0.4 + s.success_rate * 0.3 + CASE WHEN $budget IS NOT NULL THEN (100 - (s.monthly_cost / $budget * 100)) * 0.3 ELSE 30 END) AS base_score WITH s, frontend, backend, database, cloud, testing, mobile, devops, ai_ml, base_score, p, CASE WHEN $preferred_techs IS NOT NULL THEN size([x IN $preferred_techs WHERE toLower(x) IN [toLower(frontend.name), toLower(backend.name), toLower(database.name), toLower(cloud.name), toLower(testing.name), toLower(mobile.name), toLower(devops.name), toLower(ai_ml.name)]]) * 5 ELSE 0 END AS preference_bonus RETURN s.name AS stack_name, s.monthly_cost AS monthly_cost, s.setup_cost AS setup_cost, s.team_size_range AS team_size, s.development_time_months AS development_time, s.satisfaction_score AS satisfaction, s.success_rate AS success_rate, s.price_tier AS price_tier, s.recommended_domains AS recommended_domains, s.description AS description, s.pros AS pros, s.cons AS cons, COALESCE(frontend.name, s.frontend_tech, 'Not specified') AS frontend, COALESCE(backend.name, s.backend_tech, 'Not specified') AS backend, COALESCE(database.name, s.database_tech, 'Not specified') AS database, COALESCE(cloud.name, s.cloud_tech, 'Not specified') AS cloud, COALESCE(testing.name, s.testing_tech, 'Not specified') AS testing, COALESCE(mobile.name, s.mobile_tech, 'Not specified') AS mobile, COALESCE(devops.name, s.devops_tech, 'Not specified') AS devops, COALESCE(ai_ml.name, s.ai_ml_tech, 'Not specified') AS ai_ml, base_score + preference_bonus AS recommendation_score ORDER BY recommendation_score DESC, s.monthly_cost ASC LIMIT 10 """, { "budget": budget, "preferred_techs": preferred_techs or [] }) logger.info(f"🔍 Found {len(existing_stacks_no_domain)} existing stacks from Neo4j without domain filtering") if existing_stacks_no_domain: logger.info("✅ Using existing Neo4j stacks (no domain filter)") return existing_stacks_no_domain # If no existing stacks, try Claude AI for intelligent recommendations logger.info("🤖 No existing stacks found, trying Claude AI...") claude_recommendations = self.get_claude_ai_recommendations(budget, domain, preferred_techs) if claude_recommendations: logger.info(f"✅ Generated {len(claude_recommendations)} Claude AI recommendations") return claude_recommendations # Final fallback to dynamic recommendations using tools and technologies logger.info("⚠️ Claude AI failed, falling back to dynamic recommendations") return self.get_dynamic_recommendations(budget, domain, preferred_techs) def get_dynamic_recommendations(self, budget: float, domain: Optional[str] = None, preferred_techs: Optional[List[str]] = None): """Create dynamic recommendations using tools and technologies""" # Normalize domain for better matching normalized_domain = domain.lower().strip() if domain else None # Get tools within budget tools_query = """ MATCH (tool:Tool)-[:BELONGS_TO_TIER]->(p:PriceTier) WHERE tool.monthly_cost_usd <= $budget RETURN tool.name as tool_name, tool.category as category, tool.monthly_cost_usd as monthly_cost, tool.total_cost_of_ownership_score as tco_score, tool.price_performance_ratio as price_performance, p.tier_name as price_tier ORDER BY tool.price_performance_ratio DESC, tool.monthly_cost_usd ASC LIMIT 20 """ tools = self.run_query(tools_query, {"budget": budget}) # Get technologies by category (without pricing constraints) tech_categories = ["frontend", "backend", "database", "cloud", "testing", "mobile", "devops", "ai_ml"] recommendations = [] # Create domain-specific recommendations domain_specific_stacks = self._create_domain_specific_stacks(normalized_domain, budget) if domain_specific_stacks: recommendations.extend(domain_specific_stacks) for category in tech_categories: tech_query = f""" MATCH (t:Technology {{category: '{category}'}}) RETURN t.name as name, t.category as category, t.maturity_score as maturity_score, t.learning_curve as learning_curve, t.performance_rating as performance_rating, t.total_cost_of_ownership_score as tco_score, t.price_performance_ratio as price_performance ORDER BY t.total_cost_of_ownership_score DESC, t.maturity_score DESC LIMIT 3 """ technologies = self.run_query(tech_query) if technologies: # Create a recommendation entry for this category best_tech = technologies[0] recommendation = { "stack_name": f"Dynamic {category.title()} Stack - {best_tech['name']}", "monthly_cost": 0.0, # Technologies don't have pricing "setup_cost": 0.0, "team_size_range": "2-5", "development_time_months": 2, "satisfaction_score": best_tech.get('tco_score') or 80, "success_rate": best_tech.get('maturity_score') or 80, "price_tier": "Custom", "budget_efficiency": 100.0, "frontend": best_tech['name'] if category == 'frontend' else 'Not specified', "backend": best_tech['name'] if category == 'backend' else 'Not specified', "database": best_tech['name'] if category == 'database' else 'Not specified', "cloud": best_tech['name'] if category == 'cloud' else 'Not specified', "testing": best_tech['name'] if category == 'testing' else 'Not specified', "mobile": best_tech['name'] if category == 'mobile' else 'Not specified', "devops": best_tech['name'] if category == 'devops' else 'Not specified', "ai_ml": best_tech['name'] if category == 'ai_ml' else 'Not specified', "recommendation_score": (best_tech.get('tco_score') or 80) + (best_tech.get('maturity_score') or 80) / 2 } recommendations.append(recommendation) # Add tool-based recommendations if tools: # Group tools by category and create recommendations tool_categories = {} for tool in tools: category = tool['category'] if category not in tool_categories: tool_categories[category] = [] tool_categories[category].append(tool) for category, category_tools in tool_categories.items(): if category_tools: best_tool = category_tools[0] total_cost = sum(t['monthly_cost'] for t in category_tools[:3]) # Top 3 tools if total_cost <= budget: recommendation = { "stack_name": f"Tool-based {category.title()} Stack - {best_tool['tool_name']}", "monthly_cost": total_cost, "setup_cost": total_cost * 0.5, "team_size_range": "1-3", "development_time_months": 1, "satisfaction_score": best_tool.get('tco_score') or 80, "success_rate": best_tool.get('price_performance') or 80, "price_tier": best_tool.get('price_tier', 'Custom'), "budget_efficiency": 100.0 - ((total_cost / budget) * 20) if budget > 0 else 100.0, "frontend": "Not specified", "backend": "Not specified", "database": "Not specified", "cloud": "Not specified", "testing": "Not specified", "mobile": "Not specified", "devops": "Not specified", "ai_ml": "Not specified", "recommendation_score": (best_tool.get('tco_score') or 80) + (best_tool.get('price_performance') or 80) / 2, "tools": [t['tool_name'] for t in category_tools[:3]] } recommendations.append(recommendation) # Sort by recommendation score and return top 10 recommendations.sort(key=lambda x: x['recommendation_score'], reverse=True) return recommendations[:10] def _create_domain_specific_stacks(self, domain: Optional[str], budget: float): """Create domain-specific technology stacks""" if not domain: return [] # Domain-specific technology mappings domain_tech_mapping = { 'healthcare': { 'frontend': 'React', 'backend': 'Django', 'database': 'PostgreSQL', 'cloud': 'AWS', 'testing': 'Jest', 'mobile': 'React Native', 'devops': 'Docker', 'ai_ml': 'TensorFlow' }, 'finance': { 'frontend': 'Angular', 'backend': 'Java Spring', 'database': 'PostgreSQL', 'cloud': 'AWS', 'testing': 'JUnit', 'mobile': 'Flutter', 'devops': 'Kubernetes', 'ai_ml': 'Scikit-learn' }, 'gaming': { 'frontend': 'Unity', 'backend': 'Node.js', 'database': 'MongoDB', 'cloud': 'AWS', 'testing': 'Unity Test Framework', 'mobile': 'Unity', 'devops': 'Docker', 'ai_ml': 'TensorFlow' }, 'education': { 'frontend': 'React', 'backend': 'Django', 'database': 'PostgreSQL', 'cloud': 'DigitalOcean', 'testing': 'Jest', 'mobile': 'React Native', 'devops': 'Docker', 'ai_ml': 'Scikit-learn' }, 'media': { 'frontend': 'Next.js', 'backend': 'Node.js', 'database': 'MongoDB', 'cloud': 'Vercel', 'testing': 'Jest', 'mobile': 'React Native', 'devops': 'Docker', 'ai_ml': 'Hugging Face' }, 'iot': { 'frontend': 'React', 'backend': 'Python', 'database': 'InfluxDB', 'cloud': 'AWS', 'testing': 'Pytest', 'mobile': 'React Native', 'devops': 'Docker', 'ai_ml': 'TensorFlow' }, 'social': { 'frontend': 'React', 'backend': 'Node.js', 'database': 'MongoDB', 'cloud': 'AWS', 'testing': 'Jest', 'mobile': 'React Native', 'devops': 'Docker', 'ai_ml': 'Hugging Face' }, 'elearning': { 'frontend': 'Vue.js', 'backend': 'Django', 'database': 'PostgreSQL', 'cloud': 'DigitalOcean', 'testing': 'Jest', 'mobile': 'Flutter', 'devops': 'Docker', 'ai_ml': 'Scikit-learn' }, 'realestate': { 'frontend': 'React', 'backend': 'Node.js', 'database': 'PostgreSQL', 'cloud': 'AWS', 'testing': 'Jest', 'mobile': 'React Native', 'devops': 'Docker', 'ai_ml': 'Not specified' }, 'travel': { 'frontend': 'React', 'backend': 'Node.js', 'database': 'MongoDB', 'cloud': 'AWS', 'testing': 'Jest', 'mobile': 'React Native', 'devops': 'Docker', 'ai_ml': 'Not specified' }, 'manufacturing': { 'frontend': 'Angular', 'backend': 'Java Spring', 'database': 'PostgreSQL', 'cloud': 'AWS', 'testing': 'JUnit', 'mobile': 'Flutter', 'devops': 'Kubernetes', 'ai_ml': 'TensorFlow' } } # Get technology mapping for domain tech_mapping = domain_tech_mapping.get(domain) if not tech_mapping: return [] # Create domain-specific stack stack = { "stack_name": f"Domain-Specific {domain.title()} Stack", "monthly_cost": min(budget * 0.8, 100.0), # Use 80% of budget or max $100 "setup_cost": min(budget * 0.4, 500.0), # Use 40% of budget or max $500 "team_size_range": "3-6", "development_time_months": 4, "satisfaction_score": 85, "success_rate": 88, "price_tier": "Custom", "recommended_domains": [domain.title()], "description": f"Specialized technology stack optimized for {domain} applications", "pros": [ f"Optimized for {domain}", "Domain-specific features", "Proven technology choices", "Good performance" ], "cons": [ "Domain-specific complexity", "Learning curve", "Customization needs" ], "frontend": tech_mapping['frontend'], "backend": tech_mapping['backend'], "database": tech_mapping['database'], "cloud": tech_mapping['cloud'], "testing": tech_mapping['testing'], "mobile": tech_mapping['mobile'], "devops": tech_mapping['devops'], "ai_ml": tech_mapping['ai_ml'], "recommendation_score": 90.0 } return [stack] def get_available_domains(self): """Get all available domains from the database""" query = """ MATCH (d:Domain) RETURN d.name as domain_name, d.project_scale as project_scale, d.team_experience_level as team_experience_level ORDER BY d.name """ return self.run_query(query) def get_technologies_by_price_tier(self, tier_name: str): """Get technologies for a specific price tier""" query = """ MATCH (t:Technology)-[:BELONGS_TO_TIER]->(p:PriceTier {tier_name: $tier_name}) RETURN t.name as name, t.category as category, t.monthly_cost_usd as monthly_cost, t.total_cost_of_ownership_score as tco_score, t.price_performance_ratio as price_performance, t.maturity_score as maturity_score, t.learning_curve as learning_curve ORDER BY t.total_cost_of_ownership_score DESC, t.monthly_cost_usd ASC """ return self.run_query(query, {"tier_name": tier_name}) def get_tools_by_price_tier(self, tier_name: str): """Get tools for a specific price tier""" query = """ MATCH (tool:Tool)-[:BELONGS_TO_TIER]->(p:PriceTier {tier_name: $tier_name}) RETURN tool.name as name, tool.category as category, tool.monthly_cost_usd as monthly_cost, tool.total_cost_of_ownership_score as tco_score, tool.price_performance_ratio as price_performance, tool.popularity_score as popularity_score ORDER BY tool.price_performance_ratio DESC, tool.monthly_cost_usd ASC """ return self.run_query(query, {"tier_name": tier_name}) def get_price_tier_analysis(self): """Get analysis of all price tiers""" query = """ MATCH (p:PriceTier) OPTIONAL MATCH (p)<-[:BELONGS_TO_TIER]-(t:Technology) OPTIONAL MATCH (p)<-[:BELONGS_TO_TIER]-(tool:Tool) OPTIONAL MATCH (p)<-[:BELONGS_TO_TIER]-(s:TechStack) RETURN p.tier_name as tier_name, p.min_price_usd as min_price, p.max_price_usd as max_price, p.target_audience as target_audience, p.typical_project_scale as project_scale, count(DISTINCT t) as technology_count, count(DISTINCT tool) as tool_count, count(DISTINCT s) as stack_count, avg(t.monthly_cost_usd) as avg_tech_cost, avg(tool.monthly_cost_usd) as avg_tool_cost ORDER BY p.min_price_usd """ return self.run_query(query) def get_optimal_combinations(self, budget: float, category: str): """Get optimal technology combinations within budget for a category""" query = """ MATCH (t:Technology {category: $category})-[:BELONGS_TO_TIER]->(p:PriceTier) WHERE t.monthly_cost_usd <= $budget RETURN t.name as name, t.monthly_cost_usd as monthly_cost, t.total_cost_of_ownership_score as tco_score, t.price_performance_ratio as price_performance, p.tier_name as price_tier, (t.total_cost_of_ownership_score * 0.6 + t.price_performance_ratio * 0.4) as combined_score ORDER BY combined_score DESC, t.monthly_cost_usd ASC LIMIT 10 """ return self.run_query(query, {"budget": budget, "category": category}) def get_compatibility_analysis(self, tech_name: str): """Get compatibility analysis for a specific technology""" query = """ MATCH (t:Technology {name: $tech_name})-[r:COMPATIBLE_WITH]-(compatible:Technology) RETURN compatible.name as compatible_tech, compatible.category as category, r.compatibility_score as score, r.integration_effort as effort, r.reason as reason ORDER BY r.compatibility_score DESC """ return self.run_query(query, {"tech_name": tech_name}) def validate_data_integrity(self): """Validate the integrity of migrated data""" query = """ MATCH (s:TechStack) RETURN s.name as stack_name, exists((s)-[:BELONGS_TO_TIER]->()) as has_price_tier, exists((s)-[:USES_FRONTEND]->()) as has_frontend, exists((s)-[:USES_BACKEND]->()) as has_backend, exists((s)-[:USES_DATABASE]->()) as has_database, exists((s)-[:USES_CLOUD]->()) as has_cloud, s.monthly_cost as monthly_cost, s.price_tier as price_tier ORDER BY s.monthly_cost """ return self.run_query(query) def get_claude_ai_recommendations(self, budget: float, domain: Optional[str] = None, preferred_techs: Optional[List[str]] = None): """Generate recommendations using Claude AI when no knowledge graph data is available""" try: client = anthropic.Anthropic(api_key=api_key) # Create a comprehensive prompt for Claude AI prompt = f""" You are a tech stack recommendation expert. Generate 5-10 technology stack recommendations based on the following requirements: **Requirements:** - Budget: ${budget:,.2f} per month - Domain: {domain or 'general'} - Preferred Technologies: {', '.join(preferred_techs) if preferred_techs else 'None specified'} **Output Format:** Return a JSON array with the following structure for each recommendation: {{ "stack_name": "Descriptive name for the tech stack", "monthly_cost": number (monthly operational cost in USD), "setup_cost": number (one-time setup cost in USD), "team_size_range": "string (e.g., '1-2', '3-5', '6-10')", "development_time_months": number (months to complete, 1-12), "satisfaction_score": number (0-100, user satisfaction score), "success_rate": number (0-100, project success rate), "price_tier": "string (e.g., 'Micro Budget', 'Startup Budget', 'Enterprise')", "budget_efficiency": number (0-100, how well it uses the budget), "frontend": "string (specific frontend technology like 'React.js', 'Vue.js', 'Angular')", "backend": "string (specific backend technology like 'Node.js', 'Django', 'Spring Boot')", "database": "string (specific database like 'PostgreSQL', 'MongoDB', 'MySQL')", "cloud": "string (specific cloud platform like 'AWS', 'DigitalOcean', 'Azure')", "testing": "string (specific testing framework like 'Jest', 'pytest', 'Cypress')", "mobile": "string (mobile technology like 'React Native', 'Flutter', 'Ionic' or 'None')", "devops": "string (devops tool like 'Docker', 'GitHub Actions', 'Jenkins')", "ai_ml": "string (AI/ML technology like 'TensorFlow', 'scikit-learn', 'PyTorch' or 'None')", "recommendation_score": number (0-100, overall recommendation score), "tools": ["array of specific tools and services"], "description": "string (brief explanation of the recommendation)" }} **Important Guidelines:** 1. Ensure all technology fields have specific, realistic technology names (not "Not specified") 2. Monthly costs should be realistic and within budget 3. Consider the domain requirements carefully 4. Include preferred technologies when possible 5. Provide diverse recommendations (different approaches, complexity levels) 6. Make sure all numeric values are realistic and consistent 7. Focus on practical, implementable solutions Generate recommendations that are: - Cost-effective and within budget - Appropriate for the domain - Include modern, proven technologies - Provide good value for money - Are realistic to implement """ response = client.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=4000, temperature=0.7, messages=[{ "role": "user", "content": prompt }] ) # Parse Claude's response content = response.content[0].text.strip() # Extract JSON from the response import re json_match = re.search(r'\[.*\]', content, re.DOTALL) if json_match: recommendations = json.loads(json_match.group()) logger.info(f"✅ Generated {len(recommendations)} Claude AI recommendations") return recommendations else: logger.warning("❌ Could not parse Claude AI response as JSON") return [] except Exception as e: logger.error(f"❌ Claude AI recommendation failed: {e}") return [] # ================================================================================================ # POSTGRESQL MIGRATION SERVICE (SAME AS BEFORE) # ================================================================================================ class PostgreSQLMigrationService: def __init__(self, host="localhost", port=5432, user="pipeline_admin", password="secure_pipeline_2024", database="dev_pipeline"): self.config = { "host": host, "port": port, "user": user, "password": password, "database": database } self.connection = None self.cursor = None self.last_error: Optional[str] = None def is_open(self) -> bool: try: return ( self.connection is not None and getattr(self.connection, "closed", 1) == 0 and self.cursor is not None and not getattr(self.cursor, "closed", True) ) except Exception: return False def connect(self): try: if self.is_open(): self.last_error = None return True self.connection = psycopg2.connect(**self.config) self.cursor = self.connection.cursor(cursor_factory=RealDictCursor) logger.info("Connected to PostgreSQL successfully") self.last_error = None return True except Exception as e: logger.error(f"Error connecting to PostgreSQL: {e}") self.last_error = str(e) return False def close(self): try: if self.cursor and not getattr(self.cursor, "closed", True): self.cursor.close() finally: self.cursor = None try: if self.connection and getattr(self.connection, "closed", 1) == 0: self.connection.close() finally: self.connection = None # ================================================================================================ # FASTAPI APPLICATION # ================================================================================================ app = FastAPI( title="Enhanced Tech Stack Selector - Migrated Version", description="Tech stack selector using PostgreSQL data migrated to Neo4j with price-based relationships", version="15.0.0" ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # ================================================================================================ # CONFIGURATION # ================================================================================================ logger.remove() logger.add(sys.stdout, level="INFO", format="{time} | {level} | {message}") CLAUDE_API_KEY = "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" if not os.getenv("CLAUDE_API_KEY") and CLAUDE_API_KEY: os.environ["CLAUDE_API_KEY"] = CLAUDE_API_KEY api_key = os.getenv("CLAUDE_API_KEY") or CLAUDE_API_KEY logger.info(f"🔑 Claude API Key loaded: {api_key[:20]}..." if api_key else "❌ No Claude API Key found") # Initialize services NEO4J_URI = os.getenv("NEO4J_URI", "bolt://localhost:7687") NEO4J_USER = os.getenv("NEO4J_USER", "neo4j") NEO4J_PASSWORD = os.getenv("NEO4J_PASSWORD", "password") neo4j_service = MigratedNeo4jService( uri=NEO4J_URI, user=NEO4J_USER, password=NEO4J_PASSWORD ) postgres_migration_service = PostgreSQLMigrationService( host=os.getenv("POSTGRES_HOST", "localhost"), port=int(os.getenv("POSTGRES_PORT", "5432")), user=os.getenv("POSTGRES_USER", "pipeline_admin"), password=os.getenv("POSTGRES_PASSWORD", "secure_pipeline_2024"), database=os.getenv("POSTGRES_DB", "dev_pipeline") ) # ================================================================================================ # SHUTDOWN HANDLER # ================================================================================================ @app.on_event("shutdown") async def shutdown_event(): neo4j_service.close() postgres_migration_service.close() atexit.register(lambda: neo4j_service.close()) atexit.register(lambda: postgres_migration_service.close()) # ================================================================================================ # ENDPOINTS # ================================================================================================ @app.get("/health") async def health_check(): return { "status": "healthy", "service": "enhanced-tech-stack-selector-migrated", "version": "15.0.0", "features": ["migrated_neo4j", "postgresql_source", "claude_ai", "price_based_relationships"] } @app.get("/api/diagnostics") async def diagnostics(): diagnostics_result = { "service": "enhanced-tech-stack-selector-migrated", "version": "15.0.0", "timestamp": datetime.utcnow().isoformat(), "checks": {} } # Check Neo4j neo4j_check = {"status": "unknown"} try: with neo4j_service.driver.session() as session: result = session.run("MATCH (n) RETURN count(n) AS count") node_count = result.single().get("count", 0) neo4j_check.update({ "status": "ok", "node_count": int(node_count) }) except Exception as e: neo4j_check.update({ "status": "error", "error": str(e) }) diagnostics_result["checks"]["neo4j"] = neo4j_check # Check data integrity try: integrity = neo4j_service.validate_data_integrity() neo4j_check["data_integrity"] = { "total_stacks": len(integrity), "complete_stacks": len([s for s in integrity if all([ s["has_price_tier"], s["has_frontend"], s["has_backend"], s["has_database"], s["has_cloud"] ])]) } except Exception as e: neo4j_check["data_integrity"] = {"error": str(e)} return diagnostics_result # ================================================================================================ # RECOMMENDATION ENDPOINTS # ================================================================================================ class RecommendBestRequest(BaseModel): domain: Optional[str] = None budget: Optional[float] = None preferredTechnologies: Optional[List[str]] = None @app.post("/recommend/best") async def recommend_best(req: RecommendBestRequest): """Get recommendations using migrated data with price-based relationships""" try: if not req.budget or req.budget <= 0: raise HTTPException(status_code=400, detail="Budget must be greater than 0") recommendations = neo4j_service.get_recommendations_by_budget( budget=req.budget, domain=req.domain, preferred_techs=req.preferredTechnologies ) return { "success": True, "recommendations": recommendations, "count": len(recommendations), "budget": req.budget, "domain": req.domain, "data_source": "migrated_postgresql" } except Exception as e: logger.error(f"Error in recommendations: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.get("/api/price-tiers") async def get_price_tiers(): """Get all price tiers with analysis""" try: analysis = neo4j_service.get_price_tier_analysis() return { "success": True, "price_tiers": analysis, "count": len(analysis) } except Exception as e: logger.error(f"Error fetching price tiers: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.get("/api/technologies/{tier_name}") async def get_technologies_by_tier(tier_name: str): """Get technologies for a specific price tier""" try: technologies = neo4j_service.get_technologies_by_price_tier(tier_name) return { "success": True, "tier_name": tier_name, "technologies": technologies, "count": len(technologies) } except Exception as e: logger.error(f"Error fetching technologies for tier {tier_name}: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.get("/api/tools/{tier_name}") async def get_tools_by_tier(tier_name: str): """Get tools for a specific price tier""" try: tools = neo4j_service.get_tools_by_price_tier(tier_name) return { "success": True, "tier_name": tier_name, "tools": tools, "count": len(tools) } except Exception as e: logger.error(f"Error fetching tools for tier {tier_name}: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.get("/api/combinations/optimal") async def get_optimal_combinations(budget: float, category: str): """Get optimal technology combinations within budget""" try: if budget <= 0: raise HTTPException(status_code=400, detail="Budget must be greater than 0") combinations = neo4j_service.get_optimal_combinations(budget, category) return { "success": True, "combinations": combinations, "count": len(combinations), "budget": budget, "category": category } except Exception as e: logger.error(f"Error finding optimal combinations: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.get("/api/compatibility/{tech_name}") async def get_compatibility_analysis(tech_name: str): """Get compatibility analysis for a technology""" try: compatibility = neo4j_service.get_compatibility_analysis(tech_name) return { "success": True, "tech_name": tech_name, "compatible_technologies": compatibility, "count": len(compatibility) } except Exception as e: logger.error(f"Error fetching compatibility for {tech_name}: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.get("/api/validate/integrity") async def validate_data_integrity(): """Validate data integrity of migrated data""" try: integrity = neo4j_service.validate_data_integrity() return { "success": True, "integrity_check": integrity, "summary": { "total_stacks": len(integrity), "complete_stacks": len([s for s in integrity if all([ s["has_price_tier"], s["has_frontend"], s["has_backend"], s["has_database"], s["has_cloud"] ])]), "incomplete_stacks": len([s for s in integrity if not all([ s["has_price_tier"], s["has_frontend"], s["has_backend"], s["has_database"], s["has_cloud"] ])]) } } except Exception as e: logger.error(f"Error validating data integrity: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.get("/api/domains") async def get_available_domains(): """Get all available domains""" try: domains = neo4j_service.get_available_domains() return { "success": True, "domains": domains, "count": len(domains) } except Exception as e: logger.error(f"Error fetching domains: {e}") raise HTTPException(status_code=500, detail=str(e)) # ================================================================================================ # MAIN ENTRY POINT # ================================================================================================ if __name__ == "__main__": import uvicorn logger.info("="*60) logger.info("🚀 ENHANCED TECH STACK SELECTOR v15.0 - MIGRATED VERSION") logger.info("="*60) logger.info("✅ Migrated PostgreSQL data to Neo4j") logger.info("✅ Price-based relationships") logger.info("✅ Real data from PostgreSQL") logger.info("✅ Claude AI recommendations") logger.info("✅ Comprehensive pricing analysis") logger.info("="*60) uvicorn.run("main_migrated:app", host="0.0.0.0", port=8002, log_level="info")