重构配置逻辑

reelforge/config/manager.py

@@ -0,0 +1,119 @@
+"""
+Configuration Manager - Singleton pattern
+
+Provides unified access to configuration with automatic validation.
+"""
+from pathlib import Path
+from typing import Any, Optional
+from loguru import logger
+from .schema import ReelForgeConfig
+from .loader import load_config_dict, save_config_dict
+
+
+class ConfigManager:
+    """
+    Configuration Manager (Singleton)
+    
+    Provides unified access to configuration with automatic validation.
+    """
+    _instance: Optional['ConfigManager'] = None
+    
+    def __new__(cls, config_path: str = "config.yaml"):
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+        return cls._instance
+    
+    def __init__(self, config_path: str = "config.yaml"):
+        # Only initialize once
+        if hasattr(self, '_initialized'):
+            return
+        
+        self.config_path = Path(config_path)
+        self.config: ReelForgeConfig = self._load()
+        self._initialized = True
+    
+    def _load(self) -> ReelForgeConfig:
+        """Load configuration from file"""
+        data = load_config_dict(str(self.config_path))
+        return ReelForgeConfig(**data)
+    
+    def reload(self):
+        """Reload configuration from file"""
+        self.config = self._load()
+        logger.info("Configuration reloaded")
+    
+    def save(self):
+        """Save current configuration to file"""
+        save_config_dict(self.config.to_dict(), str(self.config_path))
+    
+    def update(self, updates: dict):
+        """
+        Update configuration with new values
+        
+        Args:
+            updates: Dictionary of updates (e.g., {"llm": {"api_key": "xxx"}})
+        """
+        current = self.config.to_dict()
+        
+        # Deep merge
+        def deep_merge(base: dict, updates: dict) -> dict:
+            for key, value in updates.items():
+                if key in base and isinstance(base[key], dict) and isinstance(value, dict):
+                    deep_merge(base[key], value)
+                else:
+                    base[key] = value
+            return base
+        
+        merged = deep_merge(current, updates)
+        self.config = ReelForgeConfig(**merged)
+    
+    def get(self, key: str, default: Any = None) -> Any:
+        """Dict-like access (for backward compatibility)"""
+        return self.config.to_dict().get(key, default)
+    
+    def validate(self) -> bool:
+        """Validate configuration completeness"""
+        return self.config.validate_required()
+    
+    def get_llm_config(self) -> dict:
+        """Get LLM configuration as dict"""
+        return {
+            "api_key": self.config.llm.api_key,
+            "base_url": self.config.llm.base_url,
+            "model": self.config.llm.model,
+        }
+    
+    def set_llm_config(self, api_key: str, base_url: str, model: str):
+        """Set LLM configuration"""
+        self.update({
+            "llm": {
+                "api_key": api_key,
+                "base_url": base_url,
+                "model": model,
+            }
+        })
+    
+    def get_image_config(self) -> dict:
+        """Get image configuration as dict"""
+        return {
+            "default": self.config.image.default,
+            "comfyui_url": self.config.image.comfyui_url,
+            "runninghub_api_key": self.config.image.runninghub_api_key,
+            "prompt_prefix": self.config.image.prompt_prefix,
+        }
+    
+    def set_image_config(
+        self, 
+        comfyui_url: Optional[str] = None, 
+        runninghub_api_key: Optional[str] = None
+    ):
+        """Set image configuration"""
+        updates = {}
+        if comfyui_url is not None:
+            updates["comfyui_url"] = comfyui_url
+        if runninghub_api_key is not None:
+            updates["runninghub_api_key"] = runninghub_api_key
+        
+        if updates:
+            self.update({"image": updates})
+