refactor(phase3-session-b2): extract edit and face_swap into sub-routers
Extracted 2 endpoint groups into separate sub-router modules: - edit.py: 4 endpoints (POST /edit/process, GET /edit/operations, GET /edit/models, POST /edit/recommend) - face_swap.py: 3 endpoints (POST /face-swap/process, GET /face-swap/models, POST /face-swap/recommend) All 33 routes preserved (10 extracted in B1, 7 extracted in B2, 16 remaining in legacy).
This commit is contained in:
ajaysi
@@ -8,11 +8,15 @@ from .health import router as health_router
|
||||
from .upscale import router as upscale_router
|
||||
from .control import router as control_router
|
||||
from .social import router as social_router
|
||||
from .edit import router as edit_router
|
||||
from .face_swap import router as face_swap_router
|
||||
|
||||
legacy_router.include_router(health_router)
|
||||
legacy_router.include_router(upscale_router)
|
||||
legacy_router.include_router(control_router)
|
||||
legacy_router.include_router(social_router)
|
||||
legacy_router.include_router(edit_router)
|
||||
legacy_router.include_router(face_swap_router)
|
||||
|
||||
router = legacy_router
|
||||
|
||||
|
||||
122
backend/routers/image_studio/edit.py
Normal file
122
backend/routers/image_studio/edit.py
Normal file
@@ -0,0 +1,122 @@
|
||||
"""Edit Studio endpoints."""
|
||||
|
||||
from typing import Dict, Any, Optional
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
|
||||
from .models import (
|
||||
EditImageRequest, EditImageResponse, EditOperationsResponse,
|
||||
EditModelsResponse, EditModelRecommendationRequest, EditModelRecommendationResponse,
|
||||
)
|
||||
from .deps import get_studio_manager, _require_user_id
|
||||
from services.image_studio import ImageStudioManager, EditStudioRequest
|
||||
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("/edit/process", response_model=EditImageResponse, summary="Process Edit Studio request")
|
||||
async def process_edit_image(
|
||||
request: EditImageRequest,
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
studio_manager: ImageStudioManager = Depends(get_studio_manager),
|
||||
):
|
||||
"""Perform Edit Studio operations such as remove background, inpaint, or recolor."""
|
||||
try:
|
||||
user_id = _require_user_id(current_user, "image editing")
|
||||
logger.info(f"[Edit Image] Request from user {user_id}: operation={request.operation}")
|
||||
|
||||
edit_request = EditStudioRequest(
|
||||
image_base64=request.image_base64,
|
||||
operation=request.operation,
|
||||
prompt=request.prompt,
|
||||
negative_prompt=request.negative_prompt,
|
||||
mask_base64=request.mask_base64,
|
||||
search_prompt=request.search_prompt,
|
||||
select_prompt=request.select_prompt,
|
||||
background_image_base64=request.background_image_base64,
|
||||
lighting_image_base64=request.lighting_image_base64,
|
||||
expand_left=request.expand_left,
|
||||
expand_right=request.expand_right,
|
||||
expand_up=request.expand_up,
|
||||
expand_down=request.expand_down,
|
||||
provider=request.provider,
|
||||
model=request.model,
|
||||
style_preset=request.style_preset,
|
||||
guidance_scale=request.guidance_scale,
|
||||
steps=request.steps,
|
||||
seed=request.seed,
|
||||
output_format=request.output_format,
|
||||
options=request.options or {},
|
||||
)
|
||||
|
||||
result = await studio_manager.edit_image(edit_request, user_id=user_id)
|
||||
return EditImageResponse(**result)
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"[Edit Image] ❌ Error: {str(e)}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Image editing failed: {e}")
|
||||
|
||||
|
||||
@router.get("/edit/operations", response_model=EditOperationsResponse, summary="List Edit Studio operations")
|
||||
async def get_edit_operations(
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
studio_manager: ImageStudioManager = Depends(get_studio_manager),
|
||||
):
|
||||
"""Return metadata for supported Edit Studio operations."""
|
||||
try:
|
||||
operations = studio_manager.get_edit_operations()
|
||||
return EditOperationsResponse(operations=operations)
|
||||
except Exception as e:
|
||||
logger.error(f"[Edit Operations] ❌ Error: {str(e)}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail="Failed to load edit operations")
|
||||
|
||||
|
||||
@router.get("/edit/models", response_model=EditModelsResponse, summary="List available editing models")
|
||||
async def get_edit_models(
|
||||
operation: Optional[str] = None,
|
||||
tier: Optional[str] = None,
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
studio_manager: ImageStudioManager = Depends(get_studio_manager),
|
||||
):
|
||||
"""Get available WaveSpeed editing models with metadata.
|
||||
|
||||
Query Parameters:
|
||||
- operation: Filter by operation type (e.g., "general_edit")
|
||||
- tier: Filter by tier ("budget", "mid", "premium")
|
||||
"""
|
||||
try:
|
||||
result = studio_manager.get_edit_models(operation=operation, tier=tier)
|
||||
return EditModelsResponse(**result)
|
||||
except Exception as e:
|
||||
logger.error(f"[Edit Models] ❌ Error: {str(e)}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail="Failed to load editing models")
|
||||
|
||||
|
||||
@router.post("/edit/recommend", response_model=EditModelRecommendationResponse, summary="Get model recommendation")
|
||||
async def recommend_edit_model(
|
||||
request: EditModelRecommendationRequest,
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
studio_manager: ImageStudioManager = Depends(get_studio_manager),
|
||||
):
|
||||
"""Get recommended editing model based on operation, image resolution, and user preferences.
|
||||
|
||||
Auto-detects best model when user doesn't specify one.
|
||||
"""
|
||||
try:
|
||||
user_tier = request.user_tier
|
||||
if not user_tier and current_user:
|
||||
user_tier = current_user.get("tier") or current_user.get("subscription_tier")
|
||||
|
||||
result = studio_manager.recommend_edit_model(
|
||||
operation=request.operation,
|
||||
image_resolution=request.image_resolution,
|
||||
user_tier=user_tier,
|
||||
preferences=request.preferences,
|
||||
)
|
||||
return EditModelRecommendationResponse(**result)
|
||||
except Exception as e:
|
||||
logger.error(f"[Edit Recommend] ❌ Error: {str(e)}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Failed to get recommendation: {e}")
|
||||
89
backend/routers/image_studio/face_swap.py
Normal file
89
backend/routers/image_studio/face_swap.py
Normal file
@@ -0,0 +1,89 @@
|
||||
"""Face Swap Studio endpoints."""
|
||||
|
||||
from typing import Dict, Any, Optional
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
|
||||
from .models import (
|
||||
FaceSwapRequest, FaceSwapResponse, FaceSwapModelsResponse,
|
||||
FaceSwapModelRecommendationRequest, FaceSwapModelRecommendationResponse,
|
||||
)
|
||||
from .deps import get_studio_manager, _require_user_id
|
||||
from services.image_studio import ImageStudioManager
|
||||
from services.image_studio.face_swap_service import FaceSwapStudioRequest
|
||||
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("/face-swap/process", response_model=FaceSwapResponse, summary="Process Face Swap")
|
||||
async def process_face_swap(
|
||||
request: FaceSwapRequest,
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
studio_manager: ImageStudioManager = Depends(get_studio_manager),
|
||||
):
|
||||
"""Process face swap request with auto-detection and model selection."""
|
||||
try:
|
||||
user_id = _require_user_id(current_user, "face swap")
|
||||
face_swap_request = FaceSwapStudioRequest(
|
||||
base_image_base64=request.base_image_base64,
|
||||
face_image_base64=request.face_image_base64,
|
||||
model=request.model,
|
||||
target_face_index=request.target_face_index,
|
||||
target_gender=request.target_gender,
|
||||
options=request.options,
|
||||
)
|
||||
result = await studio_manager.face_swap(face_swap_request, user_id=user_id)
|
||||
return FaceSwapResponse(**result)
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"[Face Swap] ❌ Error: {e}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Face swap failed: {e}")
|
||||
|
||||
|
||||
@router.get("/face-swap/models", response_model=FaceSwapModelsResponse, summary="List available face swap models")
|
||||
async def get_face_swap_models(
|
||||
tier: Optional[str] = None,
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
studio_manager: ImageStudioManager = Depends(get_studio_manager),
|
||||
):
|
||||
"""Get available WaveSpeed face swap models with metadata.
|
||||
|
||||
Query Parameters:
|
||||
- tier: Filter by tier ("budget", "mid", "premium")
|
||||
"""
|
||||
try:
|
||||
result = studio_manager.get_face_swap_models(tier=tier)
|
||||
return FaceSwapModelsResponse(**result)
|
||||
except Exception as e:
|
||||
logger.error(f"[Face Swap Models] ❌ Error: {str(e)}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail="Failed to load face swap models")
|
||||
|
||||
|
||||
@router.post("/face-swap/recommend", response_model=FaceSwapModelRecommendationResponse, summary="Get face swap model recommendation")
|
||||
async def recommend_face_swap_model(
|
||||
request: FaceSwapModelRecommendationRequest,
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
studio_manager: ImageStudioManager = Depends(get_studio_manager),
|
||||
):
|
||||
"""Get recommended face swap model based on image resolutions and user preferences.
|
||||
|
||||
Auto-detects best model when user doesn't specify one.
|
||||
"""
|
||||
try:
|
||||
user_tier = request.user_tier
|
||||
if not user_tier and current_user:
|
||||
user_tier = current_user.get("tier") or current_user.get("subscription_tier")
|
||||
|
||||
result = studio_manager.recommend_face_swap_model(
|
||||
base_image_resolution=request.base_image_resolution,
|
||||
face_image_resolution=request.face_image_resolution,
|
||||
user_tier=user_tier,
|
||||
preferences=request.preferences,
|
||||
)
|
||||
return FaceSwapModelRecommendationResponse(**result)
|
||||
except Exception as e:
|
||||
logger.error(f"[Face Swap Recommend] ❌ Error: {str(e)}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Failed to get recommendation: {e}")
|
||||
@@ -336,193 +336,7 @@ async def estimate_cost(
|
||||
raise HTTPException(status_code=500, detail=str(e))
|
||||
|
||||
|
||||
# ====================
|
||||
# EDIT STUDIO ENDPOINTS
|
||||
# ====================
|
||||
|
||||
@router.post("/edit/process", response_model=EditImageResponse, summary="Process Edit Studio request")
|
||||
async def process_edit_image(
|
||||
request: EditImageRequest,
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
studio_manager: ImageStudioManager = Depends(get_studio_manager),
|
||||
):
|
||||
"""Perform Edit Studio operations such as remove background, inpaint, or recolor."""
|
||||
try:
|
||||
user_id = _require_user_id(current_user, "image editing")
|
||||
logger.info(f"[Edit Image] Request from user {user_id}: operation={request.operation}")
|
||||
|
||||
edit_request = EditStudioRequest(
|
||||
image_base64=request.image_base64,
|
||||
operation=request.operation,
|
||||
prompt=request.prompt,
|
||||
negative_prompt=request.negative_prompt,
|
||||
mask_base64=request.mask_base64,
|
||||
search_prompt=request.search_prompt,
|
||||
select_prompt=request.select_prompt,
|
||||
background_image_base64=request.background_image_base64,
|
||||
lighting_image_base64=request.lighting_image_base64,
|
||||
expand_left=request.expand_left,
|
||||
expand_right=request.expand_right,
|
||||
expand_up=request.expand_up,
|
||||
expand_down=request.expand_down,
|
||||
provider=request.provider,
|
||||
model=request.model,
|
||||
style_preset=request.style_preset,
|
||||
guidance_scale=request.guidance_scale,
|
||||
steps=request.steps,
|
||||
seed=request.seed,
|
||||
output_format=request.output_format,
|
||||
options=request.options or {},
|
||||
)
|
||||
|
||||
result = await studio_manager.edit_image(edit_request, user_id=user_id)
|
||||
return EditImageResponse(**result)
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"[Edit Image] ❌ Error: {str(e)}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Image editing failed: {e}")
|
||||
|
||||
|
||||
@router.get("/edit/operations", response_model=EditOperationsResponse, summary="List Edit Studio operations")
|
||||
async def get_edit_operations(
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
studio_manager: ImageStudioManager = Depends(get_studio_manager),
|
||||
):
|
||||
"""Return metadata for supported Edit Studio operations."""
|
||||
try:
|
||||
operations = studio_manager.get_edit_operations()
|
||||
return EditOperationsResponse(operations=operations)
|
||||
except Exception as e:
|
||||
logger.error(f"[Edit Operations] ❌ Error: {str(e)}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail="Failed to load edit operations")
|
||||
|
||||
|
||||
@router.get("/edit/models", response_model=EditModelsResponse, summary="List available editing models")
|
||||
async def get_edit_models(
|
||||
operation: Optional[str] = None,
|
||||
tier: Optional[str] = None,
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
studio_manager: ImageStudioManager = Depends(get_studio_manager),
|
||||
):
|
||||
"""Get available WaveSpeed editing models with metadata.
|
||||
|
||||
Query Parameters:
|
||||
- operation: Filter by operation type (e.g., "general_edit")
|
||||
- tier: Filter by tier ("budget", "mid", "premium")
|
||||
"""
|
||||
try:
|
||||
result = studio_manager.get_edit_models(operation=operation, tier=tier)
|
||||
return EditModelsResponse(**result)
|
||||
except Exception as e:
|
||||
logger.error(f"[Edit Models] ❌ Error: {str(e)}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail="Failed to load editing models")
|
||||
|
||||
|
||||
@router.post("/edit/recommend", response_model=EditModelRecommendationResponse, summary="Get model recommendation")
|
||||
async def recommend_edit_model(
|
||||
request: EditModelRecommendationRequest,
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
studio_manager: ImageStudioManager = Depends(get_studio_manager),
|
||||
):
|
||||
"""Get recommended editing model based on operation, image resolution, and user preferences.
|
||||
|
||||
Auto-detects best model when user doesn't specify one.
|
||||
"""
|
||||
try:
|
||||
# Get user tier from current_user if available
|
||||
user_tier = request.user_tier
|
||||
if not user_tier and current_user:
|
||||
# Try to extract from user data (adjust based on your user model)
|
||||
user_tier = current_user.get("tier") or current_user.get("subscription_tier")
|
||||
|
||||
result = studio_manager.recommend_edit_model(
|
||||
operation=request.operation,
|
||||
image_resolution=request.image_resolution,
|
||||
user_tier=user_tier,
|
||||
preferences=request.preferences,
|
||||
)
|
||||
return EditModelRecommendationResponse(**result)
|
||||
except Exception as e:
|
||||
logger.error(f"[Edit Recommend] ❌ Error: {str(e)}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Failed to get recommendation: {e}")
|
||||
|
||||
|
||||
# ====================
|
||||
# FACE SWAP STUDIO ENDPOINTS
|
||||
# ====================
|
||||
|
||||
@router.post("/face-swap/process", response_model=FaceSwapResponse, summary="Process Face Swap")
|
||||
async def process_face_swap(
|
||||
request: FaceSwapRequest,
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
studio_manager: ImageStudioManager = Depends(get_studio_manager),
|
||||
):
|
||||
"""Process face swap request with auto-detection and model selection."""
|
||||
try:
|
||||
user_id = _require_user_id(current_user, "face swap")
|
||||
face_swap_request = FaceSwapStudioRequest(
|
||||
base_image_base64=request.base_image_base64,
|
||||
face_image_base64=request.face_image_base64,
|
||||
model=request.model,
|
||||
target_face_index=request.target_face_index,
|
||||
target_gender=request.target_gender,
|
||||
options=request.options,
|
||||
)
|
||||
result = await studio_manager.face_swap(face_swap_request, user_id=user_id)
|
||||
return FaceSwapResponse(**result)
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"[Face Swap] ❌ Error: {e}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Face swap failed: {e}")
|
||||
|
||||
|
||||
@router.get("/face-swap/models", response_model=FaceSwapModelsResponse, summary="List available face swap models")
|
||||
async def get_face_swap_models(
|
||||
tier: Optional[str] = None,
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
studio_manager: ImageStudioManager = Depends(get_studio_manager),
|
||||
):
|
||||
"""Get available WaveSpeed face swap models with metadata.
|
||||
|
||||
Query Parameters:
|
||||
- tier: Filter by tier ("budget", "mid", "premium")
|
||||
"""
|
||||
try:
|
||||
result = studio_manager.get_face_swap_models(tier=tier)
|
||||
return FaceSwapModelsResponse(**result)
|
||||
except Exception as e:
|
||||
logger.error(f"[Face Swap Models] ❌ Error: {str(e)}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail="Failed to load face swap models")
|
||||
|
||||
|
||||
@router.post("/face-swap/recommend", response_model=FaceSwapModelRecommendationResponse, summary="Get face swap model recommendation")
|
||||
async def recommend_face_swap_model(
|
||||
request: FaceSwapModelRecommendationRequest,
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
studio_manager: ImageStudioManager = Depends(get_studio_manager),
|
||||
):
|
||||
"""Get recommended face swap model based on image resolutions and user preferences.
|
||||
|
||||
Auto-detects best model when user doesn't specify one.
|
||||
"""
|
||||
try:
|
||||
# Get user tier from current_user if available
|
||||
user_tier = request.user_tier
|
||||
if not user_tier and current_user:
|
||||
user_tier = current_user.get("tier") or current_user.get("subscription_tier")
|
||||
|
||||
result = studio_manager.recommend_face_swap_model(
|
||||
base_image_resolution=request.base_image_resolution,
|
||||
face_image_resolution=request.face_image_resolution,
|
||||
user_tier=user_tier,
|
||||
preferences=request.preferences,
|
||||
)
|
||||
return FaceSwapModelRecommendationResponse(**result)
|
||||
except Exception as e:
|
||||
logger.error(f"[Face Swap Recommend] ❌ Error: {str(e)}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Failed to get recommendation: {e}")
|
||||
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user