All.py

import sqlite3
import sys
from geopy.geocoders import Nominatim
from geopy.exc import GeocoderUnavailable, GeocoderTimedOut
from time import sleep
from dadata import Dadata
from typing import Optional, Dict

TOKEN = "a2c67eb60ada13b05220830b7ea4611ae28e7ab9"
SECRET = "8236c69efae9d50b7c4b2c064ec863901424d4a5"
dadata = Dadata(TOKEN, SECRET)


class Geocoder:
    def __init__(self, db_path="address_cache.db"):
        self.db_path = db_path
        self._init_db()
        self.geolocator = Nominatim(user_agent="rus_geocoder")

    def _init_db(self):
        """Создаем таблицу для кеша, если её нет."""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS addresses (
                    query TEXT PRIMARY KEY,
                    address TEXT,
                    lat REAL,
                    lon REAL
                )
            """)
            conn.execute("""
                CREATE TABLE IF NOT EXISTS coordinates (
                    lat REAL,
                    lon REAL,
                    address TEXT,
                    PRIMARY KEY (lat, lon)
                )
            """)

    def _get_from_cache(self, query):
        """Ищем адрес в кеше."""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute("SELECT address, lat, lon FROM addresses WHERE query=?", (query,))
            return cursor.fetchone()

    def _save_to_cache(self, query, address, lat, lon):
        """Сохраняем результат в кеш."""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                INSERT OR REPLACE INTO addresses (query, address, lat, lon)
                VALUES (?, ?, ?, ?)
            """, (query, address, lat, lon))

    def _get_reverse_cache(self, lat, lon):
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute("SELECT address FROM coordinates WHERE lat=? AND lon=?", (lat, lon))
            result = cursor.fetchone()
            return result[0] if result else None

    def _save_reverse_cache(self, lat, lon, address):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                INSERT OR REPLACE INTO coordinates (lat, lon, address)
                VALUES (?, ?, ?)
            """, (lat, lon, address))

    def geocode(self, query, max_retries=3):
        cached = self._get_from_cache(query)
        if cached:
            return {
                "query": query,
                "address": cached[0],
                "lat": cached[1],
                "lon": cached[2],
                "source": "cache"
            }

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

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

    def reverse_geocode(self, lat, lon):
        cached = self._get_reverse_cache(lat, lon)
        if cached:
            return {
                "lat": lat,
                "lon": lon,
                "address": cached,
                "source": "cache"
            }

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

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


def reverse_data_entry():
    while True:
        try:
            lat = float(input("Введите широту (lat): ").strip())
            lon = float(input("Введите долготу (lon): ").strip())
            if -90 <= lat <= 90 and -180 <= lon <= 180:
                break
            print("Ошибка: широта должна быть от -90 до 90, долгота от -180 до 180")
        except ValueError:
            print("Ошибка: координаты должны быть числами.")

    geocoder = Geocoder()
    result = geocoder.reverse_geocode(lat, lon)
    print(format_reverse_result(result))


def format_reverse_result(result):
    if "error" in result:
        return f"\nОшибка: {result['error']}\n"

    return (
        f"\n Адрес найден ({'кеш' if result['source'] == 'cache' else 'Nominatim'}):\n"
        f"Координаты:\n"
        f"    - Широта:  {result['lat']}\n"
        f"    - Долгота: {result['lon']}\n"
        f"Адрес: {result['address']}\n"
    )


def normalize_address_with_dadata(address: str) -> Optional[Dict]:
    try:
        return dadata.clean("address", address)
    except Exception as e:
        print(f"[Ошибка Dadata] Не удалось очистить адрес: {e}")
        return None


def construct_standard_address(cleaned_data: Dict) -> Optional[str]:
    try:
        fields_order = ['street', 'house', 'city', 'region', 'country']
        parts = [cleaned_data[field] for field in fields_order if cleaned_data.get(field)]

        if not parts:
            print("Не удалось построить нормализованный адрес — все поля пусты")
            return None

        return ' '.join(parts)
    except (KeyError, TypeError) as e:
        print(f"[Ошибка сборки адреса] Неверная структура данных: {e}")
        return None


def prompt_and_construct_address():
    while True:
        try:
            city = sanitize_user_input(input("Введите город: "))
            street = sanitize_user_input(input("Введите улицу: "))
            number = sanitize_user_input(input("Введите номер дома: "))
            address = f"{street} {number} {city} Россия"

            cleaned = normalize_address_with_dadata(address)
            if not cleaned:
                print("Нормализация адреса не удалась")
                print("проверьте корректность ввода")
                continue

            normalized_address = construct_standard_address(cleaned)
            if not normalized_address:
                print("Не удалось собрать адрес из полученных данных")
                print("проверьте корректность ввода")
                continue
            return normalized_address

        except Exception as e:
            print(f"[Ошибка ввода] {e}")


def sanitize_user_input(text: str) -> str:
    return text.encode('utf-8', errors='ignore').decode('utf-8').strip()


def format_result(result):
    if "error" in result:
        return f"\n Ошибка: {result['error']}\n"

    output = [
        f"Адрес найден ({'кеш' if result['source'] == 'cache' else 'Nominatim'}):",
        f"Запрос: {result['query']}",
        f"Адрес: {result['address']}",
        f"Координаты:",
        f"    - Широта:  {result['lat']}",
        f"    - Долгота: {result['lon']}",
    ]
    return "\n".join(output) + "\n"


def run_forward_geocoding():
    geocoder = Geocoder()
    query = prompt_and_construct_address()
    result = geocoder.geocode(query)
    print(format_result(result))


def handle_user_command(command):
    if command == "exit":
        print("Выход из программы.")
        sys.exit(0)
    elif command == "start":
        run_forward_geocoding()
    elif command == "reverse":
        reverse_data_entry()
    elif command in ('-h', '--help'):
        print_help()
    else:
        print("Неправильная команда. Введите 'start', 'reverse', 'exit', '-h' или '--help'.")


def print_help():
    print("""
Геокодер для российских адресов (на основе Nominatim OpenStreetMap).

Команды:
  start       — прямой поиск (адрес → координаты)
  reverse     — обратный поиск (координаты → адрес)
  exit        — выход из интерактивного режима
  -h, --help  — показать эту справку

Использование:
  python geocoder.py         # интерактивный режим
  python geocoder.py start   # прямой поиск
  python geocoder.py reverse # обратный поиск
""")


def main():
    print("Геокодер для российских адресов.")
    print("Команды:")
    print("  start    — прямой поиск (адрес → координаты)")
    print("  reverse  — обратный поиск (координаты → адрес)")
    print("  exit     — выход")
    print(" -h или --help для получения справки")

    while True:
        print(" -h или --help для получения справки")
        command = input("\n>>> ").strip().lower()
        handle_user_command(command)
        print()


if __name__ == "__main__":
    main()