删除yaml配置,使用toml进行配置

This commit is contained in:
liuwei
2025-05-12 16:50:04 +08:00
parent dfa464ebf7
commit c3faddbd0a
6 changed files with 52 additions and 68 deletions

View File

@@ -7,3 +7,8 @@ command-format = """
token内容示例pt_key=xxx; pt_pin=xxx; token内容示例pt_key=xxx; pt_pin=xxx;
通过m.jd.com登录F12查看请求中的cookie获得 通过m.jd.com登录F12查看请求中的cookie获得
""" """
# 青龙面板配置
QL_HOST= "http://192.168.2.32:5800" # 青龙面板地址,请根据实际情况修改
CLIENT_ID= "g_jH014_iQhW" # 青龙面板应用ID请填写实际值
CLIENT_SECRET= "lSRhT-1Xs0lOEZ5YfKMMCkbl" # 青龙面板应用密钥,请填写实际值

View File

@@ -1,4 +0,0 @@
# 青龙面板配置
QL_HOST: "http://192.168.2.32:5800" # 青龙面板地址,请根据实际情况修改
CLIENT_ID: "g_jH014_iQhW" # 青龙面板应用ID请填写实际值
CLIENT_SECRET: "lSRhT-1Xs0lOEZ5YfKMMCkbl" # 青龙面板应用密钥,请填写实际值

View File

@@ -2,7 +2,6 @@ from loguru import logger
import requests import requests
import re import re
import os import os
import yaml
from json import dumps as jsonDumps from json import dumps as jsonDumps
from typing import Dict, Any, List, Optional, Tuple from typing import Dict, Any, List, Optional, Tuple
@@ -145,8 +144,6 @@ class JDTokenPlugin(MessagePluginInterface):
def __init__(self): def __init__(self):
super().__init__() super().__init__()
self.plugin_dir = os.path.dirname(os.path.abspath(__file__))
self.config_path = os.path.join(self.plugin_dir, "config.yaml")
def initialize(self, context: Dict[str, Any]) -> bool: def initialize(self, context: Dict[str, Any]) -> bool:
"""初始化插件""" """初始化插件"""
@@ -157,33 +154,22 @@ class JDTokenPlugin(MessagePluginInterface):
self.event_system = context.get("event_system") self.event_system = context.get("event_system")
self.message_util = context.get("message_util") self.message_util = context.get("message_util")
# 从TOML配置文件加载配置
self._commands = self._config.get("JD_Token", {}).get("command", ["设置京东"]) self._commands = self._config.get("JD_Token", {}).get("command", ["设置京东"])
self.command_format = self._config.get("JD_Token", {}).get("command-format", "设置京东 token内容 备注名称") self.command_format = self._config.get("JD_Token", {}).get("command-format", "设置京东 token内容 备注名称")
self.enable = self._config.get("JD_Token", {}).get("enable", True) self.enable = self._config.get("JD_Token", {}).get("enable", True)
# 加载青龙面板配置 # 从TOML配置文件加载青龙面板配置
self.load_config() ql_host = self._config.get("JD_Token", {}).get("QL_HOST", "http://localhost:5700")
self.ql = QL(self.ql_config.get("QL_HOST"), self.ql_config.get("CLIENT_ID"), self.ql_config.get("CLIENT_SECRET")) client_id = self._config.get("JD_Token", {}).get("CLIENT_ID", "")
client_secret = self._config.get("JD_Token", {}).get("CLIENT_SECRET", "")
# 初始化青龙面板连接
self.ql = QL(ql_host, client_id, client_secret)
self.LOG.info(f"[{self.name}] 插件初始化完成,指令:{self._commands}") self.LOG.info(f"[{self.name}] 插件初始化完成,指令:{self._commands}")
return True return True
def load_config(self):
"""加载配置"""
if os.path.exists(self.config_path):
with open(self.config_path, 'r', encoding='utf-8') as f:
self.ql_config = yaml.safe_load(f)
else:
# 默认配置
self.ql_config = {
"QL_HOST": "http://localhost:5700",
"CLIENT_ID": "",
"CLIENT_SECRET": ""
}
# 保存默认配置
with open(self.config_path, 'w', encoding='utf-8') as f:
yaml.dump(self.ql_config, f, default_flow_style=False, allow_unicode=True)
def start(self) -> bool: def start(self) -> bool:
"""启动插件""" """启动插件"""
self.LOG.info(f"[{self.name}] 插件已启动") self.LOG.info(f"[{self.name}] 插件已启动")
@@ -207,6 +193,7 @@ class JDTokenPlugin(MessagePluginInterface):
return command in self._commands return command in self._commands
@plugin_stats_decorator(plugin_name="京东签到Token设置") @plugin_stats_decorator(plugin_name="京东签到Token设置")
@plugin_points_cost(2, "京东签到Token设置消耗积分", Feature.UTILITY)
async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]: async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
"""处理消息""" """处理消息"""
content = str(message.get("content", "")).strip() content = str(message.get("content", "")).strip()

View File

@@ -8,3 +8,6 @@ command-format = """
城市名天气 城市名天气
城市名 天气 城市名 天气
""" """
# 和风天气API配置
API_KEY= "32e692f94e364105808ea2c0302ef162" # 请填写您的和风天气API密钥
API_DOMAIN= "https://mx564thnfv.re.qweatherapi.com" # API域名请根据实际情况修改

