159 lines
5.4 KiB
Python
159 lines
5.4 KiB
Python
import requests
|
||
import json
|
||
from datetime import datetime
|
||
|
||
class QWeatherAPI:
|
||
def __init__(self, api_key):
|
||
self.api_key = api_key
|
||
self.base_url = "https://devapi.qweather.com/v7/weather/now"
|
||
self.geo_url = "https://geoapi.qweather.com/v2/city/lookup"
|
||
|
||
def verify_api_key(self):
|
||
"""
|
||
验证API密钥是否有效
|
||
"""
|
||
try:
|
||
# 使用一个简单的城市ID测试API
|
||
test_params = {
|
||
"location": "101010100", # 北京的城市ID
|
||
"key": self.api_key
|
||
}
|
||
|
||
response = requests.get(self.base_url, params=test_params)
|
||
data = response.json()
|
||
|
||
if data["code"] == "402":
|
||
print("错误: API密钥无效或已过期")
|
||
print("请检查:")
|
||
print("1. API密钥是否正确")
|
||
print("2. 是否使用了正确的API版本（开发版/正式版）")
|
||
print("3. API密钥是否已过期")
|
||
return False
|
||
elif data["code"] == "200":
|
||
return True
|
||
else:
|
||
print(f"API返回错误: {data['code']}")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"验证API密钥时发生错误: {e}")
|
||
return False
|
||
|
||
def search_city(self, location_name, range=None, number=1):
|
||
"""
|
||
搜索城市
|
||
:param location_name: 城市名称，如 "北京"、"上海" 等
|
||
:param range: 搜索范围，使用ISO 3166国家代码，如 "cn"
|
||
:param number: 返回结果数量，范围1-20，默认1
|
||
:return: 城市信息列表
|
||
"""
|
||
try:
|
||
params = {
|
||
"location": location_name,
|
||
"key": self.api_key,
|
||
"number": min(max(1, number), 20) # 确保number在1-20之间
|
||
}
|
||
|
||
# 添加可选参数
|
||
if range:
|
||
params["range"] = range
|
||
|
||
response = requests.get(self.geo_url, params=params)
|
||
response.raise_for_status()
|
||
|
||
data = response.json()
|
||
|
||
if data["code"] == "402":
|
||
print(data)
|
||
print("错误: API密钥无效或已过期")
|
||
print("请检查:")
|
||
print("1. API密钥是否正确")
|
||
print("2. 是否使用了正确的API版本（开发版/正式版）")
|
||
print("3. API密钥是否已过期")
|
||
return None
|
||
elif data["code"] != "200":
|
||
print(f"城市查询API返回错误: {data['code']}")
|
||
return None
|
||
|
||
return data.get("location", [])
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"请求错误: {e}")
|
||
return None
|
||
except json.JSONDecodeError as e:
|
||
print(f"JSON解析错误: {e}")
|
||
return None
|
||
except Exception as e:
|
||
print(f"未知错误: {e}")
|
||
return None
|
||
|
||
def get_weather(self, location):
|
||
"""
|
||
获取实时天气数据
|
||
:param location: 城市ID或经纬度坐标，如 "101010100" 或 "116.41,39.92"
|
||
:return: 天气数据字典
|
||
"""
|
||
try:
|
||
params = {
|
||
"location": location,
|
||
"key": self.api_key
|
||
}
|
||
|
||
response = requests.get(self.base_url, params=params)
|
||
response.raise_for_status() # 检查HTTP错误
|
||
|
||
data = response.json()
|
||
|
||
if data["code"] == "402":
|
||
print("错误: API密钥无效或已过期")
|
||
print("请检查:")
|
||
print("1. API密钥是否正确")
|
||
print("2. 是否使用了正确的API版本（开发版/正式版）")
|
||
print("3. API密钥是否已过期")
|
||
return None
|
||
elif data["code"] != "200":
|
||
print(f"API返回错误: {data['code']}")
|
||
return None
|
||
|
||
return data
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"请求错误: {e}")
|
||
return None
|
||
except json.JSONDecodeError as e:
|
||
print(f"JSON解析错误: {e}")
|
||
return None
|
||
except Exception as e:
|
||
print(f"未知错误: {e}")
|
||
return None
|
||
|
||
def format_weather_data(data, city_info):
|
||
"""
|
||
格式化天气数据为易读的格式
|
||
"""
|
||
if not data or "now" not in data:
|
||
return "无法获取天气数据"
|
||
|
||
now = data["now"]
|
||
return f"""
|
||
城市信息:
|
||
名称: {city_info['name']}
|
||
所属地区: {city_info.get('adm2', 'N/A')}
|
||
所属省份: {city_info.get('adm1', 'N/A')}
|
||
所属国家: {city_info.get('country', 'N/A')}
|
||
|
||
实时天气信息:
|
||
观测时间: {now['obsTime']}
|
||
温度: {now['temp']}°C
|
||
体感温度: {now['feelsLike']}°C
|
||
天气状况: {now['text']}
|
||
风向: {now['windDir']}
|
||
风力等级: {now['windScale']}级
|
||
风速: {now['windSpeed']}km/h
|
||
相对湿度: {now['humidity']}%
|
||
降水量: {now['precip']}mm
|
||
大气压强: {now['pressure']}hPa
|
||
能见度: {now['vis']}km
|
||
云量: {now['cloud']}%
|
||
露点温度: {now['dew']}°C
|
||
""" |