80 lines
2.4 KiB
Python
80 lines
2.4 KiB
Python
import os
|
||
import yaml
|
||
from typing import Dict, Any
|
||
|
||
|
||
class ConfigLoader:
|
||
"""配置加载器，用于读取和获取配置信息"""
|
||
|
||
def __init__(self, config_path: str = None):
|
||
"""
|
||
初始化配置加载器
|
||
|
||
Args:
|
||
config_path: 配置文件路径，如果为None，则使用默认路径
|
||
"""
|
||
if config_path is None:
|
||
# 获取当前文件的目录
|
||
current_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
config_path = os.path.join(current_dir, "config", "config.yaml")
|
||
|
||
self.config_path = config_path
|
||
self.config_data = self._load_config()
|
||
|
||
def _load_config(self) -> Dict[str, Any]:
|
||
"""
|
||
加载配置文件
|
||
|
||
Returns:
|
||
配置数据字典
|
||
"""
|
||
try:
|
||
with open(self.config_path, 'r', encoding='utf-8') as file:
|
||
config_data = yaml.safe_load(file)
|
||
return config_data
|
||
except Exception as e:
|
||
print(f"加载配置文件时出错: {e}")
|
||
return {}
|
||
|
||
def get_config(self, section: str = None) -> Dict[str, Any]:
|
||
"""
|
||
获取配置数据
|
||
|
||
Args:
|
||
section: 配置部分名称，如果为None，则返回整个配置
|
||
|
||
Returns:
|
||
配置数据或指定部分的配置数据
|
||
"""
|
||
if section is None:
|
||
return self.config_data
|
||
|
||
return self.config_data.get(section, {})
|
||
|
||
def get_binance_config(self) -> Dict[str, Any]:
|
||
"""获取Binance API配置"""
|
||
return self.get_config('binance')
|
||
|
||
def get_deepseek_config(self) -> Dict[str, Any]:
|
||
"""获取DeepSeek API配置"""
|
||
return self.get_config('deepseek')
|
||
|
||
def get_crypto_config(self) -> Dict[str, Any]:
|
||
"""获取加密货币配置"""
|
||
return self.get_config('crypto')
|
||
|
||
def get_data_config(self) -> Dict[str, Any]:
|
||
"""获取数据配置"""
|
||
return self.get_config('data')
|
||
|
||
def get_agent_config(self) -> Dict[str, Any]:
|
||
"""获取Agent配置"""
|
||
return self.get_config('agent')
|
||
|
||
def get_logging_config(self) -> Dict[str, Any]:
|
||
"""获取日志配置"""
|
||
return self.get_config('logging')
|
||
|
||
def get_dingtalk_config(self) -> Dict[str, Any]:
|
||
"""获取钉钉机器人配置"""
|
||
return self.get_config('dingtalk') |