refactor: 重构config模块
Oct-autumn
parent
5974b51754
commit
36305f226c
|
|
@ -1,93 +0,0 @@
|
|||
import os
|
||||
import sys
|
||||
import tomli
|
||||
import shutil
|
||||
from .logger import logger
|
||||
from typing import Optional
|
||||
|
||||
|
||||
class Config:
|
||||
platform: str = "qq"
|
||||
nickname: Optional[str] = None
|
||||
server_host: str = "localhost"
|
||||
server_port: int = 8095
|
||||
napcat_heartbeat_interval: int = 30
|
||||
|
||||
def __init__(self):
|
||||
self._get_config_path()
|
||||
|
||||
def _get_config_path(self):
|
||||
current_file_path = os.path.abspath(__file__)
|
||||
src_path = os.path.dirname(current_file_path)
|
||||
self.root_path = os.path.join(src_path, "..")
|
||||
self.config_path = os.path.join(self.root_path, "config.toml")
|
||||
|
||||
def load_config(self): # sourcery skip: extract-method, move-assign
|
||||
include_configs = ["Napcat_Server", "MaiBot_Server", "Chat", "Voice", "Debug"]
|
||||
if not os.path.exists(self.config_path):
|
||||
logger.error("配置文件不存在！")
|
||||
logger.info("正在创建配置文件...")
|
||||
shutil.copy(
|
||||
os.path.join(self.root_path, "template", "template_config.toml"),
|
||||
os.path.join(self.root_path, "config.toml"),
|
||||
)
|
||||
logger.info("配置文件创建成功，请修改配置文件后重启程序。")
|
||||
sys.exit(1)
|
||||
with open(self.config_path, "rb") as f:
|
||||
try:
|
||||
raw_config = tomli.load(f)
|
||||
except tomli.TOMLDecodeError as e:
|
||||
logger.critical(f"配置文件bot_config.toml填写有误，请检查第{e.lineno}行第{e.colno}处：{e.msg}")
|
||||
sys.exit(1)
|
||||
for key in include_configs:
|
||||
if key not in raw_config:
|
||||
logger.error(f"配置文件中缺少必需的字段: '{key}'")
|
||||
logger.error("你的配置文件可能过时，请尝试手动更新配置文件。")
|
||||
sys.exit(1)
|
||||
|
||||
self.server_host = raw_config["Napcat_Server"].get("host", "localhost")
|
||||
self.server_port = raw_config["Napcat_Server"].get("port", 8095)
|
||||
self.napcat_heartbeat_interval = raw_config["Napcat_Server"].get("heartbeat", 30)
|
||||
|
||||
self.mai_host = raw_config["MaiBot_Server"].get("host", "localhost")
|
||||
self.mai_port = raw_config["MaiBot_Server"].get("port", 8000)
|
||||
self.platform = raw_config["MaiBot_Server"].get("platform_name")
|
||||
if not self.platform:
|
||||
logger.critical("请在配置文件中指定平台")
|
||||
sys.exit(1)
|
||||
|
||||
self.group_list_type: str = raw_config["Chat"].get("group_list_type")
|
||||
self.group_list: list = raw_config["Chat"].get("group_list", [])
|
||||
self.private_list_type: str = raw_config["Chat"].get("private_list_type")
|
||||
self.private_list: list = raw_config["Chat"].get("private_list", [])
|
||||
self.ban_user_id: list = raw_config["Chat"].get("ban_user_id", [])
|
||||
self.enable_poke: bool = raw_config["Chat"].get("enable_poke", True)
|
||||
if self.group_list_type not in ["whitelist", "blacklist"]:
|
||||
logger.critical("请在配置文件中指定group_list_type或group_list_type填写错误")
|
||||
sys.exit(1)
|
||||
if self.private_list_type not in ["whitelist", "blacklist"]:
|
||||
logger.critical("请在配置文件中指定private_list_type或private_list_type填写错误")
|
||||
sys.exit(1)
|
||||
|
||||
self.use_tts = raw_config["Voice"].get("use_tts", False)
|
||||
|
||||
self.debug_level = raw_config["Debug"].get("level", "INFO")
|
||||
if self.debug_level == "DEBUG":
|
||||
logger.debug("原始配置文件内容:")
|
||||
logger.debug(raw_config)
|
||||
logger.debug("读取到的配置内容：")
|
||||
logger.debug(f"平台: {self.platform}")
|
||||
logger.debug(f"MaiBot服务器地址: {self.mai_host}:{self.mai_port}")
|
||||
logger.debug(f"Napcat服务器地址: {self.server_host}:{self.server_port}")
|
||||
logger.debug(f"心跳间隔: {self.napcat_heartbeat_interval}秒")
|
||||
logger.debug(f"群聊列表类型: {self.group_list_type}")
|
||||
logger.debug(f"群聊列表: {self.group_list}")
|
||||
logger.debug(f"私聊列表类型: {self.private_list_type}")
|
||||
logger.debug(f"私聊列表: {self.private_list}")
|
||||
logger.debug(f"禁用用户ID列表: {self.ban_user_id}")
|
||||
logger.debug(f"是否启用TTS: {self.use_tts}")
|
||||
logger.debug(f"调试级别: {self.debug_level}")
|
||||
|
||||
|
||||
global_config = Config()
|
||||
global_config.load_config()
|
||||
|
|
@ -0,0 +1,5 @@
|
|||
from .config import global_config
|
||||
|
||||
__all__ = [
|
||||
"global_config",
|
||||
]
|
||||
|
|
@ -0,0 +1,140 @@
|
|||
import os
|
||||
from dataclasses import dataclass
|
||||
|
||||
import tomlkit
|
||||
import shutil
|
||||
|
||||
from tomlkit import TOMLDocument
|
||||
from tomlkit.items import Table
|
||||
from ..logger import logger
|
||||
from rich.traceback import install
|
||||
|
||||
from src.config.config_base import ConfigBase
|
||||
from src.config.official_configs import (
|
||||
ChatConfig,
|
||||
DebugConfig,
|
||||
MaiBotServerConfig,
|
||||
NapcatServerConfig,
|
||||
NicknameConfig,
|
||||
VoiceConfig,
|
||||
)
|
||||
|
||||
install(extra_lines=3)
|
||||
|
||||
TEMPLATE_DIR = "template"
|
||||
|
||||
|
||||
def update_config():
|
||||
# 定义文件路径
|
||||
template_path = f"{TEMPLATE_DIR}/template_config.toml"
|
||||
old_config_path = "config.toml"
|
||||
new_config_path = "config.toml"
|
||||
|
||||
# 检查配置文件是否存在
|
||||
if not os.path.exists(old_config_path):
|
||||
logger.info("配置文件不存在，从模板创建新配置")
|
||||
shutil.copy2(template_path, old_config_path) # 复制模板文件
|
||||
logger.info(f"已创建新配置文件，请填写后重新运行: {old_config_path}")
|
||||
# 如果是新创建的配置文件,直接返回
|
||||
quit()
|
||||
|
||||
# 读取旧配置文件和模板文件
|
||||
with open(old_config_path, "r", encoding="utf-8") as f:
|
||||
old_config = tomlkit.load(f)
|
||||
with open(template_path, "r", encoding="utf-8") as f:
|
||||
new_config = tomlkit.load(f)
|
||||
|
||||
# 检查version是否相同
|
||||
if old_config and "inner" in old_config and "inner" in new_config:
|
||||
old_version = old_config["inner"].get("version")
|
||||
new_version = new_config["inner"].get("version")
|
||||
if old_version and new_version and old_version == new_version:
|
||||
logger.info(f"检测到配置文件版本号相同 (v{old_version})，跳过更新")
|
||||
return
|
||||
else:
|
||||
logger.info(f"检测到版本号不同: 旧版本 v{old_version} -> 新版本 v{new_version}")
|
||||
else:
|
||||
logger.info("已有配置文件未检测到版本号，可能是旧版本。将进行更新")
|
||||
|
||||
# 备份文件名
|
||||
old_backup_path = "config.toml.back"
|
||||
|
||||
# 备份旧配置文件
|
||||
shutil.move(old_config_path, old_backup_path)
|
||||
logger.info(f"已备份旧配置文件到: {old_backup_path}")
|
||||
|
||||
# 复制模板文件到配置目录
|
||||
shutil.copy2(template_path, new_config_path)
|
||||
logger.info(f"已创建新配置文件: {new_config_path}")
|
||||
|
||||
def update_dict(target: TOMLDocument | dict, source: TOMLDocument | dict):
|
||||
"""
|
||||
将source字典的值更新到target字典中（如果target中存在相同的键）
|
||||
"""
|
||||
for key, value in source.items():
|
||||
# 跳过version字段的更新
|
||||
if key == "version":
|
||||
continue
|
||||
if key in target:
|
||||
if isinstance(value, dict) and isinstance(target[key], (dict, Table)):
|
||||
update_dict(target[key], value)
|
||||
else:
|
||||
try:
|
||||
# 对数组类型进行特殊处理
|
||||
if isinstance(value, list):
|
||||
# 如果是空数组，确保它保持为空数组
|
||||
target[key] = tomlkit.array(str(value)) if value else tomlkit.array()
|
||||
else:
|
||||
# 其他类型使用item方法创建新值
|
||||
target[key] = tomlkit.item(value)
|
||||
except (TypeError, ValueError):
|
||||
# 如果转换失败，直接赋值
|
||||
target[key] = value
|
||||
|
||||
# 将旧配置的值更新到新配置中
|
||||
logger.info("开始合并新旧配置...")
|
||||
update_dict(new_config, old_config)
|
||||
|
||||
# 保存更新后的配置（保留注释和格式）
|
||||
with open(new_config_path, "w", encoding="utf-8") as f:
|
||||
f.write(tomlkit.dumps(new_config))
|
||||
logger.info("配置文件更新完成，建议检查新配置文件中的内容，以免丢失重要信息")
|
||||
quit()
|
||||
|
||||
|
||||
@dataclass
|
||||
class Config(ConfigBase):
|
||||
"""总配置类"""
|
||||
|
||||
nickname: NicknameConfig
|
||||
napcat_server: NapcatServerConfig
|
||||
maibot_server: MaiBotServerConfig
|
||||
chat: ChatConfig
|
||||
voice: VoiceConfig
|
||||
debug: DebugConfig
|
||||
|
||||
|
||||
def load_config(config_path: str) -> Config:
|
||||
"""
|
||||
加载配置文件
|
||||
:param config_path: 配置文件路径
|
||||
:return: Config对象
|
||||
"""
|
||||
# 读取配置文件
|
||||
with open(config_path, "r", encoding="utf-8") as f:
|
||||
config_data = tomlkit.load(f)
|
||||
|
||||
# 创建Config对象
|
||||
try:
|
||||
return Config.from_dict(config_data)
|
||||
except Exception as e:
|
||||
logger.critical("配置文件解析失败")
|
||||
raise e
|
||||
|
||||
|
||||
# 更新配置
|
||||
update_config()
|
||||
|
||||
logger.info("正在品鉴配置文件...")
|
||||
global_config = load_config(config_path="config.toml")
|
||||
logger.info("非常的新鲜，非常的美味！")
|
||||
|
|
@ -0,0 +1,128 @@
|
|||
from dataclasses import dataclass, fields, MISSING
|
||||
from typing import TypeVar, Type, Any, get_origin, get_args, Literal
|
||||
|
||||
T = TypeVar("T", bound="ConfigBase")
|
||||
|
||||
TOML_DICT_TYPE = {
|
||||
int,
|
||||
float,
|
||||
str,
|
||||
bool,
|
||||
list,
|
||||
dict,
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class ConfigBase:
|
||||
"""配置类的基类"""
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls: Type[T], data: dict[str, Any]) -> T:
|
||||
"""从字典加载配置字段"""
|
||||
if not isinstance(data, dict):
|
||||
raise TypeError(f"Expected a dictionary, got {type(data).__name__}")
|
||||
|
||||
init_args: dict[str, Any] = {}
|
||||
|
||||
for f in fields(cls):
|
||||
field_name = f.name
|
||||
|
||||
if field_name.startswith("_"):
|
||||
# 跳过以 _ 开头的字段
|
||||
continue
|
||||
|
||||
if field_name not in data:
|
||||
if f.default is not MISSING or f.default_factory is not MISSING:
|
||||
# 跳过未提供且有默认值/默认构造方法的字段
|
||||
continue
|
||||
else:
|
||||
raise ValueError(f"Missing required field: '{field_name}'")
|
||||
|
||||
value = data[field_name]
|
||||
field_type = f.type
|
||||
|
||||
try:
|
||||
init_args[field_name] = cls._convert_field(value, field_type)
|
||||
except TypeError as e:
|
||||
raise TypeError(f"Field '{field_name}' has a type error: {e}") from e
|
||||
except Exception as e:
|
||||
raise RuntimeError(f"Failed to convert field '{field_name}' to target type: {e}") from e
|
||||
|
||||
return cls(**init_args)
|
||||
|
||||
@classmethod
|
||||
def _convert_field(cls, value: Any, field_type: Type[Any]) -> Any:
|
||||
"""
|
||||
转换字段值为指定类型
|
||||
|
||||
1. 对于嵌套的 dataclass，递归调用相应的 from_dict 方法
|
||||
2. 对于泛型集合类型（list, set, tuple），递归转换每个元素
|
||||
3. 对于基础类型（int, str, float, bool），直接转换
|
||||
4. 对于其他类型，尝试直接转换，如果失败则抛出异常
|
||||
"""
|
||||
|
||||
# 如果是嵌套的 dataclass，递归调用 from_dict 方法
|
||||
if isinstance(field_type, type) and issubclass(field_type, ConfigBase):
|
||||
if not isinstance(value, dict):
|
||||
raise TypeError(f"Expected a dictionary for {field_type.__name__}, got {type(value).__name__}")
|
||||
return field_type.from_dict(value)
|
||||
|
||||
# 处理泛型集合类型（list, set, tuple）
|
||||
field_origin_type = get_origin(field_type)
|
||||
field_type_args = get_args(field_type)
|
||||
|
||||
if field_origin_type in {list, set, tuple}:
|
||||
# 检查提供的value是否为list
|
||||
if not isinstance(value, list):
|
||||
raise TypeError(f"Expected an list for {field_type.__name__}, got {type(value).__name__}")
|
||||
|
||||
if field_origin_type is list:
|
||||
return [cls._convert_field(item, field_type_args[0]) for item in value]
|
||||
elif field_origin_type is set:
|
||||
return {cls._convert_field(item, field_type_args[0]) for item in value}
|
||||
elif field_origin_type is tuple:
|
||||
# 检查提供的value长度是否与类型参数一致
|
||||
if len(value) != len(field_type_args):
|
||||
raise TypeError(
|
||||
f"Expected {len(field_type_args)} items for {field_type.__name__}, got {len(value)}"
|
||||
)
|
||||
return tuple(cls._convert_field(item, arg) for item, arg in zip(value, field_type_args))
|
||||
|
||||
if field_origin_type is dict:
|
||||
# 检查提供的value是否为dict
|
||||
if not isinstance(value, dict):
|
||||
raise TypeError(f"Expected a dictionary for {field_type.__name__}, got {type(value).__name__}")
|
||||
|
||||
# 检查字典的键值类型
|
||||
if len(field_type_args) != 2:
|
||||
raise TypeError(f"Expected a dictionary with two type arguments for {field_type.__name__}")
|
||||
key_type, value_type = field_type_args
|
||||
|
||||
return {cls._convert_field(k, key_type): cls._convert_field(v, value_type) for k, v in value.items()}
|
||||
|
||||
# 处理基础类型，例如 int, str 等
|
||||
if field_origin_type is type(None) and value is None: # 处理Optional类型
|
||||
return None
|
||||
|
||||
# 处理Literal类型
|
||||
if field_origin_type is Literal or get_origin(field_type) is Literal:
|
||||
# 获取Literal的允许值
|
||||
allowed_values = get_args(field_type)
|
||||
if value in allowed_values:
|
||||
return value
|
||||
else:
|
||||
raise TypeError(f"Value '{value}' is not in allowed values {allowed_values} for Literal type")
|
||||
|
||||
if field_type is Any or isinstance(value, field_type):
|
||||
return value
|
||||
|
||||
# 其他类型，尝试直接转换
|
||||
try:
|
||||
return field_type(value)
|
||||
except (ValueError, TypeError) as e:
|
||||
raise TypeError(f"Cannot convert {type(value).__name__} to {field_type.__name__}") from e
|
||||
|
||||
def __str__(self):
|
||||
"""返回配置类的字符串表示"""
|
||||
return f"{self.__class__.__name__}({', '.join(f'{f.name}={getattr(self, f.name)}' for f in fields(self))})"
|
||||
|
|
@ -0,0 +1,77 @@
|
|||
from dataclasses import dataclass, field
|
||||
from typing import Literal
|
||||
|
||||
from src.config.config_base import ConfigBase
|
||||
|
||||
"""
|
||||
须知：
|
||||
1. 本文件中记录了所有的配置项
|
||||
2. 所有新增的class都需要继承自ConfigBase
|
||||
3. 所有新增的class都应在config.py中的Config类中添加字段
|
||||
4. 对于新增的字段，若为可选项，则应在其后添加field()并设置default_factory或default
|
||||
"""
|
||||
|
||||
ADAPTER_PLATFORM = "qq"
|
||||
|
||||
|
||||
@dataclass
|
||||
class NicknameConfig(ConfigBase):
|
||||
nickname: str
|
||||
"""机器人昵称"""
|
||||
|
||||
|
||||
@dataclass
|
||||
class NapcatServerConfig(ConfigBase):
|
||||
host: str = "localhost"
|
||||
"""Napcat服务端的主机地址"""
|
||||
|
||||
port: int = 8095
|
||||
"""Napcat服务端的端口号"""
|
||||
|
||||
heartbeat_interval: int = 30
|
||||
"""Napcat心跳间隔时间，单位为秒"""
|
||||
|
||||
|
||||
@dataclass
|
||||
class MaiBotServerConfig(ConfigBase):
|
||||
platform_name: str = field(default=ADAPTER_PLATFORM, init=False)
|
||||
"""平台名称，“qq”"""
|
||||
|
||||
host: str = "localhost"
|
||||
"""MaiMCore的主机地址"""
|
||||
|
||||
port: int = 8000
|
||||
"""MaiMCore的端口号"""
|
||||
|
||||
|
||||
@dataclass
|
||||
class ChatConfig(ConfigBase):
|
||||
group_list_type: Literal["whitelist", "blacklist"] = "whitelist"
|
||||
"""群聊列表类型 白名单/黑名单"""
|
||||
|
||||
group_list: list[str] = field(default_factory=[])
|
||||
"""群聊列表"""
|
||||
|
||||
private_list_type: Literal["whitelist", "blacklist"] = "whitelist"
|
||||
"""私聊列表类型 白名单/黑名单"""
|
||||
|
||||
private_list: list[str] = field(default_factory=[])
|
||||
"""私聊列表"""
|
||||
|
||||
ban_user_id: list[str] = field(default_factory=[])
|
||||
"""被封禁的用户ID列表，封禁后将无法与其进行交互"""
|
||||
|
||||
enable_poke: bool = True
|
||||
"""是否启用戳一戳功能"""
|
||||
|
||||
|
||||
@dataclass
|
||||
class VoiceConfig(ConfigBase):
|
||||
use_tts: bool = False
|
||||
"""是否启用TTS功能"""
|
||||
|
||||
|
||||
@dataclass
|
||||
class DebugConfig(ConfigBase):
|
||||
level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO"
|
||||
"""日志级别，默认为INFO"""
|
||||
|
|
@ -5,6 +5,6 @@ import sys
|
|||
logger.remove()
|
||||
logger.add(
|
||||
sys.stderr,
|
||||
level=global_config.debug_level,
|
||||
level=global_config.debug.level,
|
||||
format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
|
||||
)
|
||||
|
|
|
|||
|
|
@ -5,8 +5,8 @@ from .send_handler import send_handler
|
|||
|
||||
route_config = RouteConfig(
|
||||
route_config={
|
||||
global_config.platform: TargetConfig(
|
||||
url=f"ws://{global_config.mai_host}:{global_config.mai_port}/ws",
|
||||
global_config.maibot_server.platform_name: TargetConfig(
|
||||
url=f"ws://{global_config.maibot_server.host}:{global_config.maibot_server.port}/ws",
|
||||
token=None,
|
||||
)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -36,7 +36,7 @@ class RecvHandler:
|
|||
|
||||
def __init__(self):
|
||||
self.server_connection: Server.ServerConnection = None
|
||||
self.interval = global_config.napcat_heartbeat_interval
|
||||
self.interval = global_config.napcat_server.heartbeat_interval
|
||||
|
||||
async def handle_meta_event(self, message: dict) -> None:
|
||||
event_type = message.get("meta_event_type")
|
||||
|
|
@ -77,20 +77,20 @@ class RecvHandler:
|
|||
"""
|
||||
logger.debug(f"群聊id: {group_id}, 用户id: {user_id}")
|
||||
if group_id:
|
||||
if global_config.group_list_type == "whitelist" and group_id not in global_config.group_list:
|
||||
if global_config.chat.group_list_type == "whitelist" and group_id not in global_config.chat.group_list:
|
||||
logger.warning("群聊不在聊天白名单中，消息被丢弃")
|
||||
return False
|
||||
elif global_config.group_list_type == "blacklist" and group_id in global_config.group_list:
|
||||
elif global_config.chat.group_list_type == "blacklist" and group_id in global_config.chat.group_list:
|
||||
logger.warning("群聊在聊天黑名单中，消息被丢弃")
|
||||
return False
|
||||
else:
|
||||
if global_config.private_list_type == "whitelist" and user_id not in global_config.private_list:
|
||||
if global_config.chat.private_list_type == "whitelist" and user_id not in global_config.chat.private_list:
|
||||
logger.warning("私聊不在聊天白名单中，消息被丢弃")
|
||||
return False
|
||||
elif global_config.private_list_type == "blacklist" and user_id in global_config.private_list:
|
||||
elif global_config.chat.private_list_type == "blacklist" and user_id in global_config.chat.private_list:
|
||||
logger.warning("私聊在聊天黑名单中，消息被丢弃")
|
||||
return False
|
||||
if user_id in global_config.ban_user_id:
|
||||
if user_id in global_config.chat.ban_user_id:
|
||||
logger.warning("用户在全局黑名单中，消息被丢弃")
|
||||
return False
|
||||
return True
|
||||
|
|
@ -123,7 +123,7 @@ class RecvHandler:
|
|||
|
||||
# 发送者用户信息
|
||||
user_info: UserInfo = UserInfo(
|
||||
platform=global_config.platform,
|
||||
platform=global_config.maibot_server.platform_name,
|
||||
user_id=sender_info.get("user_id"),
|
||||
user_nickname=sender_info.get("nickname"),
|
||||
user_cardname=sender_info.get("card"),
|
||||
|
|
@ -149,7 +149,7 @@ class RecvHandler:
|
|||
nickname = fetched_member_info.get("nickname") if fetched_member_info else None
|
||||
# 发送者用户信息
|
||||
user_info: UserInfo = UserInfo(
|
||||
platform=global_config.platform,
|
||||
platform=global_config.maibot_server.platform_name,
|
||||
user_id=sender_info.get("user_id"),
|
||||
user_nickname=nickname,
|
||||
user_cardname=None,
|
||||
|
|
@ -164,7 +164,7 @@ class RecvHandler:
|
|||
group_name = fetched_group_info.get("group_name")
|
||||
|
||||
group_info: GroupInfo = GroupInfo(
|
||||
platform=global_config.platform,
|
||||
platform=global_config.maibot_server.platform_name,
|
||||
group_id=raw_message.get("group_id"),
|
||||
group_name=group_name,
|
||||
)
|
||||
|
|
@ -182,7 +182,7 @@ class RecvHandler:
|
|||
|
||||
# 发送者用户信息
|
||||
user_info: UserInfo = UserInfo(
|
||||
platform=global_config.platform,
|
||||
platform=global_config.maibot_server.platform_name,
|
||||
user_id=sender_info.get("user_id"),
|
||||
user_nickname=sender_info.get("nickname"),
|
||||
user_cardname=sender_info.get("card"),
|
||||
|
|
@ -195,7 +195,7 @@ class RecvHandler:
|
|||
group_name = fetched_group_info.get("group_name")
|
||||
|
||||
group_info: GroupInfo = GroupInfo(
|
||||
platform=global_config.platform,
|
||||
platform=global_config.maibot_server.platform_name,
|
||||
group_id=raw_message.get("group_id"),
|
||||
group_name=group_name,
|
||||
)
|
||||
|
|
@ -205,12 +205,12 @@ class RecvHandler:
|
|||
return None
|
||||
|
||||
additional_config: dict = {}
|
||||
if global_config.use_tts:
|
||||
if global_config.voice.use_tts:
|
||||
additional_config["allow_tts"] = True
|
||||
|
||||
# 消息信息
|
||||
message_info: BaseMessageInfo = BaseMessageInfo(
|
||||
platform=global_config.platform,
|
||||
platform=global_config.maibot_server.platform_name,
|
||||
message_id=message_id,
|
||||
time=message_time,
|
||||
user_info=user_info,
|
||||
|
|
@ -500,7 +500,7 @@ class RecvHandler:
|
|||
sub_type = raw_message.get("sub_type")
|
||||
match sub_type:
|
||||
case NoticeType.Notify.poke:
|
||||
if global_config.enable_poke:
|
||||
if global_config.chat.enable_poke:
|
||||
handled_message: Seg = await self.handle_poke_notify(raw_message)
|
||||
else:
|
||||
logger.warning("戳一戳消息被禁用，取消戳一戳处理")
|
||||
|
|
@ -532,7 +532,7 @@ class RecvHandler:
|
|||
source_name = "QQ用户"
|
||||
|
||||
user_info: UserInfo = UserInfo(
|
||||
platform=global_config.platform,
|
||||
platform=global_config.maibot_server.platform_name,
|
||||
user_id=user_id,
|
||||
user_nickname=source_name,
|
||||
user_cardname=source_cardname,
|
||||
|
|
@ -547,13 +547,13 @@ class RecvHandler:
|
|||
else:
|
||||
logger.warning("无法获取戳一戳消息所在群的名称")
|
||||
group_info = GroupInfo(
|
||||
platform=global_config.platform,
|
||||
platform=global_config.maibot_server.platform_name,
|
||||
group_id=group_id,
|
||||
group_name=group_name,
|
||||
)
|
||||
|
||||
message_info: BaseMessageInfo = BaseMessageInfo(
|
||||
platform=global_config.platform,
|
||||
platform=global_config.maibot_server.platform_name,
|
||||
message_id="notice",
|
||||
time=message_time,
|
||||
user_info=user_info,
|
||||
|
|
|
|||
|
|
@ -35,10 +35,10 @@ async def check_timeout_response() -> None:
|
|||
cleaned_message_count: int = 0
|
||||
now_time = time.time()
|
||||
for echo_id, response_time in list(response_time_dict.items()):
|
||||
if now_time - response_time > global_config.napcat_heartbeat_interval:
|
||||
if now_time - response_time > global_config.napcat_server.heartbeat_interval:
|
||||
cleaned_message_count += 1
|
||||
response_dict.pop(echo_id)
|
||||
response_time_dict.pop(echo_id)
|
||||
logger.warning(f"响应消息 {echo_id} 超时，已删除")
|
||||
logger.info(f"已删除 {cleaned_message_count} 条超时响应消息")
|
||||
await asyncio.sleep(global_config.napcat_heartbeat_interval)
|
||||
await asyncio.sleep(global_config.napcat_server.heartbeat_interval)
|
||||
|
|
|
|||
|
|
@ -209,7 +209,7 @@ class SendHandler:
|
|||
|
||||
def handle_voice_message(self, encoded_voice: str) -> dict:
|
||||
"""处理语音消息"""
|
||||
if not global_config.use_tts:
|
||||
if not global_config.voice.use_tts:
|
||||
logger.warning("未启用语音消息处理")
|
||||
return {}
|
||||
if not encoded_voice:
|
||||
|
|
|
|||
|
|
@ -1,30 +1,33 @@
|
|||
[Nickname] # 现在没用
|
||||
[inner]
|
||||
version = "0.1.0" # 版本号
|
||||
# 请勿修改版本号，除非你知道自己在做什么
|
||||
|
||||
[nickname] # 现在没用
|
||||
nickname = ""
|
||||
|
||||
[Napcat_Server] # Napcat连接的ws服务设置
|
||||
host = "localhost" # Napcat设定的主机地址
|
||||
port = 8095 # Napcat设定的端口
|
||||
heartbeat = 30 # 与Napcat设置的心跳相同（按秒计）
|
||||
[napcat_server] # Napcat连接的ws服务设置
|
||||
host = "localhost" # Napcat设定的主机地址
|
||||
port = 8095 # Napcat设定的端口
|
||||
heartbeat_interval = 30 # 与Napcat设置的心跳相同（按秒计）
|
||||
|
||||
[MaiBot_Server] # 连接麦麦的ws服务设置
|
||||
platform_name = "qq" # 标识adapter的名称（必填）
|
||||
host = "localhost" # 麦麦在.env文件中设置的主机地址，即HOST字段
|
||||
port = 8000 # 麦麦在.env文件中设置的端口，即PORT字段
|
||||
[maibot_server] # 连接麦麦的ws服务设置
|
||||
host = "localhost" # 麦麦在.env文件中设置的主机地址，即HOST字段
|
||||
port = 8000 # 麦麦在.env文件中设置的端口，即PORT字段
|
||||
|
||||
[Chat] # 黑白名单功能
|
||||
[chat] # 黑白名单功能
|
||||
group_list_type = "whitelist" # 群组名单类型，可选为：whitelist, blacklist
|
||||
group_list = [] # 群组名单
|
||||
group_list = [] # 群组名单
|
||||
# 当group_list_type为whitelist时，只有群组名单中的群组可以聊天
|
||||
# 当group_list_type为blacklist时，群组名单中的任何群组无法聊天
|
||||
private_list_type = "whitelist" # 私聊名单类型，可选为：whitelist, blacklist
|
||||
private_list = [] # 私聊名单
|
||||
private_list = [] # 私聊名单
|
||||
# 当private_list_type为whitelist时，只有私聊名单中的用户可以聊天
|
||||
# 当private_list_type为blacklist时，私聊名单中的任何用户无法聊天
|
||||
ban_user_id = [] # 全局禁止名单（全局禁止名单中的用户无法进行任何聊天）
|
||||
ban_user_id = [] # 全局禁止名单（全局禁止名单中的用户无法进行任何聊天）
|
||||
enable_poke = true # 是否启用戳一戳功能
|
||||
|
||||
[Voice] # 发送语音设置
|
||||
[voice] # 发送语音设置
|
||||
use_tts = false # 是否使用tts语音（请确保你配置了tts并有对应的adapter）
|
||||
|
||||
[Debug]
|
||||
level = "INFO" # 日志等级（DEBUG, INFO, WARNING, ERROR）
|
||||
[debug]
|
||||
level = "INFO" # 日志等级（DEBUG, INFO, WARNING, ERROR, CRITICAL）
|
||||
|
|
|
|||
Loading…
Reference in New Issue