first commit

globals/ex_apis.py

@@ -0,0 +1,298 @@
+# 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)