299 lines
9.9 KiB
Python
299 lines
9.9 KiB
Python
# 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)
|