locate-the-news / backend / app / services / geocoder.py
geocoder.py
Raw
"""
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