# external_apis.py import requests import json import logging import random from typing import Dict, Tuple, Optional import time class WeatherAPI: def __init__(self): self.logger = logging.getLogger(__name__) # 使用腾讯天气API self.base_url = "https://wis.qq.com/weather/common" def parse_city(self, city_string: str) -> Tuple[str, str, str]: """ 解析城市字符串，返回省份、城市、区县 参数: city_string: 完整的地址字符串 返回: (province, city, county) """ # 匹配省份或自治区 province_regex = r"(.*?)(省|自治区)" # 匹配城市或州 city_regex = r"(.*?省|.*?自治区)(.*?市|.*?州)" # 匹配区、县或镇 county_regex = r"(.*?市|.*?州)(.*?)(区|县|镇)" province = "" city = "" county = "" import re # 先尝试匹配省份或自治区 province_match = re.search(province_regex, city_string) if province_match: province = province_match.group(1).strip() # 然后尝试匹配城市或州 city_match = re.search(city_regex, city_string) if city_match: city = city_match.group(2).strip() else: # 如果没有匹配到城市，则可能是直辖市或者直接是区/县 city = city_string # 最后尝试匹配区、县或镇 county_match = re.search(county_regex, city_string) if county_match: county = county_match.group(2).strip() # 如果有区、县或镇，那么前面的城市部分需要重新解析 if city_match: city = city_match.group(2).strip() # 特殊情况处理，去除重复的省市名称 if city and province and city.startswith(province): city = city.replace(province, "").strip() if county and city and county.startswith(city): county = county.replace(city, "").strip() # 去除后缀 city = city.rstrip('市州') if county: county = county.rstrip('区县镇') # self.logger.info(f"解析结果 - 省份: {province}, 城市: {city}, 区县: {county}") return province, city, county def get_weather_by_qq_api(self, province: str, city: str, county: str) -> Optional[Dict]: """ 使用腾讯天气API获取天气信息 参数: province: 省份 city: 城市 county: 区县 返回: 天气信息字典 or None """ try: params = { 'source': 'pc', 'weather_type': 'observe', 'province': province, 'city': city, 'county': county } response = requests.get(self.base_url, params=params, timeout=10) response.raise_for_status() data = response.json() if data.get('status') == 200: observe_data = data.get('data', {}).get('observe', {}) return { 'weather': observe_data.get('weather', ''), 'temperature': observe_data.get('degree', ''), 'pressure': observe_data.get('pressure', '1013') } else: self.logger.error(f"腾讯天气API错误: {data.get('message')}") return None except Exception as e: self.logger.error(f"腾讯天气API调用失败: {str(e)}") return None def normalize_weather_text(self, weather_text: str) -> str: """ 将天气描述标准化为: 晴；阴；雨；雪；风；其他 参数: weather_text: 原始天气描述 返回: 标准化后的天气文本 """ if not weather_text: return '其他' weather_text_lower = weather_text.lower() # 晴 if any(word in weather_text_lower for word in ['晴', 'sunny', 'clear']): return '晴' # 阴 elif any(word in weather_text_lower for word in ['阴', '多云', 'cloudy', 'overcast']): return '阴' # 雨 elif any(word in weather_text_lower for word in ['雨', 'rain', 'drizzle', 'shower']): return '雨' # 雪 elif any(word in weather_text_lower for word in ['雪', 'snow']): return '雪' # 风 elif any(word in weather_text_lower for word in ['风', 'wind']): return '风' # 其他 else: return '其他' def adjust_pressure(self, pressure: float) -> float: """ 调整气压值：低于700时，填700-750之间随机一个；高于700就按实际情况填 参数: pressure: 原始气压值 返回: 调整后的气压值 """ try: pressure_float = float(pressure) if pressure_float < 700: adjusted_pressure = random.randint(700, 750) self.logger.info(f"气压值 {pressure_float} 低于700，调整为: {adjusted_pressure:.1f}") return round(adjusted_pressure, 1) else: self.logger.info(f"使用实际气压值: {pressure_float}") return round(pressure_float, 1) except (ValueError, TypeError): self.logger.warning(f"气压值格式错误: {pressure}，使用默认值720") return round(random.randint(700, 750), 1) def get_weather_by_address(self, address: str, max_retries: int = 2) -> Optional[Dict]: """ 根据地址获取天气信息 参数: address: 地址字符串 max_retries: 最大重试次数 返回: { 'weather': '晴/阴/雨/雪/风/其他', 'temperature': 温度值, 'pressure': 气压值 } or None """ # self.logger.info(f"开始获取地址 '{address}' 的天气信息") # 首先解析地址 province, city, county = self.parse_city(address) if not province and not city: self.logger.error("无法解析地址") return self.get_fallback_weather() # 获取天气信息 weather_data = None for attempt in range(max_retries): try: # self.logger.info(f"尝试获取天气信息 (第{attempt + 1}次)") weather_data = self.get_weather_by_qq_api(province, city, county) if weather_data: break time.sleep(1) # 短暂延迟后重试 except Exception as e: self.logger.warning(f"第{attempt + 1}次尝试失败: {str(e)}") time.sleep(1) if not weather_data: self.logger.warning("获取天气信息失败，使用备用数据") return self.get_fallback_weather() # 处理天气数据 try: # 标准化天气文本 normalized_weather = self.normalize_weather_text(weather_data['weather']) # 调整气压值 adjusted_pressure = self.adjust_pressure(weather_data['pressure']) # 处理温度 temperature = float(weather_data['temperature']) result = { 'weather': normalized_weather, 'temperature': round(temperature, 1), 'pressure': adjusted_pressure } self.logger.info(f"成功获取天气信息: {result}") return result except Exception as e: self.logger.error(f"处理天气数据时出错: {str(e)}") return self.get_fallback_weather() def get_fallback_weather(self) -> Dict: """ 获取备用天气数据（当所有API都失败时使用） 返回: 默认天气数据 """ self.logger.info("使用备用天气数据") return { 'weather': '阴', 'temperature': round(random.randint(15, 30), 1), 'pressure': round(random.randint(700, 750), 1) } def get_weather_simple(self, address: str) -> Tuple[str, float, float]: """ 简化接口：直接返回天气、温度、气压 参数: address: 地址字符串 返回: (weather, temperature, pressure) """ weather_data = self.get_weather_by_address(address) if weather_data: return weather_data['weather'], weather_data['temperature'], weather_data['pressure'] else: fallback = self.get_fallback_weather() return fallback['weather'], fallback['temperature'], fallback['pressure'] # 创建全局实例 weather_api = WeatherAPI() # 直接可用的函数 def get_weather_by_address(address: str) -> Optional[Dict]: """ 根据地址获取天气信息（直接调用函数） 参数: address: 地址字符串 返回: { 'weather': '晴/阴/雨/雪/风/其他', 'temperature': 温度值, 'pressure': 气压值 } or None """ return weather_api.get_weather_by_address(address) def get_weather_simple(address: str) -> Tuple[str, float, float]: """ 简化接口：直接返回天气、温度、气压 参数: address: 地址字符串 返回: (weather, temperature, pressure) """ return weather_api.get_weather_simple(address)