1
0
forked from bot/app

🐛 fix 通道无法在进程内传递消息的问题

This commit is contained in:
2024-08-12 02:40:51 +08:00
parent c3fc5d429b
commit 37b8d969b1
36 changed files with 430 additions and 238 deletions

View File

@ -1,12 +1,24 @@
"""
该模块用于常用配置文件的加载
多配置文件编写原则:
1.尽量不要冲突: 一个键不要多次出现
2.分工明确: 每个配置文件给一个或一类服务提供配置
3.扁平化编写: 配置文件尽量扁平化,不要出现过多的嵌套
4.注意冲突时的优先级: 项目目录下的配置文件优先级高于config目录下的配置文件
5.请不要将需要动态加载的内容写入配置文件,你应该使用其他储存方式
"""
import copy
import json
import os
from typing import List
from typing import Any, List
import nonebot
import toml
import yaml
from pydantic import BaseModel
from liteyuki import logger
config = {} # 主进程全局配置,确保加载后读取
_SUPPORTED_CONFIG_FORMATS = (".yaml", ".yml", ".json", ".toml")
class SatoriNodeConfig(BaseModel):
@ -32,18 +44,100 @@ class BasicConfig(BaseModel):
data_path: str = "data/liteyuki"
def load_from_yaml(file: str) -> dict:
global config
nonebot.logger.debug("Loading config from %s" % file)
if not os.path.exists(file):
nonebot.logger.warning(f"Config file {file} not found, created default config, please modify it and restart")
with open(file, "w", encoding="utf-8") as f:
yaml.dump(BasicConfig().dict(), f, default_flow_style=False)
def flat_config(config: dict[str, Any]) -> dict[str, Any]:
"""
扁平化配置文件
with open(file, "r", encoding="utf-8") as f:
conf = yaml.load(f, Loader=yaml.FullLoader)
config = conf
if conf is None:
nonebot.logger.warning(f"Config file {file} is empty, use default config. please modify it and restart")
conf = BasicConfig().dict()
return conf
{a:{b:{c:1}}} -> {"a.b.c": 1}
Args:
config: 配置项目
Returns:
扁平化后的配置文件,但也包含原有的键值对
"""
new_config = copy.deepcopy(config)
for key, value in config.items():
if isinstance(value, dict):
for k, v in flat_config(value).items():
new_config[f"{key}.{k}"] = v
return new_config
def load_from_yaml(file: str) -> dict[str, Any]:
"""
Load config from yaml file
"""
logger.debug(f"Loading YAML config from {file}")
config = yaml.safe_load(open(file, "r", encoding="utf-8"))
return flat_config(config if config is not None else {})
def load_from_json(file: str) -> dict[str, Any]:
"""
Load config from json file
"""
logger.debug(f"Loading JSON config from {file}")
config = json.load(open(file, "r", encoding="utf-8"))
return flat_config(config if config is not None else {})
def load_from_toml(file: str) -> dict[str, Any]:
"""
Load config from toml file
"""
logger.debug(f"Loading TOML config from {file}")
config = toml.load(open(file, "r", encoding="utf-8"))
return flat_config(config if config is not None else {})
def load_from_files(*files: str, no_warning: bool = False) -> dict[str, Any]:
"""
从指定文件加载配置项,会自动识别文件格式
默认执行扁平化选项
"""
config = {}
for file in files:
if os.path.exists(file):
if file.endswith((".yaml", "yml")):
config.update(load_from_yaml(file))
elif file.endswith(".json"):
config.update(load_from_json(file))
elif file.endswith(".toml"):
config.update(load_from_toml(file))
else:
if not no_warning:
logger.warning(f"Unsupported config file format: {file}")
else:
if not no_warning:
logger.warning(f"Config file not found: {file}")
return config
def load_configs_from_dirs(*directories: str, no_waring: bool = False) -> dict[str, Any]:
"""
从目录下加载配置文件,不递归
按照读取文件的优先级反向覆盖
默认执行扁平化选项
"""
config = {}
for directory in directories:
if not os.path.exists(directory):
if not no_waring:
logger.warning(f"Directory not found: {directory}")
continue
for file in os.listdir(directory):
if file.endswith(_SUPPORTED_CONFIG_FORMATS):
config.update(load_from_files(os.path.join(directory, file), no_warning=no_waring))
return config
def load_config_in_default(no_waring: bool = False) -> dict[str, Any]:
"""
从一个标准的轻雪项目加载配置文件
项目目录下的config.*和config目录下的所有配置文件
项目目录下的配置文件优先
"""
config = load_configs_from_dirs("config", no_waring=no_waring)
config.update(load_from_files("config.yaml", "config.toml", "config.json", "config.yml", no_warning=no_waring))
return config