Something went wrong. Try again.
forked from cameron.stream/void
Something went wrong. Try again.
void migrate_config.py
12 kB · 322 lines
Python
at contributing
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323#!/usr/bin/env python3"""Configuration Migration Script for Void BotMigrates from .env environment variables to config.yaml YAML configuration."""
import osimport shutilfrom pathlib import Pathimport yamlfrom datetime import datetime
def load_env_file(env_path=".env"): """Load environment variables from .env file.""" env_vars = {} if not os.path.exists(env_path): return env_vars try: with open(env_path, 'r', encoding='utf-8') as f: for line_num, line in enumerate(f, 1): line = line.strip() # Skip empty lines and comments if not line or line.startswith('#'): continue # Parse KEY=VALUE format if '=' in line: key, value = line.split('=', 1) key = key.strip() value = value.strip() # Remove quotes if present if value.startswith('"') and value.endswith('"'): value = value[1:-1] elif value.startswith("'") and value.endswith("'"): value = value[1:-1] env_vars[key] = value else: print(f"⚠️ Warning: Skipping malformed line {line_num} in .env: {line}") except Exception as e: print(f"❌ Error reading .env file: {e}") return env_vars
def create_config_from_env(env_vars, existing_config=None): """Create YAML configuration from environment variables.""" # Start with existing config if available, otherwise use defaults if existing_config: config = existing_config.copy() else: config = {} # Ensure all sections exist if 'letta' not in config: config['letta'] = {} if 'bluesky' not in config: config['bluesky'] = {} if 'bot' not in config: config['bot'] = {} # Map environment variables to config structure env_mapping = { 'LETTA_API_KEY': ('letta', 'api_key'), 'BSKY_USERNAME': ('bluesky', 'username'), 'BSKY_PASSWORD': ('bluesky', 'password'), 'PDS_URI': ('bluesky', 'pds_uri'), } migrated_vars = [] for env_var, (section, key) in env_mapping.items(): if env_var in env_vars: config[section][key] = env_vars[env_var] migrated_vars.append(env_var) # Set some sensible defaults if not already present if 'timeout' not in config['letta']: config['letta']['timeout'] = 600 if 'pds_uri' not in config['bluesky']: config['bluesky']['pds_uri'] = "https://bsky.social" # Add bot configuration defaults if not present if 'fetch_notifications_delay' not in config['bot']: config['bot']['fetch_notifications_delay'] = 30 if 'max_processed_notifications' not in config['bot']: config['bot']['max_processed_notifications'] = 10000 if 'max_notification_pages' not in config['bot']: config['bot']['max_notification_pages'] = 20 return config, migrated_vars
def backup_existing_files(): """Create backups of existing configuration files.""" timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") backups = [] # Backup existing config.yaml if it exists if os.path.exists("config.yaml"): backup_path = f"config.yaml.backup_{timestamp}" shutil.copy2("config.yaml", backup_path) backups.append(("config.yaml", backup_path)) # Backup .env if it exists if os.path.exists(".env"): backup_path = f".env.backup_{timestamp}" shutil.copy2(".env", backup_path) backups.append((".env", backup_path)) return backups
def load_existing_config(): """Load existing config.yaml if it exists.""" if not os.path.exists("config.yaml"): return None try: with open("config.yaml", 'r', encoding='utf-8') as f: return yaml.safe_load(f) or {} except Exception as e: print(f"⚠️ Warning: Could not read existing config.yaml: {e}") return None
def write_config_yaml(config): """Write the configuration to config.yaml.""" try: with open("config.yaml", 'w', encoding='utf-8') as f: # Write header comment f.write("# Void Bot Configuration\n") f.write("# Generated by migration script\n") f.write(f"# Created: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") f.write("# See config.yaml.example for all available options\n\n") # Write YAML content yaml.dump(config, f, default_flow_style=False, allow_unicode=True, indent=2) return True except Exception as e: print(f"❌ Error writing config.yaml: {e}") return False
def main(): """Main migration function.""" print("🔄 Void Bot Configuration Migration Tool") print("=" * 50) print("This tool migrates from .env environment variables to config.yaml") print() # Check what files exist has_env = os.path.exists(".env") has_config = os.path.exists("config.yaml") has_example = os.path.exists("config.yaml.example") print("📋 Current configuration files:") print(f" - .env file: {'✅ Found' if has_env else '❌ Not found'}") print(f" - config.yaml: {'✅ Found' if has_config else '❌ Not found'}") print(f" - config.yaml.example: {'✅ Found' if has_example else '❌ Not found'}") print() # If no .env file, suggest creating config from example if not has_env: if not has_config and has_example: print("💡 No .env file found. Would you like to create config.yaml from the example?") response = input("Create config.yaml from example? (y/n): ").lower().strip() if response in ['y', 'yes']: try: shutil.copy2("config.yaml.example", "config.yaml") print("✅ Created config.yaml from config.yaml.example") print("📝 Please edit config.yaml to add your credentials") return except Exception as e: print(f"❌ Error copying example file: {e}") return else: print("👋 Migration cancelled") return else: print("ℹ️ No .env file found and config.yaml already exists or no example available") print(" If you need to set up configuration, see CONFIG.md") return # Load environment variables from .env print("🔍 Reading .env file...") env_vars = load_env_file() if not env_vars: print("⚠️ No environment variables found in .env file") return print(f" Found {len(env_vars)} environment variables") for key in env_vars.keys(): # Mask sensitive values if 'KEY' in key or 'PASSWORD' in key: value_display = f"***{env_vars[key][-4:]}" if len(env_vars[key]) > 4 else "***" else: value_display = env_vars[key] print(f" - {key}={value_display}") print() # Load existing config if present existing_config = load_existing_config() if existing_config: print("📄 Found existing config.yaml - will merge with .env values") # Create configuration print("🏗️ Building configuration...") config, migrated_vars = create_config_from_env(env_vars, existing_config) if not migrated_vars: print("⚠️ No recognized configuration variables found in .env") print(" Recognized variables: LETTA_API_KEY, BSKY_USERNAME, BSKY_PASSWORD, PDS_URI") return print(f" Migrating {len(migrated_vars)} variables: {', '.join(migrated_vars)}") # Show preview print("\n📋 Configuration preview:") print("-" * 30) # Show Letta section if 'letta' in config and config['letta']: print("🔧 Letta:") for key, value in config['letta'].items(): if 'key' in key.lower(): display_value = f"***{value[-8:]}" if len(str(value)) > 8 else "***" else: display_value = value print(f" {key}: {display_value}") # Show Bluesky section if 'bluesky' in config and config['bluesky']: print("🐦 Bluesky:") for key, value in config['bluesky'].items(): if 'password' in key.lower(): display_value = f"***{value[-4:]}" if len(str(value)) > 4 else "***" else: display_value = value print(f" {key}: {display_value}") print() # Confirm migration response = input("💾 Proceed with migration? This will update config.yaml (y/n): ").lower().strip() if response not in ['y', 'yes']: print("👋 Migration cancelled") return # Create backups print("💾 Creating backups...") backups = backup_existing_files() for original, backup in backups: print(f" Backed up {original} → {backup}") # Write new configuration print("✍️ Writing config.yaml...") if write_config_yaml(config): print("✅ Successfully created config.yaml") # Test the new configuration print("\n🧪 Testing new configuration...") try: from config_loader import get_config test_config = get_config() print("✅ Configuration loads successfully") # Test specific sections try: from config_loader import get_letta_config letta_config = get_letta_config() print("✅ Letta configuration valid") except Exception as e: print(f"⚠️ Letta config issue: {e}") try: from config_loader import get_bluesky_config bluesky_config = get_bluesky_config() print("✅ Bluesky configuration valid") except Exception as e: print(f"⚠️ Bluesky config issue: {e}") except Exception as e: print(f"❌ Configuration test failed: {e}") return # Success message and next steps print("\n🎉 Migration completed successfully!") print("\n📖 Next steps:") print(" 1. Run: python test_config.py") print(" 2. Test the bot: python bsky.py --test") print(" 3. If everything works, you can optionally remove the .env file") print(" 4. See CONFIG.md for more configuration options") if backups: print(f"\n🗂️ Backup files created:") for original, backup in backups: print(f" {backup}") print(" These can be deleted once you verify everything works") else: print("❌ Failed to write config.yaml") if backups: print("🔄 Restoring backups...") for original, backup in backups: try: if original != ".env": # Don't restore .env, keep it as fallback shutil.move(backup, original) print(f" Restored {backup} → {original}") except Exception as e: print(f" ❌ Failed to restore {backup}: {e}")
if __name__ == "__main__": main()