#!/usr/bin/env python3 """ Configuration Manager Handles all configuration settings for the NZ Legislation Loophole Analysis application. Provides default configurations, persistent storage, and validation. """ import json import os from pathlib import Path from typing import Dict, Any, Optional import streamlit as st class ConfigManager: """Configuration manager for the application""" def __init__(self, config_file: str = None): """ Initialize configuration manager Args: config_file: Path to configuration file (optional) """ if config_file is None: config_dir = Path(__file__).parent.parent / 'config' config_dir.mkdir(exist_ok=True) config_file = config_dir / 'app_config.json' self.config_file = Path(config_file) self._config = {} self._load_config() def _load_config(self): """Load configuration from file or use defaults""" if self.config_file.exists(): try: with open(self.config_file, 'r', encoding='utf-8') as f: self._config = json.load(f) # Validate and merge with defaults self._config = self._merge_with_defaults(self._config) except (json.JSONDecodeError, IOError) as e: print(f"Warning: Could not load config file: {e}") self._config = self._get_default_config() else: self._config = self._get_default_config() def _get_default_config(self) -> Dict[str, Any]: """Get default configuration""" return { 'model': { 'path': 'qwen3.gguf', 'repo_id': 'DavidAU/Qwen3-Zero-Coder-Reasoning-0.8B-NEO-EX-GGUF', 'filename': 'Qwen3-Zero-Coder-Reasoning-0.8B-NEO-EX-D_AU-IQ4_XS-imat.gguf', 'context_length': 40960, 'max_tokens': 4096, 'temperature': 0.3, 'top_p': 0.85, 'top_k': 50, 'repeat_penalty': 1.15 }, 'processing': { 'chunk_size': 4096, 'chunk_overlap': 256, 'batch_size': 16, 'clean_text': True, 'preserve_structure': True }, 'cache': { 'enabled': True, 'max_size_mb': 1024, 'ttl_hours': 24, 'persistent': True }, 'analysis': { 'depth': 'Standard', 'include_recommendations': True, 'focus_areas': ['loopholes', 'ambiguities', 'unintended_consequences'], 'legal_domains': ['constitutional', 'administrative', 'criminal', 'civil'] }, 'ui': { 'theme': 'Auto', 'show_progress': True, 'auto_refresh': False, 'max_display_items': 50 }, 'advanced': { 'debug_mode': False, 'log_level': 'INFO', 'memory_limit_mb': 8192, 'thread_pool_size': 4, 'save_intermediate_results': True } } def _merge_with_defaults(self, user_config: Dict[str, Any]) -> Dict[str, Any]: """Merge user configuration with defaults""" default_config = self._get_default_config() def merge_dicts(default: Dict[str, Any], user: Dict[str, Any]) -> Dict[str, Any]: merged = default.copy() for key, value in user.items(): if key in merged and isinstance(merged[key], dict) and isinstance(value, dict): merged[key] = merge_dicts(merged[key], value) else: merged[key] = value return merged return merge_dicts(default_config, user_config) def get_config(self) -> Dict[str, Any]: """Get current configuration""" return self._config.copy() def update_config(self, new_config: Dict[str, Any]): """Update configuration with validation""" # Validate configuration if self._validate_config(new_config): self._config = self._merge_with_defaults(new_config) self._save_config() else: raise ValueError("Invalid configuration provided") def _validate_config(self, config: Dict[str, Any]) -> bool: """Validate configuration values""" try: # Model validation model_config = config.get('model', {}) if model_config.get('context_length', 0) < 1024: return False if model_config.get('max_tokens', 0) < 64: return False if not (0 <= model_config.get('temperature', 0) <= 2): return False # Processing validation proc_config = config.get('processing', {}) if proc_config.get('chunk_size', 0) < 256: return False if proc_config.get('chunk_overlap', 0) >= proc_config.get('chunk_size', 1): return False if proc_config.get('batch_size', 0) < 1: return False # Cache validation cache_config = config.get('cache', {}) if cache_config.get('max_size_mb', 0) < 100: return False if cache_config.get('ttl_hours', 0) < 1: return False return True except Exception: return False def _save_config(self): """Save configuration to file""" try: self.config_file.parent.mkdir(exist_ok=True) with open(self.config_file, 'w', encoding='utf-8') as f: json.dump(self._config, f, indent=2, ensure_ascii=False) except IOError as e: print(f"Warning: Could not save config file: {e}") def reset_to_defaults(self): """Reset configuration to defaults""" self._config = self._get_default_config() self._save_config() def get_section(self, section: str) -> Dict[str, Any]: """Get a specific configuration section""" return self._config.get(section, {}) def update_section(self, section: str, values: Dict[str, Any]): """Update a specific configuration section""" if section not in self._config: self._config[section] = {} self._config[section].update(values) # Validate the updated config if self._validate_config(self._config): self._save_config() else: raise ValueError(f"Invalid configuration for section: {section}") def export_config(self, filepath: str) -> bool: """Export configuration to file""" try: with open(filepath, 'w', encoding='utf-8') as f: json.dump(self._config, f, indent=2, ensure_ascii=False) return True except IOError: return False def import_config(self, filepath: str) -> bool: """Import configuration from file""" try: with open(filepath, 'r', encoding='utf-8') as f: imported_config = json.load(f) if self._validate_config(imported_config): self._config = self._merge_with_defaults(imported_config) self._save_config() return True else: return False except (IOError, json.JSONDecodeError): return False def get_model_config(self) -> Dict[str, Any]: """Get model-specific configuration""" return self._config.get('model', {}) def get_processing_config(self) -> Dict[str, Any]: """Get processing-specific configuration""" return self._config.get('processing', {}) def get_cache_config(self) -> Dict[str, Any]: """Get cache-specific configuration""" return self._config.get('cache', {}) def get_ui_config(self) -> Dict[str, Any]: """Get UI-specific configuration""" return self._config.get('ui', {}) def get_advanced_config(self) -> Dict[str, Any]: """Get advanced configuration""" return self._config.get('advanced', {}) # Global configuration instance _config_instance = None def get_config_manager(config_file: str = None) -> ConfigManager: """Get or create global configuration manager instance""" global _config_instance if _config_instance is None: _config_instance = ConfigManager(config_file) return _config_instance