3b7c00377f
Apply Black formatter to the entire codebase for consistent code style. Co-Authored-By: Claude (glm-5) <noreply@anthropic.com>
150 lines
3.8 KiB
Python
150 lines
3.8 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
配置管理器
|
|
|
|
负责加载、保存和管理用户配置。
|
|
支持从环境变量和 .env 文件加载配置。
|
|
"""
|
|
|
|
import os
|
|
from typing import TYPE_CHECKING
|
|
from config.loader import ConfigLoader
|
|
from config.schema import AppConfig
|
|
from config.defaults import DEFAULT_SETTINGS_DICT
|
|
|
|
# 类型提示时导入，避免循环引用
|
|
if TYPE_CHECKING:
|
|
from config.schema import ERPConfig, DatabaseConfig, PathConfig, ExtractionConfig
|
|
|
|
|
|
class ConfigManager:
|
|
"""配置管理器"""
|
|
|
|
def __init__(
|
|
self, config_file: str = "config/user_settings.json", use_env: bool = True
|
|
):
|
|
"""
|
|
初始化配置管理器
|
|
|
|
Args:
|
|
config_file: 配置文件路径（向后兼容）
|
|
use_env: 是否使用环境变量，默认为 True
|
|
"""
|
|
self.config_file = config_file
|
|
self.use_env = use_env
|
|
self.config: AppConfig = ConfigLoader.load(config_file, use_env=use_env)
|
|
|
|
# 验证配置
|
|
errors = self.config.validate()
|
|
if errors:
|
|
print("配置验证失败：")
|
|
for error in errors:
|
|
print(f" - {error}")
|
|
|
|
def save(self) -> bool:
|
|
"""
|
|
保存配置到文件
|
|
|
|
如果使用环境变量，则保存到 .env 文件
|
|
否则保存到 JSON 文件（向后兼容）
|
|
|
|
Returns:
|
|
保存是否成功
|
|
"""
|
|
if self.use_env:
|
|
return ConfigLoader.save_to_env(self.config, ".env")
|
|
else:
|
|
return ConfigLoader.save(self.config, self.config_file)
|
|
|
|
def get(self, key: str, default=None):
|
|
"""
|
|
获取配置项
|
|
|
|
支持点号分隔的路径，如 "erp.url"
|
|
|
|
Args:
|
|
key: 配置键
|
|
default: 默认值
|
|
|
|
Returns:
|
|
配置值
|
|
"""
|
|
keys = key.split(".")
|
|
value = self.config
|
|
|
|
try:
|
|
for k in keys:
|
|
value = getattr(value, k)
|
|
return value
|
|
except (AttributeError, TypeError):
|
|
return default
|
|
|
|
def set(self, key: str, value) -> None:
|
|
"""
|
|
设置配置项
|
|
|
|
支持点号分隔的路径，如 "erp.url"
|
|
|
|
Args:
|
|
key: 配置键
|
|
value: 配置值
|
|
"""
|
|
keys = key.split(".")
|
|
obj = self.config
|
|
|
|
# 导航到父对象
|
|
for k in keys[:-1]:
|
|
obj = getattr(obj, k)
|
|
|
|
# 获取目标字段信息用于类型转换
|
|
target_field = keys[-1]
|
|
field_type = type(getattr(obj, target_field))
|
|
|
|
# 如果是字符串且目标字段是枚举类型，进行转换
|
|
if isinstance(value, str) and hasattr(
|
|
field_type, "__members__"
|
|
): # 它是一个 Enum
|
|
try:
|
|
value = field_type(value)
|
|
except ValueError:
|
|
# 无效的枚举值，保持当前值不变
|
|
value = getattr(obj, target_field)
|
|
|
|
# 设置最终值
|
|
setattr(obj, target_field, value)
|
|
|
|
def reset_to_defaults(self) -> None:
|
|
"""重置为默认配置"""
|
|
self.config = AppConfig.from_env() # 重新从环境变量加载默认配置
|
|
self.save()
|
|
|
|
@property
|
|
def erp(self):
|
|
"""获取 ERP 配置"""
|
|
return self.config.erp
|
|
|
|
@property
|
|
def database(self):
|
|
"""获取数据库配置"""
|
|
return self.config.database
|
|
|
|
@property
|
|
def paths(self):
|
|
"""获取路径配置"""
|
|
return self.config.paths
|
|
|
|
@property
|
|
def extraction(self):
|
|
"""获取提取配置"""
|
|
return self.config.extraction
|
|
|
|
@property
|
|
def execution(self):
|
|
"""获取执行配置"""
|
|
return self.config.execution
|
|
|
|
|
|
# 为了向后兼容，保留旧版本的导入
|
|
DEFAULT_SETTINGS = DEFAULT_SETTINGS_DICT
|