features/mcp/tools/location/geocoding_tool.py

"""
地點名稱轉座標工具（Forward Geocoding）
優先使用 TDX 官方定位服務，失敗時 fallback 到 Nominatim（OSM）
"""

import aiohttp
import asyncio
import logging
from typing import Dict, Any, List
from urllib.parse import quote

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

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

POI_HINTS = ("車站", "捷運站", "站", "大學", "醫院", "百貨", "大樓", "機場", "公園", "學校")


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
            )

        data = await cls._forward_geocode_tdx(query)
        source = "tdx"
        if not data:
            data = await cls._forward_geocode_nominatim(query, limit)
            source = "nominatim"
        if not data or len(data) == 0:
            raise ExecutionError(f"找不到地點「{query}」，請確認地點名稱是否正確")

        prefer_poi = any(hint in query for hint in POI_HINTS)
        prefer_address = any(ch.isdigit() for ch in query) or "號" in query

        # 解析結果
        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,
                "geocode_source": source,
                "_kind": item.get("_kind", ""),
            })

        for result in results:
            score = float(result.get("importance", 0))
            name = result.get("name", "") or ""
            label = result.get("label", "") or ""
            display_name = result.get("display_name", "") or ""
            kind = result.get("_kind", "")
            text = f"{name} {label} {display_name}"

            if query and query in text:
                score += 5.0
            if prefer_poi and kind == "markname":
                score += 8.0
            if prefer_address and kind == "address":
                score += 8.0
            if "出入口" in text:
                score -= 1.5
            if "交叉口" in display_name and prefer_poi:
                score -= 1.0
            result["_score"] = score

        results.sort(key=lambda x: x.get("_score", 0), reverse=True)
        best_match = results[0]

        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)

    @staticmethod
    async def _forward_geocode_tdx(query: str) -> List[Dict[str, Any]] | None:
        try:
            address_endpoint = f"V3/Map/GeoCode/Coordinate/Address/{quote(query, safe='')}"
            rows = await TDXBaseAPI.call_api(
                address_endpoint,
                {"$format": "JSON"},
                cache_ttl=86400,
                api_version="",
                api_family="advanced",
            )
            parsed_address = ForwardGeocodeTool._parse_tdx_rows(rows, kind="address")

            mark_endpoint = f"V3/Map/GeoCode/Coordinate/Markname/{quote(query, safe='')}"
            rows = await TDXBaseAPI.call_api(
                mark_endpoint,
                {"$format": "JSON"},
                cache_ttl=86400,
                api_version="",
                api_family="advanced",
            )
            parsed_mark = ForwardGeocodeTool._parse_tdx_rows(rows, kind="markname")
            combined = (parsed_mark or []) + (parsed_address or [])
            return combined
        except Exception as exc:
            logger.warning("TDX forward geocode 失敗，回退 Nominatim: %s", exc)
            return None

    @staticmethod
    def _parse_tdx_rows(rows: Any, *, kind: str) -> List[Dict[str, Any]]:
        results = []
        for row in rows or []:
            lon = row.get("LocationX") or row.get("PositionLon")
            lat = row.get("LocationY") or row.get("PositionLat")
            geometry = row.get("Geometry") or ""
            if (lon is None or lat is None) and isinstance(geometry, str) and geometry.startswith("POINT"):
                try:
                    point_text = geometry.removeprefix("POINT").strip().strip("()")
                    point_lon, point_lat = point_text.split()
                    lon = float(point_lon)
                    lat = float(point_lat)
                except Exception:
                    lon = lon
                    lat = lat
            if lon is None or lat is None:
                continue
            address = row.get("Address") or row.get("RoadName") or row.get("LandMarkName") or ""
            name = row.get("Name") or row.get("LandMarkName") or row.get("LocationDescription") or ""
            city = row.get("City") or ""
            town = row.get("Town") or ""
            label = name or address or ""
            results.append({
                "lat": float(lat),
                "lon": float(lon),
                "display_name": address or label,
                "label": label,
                "importance": 1.0,
                "name": name,
                "road": row.get("RoadName") or "",
                "house_number": str(row.get("AddressNo") or "").strip(),
                "suburb": "",
                "city_district": town,
                "city": city,
                "admin": city,
                "postcode": row.get("ZipCode") or "",
                "amenity": "",
                "shop": "",
                "building": "",
                "detailed_address": address or label,
                "_kind": kind,
            })
        return results

    @staticmethod
    async def _forward_geocode_nominatim(query: str, limit: int) -> List[Dict[str, Any]]:
        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}")
                    return await resp.json()
        except asyncio.TimeoutError:
            raise ExecutionError("地點查詢逾時，請稍後再試")
        except aiohttp.ClientError as e:
            raise ExecutionError(f"網路連接錯誤: {str(e)}")