geocoder.py

from geopy.geocoders import Nominatim
from geopy.exc import GeocoderUnavailable, GeocoderTimedOut
from time import sleep
from typing import Dict
from cache import AddressCache
from constants import DEFAULT_DB_PATH


class Geocoder:
    def __init__(self, db_path: str = DEFAULT_DB_PATH):
        self.cache = AddressCache(db_path)
        self.geolocator = Nominatim(user_agent="rus_geocoder")

    def geocode(self, query: str, max_retries: int = 3) -> Dict:
        if cached := self.cache.get_address(query):
            return self._format_result(query, *cached, "cache")

        for attempt in range(max_retries):
            try:
                location = self.geolocator.geocode(query,
                                                   exactly_one=True,
                                                   language="ru")
                sleep(1)
                if location:
                    self.cache.save_address(query,
                                            location.address,
                                            location.latitude,
                                            location.longitude)
                    return self._format_result(
                        query,
                        location.address,
                        location.latitude,
                        location.longitude,
                        "nominatim"
                    )
            except (GeocoderUnavailable, GeocoderTimedOut) as e:
                if attempt == max_retries - 1:
                    return {"error": f"Адрес не найден. Ошибка сервиса: "
                                     f"{str(e)}"}
                sleep(2)

        return {"error": "Адрес не найден после нескольких попыток"}

    def reverse_geocode(self, lat: float, lon: float) -> Dict:
        if cached := self.cache.get_reverse_address(lat, lon):
            return self._format_reverse_result(lat, lon, cached, "cache")

        try:
            location = self.geolocator.reverse((lat, lon),
                                               exactly_one=True,
                                               language="ru")
            sleep(1)
            if location:
                self.cache.save_reverse_address(lat, lon, location.address)
                return self._format_reverse_result(lat, lon,
                                                   location.address,
                                                   "nominatim")
        except (GeocoderUnavailable, GeocoderTimedOut):
            pass

        return {"error": "Адрес не найден по координатам"}

    @staticmethod
    def _format_result(query: str, address: str, lat: float, lon: float,
                       source: str) -> Dict:
        return {
            "query": query,
            "address": address,
            "lat": lat,
            "lon": lon,
            "source": source
        }

    @staticmethod
    def _format_reverse_result(lat: float, lon: float,
                               address: str, source: str) -> Dict:
        return {
            "lat": lat,
            "lon": lon,
            "address": address,
            "source": source
        }