mirror of
https://github.com/pacnpal/thrillwiki_django_no_react.git
synced 2025-12-20 05:51:08 -05:00
e4e36c7899
- Created ParkPhoto and ParkPhotoEvent models in the parks app, including fields for image, caption, alt text, and relationships to the Park model. - Implemented triggers for insert and update operations on ParkPhoto to log changes in ParkPhotoEvent. - Created RidePhoto and RidePhotoEvent models in the rides app, with similar structure and functionality as ParkPhoto. - Added fields for photo type in RidePhoto and implemented corresponding triggers for logging changes. - Established necessary indexes and unique constraints for both models to ensure data integrity and optimize queries.
492 lines
16 KiB
Python
492 lines
16 KiB
Python
"""
|
|
Parks-specific location services with OpenStreetMap integration.
|
|
Handles geocoding, reverse geocoding, and location search for parks.
|
|
"""
|
|
|
|
import requests
|
|
from typing import List, Dict, Any, Optional
|
|
from django.core.cache import cache
|
|
from django.db import transaction
|
|
import logging
|
|
|
|
from ..models import ParkLocation
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ParkLocationService:
|
|
"""
|
|
Location service specifically for parks using OpenStreetMap Nominatim API.
|
|
"""
|
|
|
|
NOMINATIM_BASE_URL = "https://nominatim.openstreetmap.org"
|
|
USER_AGENT = "ThrillWiki/1.0 (https://thrillwiki.com)"
|
|
|
|
@classmethod
|
|
def search_locations(cls, query: str, limit: int = 10) -> Dict[str, Any]:
|
|
"""
|
|
Search for locations using OpenStreetMap Nominatim API.
|
|
Optimized for finding theme parks and amusement parks.
|
|
|
|
Args:
|
|
query: Search query string
|
|
limit: Maximum number of results (default: 10, max: 25)
|
|
|
|
Returns:
|
|
Dictionary with search results
|
|
"""
|
|
if not query.strip():
|
|
return {"count": 0, "results": [], "query": query}
|
|
|
|
# Limit the number of results
|
|
limit = min(limit, 25)
|
|
|
|
# Check cache first
|
|
cache_key = f"park_location_search:{query.lower()}:{limit}"
|
|
cached_result = cache.get(cache_key)
|
|
if cached_result:
|
|
return cached_result
|
|
|
|
try:
|
|
params = {
|
|
"q": query,
|
|
"format": "json",
|
|
"limit": limit,
|
|
"addressdetails": 1,
|
|
"extratags": 1,
|
|
"namedetails": 1,
|
|
"accept-language": "en",
|
|
# Prioritize places that might be parks or entertainment venues
|
|
"featuretype": "settlement,leisure,tourism",
|
|
}
|
|
|
|
headers = {
|
|
"User-Agent": cls.USER_AGENT,
|
|
}
|
|
|
|
response = requests.get(
|
|
f"{cls.NOMINATIM_BASE_URL}/search",
|
|
params=params,
|
|
headers=headers,
|
|
timeout=10,
|
|
)
|
|
response.raise_for_status()
|
|
|
|
osm_results = response.json()
|
|
|
|
# Transform OSM results to our format
|
|
results = []
|
|
for item in osm_results:
|
|
result = cls._transform_osm_result(item)
|
|
if result:
|
|
results.append(result)
|
|
|
|
result_data = {"count": len(results), "results": results, "query": query}
|
|
|
|
# Cache for 1 hour
|
|
cache.set(cache_key, result_data, 3600)
|
|
|
|
return result_data
|
|
|
|
except requests.RequestException as e:
|
|
logger.error(f"Error searching park locations: {str(e)}")
|
|
return {
|
|
"count": 0,
|
|
"results": [],
|
|
"query": query,
|
|
"error": "Location search service temporarily unavailable",
|
|
}
|
|
|
|
@classmethod
|
|
def reverse_geocode(cls, latitude: float, longitude: float) -> Dict[str, Any]:
|
|
"""
|
|
Reverse geocode coordinates to get location information using OSM.
|
|
|
|
Args:
|
|
latitude: Latitude coordinate
|
|
longitude: Longitude coordinate
|
|
|
|
Returns:
|
|
Dictionary with location information
|
|
"""
|
|
# Validate coordinates
|
|
if not (-90 <= latitude <= 90) or not (-180 <= longitude <= 180):
|
|
return {"error": "Invalid coordinates"}
|
|
|
|
# Check cache first
|
|
cache_key = f"park_reverse_geocode:{latitude:.6f}:{longitude:.6f}"
|
|
cached_result = cache.get(cache_key)
|
|
if cached_result:
|
|
return cached_result
|
|
|
|
try:
|
|
params = {
|
|
"lat": latitude,
|
|
"lon": longitude,
|
|
"format": "json",
|
|
"addressdetails": 1,
|
|
"extratags": 1,
|
|
"namedetails": 1,
|
|
"accept-language": "en",
|
|
}
|
|
|
|
headers = {
|
|
"User-Agent": cls.USER_AGENT,
|
|
}
|
|
|
|
response = requests.get(
|
|
f"{cls.NOMINATIM_BASE_URL}/reverse",
|
|
params=params,
|
|
headers=headers,
|
|
timeout=10,
|
|
)
|
|
response.raise_for_status()
|
|
|
|
osm_result = response.json()
|
|
|
|
if "error" in osm_result:
|
|
return {"error": "Location not found"}
|
|
|
|
result = cls._transform_osm_reverse_result(osm_result)
|
|
|
|
# Cache for 24 hours
|
|
cache.set(cache_key, result, 86400)
|
|
|
|
return result
|
|
|
|
except requests.RequestException as e:
|
|
logger.error(f"Error reverse geocoding park location: {str(e)}")
|
|
return {"error": "Reverse geocoding service temporarily unavailable"}
|
|
|
|
@classmethod
|
|
def geocode_address(cls, address: str) -> Dict[str, Any]:
|
|
"""
|
|
Geocode an address to get coordinates using OSM.
|
|
|
|
Args:
|
|
address: Address string to geocode
|
|
|
|
Returns:
|
|
Dictionary with coordinates and location information
|
|
"""
|
|
if not address.strip():
|
|
return {"error": "Address is required"}
|
|
|
|
# Use search_locations for geocoding
|
|
results = cls.search_locations(address, limit=1)
|
|
|
|
if results["count"] > 0:
|
|
return results["results"][0]
|
|
else:
|
|
return {"error": "Address not found"}
|
|
|
|
@classmethod
|
|
def create_park_location(
|
|
cls,
|
|
*,
|
|
park,
|
|
latitude: Optional[float] = None,
|
|
longitude: Optional[float] = None,
|
|
street_address: str = "",
|
|
city: str = "",
|
|
state: str = "",
|
|
country: str = "USA",
|
|
postal_code: str = "",
|
|
highway_exit: str = "",
|
|
parking_notes: str = "",
|
|
seasonal_notes: str = "",
|
|
osm_id: Optional[int] = None,
|
|
osm_type: str = "",
|
|
) -> ParkLocation:
|
|
"""
|
|
Create a location for a park with OSM integration.
|
|
|
|
Args:
|
|
park: Park instance
|
|
latitude: Latitude coordinate
|
|
longitude: Longitude coordinate
|
|
street_address: Street address
|
|
city: City name
|
|
state: State/region name
|
|
country: Country name (default: USA)
|
|
postal_code: Postal/ZIP code
|
|
highway_exit: Highway exit information
|
|
parking_notes: Parking information
|
|
seasonal_notes: Seasonal access notes
|
|
osm_id: OpenStreetMap ID
|
|
osm_type: OpenStreetMap type (node, way, relation)
|
|
|
|
Returns:
|
|
Created ParkLocation instance
|
|
"""
|
|
with transaction.atomic():
|
|
park_location = ParkLocation(
|
|
park=park,
|
|
street_address=street_address,
|
|
city=city,
|
|
state=state,
|
|
country=country,
|
|
postal_code=postal_code,
|
|
highway_exit=highway_exit,
|
|
parking_notes=parking_notes,
|
|
seasonal_notes=seasonal_notes,
|
|
osm_id=osm_id,
|
|
osm_type=osm_type,
|
|
)
|
|
|
|
# Set coordinates if provided
|
|
if latitude is not None and longitude is not None:
|
|
park_location.set_coordinates(latitude, longitude)
|
|
|
|
park_location.full_clean()
|
|
park_location.save()
|
|
|
|
return park_location
|
|
|
|
@classmethod
|
|
def update_park_location(
|
|
cls, park_location: ParkLocation, **updates
|
|
) -> ParkLocation:
|
|
"""
|
|
Update park location with validation.
|
|
|
|
Args:
|
|
park_location: ParkLocation instance to update
|
|
**updates: Fields to update
|
|
|
|
Returns:
|
|
Updated ParkLocation instance
|
|
"""
|
|
with transaction.atomic():
|
|
# Handle coordinates separately
|
|
latitude = updates.pop("latitude", None)
|
|
longitude = updates.pop("longitude", None)
|
|
|
|
# Update regular fields
|
|
for field, value in updates.items():
|
|
if hasattr(park_location, field):
|
|
setattr(park_location, field, value)
|
|
|
|
# Update coordinates if provided
|
|
if latitude is not None and longitude is not None:
|
|
park_location.set_coordinates(latitude, longitude)
|
|
|
|
park_location.full_clean()
|
|
park_location.save()
|
|
|
|
return park_location
|
|
|
|
@classmethod
|
|
def find_nearby_parks(
|
|
cls, latitude: float, longitude: float, radius_km: float = 50
|
|
) -> List[ParkLocation]:
|
|
"""
|
|
Find parks near given coordinates using PostGIS.
|
|
|
|
Args:
|
|
latitude: Center latitude
|
|
longitude: Center longitude
|
|
radius_km: Search radius in kilometers
|
|
|
|
Returns:
|
|
List of nearby ParkLocation instances
|
|
"""
|
|
from django.contrib.gis.geos import Point
|
|
from django.contrib.gis.measure import Distance
|
|
|
|
center_point = Point(longitude, latitude, srid=4326)
|
|
|
|
return list(
|
|
ParkLocation.objects.filter(
|
|
point__distance_lte=(center_point, Distance(km=radius_km))
|
|
)
|
|
.select_related("park", "park__operator")
|
|
.order_by("point__distance")
|
|
)
|
|
|
|
@classmethod
|
|
def enrich_location_from_osm(cls, park_location: ParkLocation) -> ParkLocation:
|
|
"""
|
|
Enrich park location data using OSM reverse geocoding.
|
|
|
|
Args:
|
|
park_location: ParkLocation instance to enrich
|
|
|
|
Returns:
|
|
Updated ParkLocation instance
|
|
"""
|
|
if not park_location.point:
|
|
return park_location
|
|
|
|
# Get detailed location info from OSM
|
|
osm_data = cls.reverse_geocode(park_location.latitude, park_location.longitude)
|
|
|
|
if "error" not in osm_data:
|
|
updates = {}
|
|
|
|
# Update missing address components
|
|
if not park_location.street_address and osm_data.get("street_address"):
|
|
updates["street_address"] = osm_data["street_address"]
|
|
if not park_location.city and osm_data.get("city"):
|
|
updates["city"] = osm_data["city"]
|
|
if not park_location.state and osm_data.get("state"):
|
|
updates["state"] = osm_data["state"]
|
|
if not park_location.country and osm_data.get("country"):
|
|
updates["country"] = osm_data["country"]
|
|
if not park_location.postal_code and osm_data.get("postal_code"):
|
|
updates["postal_code"] = osm_data["postal_code"]
|
|
|
|
# Update OSM metadata
|
|
if osm_data.get("osm_id"):
|
|
updates["osm_id"] = osm_data["osm_id"]
|
|
if osm_data.get("osm_type"):
|
|
updates["osm_type"] = osm_data["osm_type"]
|
|
|
|
if updates:
|
|
return cls.update_park_location(park_location, **updates)
|
|
|
|
return park_location
|
|
|
|
@classmethod
|
|
def _transform_osm_result(
|
|
cls, osm_item: Dict[str, Any]
|
|
) -> Optional[Dict[str, Any]]:
|
|
"""Transform OSM search result to our standard format."""
|
|
try:
|
|
address = osm_item.get("address", {})
|
|
|
|
# Extract address components
|
|
street_number = address.get("house_number", "")
|
|
street_name = address.get("road", "")
|
|
street_address = f"{street_number} {street_name}".strip()
|
|
|
|
city = (
|
|
address.get("city")
|
|
or address.get("town")
|
|
or address.get("village")
|
|
or address.get("municipality")
|
|
or ""
|
|
)
|
|
|
|
state = (
|
|
address.get("state")
|
|
or address.get("province")
|
|
or address.get("region")
|
|
or ""
|
|
)
|
|
|
|
country = address.get("country", "")
|
|
postal_code = address.get("postcode", "")
|
|
|
|
# Build formatted address
|
|
address_parts = []
|
|
if street_address:
|
|
address_parts.append(street_address)
|
|
if city:
|
|
address_parts.append(city)
|
|
if state:
|
|
address_parts.append(state)
|
|
if postal_code:
|
|
address_parts.append(postal_code)
|
|
if country:
|
|
address_parts.append(country)
|
|
|
|
formatted_address = ", ".join(address_parts)
|
|
|
|
# Check if this might be a theme park or entertainment venue
|
|
place_type = osm_item.get("type", "").lower()
|
|
extratags = osm_item.get("extratags", {})
|
|
|
|
is_park_related = any(
|
|
[
|
|
"park" in place_type,
|
|
"theme" in place_type,
|
|
"amusement" in place_type,
|
|
"attraction" in place_type,
|
|
extratags.get("tourism") == "theme_park",
|
|
extratags.get("leisure") == "amusement_arcade",
|
|
extratags.get("amenity") == "amusement_arcade",
|
|
]
|
|
)
|
|
|
|
return {
|
|
"name": osm_item.get("display_name", ""),
|
|
"latitude": float(osm_item["lat"]),
|
|
"longitude": float(osm_item["lon"]),
|
|
"formatted_address": formatted_address,
|
|
"street_address": street_address,
|
|
"city": city,
|
|
"state": state,
|
|
"country": country,
|
|
"postal_code": postal_code,
|
|
"osm_id": osm_item.get("osm_id"),
|
|
"osm_type": osm_item.get("osm_type"),
|
|
"place_type": place_type,
|
|
"importance": osm_item.get("importance", 0),
|
|
"is_park_related": is_park_related,
|
|
}
|
|
|
|
except (KeyError, ValueError, TypeError) as e:
|
|
logger.warning(f"Error transforming OSM result: {str(e)}")
|
|
return None
|
|
|
|
@classmethod
|
|
def _transform_osm_reverse_result(
|
|
cls, osm_result: Dict[str, Any]
|
|
) -> Dict[str, Any]:
|
|
"""Transform OSM reverse geocoding result to our standard format."""
|
|
address = osm_result.get("address", {})
|
|
|
|
# Extract address components
|
|
street_number = address.get("house_number", "")
|
|
street_name = address.get("road", "")
|
|
street_address = f"{street_number} {street_name}".strip()
|
|
|
|
city = (
|
|
address.get("city")
|
|
or address.get("town")
|
|
or address.get("village")
|
|
or address.get("municipality")
|
|
or ""
|
|
)
|
|
|
|
state = (
|
|
address.get("state")
|
|
or address.get("province")
|
|
or address.get("region")
|
|
or ""
|
|
)
|
|
|
|
country = address.get("country", "")
|
|
postal_code = address.get("postcode", "")
|
|
|
|
# Build formatted address
|
|
address_parts = []
|
|
if street_address:
|
|
address_parts.append(street_address)
|
|
if city:
|
|
address_parts.append(city)
|
|
if state:
|
|
address_parts.append(state)
|
|
if postal_code:
|
|
address_parts.append(postal_code)
|
|
if country:
|
|
address_parts.append(country)
|
|
|
|
formatted_address = ", ".join(address_parts)
|
|
|
|
return {
|
|
"name": osm_result.get("display_name", ""),
|
|
"latitude": float(osm_result["lat"]),
|
|
"longitude": float(osm_result["lon"]),
|
|
"formatted_address": formatted_address,
|
|
"street_address": street_address,
|
|
"city": city,
|
|
"state": state,
|
|
"country": country,
|
|
"postal_code": postal_code,
|
|
"osm_id": osm_result.get("osm_id"),
|
|
"osm_type": osm_result.get("osm_type"),
|
|
"place_type": osm_result.get("type", ""),
|
|
}
|