93 lines
3.3 KiB
Python
93 lines
3.3 KiB
Python
"""天气查询工具"""
|
||
import httpx
|
||
from typing import Optional
|
||
|
||
|
||
async def get_weather(
|
||
city: str,
|
||
units: str = "metric",
|
||
language: str = "zh_cn"
|
||
) -> dict:
|
||
"""
|
||
获取指定城市的天气信息
|
||
|
||
Args:
|
||
city: 城市名称，支持中文城市名（如"北京"、"上海"）或英文城市名（如"Beijing"、"London"）
|
||
units: 单位系统，"metric"（摄氏度）或 "imperial"（华氏度），默认为 "metric"
|
||
language: 返回结果的语言，默认为中文（zh_cn）
|
||
|
||
Returns:
|
||
dict: 包含天气信息的字典，包括：
|
||
- city: 城市名称
|
||
- temperature: 当前温度
|
||
- description: 天气描述
|
||
- humidity: 湿度百分比
|
||
- wind_speed: 风<><E9A38E>
|
||
- feels_like: 体感温度
|
||
- temp_min: 最低温度
|
||
- temp_max: 最高温度
|
||
|
||
Example:
|
||
>>> result = await get_weather("北京")
|
||
>>> print(result['temperature'])
|
||
15
|
||
|
||
>>> result = await get_weather("Tokyo", units="metric")
|
||
>>> print(result['description'])
|
||
晴朗
|
||
"""
|
||
# 使用 OpenWeatherMap API 的免费端点
|
||
# 注意：实际使用需要申请 API key，这里使用 wttr.in 作为免费替代
|
||
base_url = "https://wttr.in"
|
||
|
||
params = {
|
||
"format": "j1", # JSON 格式
|
||
"lang": language,
|
||
}
|
||
|
||
# 处理单位系统
|
||
if units == "imperial":
|
||
params["u"] = "f" # 华氏度
|
||
else:
|
||
params["u"] = "c" # 摄氏度（默认）
|
||
|
||
try:
|
||
async with httpx.AsyncClient(timeout=10.0) as client:
|
||
response = await client.get(f"{base_url}/{city}", params=params)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
|
||
# 解析当前天气
|
||
current = data.get("current_condition", [{}])[0]
|
||
|
||
# 解析位置信息
|
||
location = data.get("nearest_area", [{}])[0]
|
||
|
||
return {
|
||
"city": location.get("areaName", [{}])[0].get("value", city),
|
||
"country": location.get("country", [{}])[0].get("value", ""),
|
||
"temperature": float(current.get("temp_C", 0) if units == "metric" else current.get("temp_F", 0)),
|
||
"description": current.get("weatherDesc", [{}])[0].get("value", "未知"),
|
||
"humidity": int(current.get("humidity", 0)),
|
||
"wind_speed": float(current.get("windspeedKmph", 0)),
|
||
"feels_like": float(current.get("FeelsLikeC", 0) if units == "metric" else current.get("FeelsLikeF", 0)),
|
||
"temp_min": float(current.get("temp_C", 0) if units == "metric" else current.get("temp_F", 0)), # wttr.in 不提供 min/max，使用当前温度
|
||
"temp_max": float(current.get("temp_C", 0) if units == "metric" else current.get("temp_F", 0)),
|
||
"uv_index": int(current.get("uvIndex", 0)),
|
||
}
|
||
except httpx.TimeoutException:
|
||
return {
|
||
"error": "请求超时，请稍后重试",
|
||
"city": city
|
||
}
|
||
except httpx.HTTPError as e:
|
||
return {
|
||
"error": f"获取天气信息失败: {str(e)}",
|
||
"city": city
|
||
}
|
||
except (KeyError, IndexError, ValueError) as e:
|
||
return {
|
||
"error": f"解析天气数据失败: {str(e)}",
|
||
"city": city
|
||
}
|