from fastapi import APIRouter, HTTPException, Request from pydantic import BaseModel import httpx import logging from config.settings import get_settings from services.ai_router import route_analysis from services.quotas import check_quota, check_guest_quota logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/analyze", tags=["analyze"]) settings = get_settings() JDOODLE_LANGUAGE_MAP = { "python": {"language": "python3", "versionIndex": "4"}, "javascript": {"language": "nodejs", "versionIndex": "4"}, "typescript": {"language": "typescript", "versionIndex": "1"}, "java": {"language": "java", "versionIndex": "4"}, "c++": {"language": "cpp17", "versionIndex": "1"}, "c": {"language": "c", "versionIndex": "5"}, "go": {"language": "go", "versionIndex": "4"}, "rust": {"language": "rust", "versionIndex": "4"}, "ruby": {"language": "ruby", "versionIndex": "4"}, "php": {"language": "php", "versionIndex": "4"}, } class RunRequest(BaseModel): language: str version: str code: str stdin: str = "" class AnalyzeRequest(BaseModel): model_config = {"protected_namespaces": ()} language: str code: str model_choice: str = "auto" @router.get("/config") async def get_ai_config(): from services.settings import is_ai_enabled return {"ai_features_enabled": is_ai_enabled()} @router.post("/run") async def run_code(data: RunRequest): try: lang_config = JDOODLE_LANGUAGE_MAP.get(data.language.lower()) if not lang_config: raise HTTPException(status_code=400, detail=f"Language {data.language} not supported") async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post( "https://api.jdoodle.com/v1/execute", json={ "clientId": settings.jdoodle_client_id, "clientSecret": settings.jdoodle_client_secret, "script": data.code, "stdin": data.stdin, "language": lang_config["language"], "versionIndex": lang_config["versionIndex"], } ) result = response.json() output = result.get("output", "No output") return {"output": output} except HTTPException: raise except Exception as e: logger.error(f"Analysis failed: {str(e)}") raise HTTPException(status_code=500, detail="AI analysis failed. Please try again later.") @router.post("/analyze") async def analyze_code(data: AnalyzeRequest, request: Request): from services.settings import is_ai_enabled from middleware.auth_guard import get_current_user_optional, get_current_user if not is_ai_enabled(): raise HTTPException(status_code=403, detail="AI analysis features are currently disabled by the administrator.") try: # Get user user = await get_current_user_optional(request) # 1. Tier Enforcement # Free Tier (4 Models + Auto): 'auto', 'gemma-4-31b', 'llama-3.1', 'qwen-2.5', 'nemotron-120b' # Pro Tier (4 Elite Models): 'minimax-2.5', 'mistral-large', 'groq-70b', 'gemini-flash' effective_model = data.model_choice user_role = user.role if user else "guest" free_models = ["auto", "gemma-4-31b", "llama-3.1", "qwen-2.5", "nemotron-120b"] if user_role != "pro" and data.model_choice not in free_models: logger.info(f"User {user.id if user else 'guest'} requested {data.model_choice} but is not Pro. Defaulting to auto.") effective_model = "auto" # 2. Quota Check if user: await check_quota(str(user.id), "analysis") else: await check_guest_quota(request.client.host, "analysis") prompt = f"""[CRITICAL: ELITE SYSTEMS ARCHITECT PERSONA] Analyze this {data.language} code with the precision of a Lead Performance Engineer. CODE TO ANALYZE: {data.code} RETURN ONLY THIS JSON STRUCTURE: {{ "time_complexity": "string (Big O)", "time_explanation": "Elite technical insight (e.g. 'Constant time access via hash map—Excellent speed.')", "space_complexity": "string (Big O)", "space_explanation": "Elite technical insight (e.g. 'Minimal auxiliary space—Optimal memory footprint.')", "issues": ["list of sharp technical issues"], "suggestions": ["list of architectural improvements - MINIMUM 3"], "optimized_code": "string (the superior solution)", "optimized_time_complexity": "string (Big O of optimized version)", "optimized_time_explanation": "Technical optimization insight", "optimized_space_complexity": "string (Big O of optimized version)", "optimized_space_explanation": "Memory optimization insight" }} [NOTE: Return valid JSON only. If the code is already optimally written, structure/format it securely into standard format for `optimized_code`, and return the same complexities.]""" # Using the robust AI router with Smart Recovery Failover result_with_meta = await route_analysis(prompt, model_choice=effective_model) # Extract the actual model used from the metadata actual_model = result_with_meta.get("_actual_model", effective_model) # Remove internal metadata before returning to frontend parsed = {k: v for k, v in result_with_meta.items() if not k.startswith("_")} # Save to DB if logged in if user: try: from config.database import get_supabase_admin admin_client = get_supabase_admin() admin_client.table("analyses").insert({ "user_id": str(user.id), "language": data.language, "code": data.code, "ai_result": parsed, "time_complexity": parsed.get("time_complexity", ""), "space_complexity": parsed.get("space_complexity", ""), "model_used": actual_model # Use the ACTUAL model that responded }).execute() logger.info(f"Analysis saved for user {user.id} (Actual model: {actual_model})") except Exception as save_err: logger.error(f"Failed to save analysis to DB for user {user.id}: {save_err}") # Return result with actual model name so frontend can know return {**parsed, "actual_model_used": actual_model} except HTTPException: raise except Exception as e: logger.error(f"Analysis process failed: {str(e)}") raise HTTPException(status_code=500, detail="AI analysis failed. Please try again later.")