""" 环境配置模块,用于控制应用在不同环境(开发/生产)中的行为 """ import os import logging import json from pathlib import Path from typing import Dict, Any # 默认配置 DEFAULT_CONFIG = { "environment": "development", # 可选:development, production "log_level": {"development": "DEBUG", "production": "WARNING"}, "production_headless": True, # 生产环境是否使用无头浏览器 "data_dir": "data", # 数据存储目录 "config_dir": "config", # 配置文件目录 "log_dir": "logs", # 日志文件目录 } class EnvironmentManager: """环境配置管理类""" def __init__(self, config_file: str = None): # 默认配置文件路径 self.app_dir = Path(__file__).parent.parent.parent self.config_file = config_file or self.app_dir / "config" / "environment.json" self.config = DEFAULT_CONFIG.copy() # 尝试加载配置文件 self._load_config() # 初始化日志配置 self._setup_logging() def _load_config(self) -> Dict[str, Any]: """加载配置文件""" try: if os.path.exists(self.config_file): with open(self.config_file, "r", encoding="utf-8") as f: file_config = json.load(f) self.config.update(file_config) return self.config except Exception as e: print(f"加载配置文件失败: {str(e)},使用默认配置") return self.config def _setup_logging(self): """设置日志级别""" env = self.config.get("environment", "development") log_level_name = self.config.get("log_level", {}).get(env, "INFO") # 将字符串转换为日志级别 log_level = getattr(logging, log_level_name, logging.INFO) # 设置根日志记录器的级别 root_logger = logging.getLogger("douyin_imbot") root_logger.setLevel(log_level) # 如果没有处理器,添加一个控制台处理器 if not root_logger.handlers: handler = logging.StreamHandler() formatter = logging.Formatter( "%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) handler.setFormatter(formatter) root_logger.addHandler(handler) # 如果是生产环境,还添加一个文件处理器 if env == "production": log_dir = self.app_dir / self.config.get("log_dir", "logs") os.makedirs(log_dir, exist_ok=True) log_file = log_dir / f"douyin_imbot.log" file_handler = logging.FileHandler(log_file) file_handler.setFormatter(formatter) root_logger.addHandler(file_handler) def get_setting(self, key: str, default=None) -> Any: """获取特定配置项的值""" return self.config.get(key, default) def get_all_settings(self) -> Dict[str, Any]: """获取所有配置""" return self.config.copy() def update_setting(self, key: str, value: Any) -> bool: """更新配置项""" try: self.config[key] = value return self._save_config() except Exception as e: print(f"更新配置失败: {str(e)}") return False def _save_config(self) -> bool: """保存配置到文件""" try: # 确保目录存在 os.makedirs(os.path.dirname(self.config_file), exist_ok=True) with open(self.config_file, "w", encoding="utf-8") as f: json.dump(self.config, f, indent=2, ensure_ascii=False) return True except Exception as e: print(f"保存配置文件失败: {str(e)}") return False @property def is_production(self) -> bool: """是否为生产环境""" return self.config.get("environment", "").lower() == "production" @property def is_development(self) -> bool: """是否为开发环境""" return not self.is_production # 创建一个单例实例 env_manager = EnvironmentManager() def get_environment_settings(): """获取环境设置的辅助函数""" return env_manager.get_all_settings()