Files
2025-04-02 21:01:25 +08:00

128 lines
4.2 KiB
Python

"""
环境配置模块,用于控制应用在不同环境(开发/生产)中的行为
"""
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()