mirror of
https://github.com/pacnpal/thrilltrack-explorer.git
synced 2025-12-20 12:11:17 -05:00
d6ff4cc3a3
- Created a base email template (base.html) for consistent styling across all emails. - Added moderation approval email template (moderation_approved.html) to notify users of approved submissions. - Added moderation rejection email template (moderation_rejected.html) to inform users of required changes for their submissions. - Created password reset email template (password_reset.html) for users requesting to reset their passwords. - Developed a welcome email template (welcome.html) to greet new users and provide account details and tips for using ThrillWiki.
387 lines
14 KiB
Python
387 lines
14 KiB
Python
"""
|
|
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]
|