first commit
This commit is contained in:
commit
4890793ce5
1029 files changed
+5992
No files matched your search
@@ -0,0 +1,127 @@
|
||||
"""
|
||||
环境配置模块,用于控制应用在不同环境(开发/生产)中的行为
|
||||
"""
|
||||
|
||||
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()
|
||||
Reference in new issue
Block a user