# WORKING TECH STACK SELECTOR - STRUCTURED JSON VERSION # Simple, effective feature extraction and Claude analysis with structured JSON output # NO complex logic, just works with n8n data 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 # AI integration try: import anthropic CLAUDE_AVAILABLE = True except ImportError: CLAUDE_AVAILABLE = False # Configure logging logger.remove() logger.add(sys.stdout, level="INFO", format="{time} | {level} | {message}") # API Key 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 # ================================================================================================ # WORKING TECH STACK SELECTOR - SIMPLE AND EFFECTIVE # ================================================================================================ class WorkingTechStackSelector: """Simple selector that works with n8n data and Claude""" def __init__(self): self.claude_client = anthropic.Anthropic(api_key=CLAUDE_API_KEY) if CLAUDE_AVAILABLE else None logger.info("Working Tech Stack Selector initialized") # ================================================================================================ # FASTAPI APPLICATION # ================================================================================================ app = FastAPI( title="Working Tech Stack Selector", description="Simple, effective tech stack recommendations with structured JSON output", version="9.0.0" ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Initialize working selector working_selector = WorkingTechStackSelector() @app.get("/health") async def health_check(): """Health check""" return { "status": "healthy", "service": "working-tech-stack-selector", "version": "9.0.0", "approach": "structured_json_claude_analysis" } @app.post("/api/v1/select") async def select_working_tech_stack(request: Request): """STRUCTURED JSON VERSION - Claude returns structured JSON recommendations""" try: request_data = await request.json() # Log exactly what we receive logger.info("=== RECEIVED DATA START ===") logger.info(json.dumps(request_data, indent=2)) logger.info("=== RECEIVED DATA END ===") # Try EVERY possible path to find features all_features = [] project_name = "Unknown Project" scale_info = {} # Path 1: request_data["data"]["all_features"] if isinstance(request_data, dict) and "data" in request_data: if isinstance(request_data["data"], dict) and "all_features" in request_data["data"]: all_features = request_data["data"]["all_features"] project_name = request_data["data"].get("project_name", "Unknown Project") scale_info = request_data["data"].get("scale_information", {}) logger.info(f"✅ Found features via path 1: {len(all_features)} features") # Path 2: request_data["all_features"] if not all_features and isinstance(request_data, dict) and "all_features" in request_data: all_features = request_data["all_features"] project_name = request_data.get("project_name", "Unknown Project") scale_info = request_data.get("scale_information", {}) logger.info(f"✅ Found features via path 2: {len(all_features)} features") # Path 3: success wrapper format if not all_features and isinstance(request_data, dict) and "success" in request_data and "data" in request_data: data_section = request_data["data"] if isinstance(data_section, dict) and "all_features" in data_section: all_features = data_section["all_features"] project_name = data_section.get("project_name", "Unknown Project") scale_info = data_section.get("scale_information", {}) logger.info(f"✅ Found features via path 3: {len(all_features)} features") # Path 4: Deep recursive search if not all_features: def find_all_features(obj, path="root"): if isinstance(obj, dict): if "all_features" in obj and isinstance(obj["all_features"], list): logger.info(f"✅ Found all_features at path: {path}") return obj["all_features"], obj.get("project_name", "Unknown Project"), obj.get("scale_information", {}) for key, value in obj.items(): result = find_all_features(value, f"{path}.{key}") if result[0]: return result elif isinstance(obj, list): for i, item in enumerate(obj): result = find_all_features(item, f"{path}[{i}]") if result[0]: return result return [], "Unknown Project", {} all_features, project_name, scale_info = find_all_features(request_data) if all_features: logger.info(f"✅ Found features via deep search: {len(all_features)} features") logger.info(f"🎯 FINAL RESULTS:") logger.info(f" Features found: {len(all_features)}") logger.info(f" Project name: {project_name}") logger.info(f" Scale info: {scale_info}") if all_features: logger.info(f" First 10 features: {all_features[:10]}") if not all_features: logger.error("❌ NO FEATURES FOUND ANYWHERE") return { "error": "Still no features found after exhaustive search", "received_data_keys": list(request_data.keys()) if isinstance(request_data, dict) else "not_dict", "received_data_sample": str(request_data)[:500] + "..." if len(str(request_data)) > 500 else str(request_data) } # SUCCESS - Call Claude with found features logger.info(f"🚀 Calling Claude with {len(all_features)} features") features_text = "\n".join([f"- {feature.replace('_', ' ').title()}" for feature in all_features]) # STRUCTURED JSON PROMPT - Claude returns structured JSON prompt = f"""You are an expert software architect. Analyze these functional requirements and recommend the optimal technology stack. PROJECT: {project_name} FUNCTIONAL REQUIREMENTS TO IMPLEMENT ({len(all_features)} features): {features_text} SCALE & CONTEXT: {json.dumps(scale_info, indent=2) if scale_info else "Enterprise-scale application"} CRITICAL: Return your response as a valid JSON object with this exact structure: {{ "technology_recommendations": {{ "frontend": {{ "framework": "recommended framework with reasoning", "libraries": ["library1", "library2", "library3"], "reasoning": "detailed reasoning for frontend choices" }}, "backend": {{ "framework": "recommended backend framework", "language": "programming language", "libraries": ["library1", "library2", "library3"], "reasoning": "detailed reasoning for backend choices" }}, "database": {{ "primary": "primary database", "secondary": ["cache", "search", "analytics"], "reasoning": "detailed reasoning for database choices" }}, "infrastructure": {{ "cloud_provider": "recommended cloud provider", "orchestration": "container orchestration", "services": ["service1", "service2", "service3"], "reasoning": "detailed reasoning for infrastructure choices" }}, "testing": {{ "unit_testing": "unit testing framework", "integration_testing": "integration testing tools", "e2e_testing": "end-to-end testing framework", "performance_testing": "performance testing tools", "reasoning": "detailed reasoning for testing strategy" }}, "third_party_services": {{ "authentication": "auth service recommendation", "communication": "communication service", "monitoring": "monitoring solution", "payment": "payment processing", "other_services": ["service1", "service2"], "reasoning": "detailed reasoning for third-party choices" }} }}, "implementation_strategy": {{ "architecture_pattern": "recommended architecture pattern", "development_phases": ["phase1", "phase2", "phase3"], "deployment_strategy": "deployment approach", "scalability_approach": "how to handle scale" }}, "justification": {{ "why_this_stack": "overall reasoning for this technology combination", "scalability_benefits": "how this stack handles the scale requirements", "team_benefits": "how this stack benefits a {scale_info.get('team_size', 'large')} team", "compliance_considerations": "how this stack meets compliance requirements" }} }} IMPORTANT: - Return ONLY valid JSON, no additional text - Base all recommendations on the {len(all_features)} functional requirements provided - Consider the scale: {scale_info.get('expected_users', 'enterprise scale')} users - Ensure all technologies work together seamlessly - Provide specific technology names, not generic descriptions""" # Call Claude if working_selector.claude_client: logger.info("📞 Calling Claude API for structured JSON response...") message = working_selector.claude_client.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=8000, temperature=0.1, messages=[{"role": "user", "content": prompt}] ) claude_response = message.content[0].text logger.info("✅ Successfully received Claude response") # Try to parse Claude's JSON response try: claude_json = json.loads(claude_response) logger.info("✅ Successfully parsed Claude JSON response") return { "success": True, "features_analyzed": len(all_features), "project_name": project_name, "scale_context": scale_info, "all_features": all_features, "claude_recommendations": claude_json, "analysis_timestamp": datetime.utcnow().isoformat() } except json.JSONDecodeError as e: logger.error(f"❌ Failed to parse Claude JSON: {e}") # Fallback to text response return { "success": True, "features_analyzed": len(all_features), "project_name": project_name, "scale_context": scale_info, "all_features": all_features, "claude_recommendations": claude_response, "analysis_timestamp": datetime.utcnow().isoformat(), "json_parse_error": str(e) } else: logger.error("❌ Claude client not available") return { "error": "Claude AI not available", "features_found": len(all_features), "debug": "Check Claude API key configuration" } except Exception as e: logger.error(f"💥 ERROR in tech stack selection: {e}") return { "error": str(e), "debug": "Check service logs for detailed error information" } @app.post("/api/v1/debug-n8n") async def debug_n8n_data(request: Request): """Debug endpoint to see exactly what n8n sends""" try: request_data = await request.json() # Extract data if present if "data" in request_data: data_section = request_data["data"] all_features = data_section.get("all_features", []) else: data_section = request_data all_features = request_data.get("all_features", []) return { "raw_data_keys": list(request_data.keys()) if isinstance(request_data, dict) else "not_dict", "data_section_keys": list(data_section.keys()) if isinstance(data_section, dict) else "not_dict", "features_found": len(all_features), "first_5_features": all_features[:5] if all_features else "none", "data_structure": { "has_success": "success" in request_data, "has_data": "data" in request_data, "has_all_features": "all_features" in data_section if isinstance(data_section, dict) else False } } except Exception as e: return {"error": str(e)} if __name__ == "__main__": import uvicorn logger.info("="*60) logger.info("🚀 WORKING TECH STACK SELECTOR v9.0 - STRUCTURED JSON VERSION") logger.info("="*60) logger.info("✅ Comprehensive logging enabled") logger.info("✅ Multiple feature extraction paths") logger.info("✅ Deep recursive search capability") logger.info("✅ Claude integration with structured JSON output") logger.info("✅ JSON parsing and validation") logger.info("="*60) uvicorn.run("main:app", host="0.0.0.0", port=8002, log_level="info")