bot/utils/geocoder.py - master - zimaaadev/Parser

bot/utils/geocoder.py

63 строки2 KB

zimaaadev

Ищет природные объекты, набнрежные и т.п.

24 дек 2025, 17:18

24 дек 2025, 17:182d73bc2

# bot/utils/geocoder.py
import httpx
import logging
import asyncio
from typing import Tuple, Optional
from urllib.parse import quote

from config import GEOCODE_API_KEY  # ← импортируем ключ
from utils.cache import AsyncSQLiteCache

logger = logging.getLogger(__name__)
cache = AsyncSQLiteCache("request_cache.sqlite")

GEOCODE_URL = "https://geocode.maps.co/search"

HEADERS = {
    "User-Agent": "RealtyBot/1.0",
    "Accept-Language": "ru"
}

async def geocode_address(address: str) -> Optional[Tuple[float, float]]:
    if not address or len(address.strip()) < 3:
        return None
    address = address.strip()

    # Ключ кэша
    cache_key = f"geo:{address.lower()}"
    cached = await cache.get(cache_key)
    if cached:
        logger.info(f"📦 Кэш: {address} → {cached}")
        return tuple(cached)

    # Задержка
    await asyncio.sleep(1.0)

    try:
        encoded_address = quote(address, encoding='utf-8')
        # ✅ Добавляем api_key
        url = f"{GEOCODE_URL}?q={encoded_address}&api_key={GEOCODE_API_KEY}"

        async with httpx.AsyncClient(timeout=10.0, headers=HEADERS) as client:
            resp = await client.get(url)

        if resp.status_code != 200:
            logger.warning(f"❌ HTTP {resp.status_code}: {resp.text}")
            return None

        data = resp.json()
        if not data:
            logger.warning(f"❌ Адрес не найден: {address}")
            return None

        lat = float(data[0]["lat"])
        lon = float(data[0]["lon"])

        # Сохраняем в кэш
        await cache.set(cache_key, [lat, lon], ttl=86400)
        logger.info(f"📍 Успех: {address} → {lat}, {lon}")
        return lat, lon

    except Exception as e:
        logger.error(f"❌ Ошибка геокодинга: {e}")
        return None