#!/usr/bin/env python # -*- coding: utf-8 -*- import os import yaml from typing import Dict, Any, Optional class ConfigLoader: """配置加载器,用于读取和获取配置信息""" def __init__(self, config_path: Optional[str] = None): """ 初始化配置加载器 Args: config_path: 配置文件路径,如果为None,则使用默认路径 """ if config_path is None: # 默认配置文件路径 current_dir = os.path.dirname(os.path.abspath(__file__)) config_dir = os.path.join(os.path.dirname(current_dir), 'config') config_path = os.path.join(config_dir, 'config.yaml') # 如果默认配置文件不存在,则使用示例配置文件 if not os.path.exists(config_path): config_path = os.path.join(config_dir, 'config.example.yaml') print(f"配置文件 config.yaml 不存在,使用示例配置文件: {config_path}") self.config_path = config_path # 加载配置 self.config = self._load_config() def _load_config(self) -> Dict[str, Any]: """ 加载配置文件 Returns: 配置字典 """ try: with open(self.config_path, 'r', encoding='utf-8') as f: config = yaml.safe_load(f) print(f"已加载配置文件: {self.config_path}") return config except Exception as e: print(f"加载配置文件失败: {e}") return {} def get_config(self, section: str) -> Dict[str, Any]: """ 获取指定部分的配置 Args: section: 配置部分名称 Returns: 配置字典 """ return self.config.get(section, {}) def get_binance_config(self) -> Dict[str, Any]: """获取Binance配置""" return self.get_config('binance') def get_okx_config(self) -> Dict[str, Any]: """获取OKX配置""" return self.get_config('okx') def get_deepseek_config(self) -> Dict[str, Any]: """获取DeepSeek配置""" return self.get_config('deepseek') def get_gold_config(self) -> Dict[str, Any]: """获取黄金配置""" return self.get_config('gold') def get_alltick_config(self) -> Dict[str, Any]: """获取AllTick配置""" return self.get_config('alltick') 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_logging_config(self) -> Dict[str, Any]: """获取日志配置""" return self.get_config('logging') def get_dingtalk_config(self) -> Dict[str, Any]: """获取钉钉机器人配置""" return self.get_config('dingtalk') def get_discord_config(self) -> Dict[str, Any]: """获取Discord机器人配置""" return self.get_config('discord') def get_database_config(self) -> Dict[str, Any]: """获取数据库配置""" return self.get_config('database')