""" 反地理與時區工具（免費 API 優先） - reverse_geocode: 使用 Nominatim（OSM）反查城市/行政區（先查 DB/記憶體快取） """ import aiohttp import asyncio import logging from typing import Dict, Any 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.geocode") class ReverseGeocodeTool(MCPTool): NAME = "reverse_geocode" DESCRIPTION = "Convert coordinates (latitude/longitude) to city/district names (uses cache when available)" CATEGORY = "地理定位" TAGS = ["geocode", "reverse", "city"] KEYWORDS = ["座標", "經緯度", "反查", "地址", "我在哪"] USAGE_TIPS = ["提供 lat/lon 即可"] @classmethod def get_input_schema(cls) -> Dict[str, Any]: return StandardToolSchemas.create_input_schema({ "lat": {"type": "number", "description": "緯度"}, "lon": {"type": "number", "description": "經度"} }, required=["lat", "lon"]) @classmethod def get_output_schema(cls) -> Dict[str, Any]: schema = StandardToolSchemas.create_output_schema() schema["properties"].update({ "city": {"type": "string"}, "admin": {"type": "string"}, "country_code": {"type": "string"} }) return schema @classmethod async def execute(cls, arguments: Dict[str, Any]) -> Dict[str, Any]: lat = arguments.get("lat") lon = arguments.get("lon") if lat is None or lon is None: raise ExecutionError("缺少經緯度") # 先用 geohash7 當鍵查快取 try: from geohash2 import encode as gh_encode geokey = gh_encode(lat, lon, precision=7) except Exception: geokey = f"{round(lat,4)},{round(lon,4)}" # 記憶體快取 cached = await db_cache.get_geo_cached(geokey) if cached: # 補齊缺失欄位（兼容舊版快取） cached.setdefault("lat", lat) # ← 補上座標 cached.setdefault("lon", lon) # ← 補上座標 cached.setdefault("name", "") cached.setdefault("detailed_address", cached.get("label") or cached.get("display_name") or "") cached.setdefault("postcode", "") cached.setdefault("city_district", "") cached.setdefault("amenity", "") cached.setdefault("shop", "") cached.setdefault("building", "") cached.setdefault("office", "") cached.setdefault("leisure", "") cached.setdefault("tourism", "") return cls.create_success_response( content=cached.get("display_name") or f"{cached.get('city')}, {cached.get('admin')}", data=cached ) # DB 快取 db_cached = await get_geo_cache(geokey) if db_cached: # 補齊缺失欄位（兼容舊版快取） db_cached.setdefault("lat", lat) # ← 補上座標 db_cached.setdefault("lon", lon) # ← 補上座標 db_cached.setdefault("name", "") db_cached.setdefault("detailed_address", db_cached.get("label") or db_cached.get("display_name") or "") db_cached.setdefault("postcode", "") db_cached.setdefault("city_district", "") db_cached.setdefault("amenity", "") db_cached.setdefault("shop", "") db_cached.setdefault("building", "") db_cached.setdefault("office", "") db_cached.setdefault("leisure", "") db_cached.setdefault("tourism", "") await db_cache.set_geo_cache(geokey, db_cached) return cls.create_success_response( content=db_cached.get("display_name") or f"{db_cached.get('city')}, {db_cached.get('admin')}", data=db_cached ) # 外呼 Nominatim（公共端點，務必節流） url = "https://nominatim.openstreetmap.org/reverse" params = { "format": "jsonv2", "lat": lat, "lon": lon, "zoom": 18, "addressdetails": 1, "extratags": 1, "namedetails": 1 } headers = { "User-Agent": "BloomWare/1.0 (contact@example.com)" } # 呼叫 Nominatim API data = None try: async with aiohttp.ClientSession(headers=headers) as session: async with session.get(url, params=params, timeout=30) as resp: if resp.status != 200: raise ExecutionError(f"Nominatim 失敗: HTTP {resp.status}") response_text = await resp.text() if not response_text or response_text.strip() == "": raise ExecutionError("Nominatim 回應為空") import json try: data = json.loads(response_text) except json.JSONDecodeError: raise ExecutionError(f"Nominatim 回應非 JSON: {response_text[:200]}") except aiohttp.ClientError as e: raise ExecutionError(f"Nominatim 網路錯誤: {e}") except asyncio.TimeoutError: raise ExecutionError("Nominatim 請求逾時") # 驗證回應 if data is None: raise ExecutionError("Nominatim 回應為 null") if not isinstance(data, dict): raise ExecutionError(f"Nominatim 回應格式錯誤: {type(data)}") if "error" in data: raise ExecutionError(f"Nominatim 錯誤: {data.get('error')}") # 解析地址資訊 addr = data.get("address") or {} extratags = data.get("extratags") or {} # 基本地址組件 road = addr.get("road") or addr.get("pedestrian") or addr.get("footway") or addr.get("cycleway") or "" house_number = addr.get("house_number") or "" suburb = addr.get("suburb") or addr.get("neighbourhood") or addr.get("quarter") 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 "" country_code = (addr.get("country_code") or "").upper() 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 "" office = addr.get("office") or extratags.get("office") or "" leisure = addr.get("leisure") or extratags.get("leisure") or "" tourism = addr.get("tourism") or extratags.get("tourism") or "" # 地點名稱（優先使用繁中） name = data.get("name") or "" namedetails = data.get("namedetails") or {} name_zh = namedetails.get("name:zh") or namedetails.get("name:zh-TW") or name display_name = data.get("display_name") or "" # 組裝精確標籤（優先顯示最精確的資訊） label_parts = [] # 1. POI 名稱（如「7-11 明倫門市」「台北101」） if name_zh and name_zh != road: label_parts.append(name_zh) # 2. 門牌號碼 + 路名（如「中正路123號」） if road and house_number: label_parts.append(f"{road}{house_number}號") elif road: # 如果沒有門牌，但有路口資訊 if "路口" in road or "交叉口" in road or "intersection" in road.lower(): label_parts.append(road) else: # 嘗試從附近找路口 label_parts.append(road) # 3. 郵遞區號（如「100」） if postcode and len(label_parts) > 0: label_parts[0] = f"〒{postcode} {label_parts[0]}" # 4. 區域（如「大安區」） if city_district and city_district not in label_parts: label_parts.append(city_district) elif suburb and suburb not in label_parts: label_parts.append(suburb) # 5. 城市（如「台北市」） if city and city not in label_parts: label_parts.append(city) # 6. 省份/州（如「台灣」） if admin and admin not in city and admin not in label_parts: label_parts.append(admin) label = ", ".join(filter(None, label_parts)) # 組裝詳細地址（用於 AI 顯示） detailed_address_parts = [] if name_zh: detailed_address_parts.append(f"地點: {name_zh}") if road and house_number: detailed_address_parts.append(f"地址: {road}{house_number}號") elif road: detailed_address_parts.append(f"路段: {road}") if suburb: detailed_address_parts.append(f"區域: {suburb}") if city: detailed_address_parts.append(f"城市: {city}") if postcode: detailed_address_parts.append(f"郵遞區號: {postcode}") detailed_address = " | ".join(detailed_address_parts) if detailed_address_parts else label payload = { "lat": lat, "lon": lon, "city": city or "", "admin": admin or "", "country_code": country_code, "display_name": display_name, "label": label or display_name, "detailed_address": detailed_address, "road": road, "house_number": house_number, "suburb": suburb, "city_district": city_district, "postcode": postcode, "amenity": amenity, "shop": shop, "building": building, "office": office, "leisure": leisure, "tourism": tourism, "name": name_zh or name, } # 回寫快取 await db_cache.set_geo_cache(geokey, payload) await set_geo_cache(geokey, payload) return cls.create_success_response(content=payload.get("label") or f"{payload['city']}, {payload['admin']}", data=payload)