""" Search service for ThrillWiki entities. Provides full-text search capabilities with PostgreSQL and fallback for SQLite. - PostgreSQL: Uses SearchVector, SearchQuery, SearchRank for full-text search - SQLite: Falls back to case-insensitive LIKE queries """ from typing import List, Optional, Dict, Any from django.db.models import Q, QuerySet, Value, CharField, F from django.db.models.functions import Concat from django.conf import settings # Conditionally import PostgreSQL search features _using_postgis = 'postgis' in settings.DATABASES['default']['ENGINE'] if _using_postgis: from django.contrib.postgres.search import SearchVector, SearchQuery, SearchRank, TrigramSimilarity from django.contrib.postgres.aggregates import StringAgg class SearchService: """Service for searching across all entity types.""" def __init__(self): self.using_postgres = _using_postgis def search_all( self, query: str, entity_types: Optional[List[str]] = None, limit: int = 20 ) -> Dict[str, Any]: """ Search across all entity types. Args: query: Search query string entity_types: Optional list to filter by entity types limit: Maximum results per entity type Returns: Dictionary with results grouped by entity type """ results = {} # Default to all entity types if not specified if not entity_types: entity_types = ['company', 'ride_model', 'park', 'ride'] if 'company' in entity_types: results['companies'] = list(self.search_companies(query, limit=limit)) if 'ride_model' in entity_types: results['ride_models'] = list(self.search_ride_models(query, limit=limit)) if 'park' in entity_types: results['parks'] = list(self.search_parks(query, limit=limit)) if 'ride' in entity_types: results['rides'] = list(self.search_rides(query, limit=limit)) return results def search_companies( self, query: str, filters: Optional[Dict[str, Any]] = None, limit: int = 20 ) -> QuerySet: """ Search companies with full-text search. Args: query: Search query string filters: Optional filters (company_types, founded_after, etc.) limit: Maximum number of results Returns: QuerySet of Company objects """ from apps.entities.models import Company if self.using_postgres: # PostgreSQL full-text search using pre-computed search_vector search_query = SearchQuery(query, search_type='websearch') results = Company.objects.annotate( rank=SearchRank(F('search_vector'), search_query) ).filter(search_vector=search_query).order_by('-rank') else: # SQLite fallback using LIKE results = Company.objects.filter( Q(name__icontains=query) | Q(description__icontains=query) ).order_by('name') # Apply additional filters if filters: if filters.get('company_types'): # Filter by company types (stored in JSONField) results = results.filter( company_types__contains=filters['company_types'] ) if filters.get('founded_after'): results = results.filter(founded_date__gte=filters['founded_after']) if filters.get('founded_before'): results = results.filter(founded_date__lte=filters['founded_before']) return results[:limit] def search_ride_models( self, query: str, filters: Optional[Dict[str, Any]] = None, limit: int = 20 ) -> QuerySet: """ Search ride models with full-text search. Args: query: Search query string filters: Optional filters (manufacturer_id, model_type, etc.) limit: Maximum number of results Returns: QuerySet of RideModel objects """ from apps.entities.models import RideModel if self.using_postgres: # PostgreSQL full-text search using pre-computed search_vector search_query = SearchQuery(query, search_type='websearch') results = RideModel.objects.select_related('manufacturer').annotate( rank=SearchRank(F('search_vector'), search_query) ).filter(search_vector=search_query).order_by('-rank') else: # SQLite fallback using LIKE results = RideModel.objects.select_related('manufacturer').filter( Q(name__icontains=query) | Q(manufacturer__name__icontains=query) | Q(description__icontains=query) ).order_by('manufacturer__name', 'name') # Apply additional filters if filters: if filters.get('manufacturer_id'): results = results.filter(manufacturer_id=filters['manufacturer_id']) if filters.get('model_type'): results = results.filter(model_type=filters['model_type']) return results[:limit] def search_parks( self, query: str, filters: Optional[Dict[str, Any]] = None, limit: int = 20 ) -> QuerySet: """ Search parks with full-text search and location filtering. Args: query: Search query string filters: Optional filters (status, park_type, location, radius, etc.) limit: Maximum number of results Returns: QuerySet of Park objects """ from apps.entities.models import Park if self.using_postgres: # PostgreSQL full-text search using pre-computed search_vector search_query = SearchQuery(query, search_type='websearch') results = Park.objects.annotate( rank=SearchRank(F('search_vector'), search_query) ).filter(search_vector=search_query).order_by('-rank') else: # SQLite fallback using LIKE results = Park.objects.filter( Q(name__icontains=query) | Q(description__icontains=query) ).order_by('name') # Apply additional filters if filters: if filters.get('status'): results = results.filter(status=filters['status']) if filters.get('park_type'): results = results.filter(park_type=filters['park_type']) if filters.get('operator_id'): results = results.filter(operator_id=filters['operator_id']) if filters.get('opening_after'): results = results.filter(opening_date__gte=filters['opening_after']) if filters.get('opening_before'): results = results.filter(opening_date__lte=filters['opening_before']) # Location-based filtering (PostGIS only) if self.using_postgres and filters.get('location') and filters.get('radius'): from django.contrib.gis.geos import Point from django.contrib.gis.measure import D longitude, latitude = filters['location'] point = Point(longitude, latitude, srid=4326) radius_km = filters['radius'] # Use distance filter results = results.filter( location_point__distance_lte=(point, D(km=radius_km)) ).annotate( distance=F('location_point__distance') ).order_by('distance') return results[:limit] def search_rides( self, query: str, filters: Optional[Dict[str, Any]] = None, limit: int = 20 ) -> QuerySet: """ Search rides with full-text search. Args: query: Search query string filters: Optional filters (park_id, manufacturer_id, status, etc.) limit: Maximum number of results Returns: QuerySet of Ride objects """ from apps.entities.models import Ride if self.using_postgres: # PostgreSQL full-text search using pre-computed search_vector search_query = SearchQuery(query, search_type='websearch') results = Ride.objects.select_related('park', 'manufacturer', 'model').annotate( rank=SearchRank(F('search_vector'), search_query) ).filter(search_vector=search_query).order_by('-rank') else: # SQLite fallback using LIKE results = Ride.objects.select_related('park', 'manufacturer', 'model').filter( Q(name__icontains=query) | Q(park__name__icontains=query) | Q(manufacturer__name__icontains=query) | Q(description__icontains=query) ).order_by('park__name', 'name') # Apply additional filters if filters: if filters.get('park_id'): results = results.filter(park_id=filters['park_id']) if filters.get('manufacturer_id'): results = results.filter(manufacturer_id=filters['manufacturer_id']) if filters.get('model_id'): results = results.filter(model_id=filters['model_id']) if filters.get('status'): results = results.filter(status=filters['status']) if filters.get('ride_category'): results = results.filter(ride_category=filters['ride_category']) if filters.get('is_coaster') is not None: results = results.filter(is_coaster=filters['is_coaster']) if filters.get('opening_after'): results = results.filter(opening_date__gte=filters['opening_after']) if filters.get('opening_before'): results = results.filter(opening_date__lte=filters['opening_before']) # Height/speed filters if filters.get('min_height'): results = results.filter(height__gte=filters['min_height']) if filters.get('max_height'): results = results.filter(height__lte=filters['max_height']) if filters.get('min_speed'): results = results.filter(speed__gte=filters['min_speed']) if filters.get('max_speed'): results = results.filter(speed__lte=filters['max_speed']) return results[:limit] def autocomplete( self, query: str, entity_type: Optional[str] = None, limit: int = 10 ) -> List[Dict[str, Any]]: """ Get autocomplete suggestions for search. Args: query: Partial search query entity_type: Optional specific entity type limit: Maximum number of suggestions Returns: List of suggestion dictionaries with name and entity_type """ suggestions = [] if not query or len(query) < 2: return suggestions # Search in names only for autocomplete if entity_type == 'company' or not entity_type: from apps.entities.models import Company companies = Company.objects.filter( name__istartswith=query ).values('id', 'name', 'slug')[:limit] for company in companies: suggestions.append({ 'id': company['id'], 'name': company['name'], 'slug': company['slug'], 'entity_type': 'company' }) if entity_type == 'park' or not entity_type: from apps.entities.models import Park parks = Park.objects.filter( name__istartswith=query ).values('id', 'name', 'slug')[:limit] for park in parks: suggestions.append({ 'id': park['id'], 'name': park['name'], 'slug': park['slug'], 'entity_type': 'park' }) if entity_type == 'ride' or not entity_type: from apps.entities.models import Ride rides = Ride.objects.select_related('park').filter( name__istartswith=query ).values('id', 'name', 'slug', 'park__name')[:limit] for ride in rides: suggestions.append({ 'id': ride['id'], 'name': ride['name'], 'slug': ride['slug'], 'park_name': ride['park__name'], 'entity_type': 'ride' }) if entity_type == 'ride_model' or not entity_type: from apps.entities.models import RideModel models = RideModel.objects.select_related('manufacturer').filter( name__istartswith=query ).values('id', 'name', 'slug', 'manufacturer__name')[:limit] for model in models: suggestions.append({ 'id': model['id'], 'name': model['name'], 'slug': model['slug'], 'manufacturer_name': model['manufacturer__name'], 'entity_type': 'ride_model' }) # Sort by relevance (exact matches first, then alphabetically) suggestions.sort(key=lambda x: ( not x['name'].lower().startswith(query.lower()), x['name'].lower() )) return suggestions[:limit]