refactor(phase3-session-b1): extract upscale, control, social, health into sub-routers
Extracted 4 endpoint groups into separate sub-router modules:
- health.py: 1 endpoint (GET /health)
- upscale.py: 1 endpoint (POST /upscale)
- control.py: 2 endpoints (POST /control/process, GET /control/operations)
- social.py: 2 endpoints (POST /social/optimize, GET /social/platforms/{platform}/formats)
__init__.py now composes these sub-routers into the legacy router.
All 33 routes preserved. No functional changes.
This commit is contained in:
ajaysi
@@ -525,171 +525,10 @@ async def recommend_face_swap_model(
|
||||
raise HTTPException(status_code=500, detail=f"Failed to get recommendation: {e}")
|
||||
|
||||
|
||||
# ====================
|
||||
# UPSCALE STUDIO ENDPOINTS
|
||||
# ====================
|
||||
|
||||
@router.post("/upscale", response_model=UpscaleImageResponse, summary="Upscale Image")
|
||||
async def upscale_image(
|
||||
request: UpscaleImageRequest,
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
studio_manager: ImageStudioManager = Depends(get_studio_manager),
|
||||
):
|
||||
"""Upscale an image using Stability AI pipelines."""
|
||||
try:
|
||||
user_id = _require_user_id(current_user, "image upscaling")
|
||||
upscale_request = UpscaleStudioRequest(
|
||||
image_base64=request.image_base64,
|
||||
mode=request.mode,
|
||||
target_width=request.target_width,
|
||||
target_height=request.target_height,
|
||||
preset=request.preset,
|
||||
prompt=request.prompt,
|
||||
)
|
||||
result = await studio_manager.upscale_image(upscale_request, user_id=user_id)
|
||||
return UpscaleImageResponse(**result)
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"[Upscale Image] ❌ Error: {e}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Image upscaling failed: {e}")
|
||||
|
||||
|
||||
# ====================
|
||||
# CONTROL STUDIO ENDPOINTS
|
||||
# ====================
|
||||
|
||||
@router.post("/control/process", response_model=ControlImageResponse, summary="Process Control Studio request")
|
||||
async def process_control_image(
|
||||
request: ControlImageRequest,
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
studio_manager: ImageStudioManager = Depends(get_studio_manager),
|
||||
):
|
||||
"""Perform Control Studio operations such as sketch-to-image, structure control, style control, and style transfer."""
|
||||
try:
|
||||
user_id = _require_user_id(current_user, "image control")
|
||||
logger.info(f"[Control Image] Request from user {user_id}: operation={request.operation}")
|
||||
|
||||
control_request = ControlStudioRequest(
|
||||
operation=request.operation,
|
||||
prompt=request.prompt,
|
||||
control_image_base64=request.control_image_base64,
|
||||
style_image_base64=request.style_image_base64,
|
||||
negative_prompt=request.negative_prompt,
|
||||
control_strength=request.control_strength,
|
||||
fidelity=request.fidelity,
|
||||
style_strength=request.style_strength,
|
||||
composition_fidelity=request.composition_fidelity,
|
||||
change_strength=request.change_strength,
|
||||
aspect_ratio=request.aspect_ratio,
|
||||
style_preset=request.style_preset,
|
||||
seed=request.seed,
|
||||
output_format=request.output_format,
|
||||
)
|
||||
|
||||
result = await studio_manager.control_image(control_request, user_id=user_id)
|
||||
return ControlImageResponse(**result)
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"[Control Image] ❌ Error: {str(e)}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Image control failed: {e}")
|
||||
|
||||
|
||||
@router.get("/control/operations", response_model=ControlOperationsResponse, summary="List Control Studio operations")
|
||||
async def get_control_operations(
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
studio_manager: ImageStudioManager = Depends(get_studio_manager),
|
||||
):
|
||||
"""Return metadata for supported Control Studio operations."""
|
||||
try:
|
||||
operations = studio_manager.get_control_operations()
|
||||
return ControlOperationsResponse(operations=operations)
|
||||
except Exception as e:
|
||||
logger.error(f"[Control Operations] ❌ Error: {str(e)}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail="Failed to load control operations")
|
||||
|
||||
|
||||
# ====================
|
||||
# SOCIAL OPTIMIZER ENDPOINTS
|
||||
# ====================
|
||||
|
||||
@router.post("/social/optimize", response_model=SocialOptimizeResponse, summary="Optimize image for social platforms")
|
||||
async def optimize_for_social(
|
||||
request: SocialOptimizeRequest,
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
studio_manager: ImageStudioManager = Depends(get_studio_manager),
|
||||
):
|
||||
"""Optimize an image for multiple social media platforms with smart cropping and safe zones."""
|
||||
try:
|
||||
user_id = _require_user_id(current_user, "social optimization")
|
||||
logger.info(f"[Social Optimizer] Request from user {user_id}: platforms={request.platforms}")
|
||||
|
||||
# Convert platform strings to Platform enum
|
||||
from services.image_studio.templates import Platform
|
||||
platforms = []
|
||||
for platform_str in request.platforms:
|
||||
try:
|
||||
platforms.append(Platform(platform_str.lower()))
|
||||
except ValueError:
|
||||
logger.warning(f"[Social Optimizer] Invalid platform: {platform_str}")
|
||||
continue
|
||||
|
||||
if not platforms:
|
||||
raise HTTPException(status_code=400, detail="No valid platforms provided")
|
||||
|
||||
# Convert format_names dict keys to Platform enum
|
||||
format_names = None
|
||||
if request.format_names:
|
||||
format_names = {}
|
||||
for platform_str, format_name in request.format_names.items():
|
||||
try:
|
||||
platform = Platform(platform_str.lower())
|
||||
format_names[platform] = format_name
|
||||
except ValueError:
|
||||
logger.warning(f"[Social Optimizer] Invalid platform in format_names: {platform_str}")
|
||||
|
||||
social_request = SocialOptimizerRequest(
|
||||
image_base64=request.image_base64,
|
||||
platforms=platforms,
|
||||
format_names=format_names,
|
||||
show_safe_zones=request.show_safe_zones,
|
||||
crop_mode=request.crop_mode,
|
||||
focal_point=request.focal_point,
|
||||
output_format=request.output_format,
|
||||
options={},
|
||||
)
|
||||
|
||||
result = await studio_manager.optimize_for_social(social_request, user_id=user_id)
|
||||
return SocialOptimizeResponse(**result)
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"[Social Optimizer] ❌ Error: {str(e)}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Social optimization failed: {e}")
|
||||
|
||||
|
||||
@router.get("/social/platforms/{platform}/formats", response_model=PlatformFormatsResponse, summary="Get platform formats")
|
||||
async def get_platform_formats(
|
||||
platform: str,
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
studio_manager: ImageStudioManager = Depends(get_studio_manager),
|
||||
):
|
||||
"""Get available formats for a social media platform."""
|
||||
try:
|
||||
from services.image_studio.templates import Platform
|
||||
try:
|
||||
platform_enum = Platform(platform.lower())
|
||||
except ValueError:
|
||||
raise HTTPException(status_code=400, detail=f"Invalid platform: {platform}")
|
||||
|
||||
formats = studio_manager.get_social_platform_formats(platform_enum)
|
||||
return PlatformFormatsResponse(formats=formats)
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"[Platform Formats] ❌ Error: {str(e)}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Failed to load platform formats: {e}")
|
||||
|
||||
|
||||
# ====================
|
||||
@@ -1203,26 +1042,5 @@ async def get_format_recommendations(
|
||||
return FormatRecommendationsResponse(recommendations=recommendations)
|
||||
|
||||
|
||||
# ====================
|
||||
# HEALTH CHECK
|
||||
# ====================
|
||||
|
||||
@router.get("/health", summary="Health Check")
|
||||
async def health_check():
|
||||
"""Health check endpoint for Image Studio.
|
||||
|
||||
Returns:
|
||||
Health status
|
||||
"""
|
||||
return {
|
||||
"status": "healthy",
|
||||
"service": "image_studio",
|
||||
"version": "1.0.0",
|
||||
"modules": {
|
||||
"create_studio": "available",
|
||||
"templates": "available",
|
||||
"providers": "available",
|
||||
"compression": "available",
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user