"""
Geocoder Service - Converts place names to coordinates using Nominatim.
"""
from typing import Optional, Dict
from geopy.geocoders import Nominatim
from geopy.exc import GeocoderTimedOut, GeocoderServiceError
from cachetools import TTLCache
import time
from ..config import settings
from ..utils.logger import setup_logger
logger = setup_logger(__name__)
# Initialize geocoder with user agent
geolocator = Nominatim(
user_agent=settings.GEOCODER_USER_AGENT if hasattr(settings, 'GEOCODER_USER_AGENT') else "locate-the-news"
)
# Cache for geocoding results (TTL = 1 hour)
geocode_cache = TTLCache(maxsize=1000, ttl=3600)
def geocode_location(place: str, timeout: int = 10) -> Optional[Dict[str, float]]:
"""
Geocodes a place name to latitude and longitude coordinates.
Uses caching to avoid repeated API calls.
Args:
place: The name of the place to geocode
timeout: Request timeout in seconds
Returns:
Dictionary with 'latitude' and 'longitude' keys, or None if geocoding fails
"""
if not place or not place.strip():
return None
# Check cache first
cache_key = place.lower().strip()
if cache_key in geocode_cache:
logger.debug(f"Cache hit for '{place}'")
return geocode_cache[cache_key]
try:
# Rate limiting - Nominatim requires 1 request per second
time.sleep(1)
location = geolocator.geocode(place, timeout=timeout)
if location:
result = {
"latitude": location.latitude,
"longitude": location.longitude
}
geocode_cache[cache_key] = result
logger.info(f"Geocoded '{place}' to ({location.latitude}, {location.longitude})")
return result
else:
logger.warning(f"Could not geocode place: {place}")
return None
except GeocoderTimedOut:
logger.warning(f"Geocoding timeout for place: {place}")
return None
except GeocoderServiceError as e:
logger.error(f"Geocoding service error for place '{place}': {e}")
return None
except Exception as e:
logger.error(f"Unexpected error geocoding place '{place}': {e}")
return None
def geocode_place(place: str, timeout: int = 10) -> Optional[Dict[str, float]]:
"""
Legacy function name - calls geocode_location.
Args:
place: The name of the place to geocode
timeout: Request timeout in seconds
Returns:
Dictionary with 'lat' and 'lon' keys, or None if geocoding fails
"""
result = geocode_location(place, timeout)
if result:
return {
"lat": result["latitude"],
"lon": result["longitude"]
}
return None