""" 旅行规划插件 基于高德地图 API，提供以下功能： - 地点搜索与地理编码 - 天气查询（实况 + 4天预报） - 景点/酒店/餐厅搜索 - 路径规划（驾车/公交/步行） - 周边搜索 支持 LLM 函数调用，可与 AIChat 插件配合使用。 """ import asyncio 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"] } } }, { "type": "function", "function": { "name": "plan_detailed_trip", "description": "【必须调用】详细行程规划工具。当用户提到'规划行程'、'安排旅行'、'去XX旅游'、'帮我规划'、'我想去XX玩'时，必须调用此工具获取实时的交通、酒店、景点信息。此工具会返回：1.从用户家到火车站的详细路线（地铁几号线、哪站上哪站下）2.高铁车次和时刻 3.酒店推荐 4.景点推荐 5.餐厅推荐 6.天气预报。", "parameters": { "type": "object", "properties": { "origin_city": { "type": "string", "description": "出发城市，如：合肥、上海、北京" }, "origin_address": { "type": "string", "description": "用户的具体出发地址，如：合肥市蜀山区xxx小区、上海市浦东新区xxx路。如果用户没提供具体地址，填写城市名即可" }, "destination": { "type": "string", "description": "目的地城市，如：北京、杭州、成都" }, "days": { "type": "integer", "description": "旅行天数，默认2天" }, "departure_time": { "type": "string", "description": "出发时间偏好，如：周六早上、明天下午" }, "preferences": { "type": "string", "description": "旅行偏好，如：喜欢历史文化、想吃美食、带小孩" } }, "required": ["origin_city", "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) elif tool_name == "plan_detailed_trip": return await self._tool_plan_detailed_trip(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, "need_ai_reply": True # 让 AI 根据这些信息生成详细的行程规划 } async def _tool_plan_detailed_trip(self, args: Dict) -> Dict: """ 详细行程规划工具（优化版：并行 API 调用） """ origin_city = args.get("origin_city", "") origin_address = args.get("origin_address", "") or origin_city destination = args.get("destination", "") days = args.get("days", 2) departure_time = args.get("departure_time", "") preferences = args.get("preferences", "") info = { "origin_city": origin_city, "origin_address": origin_address, "destination": destination, "days": days } # ========== 第1步：并行获取基础信息 ========== user_geo_task = self.amap.geocode(origin_address, origin_city) dest_geo_task = self.amap.geocode(destination) weather_task = self.amap.get_weather(destination, "all") user_geo, dest_geo, weather_result = await asyncio.gather( user_geo_task, dest_geo_task, weather_task, return_exceptions=True ) # 处理地理编码结果 if isinstance(user_geo, Exception) or not user_geo.get("success"): user_geo = await self.amap.geocode(origin_city) if isinstance(dest_geo, Exception) or not dest_geo.get("success"): return {"success": False, "message": f"无法定位目的地：{destination}"} if not user_geo.get("success"): return {"success": False, "message": f"无法定位出发地：{origin_address}"} user_loc = user_geo["location"] dest_loc = dest_geo["location"] origin_city_name = user_geo.get("city") or origin_city dest_city_name = dest_geo.get("city") or destination # ========== 第2步：并行搜索火车站和目的地信息 ========== origin_stations_task = self.amap.search_poi(types="150200", city=origin_city, citylimit=True, offset=3) dest_stations_task = self.amap.search_poi(types="150200", city=destination, citylimit=True, offset=2) hotels_task = self.amap.search_poi(types="100100|100101", city=destination, citylimit=True, offset=5) attractions_task = self.amap.search_poi(types="110000", city=destination, citylimit=True, offset=6) food_task = self.amap.search_poi(types="050000", city=destination, citylimit=True, offset=5) results = await asyncio.gather( origin_stations_task, dest_stations_task, hotels_task, attractions_task, food_task, return_exceptions=True ) origin_stations, dest_stations, hotels, attractions, food = results # ========== 第3步：规划到火车站的路线（只规划1个最近的） ========== best_station = None best_route = None if not isinstance(origin_stations, Exception) and origin_stations.get("success") and origin_stations.get("pois"): station = origin_stations["pois"][0] # 只取第一个火车站 try: route = await self.amap.route_transit(user_loc, station["location"], city=origin_city_name) if route.get("success") and route.get("transits"): best_station = station best_route = route["transits"][0] except Exception as e: logger.warning(f"规划到火车站路线失败: {e}") # ========== 第4步：规划城际交通 ========== transit_info = None if best_station and not isinstance(dest_stations, Exception) and dest_stations.get("success") and dest_stations.get("pois"): try: dest_station = dest_stations["pois"][0] transit = await self.amap.route_transit( best_station["location"], dest_station["location"], city=origin_city_name, cityd=dest_city_name ) if transit.get("success") and transit.get("transits"): transit_info = transit["transits"][0] except Exception as e: logger.warning(f"城际交通规划失败: {e}") # ========== 组装输出 ========== sections = [] sections.append(f"📋 {origin_address} → {destination} {days}天行程\n") sections.append(f"📍 出发地：{user_geo.get('formatted_address', origin_address)}\n") # 天气 if not isinstance(weather_result, Exception) and weather_result.get("success"): sections.append("【天气预报】") for f in weather_result.get("forecasts", [])[:3]: sections.append(f" {f['date']}：{f['dayweather']} {f['nighttemp']}~{f['daytemp']}℃") sections.append("") # 到火车站 if best_station and best_route: sections.append("【从您家到火车站】") sections.append(f" 🚉 {best_station['name']}") sections.append(f" ⏱️ 预计：{self._format_duration(best_route['duration'])}") for seg in best_route.get("segments", []): if seg["type"] == "bus": line = seg["name"] icon = "🚇" if "地铁" in line or "号线" in line else "🚌" sections.append(f" {icon} {line}：{seg['departure_stop']} → {seg['arrival_stop']}（{seg['via_num']}站）") sections.append("") # 城际交通 if transit_info: sections.append("【城际高铁/火车】") sections.append(f" ⏱️ 全程约{self._format_duration(transit_info['duration'])}，费用约{transit_info.get('cost', '未知')}元") for seg in transit_info.get("segments", []): if seg["type"] == "railway": sections.append(f" 🚄 {seg['trip']} {seg['name']}") sections.append(f" {seg['departure_stop']} → {seg['arrival_stop']}") sections.append("") # 酒店 if not isinstance(hotels, Exception) and hotels.get("success") and hotels.get("pois"): sections.append("【酒店推荐】") for i, h in enumerate(hotels["pois"][:4], 1): rating = f"⭐{h['rating']}" if h.get("rating") else "" cost = f"¥{h['cost']}/晚" if h.get("cost") else "" sections.append(f" {i}. {h['name']} {rating} {cost}") sections.append("") # 景点 if not isinstance(attractions, Exception) and attractions.get("success") and attractions.get("pois"): sections.append("【热门景点】") for i, p in enumerate(attractions["pois"][:5], 1): rating = f"⭐{p['rating']}" if p.get("rating") else "" sections.append(f" {i}. {p['name']} {rating}") sections.append("") # 美食 if not isinstance(food, Exception) and food.get("success") and food.get("pois"): sections.append("【美食推荐】") for i, p in enumerate(food["pois"][:4], 1): cost = f"人均¥{p['cost']}" if p.get("cost") else "" sections.append(f" {i}. {p['name']} {cost}") sections.append("") # 提示 sections.append(f"📌 请根据以上信息为用户安排{days}天行程") if departure_time: sections.append(f" 出发时间偏好：{departure_time}") if preferences: sections.append(f" 用户偏好：{preferences}") return { "success": True, "message": "\n".join(sections), "data": info, "need_ai_reply": True } # ==================== 辅助方法 ==================== 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}小时"