133 lines
5.0 KiB
Python
133 lines
5.0 KiB
Python
import logging
|
|
from typing import Optional, List, Dict, Any
|
|
from fastapi import APIRouter, Depends, Query, Request, HTTPException, status
|
|
|
|
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
|
|
from src.utils.authorization import (
|
|
create_auth_context,
|
|
log_authorization_context,
|
|
get_team_filter,
|
|
AuthorizationError
|
|
)
|
|
from src.api.v1.error_handlers import handle_service_error
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(tags=["Search"], prefix="/search")
|
|
|
|
@router.get("", response_model=SearchResponse)
|
|
async def search_images(
|
|
request: Request,
|
|
search_service: SearchServiceDep,
|
|
q: str = Query(..., description="Search query for semantic image search"),
|
|
skip: int = Query(0, ge=0, description="Number of records to skip for pagination"),
|
|
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 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
|
|
skip: Number of records to skip for pagination (default: 0)
|
|
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
|
|
"""
|
|
auth_context = create_auth_context(
|
|
user=current_user,
|
|
resource_type="image",
|
|
action="search",
|
|
query=q,
|
|
skip=skip,
|
|
limit=limit,
|
|
similarity_threshold=similarity_threshold,
|
|
collection_id=collection_id,
|
|
team_filter=get_team_filter(current_user),
|
|
path=request.url.path,
|
|
method=request.method
|
|
)
|
|
log_authorization_context(auth_context, success=True)
|
|
|
|
try:
|
|
response = await search_service.search_images(
|
|
query=q,
|
|
user=current_user,
|
|
request=request,
|
|
skip=skip,
|
|
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 Exception as e:
|
|
raise handle_service_error(e, "image search")
|
|
|
|
@router.post("", response_model=SearchResponse)
|
|
async def search_images_advanced(
|
|
search_request: SearchRequest,
|
|
request: Request,
|
|
search_service: SearchServiceDep,
|
|
current_user: UserModel = Depends(get_current_user)
|
|
):
|
|
"""
|
|
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
|
|
"""
|
|
auth_context = create_auth_context(
|
|
user=current_user,
|
|
resource_type="image",
|
|
action="advanced_search",
|
|
search_request=search_request.dict(),
|
|
team_filter=get_team_filter(current_user),
|
|
path=request.url.path,
|
|
method=request.method
|
|
)
|
|
log_authorization_context(auth_context, success=True)
|
|
|
|
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 Exception as e:
|
|
raise handle_service_error(e, "advanced image search")
|