View File

@@ -1,3 +0,0 @@
# 和风天气API配置
API_KEY: "32e692f94e364105808ea2c0302ef162" # 请填写您的和风天气API密钥
API_DOMAIN: "https://mx564thnfv.re.qweatherapi.com" # API域名请根据实际情况修改

View File

@@ -1,8 +1,6 @@
from loguru import logger from loguru import logger
import aiohttp import aiohttp
import jieba
import os import os
import yaml
from typing import Dict, Any, List, Optional, Tuple from typing import Dict, Any, List, Optional, Tuple
from plugin_common.message_plugin_interface import MessagePluginInterface from plugin_common.message_plugin_interface import MessagePluginInterface
@@ -43,7 +41,6 @@ class WeatherPlugin(MessagePluginInterface):
def __init__(self): def __init__(self):
super().__init__() super().__init__()
self.plugin_dir = os.path.dirname(os.path.abspath(__file__)) self.plugin_dir = os.path.dirname(os.path.abspath(__file__))
self.config_path = os.path.join(self.plugin_dir, "config.yaml")
def initialize(self, context: Dict[str, Any]) -> bool: def initialize(self, context: Dict[str, Any]) -> bool:
"""初始化插件""" """初始化插件"""
@@ -54,6 +51,7 @@ class WeatherPlugin(MessagePluginInterface):
self.event_system = context.get("event_system") self.event_system = context.get("event_system")
self.message_util = context.get("message_util") self.message_util = context.get("message_util")
# 从TOML配置文件加载配置
self._commands = self._config.get("Weather", {}).get("command", ["天气"]) self._commands = self._config.get("Weather", {}).get("command", ["天气"])
self.command_format = self._config.get("Weather", {}).get("command-format", """⚙️获取天气: self.command_format = self._config.get("Weather", {}).get("command-format", """⚙️获取天气:
天气 城市名 天气 城市名
@@ -63,29 +61,12 @@ class WeatherPlugin(MessagePluginInterface):
self.enable = self._config.get("Weather", {}).get("enable", True) self.enable = self._config.get("Weather", {}).get("enable", True)
# 加载API配置 # 加载API配置
self.load_config() self.api_key = self._config.get("Weather", {}).get("API_KEY", "")
self.api_domain = self._config.get("Weather", {}).get("API_DOMAIN", "")
self.LOG.info(f"[{self.name}] 插件初始化完成,指令:{self._commands}") self.LOG.info(f"[{self.name}] 插件初始化完成,指令:{self._commands}")
return True return True
def load_config(self):
"""加载配置"""
if os.path.exists(self.config_path):
with open(self.config_path, 'r', encoding='utf-8') as f:
self.weather_config = yaml.safe_load(f)
else:
# 默认配置
self.weather_config = {
"API_KEY": "",
"API_DOMAIN": ""
}
# 保存默认配置
with open(self.config_path, 'w', encoding='utf-8') as f:
yaml.dump(self.weather_config, f, default_flow_style=False, allow_unicode=True)
self.api_key = self.weather_config.get("API_KEY", "")
self.api_domain = self.weather_config.get("API_DOMAIN", "")
def start(self) -> bool: def start(self) -> bool:
"""启动插件""" """启动插件"""
self.LOG.info(f"[{self.name}] 插件已启动") self.LOG.info(f"[{self.name}] 插件已启动")
@@ -113,6 +94,7 @@ class WeatherPlugin(MessagePluginInterface):
return True return True
@plugin_stats_decorator(plugin_name="天气查询") @plugin_stats_decorator(plugin_name="天气查询")
@plugin_points_cost(1, "天气查询消耗积分", Feature.UTILITY)
async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]: async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
"""处理消息""" """处理消息"""
content = str(message.get("content", "")).strip() content = str(message.get("content", "")).strip()
@@ -126,28 +108,21 @@ class WeatherPlugin(MessagePluginInterface):
if roomid and gbm.get_group_permission(roomid, Feature.WEATHER) == PermissionStatus.DISABLED: if roomid and gbm.get_group_permission(roomid, Feature.WEATHER) == PermissionStatus.DISABLED:
return False, "没有权限" return False, "没有权限"
# 处理消息内容 # 处理消息内容 - 不再使用jieba分词
content = content.replace(" ", "") city_name = self._extract_city_name(content)
command = list(jieba.cut(content))
if len(command) == 1: if not city_name:
await bot.send_text_message((roomid if roomid else sender), f"\n{self.command_format}", sender) await bot.send_text_message((roomid if roomid else sender), f"\n{self.command_format}", sender)
return False, "命令格式错误" return False, "命令格式错误"
elif len(command) > 3:
return False, "命令格式错误"
# 配置密钥检查 # 配置密钥检查
if not self.api_key: if not self.api_key:
await bot.send_text_message((roomid if roomid else sender), "\n你还没配置天气API密钥", sender) await bot.send_text_message((roomid if roomid else sender), "\n你还没配置天气API密钥请在config.toml中配置Weather.API_KEY", sender)
return False, "API密钥未配置" return False, "API密钥未配置"
try: try:
# 提取城市名
command.remove("天气")
request_loc = "".join(command)
# 获取天气信息 # 获取天气信息
weather_info = await self._get_weather_info(request_loc) weather_info = await self._get_weather_info(city_name)
if not weather_info: if not weather_info:
await bot.send_text_message((roomid if roomid else sender), "\n⚠️查询天气失败!", sender) await bot.send_text_message((roomid if roomid else sender), "\n⚠️查询天气失败!", sender)
return False, "查询天气失败" return False, "查询天气失败"
@@ -161,6 +136,27 @@ class WeatherPlugin(MessagePluginInterface):
await bot.send_text_message((roomid if roomid else sender), f"\n⚠️处理出错: {str(e)}", sender) await bot.send_text_message((roomid if roomid else sender), f"\n⚠️处理出错: {str(e)}", sender)
return False, f"处理出错: {e}" return False, f"处理出错: {e}"
def _extract_city_name(self, content: str) -> str:
"""提取城市名称替代jieba分词"""
# 去除空格
content = content.replace(" ", "")
# 处理几种常见格式
if content.startswith("天气"):
# 格式: "天气北京"
return content[2:]
elif content.endswith("天气"):
# 格式: "北京天气"
return content[:-2]
else:
# 尝试分离"天气"和城市名
parts = content.split("天气")
if len(parts) == 2:
# 选择非空的部分作为城市名
return parts[0] if parts[0] else parts[1]
return ""
async def _get_weather_info(self, city_name: str) -> str: async def _get_weather_info(self, city_name: str) -> str:
"""获取天气信息""" """获取天气信息"""
try: try: