""" ConfigManager - Gestione sicura di config.yaml per editor profili """ import yaml import os import shutil from datetime import datetime from typing import Optional, Dict, Tuple class ConfigManager: """Gestisce lettura/scrittura di config.yaml con backup e validazione""" def __init__(self, config_path: str): """ Inizializza ConfigManager Args: config_path: Path assoluto a config.yaml """ self.config_path = config_path self.config = None self.backup_dir = os.path.join( os.path.dirname(config_path), 'backups' ) # Crea directory backup se non esiste os.makedirs(self.backup_dir, exist_ok=True) def load_config(self) -> Dict: """ Carica config.yaml con error handling Returns: Dict contenente la configurazione Raises: FileNotFoundError: Se config.yaml non esiste ValueError: Se YAML è malformato o vuoto Exception: Altri errori di caricamento """ if self.config: return self.config try: with open(self.config_path, 'r', encoding='utf-8') as f: self.config = yaml.safe_load(f) if not self.config: raise ValueError("Config file vuoto o invalido") return self.config except FileNotFoundError: raise FileNotFoundError(f"File config non trovato: {self.config_path}") except yaml.YAMLError as e: raise ValueError(f"Errore parsing YAML: {e}") except Exception as e: raise Exception(f"Errore caricamento config: {e}") def get_service_config(self, profile_name: str, service_name: str) -> Optional[Dict]: """ Estrae config per servizio specifico di un profilo Args: profile_name: Nome profilo (es. 'warren', 'aurelio') service_name: Nome servizio (es. 'CognitiveService') Returns: Dict con parametri servizio, o None se non trovato """ if not self.config: self.load_config() profiles = self.config.get('profiles', {}) if profile_name not in profiles: return None services = profiles[profile_name] for service in services: if service.get('service_name') == service_name: return service.get('config', {}) return None def update_service_config( self, profile_name: str, service_name: str, updates: Dict ) -> bool: """ Aggiorna parametri di un servizio e scrive su disk Args: profile_name: Nome profilo (es. 'warren') service_name: Nome servizio (es. 'CognitiveService') updates: Dict con chiavi da aggiornare (es. {'temperature': 0.8}) Returns: True se successo Raises: ValueError: Se profilo o servizio non trovato Exception: Se scrittura fallisce """ if not self.config: self.load_config() # Trova il servizio profiles = self.config.get('profiles', {}) if profile_name not in profiles: raise ValueError(f"Profilo '{profile_name}' non trovato") services = profiles[profile_name] service_found = False for service in services: if service.get('service_name') == service_name: # Aggiorna parametri in memoria service_config = service.get('config', {}) service_config.update(updates) service['config'] = service_config service_found = True break if not service_found: raise ValueError(f"Servizio '{service_name}' non trovato in profilo '{profile_name}'") # Scrivi su disk try: with open(self.config_path, 'w', encoding='utf-8') as f: yaml.dump( self.config, f, default_flow_style=False, allow_unicode=True, sort_keys=False, width=120 ) return True except Exception as e: raise Exception(f"Errore scrittura config.yaml: {e}") def create_backup(self) -> str: """ Crea backup timestampato di config.yaml Returns: Path del file backup creato Raises: Exception: Se creazione backup fallisce """ timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") backup_filename = f"config.yaml.backup_{timestamp}" backup_path = os.path.join(self.backup_dir, backup_filename) try: shutil.copy2(self.config_path, backup_path) # Pulizia vecchi backup (mantieni ultimi 5) self._cleanup_old_backups(max_backups=5) return backup_path except Exception as e: raise Exception(f"Errore creazione backup: {e}") def _cleanup_old_backups(self, max_backups: int = 5): """ Rimuove backup vecchi, mantiene solo gli ultimi N Args: max_backups: Numero massimo di backup da mantenere """ try: backups = [ f for f in os.listdir(self.backup_dir) if f.startswith('config.yaml.backup_') ] backups.sort(reverse=True) # Più recenti prima # Rimuovi i vecchi oltre il limite for old_backup in backups[max_backups:]: os.remove(os.path.join(self.backup_dir, old_backup)) except Exception: pass # Non critico se fallisce def validate_yaml_syntax(self, yaml_text: str) -> Tuple[bool, str]: """ Valida sintassi YAML prima di scrivere Args: yaml_text: Stringa YAML da validare Returns: Tupla (is_valid, error_message) """ try: yaml.safe_load(yaml_text) return True, "" except yaml.YAMLError as e: return False, f"Errore sintassi YAML: {str(e)}" def get_all_profiles(self) -> Dict[str, Dict]: """ Carica tutti i profili da config.yaml con metadata Returns: Dict { 'profile_id': { 'name': str, 'subtitle': str, 'icon': str, 'description': str, 'color': str, 'period': str, 'services': list }, ... } """ if not self.config: self.load_config() profiles = self.config.get('profiles', {}) result = {} for profile_id, services in profiles.items(): # Estrai metadata per ogni profilo result[profile_id] = self._get_profile_metadata(profile_id, services) return result def _get_profile_metadata(self, profile_id: str, services: list) -> Dict: """ Estrae metadata per un profilo Priorità: 1. Metadata esplicito in config (futuro) 2. Inferenza da agent_prompt in CognitiveService 3. Default generati da profile_id Args: profile_id: ID del profilo services: Lista servizi del profilo Returns: Dict con metadata completi """ # 1. Cerca metadata esplicito (formato futuro) for service in services: if isinstance(service, dict) and 'metadata' in service: metadata = service['metadata'] metadata['services'] = services return metadata # 2. Cerca CognitiveService per inferire metadata cognitive_service = None for service in services: if isinstance(service, dict) and service.get('service_name') == 'CognitiveService': cognitive_service = service break if cognitive_service: agent_prompt = cognitive_service.get('config', {}).get('agent_prompt', '') if agent_prompt: metadata = self._infer_metadata_from_prompt(profile_id, agent_prompt) metadata['services'] = services return metadata # 3. Fallback a default metadata = self._generate_default_metadata(profile_id) metadata['services'] = services return metadata def _infer_metadata_from_prompt(self, profile_id: str, prompt: str) -> Dict: """ Inferisce metadata da agent_prompt Args: profile_id: ID del profilo prompt: Agent prompt da cui estrarre info Returns: Dict con metadata inferiti """ # Estrazione euristica da prompt lines = prompt.split('\n') # Cerca nome/titolo nelle prime righe name = None subtitle = None period = None for line in lines[:10]: # Analizza prime 10 righe line = line.strip() # Pattern comuni: "Sei Marco Aurelio", "Tu sei Warren Buffett" if any(word in line.lower() for word in ['sei ', 'you are', "you're"]): # Estrai nome dopo "sei/are" for sep in [' sei ', ' are ', "'re "]: if sep in line.lower(): parts = line.lower().split(sep) if len(parts) > 1: candidate = parts[1].split(',')[0].split('.')[0].strip() if len(candidate) > 0 and len(candidate) < 50: name = candidate.title() break # Pattern date: "121-180 d.C.", "1930-present" if any(char.isdigit() for char in line): if '-' in line and any(x in line.lower() for x in ['d.c.', 'a.d.', 'present', '19', '20']): period = line.strip() # Descrizione: prime 150 caratteri del prompt description = ' '.join(prompt.split()[:20]) if len(description) > 150: description = description[:147] + '...' # Icon e color basati su profile_id defaults = self._generate_default_metadata(profile_id) return { 'name': name or defaults['name'], 'subtitle': subtitle or defaults['subtitle'], 'icon': defaults['icon'], 'description': description, 'color': defaults['color'], 'period': period or defaults.get('period', '') } def _generate_default_metadata(self, profile_id: str) -> Dict: """ Genera metadata di default per un profilo Args: profile_id: ID del profilo Returns: Dict con metadata default sicuri """ # Metadata conosciuti per profili di sistema KNOWN_PROFILES = { 'aurelio': { 'name': 'Marco Aurelio', 'subtitle': 'Imperatore Filosofo', 'icon': '👑', 'description': 'Imperatore romano e filosofo stoico, autore delle Meditazioni', 'color': '#8B4513', 'period': '121-180 d.C.' }, 'warren': { 'name': 'Warren Buffett', 'subtitle': 'Oracolo di Omaha', 'icon': '📈', 'description': 'Investitore e filantropo, CEO di Berkshire Hathaway', 'color': '#2E7D32', 'period': '1930-presente' } } # Se profilo conosciuto, ritorna metadata completi if profile_id in KNOWN_PROFILES: return KNOWN_PROFILES[profile_id].copy() # Altrimenti genera default generici return { 'name': profile_id.replace('_', ' ').title(), 'subtitle': 'Profilo Personalizzato', 'icon': '🤖', 'description': f'Profilo personalizzato: {profile_id}', 'color': '#757575', 'period': '' } def delete_profile(self, profile_id: str) -> bool: """ Elimina profilo da config.yaml Crea backup automatico prima di eliminare. Protegge i profili di sistema (aurelio, warren). Args: profile_id: ID del profilo da eliminare Returns: True se successo Raises: ValueError: Se profilo è di sistema o non esiste Exception: Se salvataggio fallisce """ # Lista profili di sistema protetti SYSTEM_PROFILES = ['aurelio', 'warren'] # Validazione: profilo di sistema if profile_id in SYSTEM_PROFILES: raise ValueError( f"Impossibile eliminare profilo di sistema: {profile_id}" ) # Carica config se necessario if not self.config: self.load_config() # Validazione: profilo esiste profiles = self.config.get('profiles', {}) if profile_id not in profiles: raise ValueError(f"Profilo '{profile_id}' non esiste in config.yaml") # Crea backup prima di eliminare try: backup_path = self.create_backup() print(f"[ConfigManager] Backup creato: {backup_path}") except Exception as e: print(f"[ConfigManager] WARNING: Impossibile creare backup: {e}") # Procedi comunque (non bloccare eliminazione per backup fallito) # Elimina profilo da config del self.config['profiles'][profile_id] # Salva config aggiornato self._save_config() print(f"[ConfigManager] Profilo '{profile_id}' eliminato da config.yaml") return True def _save_config(self): """ Salva config corrente su disk Raises: Exception: Se scrittura fallisce """ try: with open(self.config_path, 'w', encoding='utf-8') as f: yaml.dump( self.config, f, default_flow_style=False, allow_unicode=True, sort_keys=False, width=120 ) except Exception as e: raise Exception(f"Errore scrittura config.yaml: {e}")