"""Create Studio, Templates, Providers, Cost Estimation, and Platform Specs endpoints.""" import base64 from typing import Dict, Any, Optional from fastapi import APIRouter, Depends, HTTPException from .models import CreateImageRequest, CostEstimationRequest from .deps import get_studio_manager, _require_user_id from services.image_studio import ImageStudioManager, CreateStudioRequest from services.image_studio.templates import Platform, TemplateCategory from middleware.auth_middleware import get_current_user from utils.logger_utils import get_service_logger logger = get_service_logger("api.image_studio") router = APIRouter(tags=["image-studio"]) @router.post("/create", summary="Generate Image") async def create_image( request: CreateImageRequest, current_user: Dict[str, Any] = Depends(get_current_user), studio_manager: ImageStudioManager = Depends(get_studio_manager) ): """Generate image(s) using Create Studio.""" try: user_id = _require_user_id(current_user, "image generation") logger.info(f"[Create Image] Request from user {user_id}: {request.prompt[:100]}") studio_request = CreateStudioRequest( prompt=request.prompt, template_id=request.template_id, provider=request.provider, model=request.model, width=request.width, height=request.height, aspect_ratio=request.aspect_ratio, style_preset=request.style_preset, quality=request.quality, negative_prompt=request.negative_prompt, guidance_scale=request.guidance_scale, steps=request.steps, seed=request.seed, num_variations=request.num_variations, enhance_prompt=request.enhance_prompt, use_persona=request.use_persona, persona_id=request.persona_id, ) result = await studio_manager.create_image(studio_request, user_id=user_id) for idx, img_result in enumerate(result["results"]): if "image_bytes" in img_result: img_result["image_base64"] = base64.b64encode(img_result["image_bytes"]).decode("utf-8") del img_result["image_bytes"] logger.info(f"[Create Image] ✅ Success: {result['total_generated']} images generated") return result except ValueError as e: logger.error(f"[Create Image] ❌ Validation error: {str(e)}") raise HTTPException(status_code=400, detail=str(e)) except RuntimeError as e: logger.error(f"[Create Image] ❌ Generation error: {str(e)}") raise HTTPException(status_code=500, detail=f"Image generation failed: {str(e)}") except Exception as e: logger.error(f"[Create Image] ❌ Unexpected error: {str(e)}", exc_info=True) raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}") @router.get("/templates", summary="Get Templates") async def get_templates( platform: Optional[Platform] = None, category: Optional[TemplateCategory] = None, current_user: Dict[str, Any] = Depends(get_current_user), studio_manager: ImageStudioManager = Depends(get_studio_manager) ): """Get available image templates.""" try: templates = studio_manager.get_templates(platform=platform, category=category) templates_dict = [ { "id": t.id, "name": t.name, "category": t.category.value, "platform": t.platform.value if t.platform else None, "aspect_ratio": { "ratio": t.aspect_ratio.ratio, "width": t.aspect_ratio.width, "height": t.aspect_ratio.height, "label": t.aspect_ratio.label, }, "description": t.description, "recommended_provider": t.recommended_provider, "style_preset": t.style_preset, "quality": t.quality, "use_cases": t.use_cases or [], } for t in templates ] return {"templates": templates_dict, "total": len(templates_dict)} except Exception as e: logger.error(f"[Get Templates] ❌ Error: {str(e)}", exc_info=True) raise HTTPException(status_code=500, detail=str(e)) @router.get("/templates/search", summary="Search Templates") async def search_templates( query: str, current_user: Dict[str, Any] = Depends(get_current_user), studio_manager: ImageStudioManager = Depends(get_studio_manager) ): """Search templates by query.""" try: templates = studio_manager.search_templates(query) templates_dict = [ { "id": t.id, "name": t.name, "category": t.category.value, "platform": t.platform.value if t.platform else None, "aspect_ratio": { "ratio": t.aspect_ratio.ratio, "width": t.aspect_ratio.width, "height": t.aspect_ratio.height, "label": t.aspect_ratio.label, }, "description": t.description, "recommended_provider": t.recommended_provider, "style_preset": t.style_preset, "quality": t.quality, "use_cases": t.use_cases or [], } for t in templates ] return {"templates": templates_dict, "total": len(templates_dict), "query": query} except Exception as e: logger.error(f"[Search Templates] ❌ Error: {str(e)}", exc_info=True) raise HTTPException(status_code=500, detail=str(e)) @router.get("/templates/recommend", summary="Recommend Templates") async def recommend_templates( use_case: str, platform: Optional[Platform] = None, current_user: Dict[str, Any] = Depends(get_current_user), studio_manager: ImageStudioManager = Depends(get_studio_manager) ): """Recommend templates based on use case.""" try: templates = studio_manager.recommend_templates(use_case, platform=platform) templates_dict = [ { "id": t.id, "name": t.name, "category": t.category.value, "platform": t.platform.value if t.platform else None, "aspect_ratio": { "ratio": t.aspect_ratio.ratio, "width": t.aspect_ratio.width, "height": t.aspect_ratio.height, "label": t.aspect_ratio.label, }, "description": t.description, "recommended_provider": t.recommended_provider, "style_preset": t.style_preset, "quality": t.quality, "use_cases": t.use_cases or [], } for t in templates ] return {"templates": templates_dict, "total": len(templates_dict), "use_case": use_case} except Exception as e: logger.error(f"[Recommend Templates] ❌ Error: {str(e)}", exc_info=True) raise HTTPException(status_code=500, detail=str(e)) @router.get("/providers", summary="Get Providers") async def get_providers( current_user: Dict[str, Any] = Depends(get_current_user), studio_manager: ImageStudioManager = Depends(get_studio_manager) ): """Get available AI providers and their capabilities.""" try: providers = studio_manager.get_providers() return {"providers": providers} except Exception as e: logger.error(f"[Get Providers] ❌ Error: {str(e)}", exc_info=True) raise HTTPException(status_code=500, detail=str(e)) @router.post("/estimate-cost", summary="Estimate Cost") async def estimate_cost( request: CostEstimationRequest, current_user: Dict[str, Any] = Depends(get_current_user), studio_manager: ImageStudioManager = Depends(get_studio_manager) ): """Estimate cost for image generation operations.""" try: resolution = None if request.width and request.height: resolution = (request.width, request.height) estimate = studio_manager.estimate_cost( provider=request.provider, model=request.model, operation=request.operation, num_images=request.num_images, resolution=resolution ) return estimate except Exception as e: logger.error(f"[Estimate Cost] ❌ Error: {str(e)}", exc_info=True) raise HTTPException(status_code=500, detail=str(e)) @router.get("/platform-specs/{platform}", summary="Get Platform Specifications") async def get_platform_specs( platform: Platform, current_user: Dict[str, Any] = Depends(get_current_user), studio_manager: ImageStudioManager = Depends(get_studio_manager) ): """Get specifications and requirements for a specific platform.""" try: specs = studio_manager.get_platform_specs(platform) if not specs: raise HTTPException(status_code=404, detail=f"Specifications not found for platform: {platform}") return specs except HTTPException: raise except Exception as e: logger.error(f"[Get Platform Specs] ❌ Error: {str(e)}", exc_info=True) raise HTTPException(status_code=500, detail=str(e))