别再问怎么给QQ机器人加功能了！手把手教你用Nonebot2写一个天气查询插件（附完整代码）

NoneBot2实战从零构建智能QQ机器人天气查询插件在当今即时通讯生态中智能机器人已成为提升社群互动效率的利器。本文将深入探讨如何基于Python的NoneBot2框架为QQ机器人开发一个功能完备的天气查询插件。不同于基础教程我们聚焦于工程化实现与最佳实践涵盖API集成、异常处理、性能优化等进阶话题。1. 开发环境与项目结构1.1 环境配置要点确保已安装以下核心组件Python 3.8推荐3.10NoneBot2 2.0.0QQ适配器如nonebot-adapter-qqpip install nonebot2 nonebot-adapter-qq httpx项目目录结构示例my_bot/ ├── bot.py # 入口文件 ├── pyproject.toml # 插件配置 └── plugins/ └── weather.py # 天气插件1.2 基础通信架构NoneBot2采用事件驱动模型其核心流程为QQ消息 → 2. 适配器解析 → 3. 事件处理器 → 4. 插件响应在bot.py中初始化机器人import nonebot from nonebot.adapters.qq import Adapter nonebot.init() driver nonebot.get_driver() driver.register_adapter(Adapter) nonebot.load_plugins(plugins) # 自动加载插件目录 if __name__ __main__: nonebot.run()2. 天气插件核心实现2.1 命令响应器设计使用on_command创建天气查询指令from nonebot import on_command from nonebot.rule import to_me from nonebot.params import CommandArg weather on_command( 天气, ruleto_me(), aliases{weather, 查天气}, priority5, blockTrue )参数解析技巧weather.handle() async def handle_location(args: str CommandArg()): if not args.extract_plain_text().strip(): await weather.finish(请输入查询地点例如/天气 北京)2.2 天气API集成推荐使用高稳定性的和风天气APIimport httpx from typing import Dict, Optional async def fetch_weather(location: str) - Optional[Dict]: base_url https://devapi.qweather.com/v7/weather/now params { location: location, key: YOUR_API_KEY, # 替换为实际key lang: zh } async with httpx.AsyncClient() as client: try: resp await client.get(base_url, paramsparams, timeout5.0) return resp.json() if resp.status_code 200 else None except (httpx.TimeoutException, httpx.NetworkError): return None提示实际开发中应将API密钥存储在环境变量中避免硬编码2.3 数据格式化输出将API响应转换为用户友好格式def format_weather(data: Dict) - str: if not data or data.get(code) ! 200: return 天气数据获取失败请稍后重试 now data[now] return ( f当前天气{now[text]}\n f温度{now[temp]}℃ | 体感{now[feelsLike]}℃\n f湿度{now[humidity]}% | 风向{now[windDir]}\n f空气质量{now[airQuality][category]} )3. 进阶功能实现3.1 缓存机制优化使用lru_cache减少API调用from functools import lru_cache import time lru_cache(maxsize32) async def get_cached_weather(location: str): # 设置10分钟缓存 if cached : cache.get(location): if time.time() - cached[timestamp] 600: return cached[data] data await fetch_weather(location) if data: cache[location] {data: data, timestamp: time.time()} return data3.2 异常处理策略完善错误处理流程weather.got(location, prompt请输入要查询的城市名称) async def handle_weather(location: str ArgPlainText()): try: if not (data : await get_cached_weather(location)): await weather.finish(服务暂时不可用) await weather.finish(format_weather(data)) except Exception as e: logger.error(fWeather query failed: {e}) await weather.finish(查询出现异常)3.3 可视化增强使用MessageSegment实现富媒体回复from nonebot.adapters.qq import MessageSegment async def send_weather_card(location: str): data await fetch_weather(location) if not data: return None return MessageSegment.markdown( f ## {location}天气 - 状态{data[now][text]} - 温度{data[now][temp]}℃ - 湿度{data[now][humidity]}%  )4. 部署与监控4.1 性能监控配置添加Prometheus监控指标from prometheus_client import Counter WEATHER_QUERIES Counter( weather_query_total, Total weather queries, [location] ) weather.handle() async def track_usage(location: str CommandArg()): WEATHER_QUERIES.labels(locationlocation).inc()4.2 日志结构化配置JSON格式日志import json_logging import logging json_logging.init_non_web() logger logging.getLogger(weather-plugin) logger.addHandler(logging.StreamHandler()) logger.setLevel(logging.INFO)典型日志输出示例{ time: 2023-08-20T14:30:00Z, level: INFO, location: 北京, response_time: 320, message: Weather query completed }5. 插件扩展思路5.1 多数据源容灾实现天气API的降级策略async def get_weather_with_fallback(location: str): for api in [fetch_heweather, fetch_openweather]: if result : await api(location): return result return None5.2 用户偏好存储使用SQLite记录用户常用地点import sqlite3 def init_db(): conn sqlite3.connect(user_prefs.db) conn.execute( CREATE TABLE IF NOT EXISTS prefs (user_id TEXT PRIMARY KEY, location TEXT) ) return conn5.3 定时天气推送利用APScheduler实现定时任务from nonebot_plugin_apscheduler import scheduler scheduler.scheduled_job(cron, hour8) async def morning_weather(): conn init_db() for user in conn.execute(SELECT user_id, location FROM prefs): await send_private_msg(user[0], await get_weather(user[1]))通过以上实现我们构建了一个具备生产级可靠性的天气查询插件。在实际项目中建议进一步考虑接口限流防止滥用敏感词过滤机制多语言支持单元测试覆盖率这种模块化设计思路同样适用于其他功能插件的开发如快递查询、日程管理等。关键在于保持代码的可扩展性和可维护性这是区分业余项目与专业开发的核心标准。