features/mcp/tools/geocoding_tool.py

"""
地點名稱轉座標工具（Forward Geocoding）
使用 Nominatim（OSM）將地點名稱轉換為經緯度座標
"""

import aiohttp
import asyncio
import logging
from typing import Dict, Any, List

from .base_tool import MCPTool, StandardToolSchemas, ExecutionError
from core.database import get_geo_cache, set_geo_cache
from core.database.cache import db_cache

logger = logging.getLogger("mcp.tools.geocoding")


class ForwardGeocodeTool(MCPTool):
    NAME = "forward_geocode"
    DESCRIPTION = "Convert place names (e.g., 'Ming Chuan University', 'Taoyuan Train Station') to coordinates (latitude/longitude)"
    CATEGORY = "地理定位"
    TAGS = ["geocode", "forward", "地點", "座標"]
    KEYWORDS = ["地點", "位置", "座標", "在哪裡", "地址查詢"]
    USAGE_TIPS = [
        "提供地點名稱即可（如「台北101」「淡水捷運站」）",
        "支援地標、車站、學校、商圈等",
        "會返回最相關的座標與詳細地址"
    ]

    @classmethod
    def get_input_schema(cls) -> Dict[str, Any]:
        return StandardToolSchemas.create_input_schema({
            "query": {
                "type": "string",
                "description": "地點名稱或地址（如「銘傳大學桃園校區」「桃園火車站」「台北101」）"
            },
            "limit": {
                "type": "integer",
                "description": "返回結果數量（預設 1，最多 5）",
                "default": 1
            }
        }, required=["query"])

    @classmethod
    def get_output_schema(cls) -> Dict[str, Any]:
        schema = StandardToolSchemas.create_output_schema()
        schema["properties"].update({
            "results": {
                "type": "array",
                "description": "地點查詢結果列表",
                "items": {
                    "type": "object",
                    "properties": {
                        "lat": {"type": "number", "description": "緯度"},
                        "lon": {"type": "number", "description": "經度"},
                        "display_name": {"type": "string", "description": "完整地址"},
                        "label": {"type": "string", "description": "簡短標籤"},
                        "importance": {"type": "number", "description": "重要性評分（0-1）"}
                    }
                }
            },
            "best_match": {
                "type": "object",
                "description": "最佳匹配結果",
                "properties": {
                    "lat": {"type": "number"},
                    "lon": {"type": "number"},
                    "label": {"type": "string"}
                }
            }
        })
        return schema

    @classmethod
    async def execute(cls, arguments: Dict[str, Any]) -> Dict[str, Any]:
        query = arguments.get("query", "").strip()
        if not query:
            raise ExecutionError("請提供地點名稱")

        limit = min(int(arguments.get("limit", 1)), 5)

        # 生成快取鍵（基於查詢文字）
        import hashlib
        cache_key = hashlib.md5(f"geocode:{query}".encode()).hexdigest()

        # 記憶體快取
        cached = await db_cache.get_geo_cached(cache_key)
        if cached:
            logger.info(f"📍 Geocoding 快取命中: {query}")
            return cls.create_success_response(
                content=f"找到地點：{cached['best_match']['label']}",
                data=cached
            )

        # DB 快取
        db_cached = await get_geo_cache(cache_key)
        if db_cached:
            await db_cache.set_geo_cache(cache_key, db_cached)
            return cls.create_success_response(
                content=f"找到地點：{db_cached['best_match']['label']}",
                data=db_cached
            )

        # 外呼 Nominatim（公共端點，務必節流）
        url = "https://nominatim.openstreetmap.org/search"
        params = {
            "format": "jsonv2",
            "q": query,
            "limit": limit,
            "addressdetails": 1,
            "extratags": 1,  # 取得額外標籤
            "namedetails": 1,  # 取得多語言名稱
            "accept-language": "zh-TW,zh"
        }
        headers = {
            "User-Agent": "BloomWare/1.0 (contact@example.com)"
        }

        try:
            async with aiohttp.ClientSession(headers=headers) as session:
                async with session.get(url, params=params, timeout=aiohttp.ClientTimeout(total=10)) as resp:
                    if resp.status != 200:
                        raise ExecutionError(f"Nominatim 查詢失敗: HTTP {resp.status}")
                    
                    data = await resp.json()
                    
                    if not data or len(data) == 0:
                        raise ExecutionError(f"找不到地點「{query}」，請確認地點名稱是否正確")

        except asyncio.TimeoutError:
            raise ExecutionError("地點查詢逾時，請稍後再試")
        except aiohttp.ClientError as e:
            raise ExecutionError(f"網路連接錯誤: {str(e)}")

        # 解析結果
        results = []
        for item in data:
            lat = float(item.get("lat", 0))
            lon = float(item.get("lon", 0))
            display_name = item.get("display_name", "")
            importance = float(item.get("importance", 0))
            
            # 解析地址組件
            addr = item.get("address", {})
            extratags = item.get("extratags", {})
            namedetails = item.get("namedetails", {})
            
            name = item.get("name", "")
            name_zh = namedetails.get("name:zh") or namedetails.get("name:zh-TW") or name
            
            # 基本地址組件
            road = addr.get("road") or addr.get("pedestrian") or addr.get("footway") or ""
            house_number = addr.get("house_number") or ""
            suburb = addr.get("suburb") or addr.get("neighbourhood") or ""
            city_district = addr.get("city_district") or ""
            city = addr.get("city") or addr.get("town") or addr.get("village") or addr.get("county") or ""
            admin = addr.get("state") or addr.get("county") or ""
            postcode = addr.get("postcode") or ""
            
            # POI 資訊
            amenity = addr.get("amenity") or extratags.get("amenity") or ""
            shop = addr.get("shop") or extratags.get("shop") or ""
            building = addr.get("building") or extratags.get("building") or ""
            
            # 組裝簡短標籤
            label_parts = []
            if name_zh and name_zh != road:
                label_parts.append(name_zh)
            
            if road and house_number:
                label_parts.append(f"{road}{house_number}號")
            elif road:
                label_parts.append(road)
            
            if city_district and city_district not in str(label_parts):
                label_parts.append(city_district)
            elif suburb and suburb not in str(label_parts):
                label_parts.append(suburb)
            
            # 添加城市/區域資訊
            if city and city not in str(label_parts):
                label_parts.append(city)
            
            label = ", ".join(filter(None, label_parts)) if label_parts else display_name
            
            # 組裝詳細地址
            detailed_parts = []
            if name_zh:
                detailed_parts.append(f"地點: {name_zh}")
            if road and house_number:
                detailed_parts.append(f"地址: {road}{house_number}號")
            elif road:
                detailed_parts.append(f"路段: {road}")
            if suburb:
                detailed_parts.append(f"區域: {suburb}")
            if city:
                detailed_parts.append(f"城市: {city}")
            if postcode:
                detailed_parts.append(f"郵遞區號: {postcode}")
            
            detailed_address = " | ".join(detailed_parts) if detailed_parts else label

            results.append({
                "lat": lat,
                "lon": lon,
                "display_name": display_name,
                "label": label,
                "detailed_address": detailed_address,
                "importance": importance,
                # 額外欄位供後續使用
                "name": name_zh or name,
                "road": road,
                "house_number": house_number,
                "suburb": suburb,
                "city_district": city_district,
                "city": city,
                "admin": admin,
                "postcode": postcode,
                "amenity": amenity,
                "shop": shop,
                "building": building,
            })

        # 最佳匹配（重要性最高）
        best_match = max(results, key=lambda x: x["importance"])

        payload = {
            "results": results,
            "best_match": best_match,
            "query": query
        }

        # 回寫快取（雙層）
        await db_cache.set_geo_cache(cache_key, payload)
        await set_geo_cache(cache_key, payload)

        logger.info(f"📍 Geocoding 成功: {query} → {best_match['label']} ({best_match['lat']:.4f}, {best_match['lon']:.4f})")

        # 組裝友善回覆
        content_parts = [f"找到地點：{best_match['label']}"]
        if len(results) > 1:
            content_parts.append(f"（共 {len(results)} 個結果，已選擇最相關的）")
        
        content = "\n".join(content_parts)

        return cls.create_success_response(content=content, data=payload)