""" Search and autocomplete endpoints for ThrillWiki API. Provides full-text search and filtering across all entity types. """ from typing import List, Optional from uuid import UUID from datetime import date from decimal import Decimal from django.http import HttpRequest from ninja import Router, Query from apps.entities.search import SearchService from apps.users.permissions import jwt_auth from api.v1.schemas import ( GlobalSearchResponse, CompanySearchResult, RideModelSearchResult, ParkSearchResult, RideSearchResult, AutocompleteResponse, AutocompleteItem, ErrorResponse, ) router = Router(tags=["Search"]) search_service = SearchService() # ============================================================================ # Helper Functions # ============================================================================ def _company_to_search_result(company) -> CompanySearchResult: """Convert Company model to search result.""" return CompanySearchResult( id=company.id, name=company.name, slug=company.slug, entity_type='company', description=company.description, image_url=company.logo_image_url or None, company_types=company.company_types or [], park_count=company.park_count, ride_count=company.ride_count, ) def _ride_model_to_search_result(model) -> RideModelSearchResult: """Convert RideModel to search result.""" return RideModelSearchResult( id=model.id, name=model.name, slug=model.slug, entity_type='ride_model', description=model.description, image_url=model.image_url or None, manufacturer_name=model.manufacturer.name if model.manufacturer else '', model_type=model.model_type, installation_count=model.installation_count, ) def _park_to_search_result(park) -> ParkSearchResult: """Convert Park model to search result.""" return ParkSearchResult( id=park.id, name=park.name, slug=park.slug, entity_type='park', description=park.description, image_url=park.banner_image_url or park.logo_image_url or None, park_type=park.park_type, status=park.status, operator_name=park.operator.name if park.operator else None, ride_count=park.ride_count, coaster_count=park.coaster_count, coordinates=park.coordinates, ) def _ride_to_search_result(ride) -> RideSearchResult: """Convert Ride model to search result.""" return RideSearchResult( id=ride.id, name=ride.name, slug=ride.slug, entity_type='ride', description=ride.description, image_url=ride.image_url or None, park_name=ride.park.name if ride.park else '', park_slug=ride.park.slug if ride.park else '', manufacturer_name=ride.manufacturer.name if ride.manufacturer else None, ride_category=ride.ride_category, status=ride.status, is_coaster=ride.is_coaster, ) # ============================================================================ # Search Endpoints # ============================================================================ @router.get( "", response={200: GlobalSearchResponse, 400: ErrorResponse}, summary="Global search across all entities" ) def search_all( request: HttpRequest, q: str = Query(..., min_length=2, max_length=200, description="Search query"), entity_types: Optional[List[str]] = Query(None, description="Filter by entity types (company, ride_model, park, ride)"), limit: int = Query(20, ge=1, le=100, description="Maximum results per entity type"), ): """ Search across all entity types with full-text search. - **q**: Search query (minimum 2 characters) - **entity_types**: Optional list of entity types to search (defaults to all) - **limit**: Maximum results per entity type (1-100, default 20) Returns results grouped by entity type. """ try: results = search_service.search_all( query=q, entity_types=entity_types, limit=limit ) # Convert to schema objects response_data = { 'query': q, 'total_results': 0, 'companies': [], 'ride_models': [], 'parks': [], 'rides': [], } if 'companies' in results: response_data['companies'] = [ _company_to_search_result(c) for c in results['companies'] ] response_data['total_results'] += len(response_data['companies']) if 'ride_models' in results: response_data['ride_models'] = [ _ride_model_to_search_result(m) for m in results['ride_models'] ] response_data['total_results'] += len(response_data['ride_models']) if 'parks' in results: response_data['parks'] = [ _park_to_search_result(p) for p in results['parks'] ] response_data['total_results'] += len(response_data['parks']) if 'rides' in results: response_data['rides'] = [ _ride_to_search_result(r) for r in results['rides'] ] response_data['total_results'] += len(response_data['rides']) return GlobalSearchResponse(**response_data) except Exception as e: return 400, ErrorResponse(detail=str(e)) @router.get( "/companies", response={200: List[CompanySearchResult], 400: ErrorResponse}, summary="Search companies" ) def search_companies( request: HttpRequest, q: str = Query(..., min_length=2, max_length=200, description="Search query"), company_types: Optional[List[str]] = Query(None, description="Filter by company types"), founded_after: Optional[date] = Query(None, description="Founded after date"), founded_before: Optional[date] = Query(None, description="Founded before date"), limit: int = Query(20, ge=1, le=100, description="Maximum results"), ): """ Search companies with optional filters. - **q**: Search query - **company_types**: Filter by types (manufacturer, operator, designer, etc.) - **founded_after/before**: Filter by founding date range - **limit**: Maximum results (1-100, default 20) """ try: filters = {} if company_types: filters['company_types'] = company_types if founded_after: filters['founded_after'] = founded_after if founded_before: filters['founded_before'] = founded_before results = search_service.search_companies( query=q, filters=filters if filters else None, limit=limit ) return [_company_to_search_result(c) for c in results] except Exception as e: return 400, ErrorResponse(detail=str(e)) @router.get( "/ride-models", response={200: List[RideModelSearchResult], 400: ErrorResponse}, summary="Search ride models" ) def search_ride_models( request: HttpRequest, q: str = Query(..., min_length=2, max_length=200, description="Search query"), manufacturer_id: Optional[UUID] = Query(None, description="Filter by manufacturer"), model_type: Optional[str] = Query(None, description="Filter by model type"), limit: int = Query(20, ge=1, le=100, description="Maximum results"), ): """ Search ride models with optional filters. - **q**: Search query - **manufacturer_id**: Filter by specific manufacturer - **model_type**: Filter by model type - **limit**: Maximum results (1-100, default 20) """ try: filters = {} if manufacturer_id: filters['manufacturer_id'] = manufacturer_id if model_type: filters['model_type'] = model_type results = search_service.search_ride_models( query=q, filters=filters if filters else None, limit=limit ) return [_ride_model_to_search_result(m) for m in results] except Exception as e: return 400, ErrorResponse(detail=str(e)) @router.get( "/parks", response={200: List[ParkSearchResult], 400: ErrorResponse}, summary="Search parks" ) def search_parks( request: HttpRequest, q: str = Query(..., min_length=2, max_length=200, description="Search query"), status: Optional[str] = Query(None, description="Filter by status"), park_type: Optional[str] = Query(None, description="Filter by park type"), operator_id: Optional[UUID] = Query(None, description="Filter by operator"), opening_after: Optional[date] = Query(None, description="Opened after date"), opening_before: Optional[date] = Query(None, description="Opened before date"), latitude: Optional[float] = Query(None, description="Search center latitude"), longitude: Optional[float] = Query(None, description="Search center longitude"), radius: Optional[float] = Query(None, ge=0, le=500, description="Search radius in km (PostGIS only)"), limit: int = Query(20, ge=1, le=100, description="Maximum results"), ): """ Search parks with optional filters including location-based search. - **q**: Search query - **status**: Filter by operational status - **park_type**: Filter by park type - **operator_id**: Filter by operator company - **opening_after/before**: Filter by opening date range - **latitude/longitude/radius**: Location-based filtering (PostGIS only) - **limit**: Maximum results (1-100, default 20) """ try: filters = {} if status: filters['status'] = status if park_type: filters['park_type'] = park_type if operator_id: filters['operator_id'] = operator_id if opening_after: filters['opening_after'] = opening_after if opening_before: filters['opening_before'] = opening_before # Location-based search (PostGIS only) if latitude is not None and longitude is not None and radius is not None: filters['location'] = (longitude, latitude) filters['radius'] = radius results = search_service.search_parks( query=q, filters=filters if filters else None, limit=limit ) return [_park_to_search_result(p) for p in results] except Exception as e: return 400, ErrorResponse(detail=str(e)) @router.get( "/rides", response={200: List[RideSearchResult], 400: ErrorResponse}, summary="Search rides" ) def search_rides( request: HttpRequest, q: str = Query(..., min_length=2, max_length=200, description="Search query"), park_id: Optional[UUID] = Query(None, description="Filter by park"), manufacturer_id: Optional[UUID] = Query(None, description="Filter by manufacturer"), model_id: Optional[UUID] = Query(None, description="Filter by model"), status: Optional[str] = Query(None, description="Filter by status"), ride_category: Optional[str] = Query(None, description="Filter by category"), is_coaster: Optional[bool] = Query(None, description="Filter coasters only"), opening_after: Optional[date] = Query(None, description="Opened after date"), opening_before: Optional[date] = Query(None, description="Opened before date"), min_height: Optional[Decimal] = Query(None, description="Minimum height in feet"), max_height: Optional[Decimal] = Query(None, description="Maximum height in feet"), min_speed: Optional[Decimal] = Query(None, description="Minimum speed in mph"), max_speed: Optional[Decimal] = Query(None, description="Maximum speed in mph"), limit: int = Query(20, ge=1, le=100, description="Maximum results"), ): """ Search rides with extensive filtering options. - **q**: Search query - **park_id**: Filter by specific park - **manufacturer_id**: Filter by manufacturer - **model_id**: Filter by specific ride model - **status**: Filter by operational status - **ride_category**: Filter by category (roller_coaster, flat_ride, etc.) - **is_coaster**: Filter to show only coasters - **opening_after/before**: Filter by opening date range - **min_height/max_height**: Filter by height range (feet) - **min_speed/max_speed**: Filter by speed range (mph) - **limit**: Maximum results (1-100, default 20) """ try: filters = {} if park_id: filters['park_id'] = park_id if manufacturer_id: filters['manufacturer_id'] = manufacturer_id if model_id: filters['model_id'] = model_id if status: filters['status'] = status if ride_category: filters['ride_category'] = ride_category if is_coaster is not None: filters['is_coaster'] = is_coaster if opening_after: filters['opening_after'] = opening_after if opening_before: filters['opening_before'] = opening_before if min_height: filters['min_height'] = min_height if max_height: filters['max_height'] = max_height if min_speed: filters['min_speed'] = min_speed if max_speed: filters['max_speed'] = max_speed results = search_service.search_rides( query=q, filters=filters if filters else None, limit=limit ) return [_ride_to_search_result(r) for r in results] except Exception as e: return 400, ErrorResponse(detail=str(e)) # ============================================================================ # Autocomplete Endpoint # ============================================================================ @router.get( "/autocomplete", response={200: AutocompleteResponse, 400: ErrorResponse}, summary="Autocomplete suggestions" ) def autocomplete( request: HttpRequest, q: str = Query(..., min_length=2, max_length=100, description="Partial search query"), entity_type: Optional[str] = Query(None, description="Filter by entity type (company, park, ride, ride_model)"), limit: int = Query(10, ge=1, le=20, description="Maximum suggestions"), ): """ Get autocomplete suggestions for search. - **q**: Partial query (minimum 2 characters) - **entity_type**: Optional entity type filter - **limit**: Maximum suggestions (1-20, default 10) Returns quick name-based suggestions for autocomplete UIs. """ try: suggestions = search_service.autocomplete( query=q, entity_type=entity_type, limit=limit ) # Convert to schema objects items = [ AutocompleteItem( id=s['id'], name=s['name'], slug=s['slug'], entity_type=s['entity_type'], park_name=s.get('park_name'), manufacturer_name=s.get('manufacturer_name'), ) for s in suggestions ] return AutocompleteResponse( query=q, suggestions=items ) except Exception as e: return 400, ErrorResponse(detail=str(e))