services/overpass_client.py

"""OpenStreetMap Overpass API クライアント"""

import time
from typing import List, Optional, Dict, Any
from dataclasses import dataclass
from cachetools import TTLCache

from config.settings import settings


@dataclass
class POI:
    """Point of Interest"""

    osm_id: int
    name: str
    lat: float
    lon: float
    poi_type: str
    tags: Dict[str, str]


class OverpassClient:
    """
    OpenStreetMap Overpass API クライアント
    POI検索とキャッシュ機能を提供
    """

    OVERPASS_URL = "https://overpass-api.de/api/interpreter"

    # 店舗タイプマッピング
    SHOP_TYPE_MAPPING = {
        "convenience_store": ["shop=convenience", "amenity=convenience"],
        "restaurant": ["amenity=restaurant", "amenity=fast_food"],
        "hospital": ["amenity=hospital", "amenity=clinic"],
        "pharmacy": ["amenity=pharmacy", "shop=chemist"],
        "gas_station": ["amenity=fuel"],
        "parking": ["amenity=parking"],
        "station": ["railway=station", "public_transport=station"],
        "bank": ["amenity=bank"],
        "post_office": ["amenity=post_office"],
        "supermarket": ["shop=supermarket"],
    }

    def __init__(self, timeout: int = 25, cache_ttl: int = 300):
        self.timeout = timeout
        self._cache = TTLCache(maxsize=100, ttl=cache_ttl)
        self._last_request_time: float = 0
        self._min_interval: float = 1.0  # 最低1秒間隔

    def _wait_for_rate_limit(self) -> None:
        """レート制限のための待機"""
        elapsed = time.time() - self._last_request_time
        if elapsed < self._min_interval:
            time.sleep(self._min_interval - elapsed)

    def _build_name_query(
        self,
        name: str,
        lat: float,
        lon: float,
        radius: int,
    ) -> str:
        """名前でPOIを検索するクエリを構築"""
        return f"""
[out:json][timeout:{self.timeout}];
(
  node["name"~"{name}",i](around:{radius},{lat},{lon});
  way["name"~"{name}",i](around:{radius},{lat},{lon});
);
out center;
"""

    def _build_type_query(
        self,
        poi_type: str,
        lat: float,
        lon: float,
        radius: int,
    ) -> str:
        """タイプでPOIを検索するクエリを構築"""
        tags = self.SHOP_TYPE_MAPPING.get(poi_type, [])
        if not tags:
            return ""

        conditions = []
        for tag in tags:
            key, value = tag.split("=")
            conditions.append(f'node["{key}"="{value}"](around:{radius},{lat},{lon});')
            conditions.append(f'way["{key}"="{value}"](around:{radius},{lat},{lon});')

        return f"""
[out:json][timeout:{self.timeout}];
(
  {chr(10).join(conditions)}
);
out center;
"""

    def _parse_response(self, data: Dict[str, Any]) -> List[POI]:
        """Overpass APIレスポンスをパース"""
        pois = []
        elements = data.get("elements", [])

        for elem in elements:
            tags = elem.get("tags", {})
            name = tags.get("name", "")

            # 座標の取得（wayの場合はcenter）
            if elem.get("type") == "way":
                center = elem.get("center", {})
                lat = center.get("lat", 0)
                lon = center.get("lon", 0)
            else:
                lat = elem.get("lat", 0)
                lon = elem.get("lon", 0)

            # POIタイプの判定
            poi_type = "unknown"
            if tags.get("shop"):
                poi_type = tags.get("shop")
            elif tags.get("amenity"):
                poi_type = tags.get("amenity")
            elif tags.get("railway"):
                poi_type = "station"

            if lat and lon:
                pois.append(
                    POI(
                        osm_id=elem.get("id", 0),
                        name=name,
                        lat=lat,
                        lon=lon,
                        poi_type=poi_type,
                        tags=tags,
                    )
                )

        return pois

    def search_by_name(
        self,
        name: str,
        lat: float,
        lon: float,
        radius: int = 500,
    ) -> List[POI]:
        """
        名前でPOIを検索

        Args:
            name: 検索名
            lat: 緯度
            lon: 経度
            radius: 検索半径（メートル）

        Returns:
            POIのリスト
        """
        cache_key = f"name:{name}:{lat:.4f}:{lon:.4f}:{radius}"
        if cache_key in self._cache:
            return self._cache[cache_key]

        query = self._build_name_query(name, lat, lon, radius)
        result = self._execute_query(query)
        self._cache[cache_key] = result
        return result

    def search_by_type(
        self,
        poi_type: str,
        lat: float,
        lon: float,
        radius: int = 500,
    ) -> List[POI]:
        """
        タイプでPOIを検索

        Args:
            poi_type: POIタイプ
            lat: 緯度
            lon: 経度
            radius: 検索半径（メートル）

        Returns:
            POIのリスト
        """
        cache_key = f"type:{poi_type}:{lat:.4f}:{lon:.4f}:{radius}"
        if cache_key in self._cache:
            return self._cache[cache_key]

        query = self._build_type_query(poi_type, lat, lon, radius)
        if not query:
            return []

        result = self._execute_query(query)
        self._cache[cache_key] = result
        return result

    def _execute_query(self, query: str) -> List[POI]:
        """Overpass APIクエリを実行"""
        try:
            import requests

            self._wait_for_rate_limit()
            self._last_request_time = time.time()

            response = requests.post(
                self.OVERPASS_URL,
                data={"data": query},
                timeout=self.timeout,
            )
            response.raise_for_status()

            return self._parse_response(response.json())

        except Exception as e:
            print(f"Overpass query error: {e}")
            return []

    def search_combined(
        self,
        names: List[str],
        types: List[str],
        lat: float,
        lon: float,
        radius: int = 500,
    ) -> Dict[str, List[POI]]:
        """
        複合検索（名前とタイプ両方）

        Returns:
            {"names": {...}, "types": {...}} の形式
        """
        result = {"names": {}, "types": {}}

        for name in names:
            pois = self.search_by_name(name, lat, lon, radius)
            if pois:
                result["names"][name] = pois

        for poi_type in types:
            pois = self.search_by_type(poi_type, lat, lon, radius)
            if pois:
                result["types"][poi_type] = pois

        return result

    def clear_cache(self) -> None:
        """キャッシュをクリア"""
        self._cache.clear()