From fbd25353d5e36a1372c6398f8cddf20cf155f5dd Mon Sep 17 00:00:00 2001 From: koalalgx <82762701+koalalgx@users.noreply.github.com> Date: Tue, 25 Mar 2025 14:29:28 +0800 Subject: [PATCH] Update get_weather.py --- .../plugins_func/functions/get_weather.py | 198 ++++++++++++------ 1 file changed, 136 insertions(+), 62 deletions(-) diff --git a/main/xiaozhi-server/plugins_func/functions/get_weather.py b/main/xiaozhi-server/plugins_func/functions/get_weather.py index 8cdd6ea1..58f63844 100644 --- a/main/xiaozhi-server/plugins_func/functions/get_weather.py +++ b/main/xiaozhi-server/plugins_func/functions/get_weather.py @@ -2,6 +2,7 @@ import requests from bs4 import BeautifulSoup from config.logger import setup_logging from plugins_func.register import register_function, ToolType, ActionResponse, Action +import json TAG = __name__ logger = setup_logging() @@ -39,85 +40,158 @@ HEADERS = { ) } -# 天气代码 https://dev.qweather.com/docs/resource/icons/#weather-icons -WEATHER_CODE_MAP = { - "100": "晴", "101": "多云", "102": "少云", "103": "晴间多云", "104": "阴", - "150": "晴", "151": "多云", "152": "少云", "153": "晴间多云", - "300": "阵雨", "301": "强阵雨", "302": "雷阵雨", "303": "强雷阵雨", "304": "雷阵雨伴有冰雹", - "305": "小雨", "306": "中雨", "307": "大雨", "308": "极端降雨", "309": "毛毛雨/细雨", - "310": "暴雨", "311": "大暴雨", "312": "特大暴雨", "313": "冻雨", "314": "小到中雨", - "315": "中到大雨", "316": "大到暴雨", "317": "暴雨到大暴雨", "318": "大暴雨到特大暴雨", - "350": "阵雨", "351": "强阵雨", "399": "雨", - "400": "小雪", "401": "中雪", "402": "大雪", "403": "暴雪", "404": "雨夹雪", - "405": "雨雪天气", "406": "阵雨夹雪", "407": "阵雪", "408": "小到中雪", "409": "中到大雪", "410": "大到暴雪", - "456": "阵雨夹雪", "457": "阵雪", "499": "雪", - "500": "薄雾", "501": "雾", "502": "霾", "503": "扬沙", "504": "浮尘", - "507": "沙尘暴", "508": "强沙尘暴", - "509": "浓雾", "510": "强浓雾", "511": "中度霾", "512": "重度霾", "513": "严重霾", "514": "大雾", "515": "特强浓雾", - "900": "热", "901": "冷", "999": "未知" -} def fetch_city_info(location, api_key): url = f"https://geoapi.qweather.com/v2/city/lookup?key={api_key}&location={location}&lang=zh" - response = requests.get(url, headers=HEADERS).json() - return response.get('location', [])[0] if response.get('location') else None + logger.bind(tag=TAG).info(f"正在请求城市信息API，URL: {url}") + try: + response = requests.get(url, headers=HEADERS) + logger.bind(tag=TAG).info(f"城市信息API响应状态码: {response.status_code}") + json_data = response.json() + logger.bind(tag=TAG).debug(f"城市信息API响应: {json.dumps(json_data, ensure_ascii=False)}") + return json_data.get('location', [])[0] if json_data.get('location') else None + except Exception as e: + logger.bind(tag=TAG).error(f"获取城市信息失败: {str(e)}") + return None def fetch_weather_page(url): - response = requests.get(url, headers=HEADERS) - return BeautifulSoup(response.text, "html.parser") if response.ok else None + logger.bind(tag=TAG).info(f"正在请求天气页面，URL: {url}") + try: + response = requests.get(url, headers=HEADERS) + logger.bind(tag=TAG).info(f"天气页面响应状态码: {response.status_code}") + if not response.ok: + logger.bind(tag=TAG).error(f"获取天气页面失败: {response.text[:200]}") + return BeautifulSoup(response.text, "html.parser") if response.ok else None + except Exception as e: + logger.bind(tag=TAG).error(f"获取天气页面失败: {str(e)}") + return None def parse_weather_info(soup): - city_name = soup.select_one("h1.c-submenu__location").get_text(strip=True) + try: + city_name = soup.select_one("h1.c-submenu__location").get_text(strip=True) + logger.bind(tag=TAG).info(f"解析到城市名称: {city_name}") - current_abstract = soup.select_one(".c-city-weather-current .current-abstract") - current_abstract = current_abstract.get_text(strip=True) if current_abstract else "未知" + current_abstract = soup.select_one(".c-city-weather-current .current-abstract") + current_abstract = current_abstract.get_text(strip=True) if current_abstract else "未知" + logger.bind(tag=TAG).info(f"当前天气概况: {current_abstract}") - current_basic = {} - for item in soup.select(".c-city-weather-current .current-basic .current-basic___item"): - parts = item.get_text(strip=True, separator=" ").split(" ") - if len(parts) == 2: - key, value = parts[1], parts[0] - current_basic[key] = value + current_basic = {} + for item in soup.select(".c-city-weather-current .current-basic .current-basic___item"): + parts = item.get_text(strip=True, separator=" ").split(" ") + if len(parts) == 2: + key, value = parts[1], parts[0] + current_basic[key] = value + logger.bind(tag=TAG).info(f"当前天气详情: {json.dumps(current_basic, ensure_ascii=False)}") - temps_list = [] - for row in soup.select(".city-forecast-tabs__row")[:7]: # 取前7天的数据 - date = row.select_one(".date-bg .date").get_text(strip=True) - weather_code = row.select_one(".date-bg .icon")["src"].split("/")[-1].split(".")[0] - weather = WEATHER_CODE_MAP.get(weather_code, "未知") - temps = [span.get_text(strip=True) for span in row.select(".tmp-cont .temp")] - high_temp, low_temp = (temps[0], temps[-1]) if len(temps) >= 2 else (None, None) - temps_list.append((date, weather, high_temp, low_temp)) + temps_list = [] + for row in soup.select(".city-forecast-tabs__row")[:7]: # 取前7天的数据 + date = row.select_one(".date-bg .date").get_text(strip=True) + temps = [span.get_text(strip=True) for span in row.select(".tmp-cont .temp")] + high_temp, low_temp = (temps[0], temps[-1]) if len(temps) >= 2 else (None, None) + temps_list.append((date, high_temp, low_temp)) + logger.bind(tag=TAG).info(f"获取到未来7天温度: {temps_list}") - return city_name, current_abstract, current_basic, temps_list + return city_name, current_abstract, current_basic, temps_list + except Exception as e: + logger.bind(tag=TAG).error(f"解析天气信息失败: {str(e)}") + return "未知城市", "未知天气", {}, [] @register_function('get_weather', GET_WEATHER_FUNCTION_DESC, ToolType.SYSTEM_CTL) def get_weather(conn, location: str = None, lang: str = "zh_CN"): - api_key = conn.config["plugins"]["get_weather"]["api_key"] - default_location = conn.config["plugins"]["get_weather"]["default_location"] - location = location or conn.client_ip_info.get("city") or default_location - logger.bind(tag=TAG).debug(f"获取天气: {location}") + logger.bind(tag=TAG).info(f"===== 天气查询函数开始执行 =====") + logger.bind(tag=TAG).info(f"查询参数: location={location}, lang={lang}") + + try: + api_key = conn.config["plugins"]["get_weather"]["api_key"] + default_location = conn.config["plugins"]["get_weather"]["default_location"] + + # 位置参数处理逻辑优化 + # 1. 如果传入了明确的位置参数且不为None/空字符串，则使用该参数 + # 2. 否则，尝试使用客户端IP地址对应的城市信息 + # 3. 如果上述都无效，则使用配置文件中的默认位置 + if location and location.strip(): + final_location = location.strip() + logger.bind(tag=TAG).info(f"使用用户明确指定的位置: {final_location}") + elif conn.client_ip_info and "city" in conn.client_ip_info and conn.client_ip_info["city"]: + final_location = conn.client_ip_info["city"] + logger.bind(tag=TAG).info(f"使用IP地址解析的位置: {final_location}") + else: + final_location = default_location + logger.bind(tag=TAG).info(f"使用配置文件中的默认位置: {final_location}") + + logger.bind(tag=TAG).info(f"最终使用的查询地点: {final_location}, API Key: {api_key}") - city_info = fetch_city_info(location, api_key) - if not city_info: - return ActionResponse(Action.REQLLM, f"未找到相关的城市: {location}，请确认地点是否正确", None) + city_info = fetch_city_info(final_location, api_key) + if not city_info: + error_msg = f"未找到相关的城市: {final_location}，请确认地点是否正确" + logger.bind(tag=TAG).error(error_msg) + # 构造一个直接回复 + direct_response = f"抱歉，我未能找到\"{final_location}\"的天气信息，请确认地点名称是否正确，或者尝试查询其他城市的天气。" + return ActionResponse(Action.RESPONSE, None, direct_response) # 使用RESPONSE动作直接返回 - soup = fetch_weather_page(city_info['fxLink']) - if not soup: - return ActionResponse(Action.REQLLM, None, "请求失败") + logger.bind(tag=TAG).info(f"获取到城市信息: {json.dumps(city_info, ensure_ascii=False)}") + logger.bind(tag=TAG).info(f"天气查询链接: {city_info['fxLink']}") + + soup = fetch_weather_page(city_info['fxLink']) + if not soup: + error_msg = "请求天气页面失败" + logger.bind(tag=TAG).error(error_msg) + direct_response = f"抱歉，在获取{final_location}的天气信息时遇到了问题，请稍后再试。" + return ActionResponse(Action.RESPONSE, None, direct_response) # 使用RESPONSE动作直接返回 - city_name, current_abstract, current_basic, temps_list = parse_weather_info(soup) - weather_report = f"根据下列数据，用{lang}回应用户的查询天气请求：\n{city_name}未来7天天气:\n" - for i, (date, weather, high, low) in enumerate(temps_list): - if high and low: - weather_report += f"{date}: {low}到{high}, {weather}\n" - weather_report += ( - f"当前天气: {current_abstract}\n" - f"当前天气参数: {current_basic}\n" - f"(确保只报告指定单日的天气情况，除非未来会出现异常天气；或者用户明确要求想要了解多日天气，如果未指定，默认报告今天的天气。" - "参数为0的值不需要报告给用户，每次都报告体感温度，根据语境选择合适的参数内容告知用户，并对参数给出相应评价)" - ) - - return ActionResponse(Action.REQLLM, weather_report, None) + city_name, current_abstract, current_basic, temps_list = parse_weather_info(soup) + + # 构建天气报告（给LLM使用的详细数据） + weather_report = f"根据下列数据，用{lang}回应用户的查询天气请求：\n{city_name}未来7天天气:\n" + for i, (date, high, low) in enumerate(temps_list): + if high and low: + weather_report += f"{date}: {low}到{high}\n" + weather_report += ( + f"当前天气: {current_abstract}\n" + f"当前天气参数: {current_basic}\n" + f"(确保只报告指定单日的气温范围，除非用户明确要求想要了解多日天气，如果未指定，默认报告今天的温度范围。" + "参数为0的值不需要报告给用户，每次都报告体感温度，根据语境选择合适的参数内容告知用户，并对参数给出相应评价)" + ) + + # 同时构建一个直接可用的人性化天气回复 + # 这用作备用，防止LLM处理失败时能够直接回复用户 + today_temp = "" + for date, high, low in temps_list: + if "今天" in date or "今日" in date: + today_temp = f"{low}到{high}" + break + if not today_temp and temps_list: + today_temp = f"{temps_list[0][2]}到{temps_list[0][1]}" + + feel_temp = current_basic.get("体感温度", "") + humidity = current_basic.get("相对湿度", "") + + direct_response = f"{city_name}今天{current_abstract}，温度{today_temp}。" + if feel_temp: + direct_response += f"体感温度{feel_temp}。" + if humidity: + direct_response += f"相对湿度{humidity}。" + + # 添加简单的建议 + if "雨" in current_abstract: + direct_response += "外出请记得带伞。" + elif temps_list and temps_list[0][1] and int(temps_list[0][1].replace("°", "")) > 30: + direct_response += "天气炎热，注意防暑。" + elif temps_list and temps_list[0][2] and int(temps_list[0][2].replace("°", "")) < 10: + direct_response += "天气较冷，注意保暖。" + + logger.bind(tag=TAG).info(f"构建的天气报告: {weather_report}") + logger.bind(tag=TAG).info(f"备用直接回复: {direct_response}") + logger.bind(tag=TAG).info(f"===== 天气查询函数执行完毕，返回结果 =====") + + # 返回详细天气报告，但添加direct_response作为备用 + return ActionResponse(Action.REQLLM, weather_report, direct_response) + except Exception as e: + error_msg = f"查询天气时发生错误: {str(e)}" + logger.bind(tag=TAG).error(error_msg) + # 出错时也提供一个直接回复 + direct_response = f"抱歉，在获取天气信息时遇到了技术问题，请稍后再试。" + return ActionResponse(Action.RESPONSE, None, direct_response) # 使用RESPONSE动作直接返回