mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-11-01 05:15:22 +08:00
129 lines
4.6 KiB
Python
129 lines
4.6 KiB
Python
# Auto generated by LangBot Plugin SDK.
|
|
# Please refer to https://langbot.app/docs/en/plugin/dev/tutor.html for more details.
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, AsyncGenerator
|
|
import logging
|
|
import traceback
|
|
|
|
from langbot_plugin.api.definition.components.command.command import Command, Subcommand
|
|
from langbot_plugin.api.entities.builtin.command.context import ExecuteContext, CommandReturn
|
|
from langbot_plugin.api.entities.builtin.command.errors import CommandError
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class Weather(Command):
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
|
|
@self.subcommand(
|
|
name="*",
|
|
help="Display weather information for a city",
|
|
usage="weather <city_name>",
|
|
aliases=["天气"],
|
|
)
|
|
async def get_weather(self, context: ExecuteContext) -> AsyncGenerator[CommandReturn, None]:
|
|
"""Get weather information for a city"""
|
|
if not context.crt_params:
|
|
yield CommandReturn(
|
|
text="请提供城市名称，例如: !weather 北京",
|
|
)
|
|
return
|
|
|
|
city = context.crt_params[0]
|
|
config = self.plugin.get_config()
|
|
|
|
try:
|
|
# Use QWeather API to get weather data
|
|
weather_text = await self._get_qweather_text(city, config)
|
|
|
|
if weather_text:
|
|
yield CommandReturn(
|
|
text=weather_text,
|
|
)
|
|
else:
|
|
yield CommandReturn(
|
|
error=CommandError("天气信息获取失败，请检查城市名称是否正确"),
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"获取天气信息失败: {str(e)}")
|
|
traceback.print_exc()
|
|
yield CommandReturn(
|
|
error=CommandError(f"获取天气信息失败: {str(e)}")
|
|
)
|
|
|
|
async def _get_qweather_text(self, city: str, config: dict) -> str:
|
|
"""Get weather using QWeather API and format as text"""
|
|
try:
|
|
from pkg.weather_data import Weather as WeatherData, CityNotFoundError
|
|
|
|
api_key = config.get('qweather_apikey', '')
|
|
api_type = int(config.get('qweather_apitype', '0'))
|
|
|
|
if not api_key:
|
|
raise ValueError("未配置和风天气 API Key，请在插件配置中填写")
|
|
|
|
# Get weather data
|
|
w_data = WeatherData(city_name=city, api_key=api_key, api_type=api_type)
|
|
await w_data.load_data()
|
|
|
|
# Format as text
|
|
weather_text = self._format_weather_text(w_data)
|
|
return weather_text
|
|
|
|
except CityNotFoundError:
|
|
logger.error(f"City not found: {city}")
|
|
return ''
|
|
except Exception as e:
|
|
logger.error(f"QWeather API error: {str(e)}")
|
|
raise
|
|
|
|
def _format_weather_text(self, weather) -> str:
|
|
"""Format weather data as text"""
|
|
lines = []
|
|
lines.append(f"📍 {weather.city_name} 天气信息")
|
|
lines.append("")
|
|
|
|
# Current weather
|
|
now = weather.now.now
|
|
lines.append(f"🌡️ 当前天气")
|
|
lines.append(f" 温度: {now.temp}°C")
|
|
lines.append(f" 天气: {now.text}")
|
|
lines.append(f" 风向: {now.windDir} {now.windScale}级")
|
|
lines.append(f" 湿度: {now.humidity}%")
|
|
lines.append(f" 能见度: {now.vis}km")
|
|
lines.append("")
|
|
|
|
# Air quality
|
|
if weather.air and weather.air.now:
|
|
air = weather.air.now
|
|
lines.append(f"💨 空气质量")
|
|
lines.append(f" AQI: {air.aqi} ({air.category})")
|
|
lines.append(f" PM2.5: {air.pm2p5}")
|
|
lines.append("")
|
|
|
|
# Forecast
|
|
if weather.daily and weather.daily.daily:
|
|
lines.append(f"📅 未来{len(weather.daily.daily)}天预报")
|
|
for day in weather.daily.daily[:3]: # Show first 3 days
|
|
lines.append(f" {day.fxDate}: {day.textDay} {day.tempMin}~{day.tempMax}°C")
|
|
lines.append("")
|
|
|
|
# Weather warnings
|
|
if weather.warning and weather.warning.warning:
|
|
lines.append(f"⚠️ 天气预警")
|
|
for warn in weather.warning.warning:
|
|
lines.append(f" {warn.type}: {warn.title}")
|
|
lines.append("")
|
|
|
|
# Sun times
|
|
if weather.sun:
|
|
lines.append(f"🌅 日出日落")
|
|
lines.append(f" 日出: {weather.sun.sunrise}")
|
|
lines.append(f" 日落: {weather.sun.sunset}")
|
|
|
|
return "\n".join(lines)
|