""" 旅行规划插件 基于高德地图 API，提供以下功能： - 地点搜索与地理编码 - 天气查询（实况 + 4天预报） - 景点/酒店/餐厅搜索 - 路径规划（驾车/公交/步行） - 周边搜索 支持 LLM 函数调用，可与 AIChat 插件配合使用。 """ import tomllib from pathlib import Path from typing import Any, Dict, List from loguru import logger from utils.plugin_base import PluginBase from .amap_client import AmapClient, AmapConfig class TravelPlanner(PluginBase): """旅行规划插件""" description = "旅行规划助手，支持天气查询、景点搜索、路线规划" author = "ShiHao" version = "1.0.0" def __init__(self): super().__init__() self.config = None self.amap: AmapClient = None async def async_init(self): """插件异步初始化""" # 读取配置 config_path = Path(__file__).parent / "config.toml" with open(config_path, "rb") as f: self.config = tomllib.load(f) # 初始化高德 API 客户端 amap_config = self.config.get("amap", {}) api_key = amap_config.get("api_key", "") secret = amap_config.get("secret", "") if not api_key: logger.warning("TravelPlanner: 未配置高德 API Key，请在 config.toml 中设置") else: self.amap = AmapClient(AmapConfig( api_key=api_key, secret=secret, timeout=amap_config.get("timeout", 30) )) if secret: logger.success(f"TravelPlanner 插件已加载，API Key: {api_key[:8]}...（已启用数字签名）") else: logger.success(f"TravelPlanner 插件已加载，API Key: {api_key[:8]}...（未配置安全密钥）") async def on_disable(self): """插件禁用时关闭连接""" await super().on_disable() if self.amap: await self.amap.close() logger.info("TravelPlanner: 已关闭高德 API 连接") # ==================== LLM 工具定义 ==================== def get_llm_tools(self) -> List[Dict]: """返回 LLM 可调用的工具列表""" return [ { "type": "function", "function": { "name": "search_location", "description": "【旅行工具】将地名转换为坐标和行政区划信息。仅当用户明确询问某个地点的位置信息时使用。", "parameters": { "type": "object", "properties": { "address": { "type": "string", "description": "地址或地名，如：北京市、西湖、故宫" }, "city": { "type": "string", "description": "所在城市，可选。填写可提高搜索精度" } }, "required": ["address"] } } }, { "type": "function", "function": { "name": "query_weather", "description": "【旅行工具】查询城市天气预报。仅当用户明确询问某城市的天气情况时使用，如'北京天气怎么样'、'杭州明天会下雨吗'。", "parameters": { "type": "object", "properties": { "city": { "type": "string", "description": "城市名称，如：北京、杭州、上海" }, "forecast": { "type": "boolean", "description": "是否查询预报天气。true=未来4天预报，false=当前实况" } }, "required": ["city"] } } }, { "type": "function", "function": { "name": "search_poi", "description": "【旅行工具】搜索地点（景点、酒店、餐厅等）。仅当用户明确要求查找某城市的景点、酒店、餐厅等时使用。", "parameters": { "type": "object", "properties": { "city": { "type": "string", "description": "搜索城市，如：杭州、北京" }, "keyword": { "type": "string", "description": "搜索关键词，如：西湖、希尔顿酒店、火锅" }, "category": { "type": "string", "enum": ["景点", "酒店", "餐厅", "购物", "交通"], "description": "POI 类别。不填则搜索所有类别" }, "limit": { "type": "integer", "description": "返回结果数量，默认10，最大20" } }, "required": ["city"] } } }, { "type": "function", "function": { "name": "search_nearby", "description": "【旅行工具】搜索某地点周边的设施。仅当用户明确要求查找某地点附近的餐厅、酒店等时使用，如'西湖附近有什么好吃的'。", "parameters": { "type": "object", "properties": { "location": { "type": "string", "description": "中心地点名称，如：西湖、故宫" }, "city": { "type": "string", "description": "所在城市" }, "keyword": { "type": "string", "description": "搜索关键词" }, "category": { "type": "string", "enum": ["景点", "酒店", "餐厅", "购物", "交通"], "description": "POI 类别" }, "radius": { "type": "integer", "description": "搜索半径（米），默认3000，最大50000" } }, "required": ["location", "city"] } } }, { "type": "function", "function": { "name": "plan_route", "description": "【旅行工具】规划两地之间的出行路线。仅当用户明确要求规划从A到B的路线时使用，如'从北京到杭州怎么走'、'上海到苏州的高铁'。", "parameters": { "type": "object", "properties": { "origin": { "type": "string", "description": "起点地名，如：北京、上海虹桥站" }, "destination": { "type": "string", "description": "终点地名，如：杭州、西湖" }, "origin_city": { "type": "string", "description": "起点所在城市" }, "destination_city": { "type": "string", "description": "终点所在城市（跨城时必填）" }, "mode": { "type": "string", "enum": ["driving", "transit", "walking"], "description": "出行方式：driving=驾车，transit=公交/高铁，walking=步行。默认 transit" } }, "required": ["origin", "destination", "origin_city"] } } }, { "type": "function", "function": { "name": "get_travel_info", "description": "【旅行工具】获取目的地城市的旅行信息（天气、景点、交通）。仅当用户明确表示要去某城市旅游并询问相关信息时使用，如'我想去杭州玩，帮我看看'、'北京旅游攻略'。", "parameters": { "type": "object", "properties": { "destination": { "type": "string", "description": "目的地城市，如：杭州、成都" }, "origin": { "type": "string", "description": "出发城市，如：北京、上海。填写后会规划交通路线" } }, "required": ["destination"] } } } ] async def execute_llm_tool( self, tool_name: str, arguments: Dict[str, Any], bot, from_wxid: str ) -> Dict[str, Any]: """执行 LLM 工具调用""" if not self.amap: return {"success": False, "message": "高德 API 未配置，请联系管理员设置 API Key"} try: if tool_name == "search_location": return await self._tool_search_location(arguments) elif tool_name == "query_weather": return await self._tool_query_weather(arguments) elif tool_name == "search_poi": return await self._tool_search_poi(arguments) elif tool_name == "search_nearby": return await self._tool_search_nearby(arguments) elif tool_name == "plan_route": return await self._tool_plan_route(arguments) elif tool_name == "get_travel_info": return await self._tool_get_travel_info(arguments) else: return {"success": False, "message": f"未知工具: {tool_name}"} except Exception as e: logger.error(f"TravelPlanner 工具执行失败: {tool_name}, 错误: {e}") return {"success": False, "message": f"工具执行失败: {str(e)}"} # ==================== 工具实现 ==================== async def _tool_search_location(self, args: Dict) -> Dict: """地点搜索工具""" address = args.get("address", "") city = args.get("city") result = await self.amap.geocode(address, city) if not result["success"]: return {"success": False, "message": result.get("error", "地点搜索失败")} return { "success": True, "message": f"已找到地点：{result['formatted_address']}", "data": { "name": address, "formatted_address": result["formatted_address"], "location": result["location"], "province": result["province"], "city": result["city"], "district": result["district"], "adcode": result["adcode"] } } async def _tool_query_weather(self, args: Dict) -> Dict: """天气查询工具""" city = args.get("city", "") forecast = args.get("forecast", True) extensions = "all" if forecast else "base" result = await self.amap.get_weather(city, extensions) if not result["success"]: return {"success": False, "message": result.get("error", "天气查询失败")} if result["type"] == "live": return { "success": True, "message": f"{result['city']}当前天气：{result['weather']}，{result['temperature']}℃", "data": { "city": result["city"], "weather": result["weather"], "temperature": result["temperature"], "humidity": result["humidity"], "wind": f"{result['winddirection']}风 {result['windpower']}级", "reporttime": result["reporttime"] } } else: forecasts = result["forecasts"] weather_text = "\n".join([ f"- {f['date']} 星期{self._weekday_cn(f['week'])}：白天{f['dayweather']} {f['daytemp']}℃，夜间{f['nightweather']} {f['nighttemp']}℃" for f in forecasts ]) return { "success": True, "message": f"{result['city']}未来天气预报：\n{weather_text}", "data": { "city": result["city"], "province": result["province"], "forecasts": forecasts, "reporttime": result["reporttime"] } } async def _tool_search_poi(self, args: Dict) -> Dict: """POI 搜索工具""" city = args.get("city", "") keyword = args.get("keyword") category = args.get("category") limit = min(args.get("limit", 10), 20) # 获取 POI 类型代码 types = None if category: poi_types = self.config.get("poi_types", {}) types = poi_types.get(category) result = await self.amap.search_poi( keywords=keyword, types=types, city=city, citylimit=True, offset=limit ) if not result["success"]: return {"success": False, "message": result.get("error", "搜索失败")} pois = result["pois"] if not pois: return {"success": False, "message": f"在{city}未找到相关地点"} # 格式化输出 poi_list = [] for i, poi in enumerate(pois, 1): info = f"{i}. {poi['name']}" if poi.get("address"): info += f" - {poi['address']}" if poi.get("rating"): info += f" ⭐{poi['rating']}" if poi.get("cost"): info += f" 人均¥{poi['cost']}" poi_list.append(info) return { "success": True, "message": f"在{city}找到{len(pois)}个结果：\n" + "\n".join(poi_list), "data": { "city": city, "category": category or "全部", "count": len(pois), "pois": pois } } async def _tool_search_nearby(self, args: Dict) -> Dict: """周边搜索工具""" location_name = args.get("location", "") city = args.get("city", "") keyword = args.get("keyword") category = args.get("category") radius = min(args.get("radius", 3000), 50000) # 先获取中心点坐标 geo_result = await self.amap.geocode(location_name, city) if not geo_result["success"]: return {"success": False, "message": f"无法定位 {location_name}"} location = geo_result["location"] # 获取 POI 类型代码 types = None if category: poi_types = self.config.get("poi_types", {}) types = poi_types.get(category) result = await self.amap.search_around( location=location, keywords=keyword, types=types, radius=radius, offset=10 ) if not result["success"]: return {"success": False, "message": result.get("error", "周边搜索失败")} pois = result["pois"] if not pois: return {"success": False, "message": f"在{location_name}周边未找到相关地点"} # 格式化输出 poi_list = [] for i, poi in enumerate(pois, 1): info = f"{i}. {poi['name']}" if poi.get("distance"): info += f" ({poi['distance']}米)" if poi.get("rating"): info += f" ⭐{poi['rating']}" poi_list.append(info) return { "success": True, "message": f"{location_name}周边{radius}米内找到{len(pois)}个结果：\n" + "\n".join(poi_list), "data": { "center": location_name, "radius": radius, "category": category or "全部", "count": len(pois), "pois": pois } } async def _tool_plan_route(self, args: Dict) -> Dict: """路线规划工具""" origin = args.get("origin", "") destination = args.get("destination", "") origin_city = args.get("origin_city", "") destination_city = args.get("destination_city", origin_city) mode = args.get("mode", "transit") # 获取起终点坐标 origin_geo = await self.amap.geocode(origin, origin_city) if not origin_geo["success"]: return {"success": False, "message": f"无法定位起点：{origin}"} dest_geo = await self.amap.geocode(destination, destination_city) if not dest_geo["success"]: return {"success": False, "message": f"无法定位终点：{destination}"} origin_loc = origin_geo["location"] dest_loc = dest_geo["location"] # 根据模式规划路线 if mode == "driving": result = await self.amap.route_driving(origin_loc, dest_loc) if not result["success"]: return {"success": False, "message": result.get("error", "驾车路线规划失败")} distance_km = result["distance"] / 1000 duration_h = result["duration"] / 3600 msg = f"🚗 驾车路线：{origin} → {destination}\n" msg += f"距离：{distance_km:.1f}公里，预计{self._format_duration(result['duration'])}\n" if result["tolls"]: msg += f"收费：约{result['tolls']}元\n" if result["taxi_cost"]: msg += f"打车费用：约{result['taxi_cost']}元" return { "success": True, "message": msg, "data": result } elif mode == "transit": result = await self.amap.route_transit( origin_loc, dest_loc, city=origin_city, cityd=destination_city if destination_city != origin_city else None ) if not result["success"]: return {"success": False, "message": result.get("error", "公交路线规划失败")} msg = f"🚄 公交/高铁路线：{origin} → {destination}\n" for i, transit in enumerate(result["transits"][:2], 1): msg += f"\n方案{i}：{self._format_duration(transit['duration'])}" if transit.get("cost"): msg += f"，约{transit['cost']}元" msg += "\n" for seg in transit["segments"]: if seg["type"] == "walking" and seg["distance"] > 100: msg += f" 🚶 步行{seg['distance']}米\n" elif seg["type"] == "bus": msg += f" 🚌 {seg['name']}：{seg['departure_stop']} → {seg['arrival_stop']}（{seg['via_num']}站）\n" elif seg["type"] == "railway": msg += f" 🚄 {seg['trip']} {seg['name']}：{seg['departure_stop']} {seg.get('departure_time', '')} → {seg['arrival_stop']} {seg.get('arrival_time', '')}\n" return { "success": True, "message": msg.strip(), "data": result } elif mode == "walking": result = await self.amap.route_walking(origin_loc, dest_loc) if not result["success"]: return {"success": False, "message": result.get("error", "步行路线规划失败")} return { "success": True, "message": f"🚶 步行路线：{origin} → {destination}\n距离：{result['distance']}米，预计{self._format_duration(result['duration'])}", "data": result } return {"success": False, "message": f"不支持的出行方式：{mode}"} async def _tool_get_travel_info(self, args: Dict) -> Dict: """一键获取旅行信息""" destination = args.get("destination", "") origin = args.get("origin") info = {"destination": destination} msg_parts = [f"📍 {destination} 旅行信息\n"] # 1. 查询天气 weather_result = await self.amap.get_weather(destination, "all") if weather_result["success"]: info["weather"] = weather_result msg_parts.append("🌤️ 天气预报：") for f in weather_result["forecasts"][:3]: msg_parts.append(f" {f['date']} {f['dayweather']} {f['nighttemp']}~{f['daytemp']}℃") # 2. 搜索热门景点 poi_result = await self.amap.search_poi( types="110000", # 景点 city=destination, citylimit=True, offset=5 ) if poi_result["success"] and poi_result["pois"]: info["attractions"] = poi_result["pois"] msg_parts.append("\n🏞️ 热门景点：") for poi in poi_result["pois"][:5]: rating = f" ⭐{poi['rating']}" if poi.get("rating") else "" msg_parts.append(f" • {poi['name']}{rating}") # 3. 规划交通路线（如果提供了出发地） if origin: origin_geo = await self.amap.geocode(origin) dest_geo = await self.amap.geocode(destination) if origin_geo["success"] and dest_geo["success"]: route_result = await self.amap.route_transit( origin_geo["location"], dest_geo["location"], city=origin_geo.get("city", origin), cityd=dest_geo.get("city", destination) ) if route_result["success"] and route_result["transits"]: info["route"] = route_result transit = route_result["transits"][0] msg_parts.append(f"\n🚄 从{origin}出发：") msg_parts.append(f" 预计{self._format_duration(transit['duration'])}") # 显示主要交通工具 for seg in transit["segments"]: if seg["type"] == "railway": msg_parts.append(f" {seg['trip']}：{seg['departure_stop']} → {seg['arrival_stop']}") break return { "success": True, "message": "\n".join(msg_parts), "data": info } # ==================== 辅助方法 ==================== def _weekday_cn(self, week: str) -> str: """星期数字转中文""" mapping = {"1": "一", "2": "二", "3": "三", "4": "四", "5": "五", "6": "六", "7": "日"} return mapping.get(str(week), week) def _format_duration(self, seconds: int) -> str: """格式化时长""" if seconds < 60: return f"{seconds}秒" elif seconds < 3600: return f"{seconds // 60}分钟" else: hours = seconds // 3600 minutes = (seconds % 3600) // 60 if minutes: return f"{hours}小时{minutes}分钟" return f"{hours}小时"