136 lines
5.3 KiB
Python
136 lines
5.3 KiB
Python
import logging
|
|
from typing import Optional, List, Dict, Any
|
|
from fastapi import APIRouter, Depends, Query, Request, HTTPException
|
|
|
|
from src.auth.security import get_current_user
|
|
from src.dependencies import SearchServiceDep
|
|
from src.models.user import UserModel
|
|
from src.schemas.search import SearchResponse, SearchRequest
|
|
from src.utils.logging import log_request
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(tags=["Search"], prefix="/search")
|
|
|
|
@router.get("", response_model=SearchResponse)
|
|
async def search_images(
|
|
request: Request,
|
|
q: str = Query(..., description="Search query for semantic image search"),
|
|
limit: int = Query(10, ge=1, le=50, description="Number of results to return (1-50)"),
|
|
similarity_threshold: float = Query(0.65, ge=0.0, le=1.0, description="Similarity threshold (0.0-1.0)"),
|
|
collection_id: Optional[str] = Query(None, description="Filter results by collection ID"),
|
|
current_user: UserModel = Depends(get_current_user),
|
|
search_service: SearchServiceDep = Depends()
|
|
):
|
|
"""
|
|
Search for images using semantic similarity
|
|
|
|
Performs a semantic search across images using AI-powered similarity matching.
|
|
Regular users can only search within their team's images, while admin users
|
|
can search across all teams.
|
|
|
|
Args:
|
|
q: The search query text to find similar images
|
|
limit: Maximum number of results to return (1-50, default: 10)
|
|
similarity_threshold: Minimum similarity score (0.0-1.0, default: 0.65)
|
|
collection_id: Optional filter to search within a specific collection
|
|
current_user: The authenticated user performing the search
|
|
search_service: Injected search service
|
|
|
|
Returns:
|
|
SearchResponse: List of matching images with similarity scores
|
|
|
|
Raises:
|
|
400: Invalid search parameters or query format
|
|
500: Search service errors
|
|
"""
|
|
log_request(
|
|
{
|
|
"path": request.url.path,
|
|
"method": request.method,
|
|
"query": q,
|
|
"limit": limit,
|
|
"similarity_threshold": similarity_threshold,
|
|
"collection_id": collection_id,
|
|
"is_admin": current_user.is_admin
|
|
},
|
|
user_id=str(current_user.id),
|
|
team_id=str(current_user.team_id)
|
|
)
|
|
|
|
try:
|
|
response = await search_service.search_images(
|
|
query=q,
|
|
user=current_user,
|
|
request=request,
|
|
limit=limit,
|
|
similarity_threshold=similarity_threshold,
|
|
collection_id=collection_id
|
|
)
|
|
logger.info(f"Search completed: '{q}' returned {len(response.results)} results for user {current_user.id}")
|
|
return response
|
|
except ValueError as e:
|
|
logger.warning(f"Invalid search parameters: {e}")
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
except RuntimeError as e:
|
|
logger.error(f"Search service error: {e}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
except Exception as e:
|
|
logger.error(f"Unexpected error in search: {e}")
|
|
raise HTTPException(status_code=500, detail="Internal server error")
|
|
|
|
@router.post("", response_model=SearchResponse)
|
|
async def search_images_advanced(
|
|
search_request: SearchRequest,
|
|
request: Request,
|
|
current_user: UserModel = Depends(get_current_user),
|
|
search_service: SearchServiceDep = Depends()
|
|
):
|
|
"""
|
|
Advanced search for images with extended options
|
|
|
|
Provides advanced search capabilities with more filtering and configuration
|
|
options than the basic search endpoint. Supports complex queries and
|
|
multiple search parameters.
|
|
|
|
Args:
|
|
search_request: Advanced search request with detailed parameters
|
|
current_user: The authenticated user performing the search
|
|
search_service: Injected search service
|
|
|
|
Returns:
|
|
SearchResponse: List of matching images with similarity scores and metadata
|
|
|
|
Raises:
|
|
400: Invalid search request or validation errors
|
|
500: Search service errors
|
|
"""
|
|
log_request(
|
|
{
|
|
"path": request.url.path,
|
|
"method": request.method,
|
|
"search_request": search_request.dict(),
|
|
"is_admin": current_user.is_admin
|
|
},
|
|
user_id=str(current_user.id),
|
|
team_id=str(current_user.team_id)
|
|
)
|
|
|
|
try:
|
|
response = await search_service.search_images_advanced(
|
|
search_request=search_request,
|
|
user=current_user,
|
|
request=request
|
|
)
|
|
logger.info(f"Advanced search completed: '{search_request.query}' returned {len(response.results)} results for user {current_user.id}")
|
|
return response
|
|
except ValueError as e:
|
|
logger.warning(f"Invalid advanced search request: {e}")
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
except RuntimeError as e:
|
|
logger.error(f"Advanced search service error: {e}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
except Exception as e:
|
|
logger.error(f"Unexpected error in advanced search: {e}")
|
|
raise HTTPException(status_code=500, detail="Internal server error")
|