danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Merge pull request #80 from Nezreka/docker-fix

Browse source 
Migrate config storage to SQLite database
This commit is contained in:
Broque Thomas 2025-12-04 11:09:24 -08:00 committed by GitHub
commit c237f354c4
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 220 additions and 32 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

252
config/settings.py
View file

@ -1,5 +1,6 @@
import json
import os
import sqlite3
from typing import Dict, Any, Optional
from cryptography.fernet import Fernet
from pathlib import Path
@ -9,8 +10,9 @@ class ConfigManager:
self.config_path = Path(config_path)
self.config_data: Dict[str, Any] = {}
self.encryption_key: Optional[bytes] = None
self.database_path = Path("database/music_library.db") # Hardcoded - same as MusicDatabase
self._load_config()
def _get_encryption_key(self) -> bytes:
key_file = self.config_path.parent / ".encryption_key"
if key_file.exists():
@ -22,48 +24,234 @@ class ConfigManager:
f.write(key)
key_file.chmod(0o600)
return key
def _ensure_database_exists(self):
"""Ensure database file and metadata table exist"""
try:
# Create database directory if it doesn't exist
self.database_path.parent.mkdir(parents=True, exist_ok=True)
# Connect to database (creates file if it doesn't exist)
conn = sqlite3.connect(str(self.database_path))
cursor = conn.cursor()
# Create metadata table if it doesn't exist
cursor.execute("""
CREATE TABLE IF NOT EXISTS metadata (
key TEXT PRIMARY KEY,
value TEXT,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.commit()
conn.close()
except Exception as e:
print(f"Warning: Could not ensure database exists: {e}")
def _load_from_database(self) -> Optional[Dict[str, Any]]:
"""Load configuration from database"""
try:
self._ensure_database_exists()
conn = sqlite3.connect(str(self.database_path))
cursor = conn.cursor()
cursor.execute("SELECT value FROM metadata WHERE key = 'app_config'")
row = cursor.fetchone()
conn.close()
if row and row[0]:
config_data = json.loads(row[0])
print("[OK] Configuration loaded from database")
return config_data
else:
return None
except Exception as e:
print(f"Warning: Could not load config from database: {e}")
return None
def _save_to_database(self, config_data: Dict[str, Any]) -> bool:
"""Save configuration to database"""
try:
self._ensure_database_exists()
conn = sqlite3.connect(str(self.database_path))
cursor = conn.cursor()
config_json = json.dumps(config_data, indent=2)
cursor.execute("""
INSERT OR REPLACE INTO metadata (key, value, updated_at)
VALUES ('app_config', ?, CURRENT_TIMESTAMP)
""", (config_json,))
conn.commit()
conn.close()
return True
except Exception as e:
print(f"Error: Could not save config to database: {e}")
return False
def _load_from_config_file(self) -> Optional[Dict[str, Any]]:
"""Load configuration from config.json file (for migration)"""
try:
if self.config_path.exists():
with open(self.config_path, 'r') as f:
config_data = json.load(f)
print(f"[OK] Configuration loaded from {self.config_path}")
return config_data
else:
return None
except Exception as e:
print(f"Warning: Could not load config from file: {e}")
return None
def _get_default_config(self) -> Dict[str, Any]:
"""Get default configuration"""
return {
"active_media_server": "plex",
"spotify": {
"client_id": "",
"client_secret": "",
"redirect_uri": "http://127.0.0.1:8888/callback"
},
"tidal": {
"client_id": "",
"client_secret": "",
"redirect_uri": "http://127.0.0.1:8889/tidal/callback"
},
"plex": {
"base_url": "",
"token": "",
"auto_detect": True
},
"jellyfin": {
"base_url": "",
"api_key": "",
"auto_detect": True
},
"navidrome": {
"base_url": "",
"username": "",
"password": "",
"auto_detect": True
},
"soulseek": {
"slskd_url": "",
"api_key": "",
"download_path": "./downloads",
"transfer_path": "./Transfer"
},
"listenbrainz": {
"token": ""
},
"logging": {
"path": "logs/app.log",
"level": "INFO"
},
"database": {
"path": "database/music_library.db",
"max_workers": 5
},
"metadata_enhancement": {
"enabled": True,
"embed_album_art": True
},
"playlist_sync": {
"create_backup": True
},
"settings": {
"audio_quality": "flac"
}
}
def _load_config(self):
if not self.config_path.exists():
raise FileNotFoundError(f"Configuration file not found: {self.config_path}")
with open(self.config_path, 'r') as f:
self.config_data = json.load(f)
"""
Load configuration with priority:
1. Database (primary storage)
2. config.json (migration from file-based config)
3. Defaults (fresh install)
"""
# Try loading from database first
config_data = self._load_from_database()
if config_data:
# Configuration exists in database
self.config_data = config_data
return
# Database is empty - try migration from config.json
config_data = self._load_from_config_file()
if config_data:
# Migrate from config.json to database
print("[MIGRATE] Migrating configuration from config.json to database...")
if self._save_to_database(config_data):
print("[OK] Configuration migrated successfully")
self.config_data = config_data
return
else:
print("[WARN] Migration failed - using file-based config")
self.config_data = config_data
return
# No config.json either - use defaults
print("[INFO] No existing configuration found - using defaults")
config_data = self._get_default_config()
# Try to save defaults to database
if self._save_to_database(config_data):
print("[OK] Default configuration saved to database")
else:
print("[WARN] Could not save defaults to database - using in-memory config")
self.config_data = config_data
def _save_config(self):
with open(self.config_path, 'w') as f:
json.dump(self.config_data, f, indent=2)
"""Save configuration to database"""
success = self._save_to_database(self.config_data)
if not success:
# Fallback: Try to save to config.json if database fails
print("[WARN] Database save failed - attempting file fallback")
try:
self.config_path.parent.mkdir(parents=True, exist_ok=True)
with open(self.config_path, 'w') as f:
json.dump(self.config_data, f, indent=2)
print("[OK] Configuration saved to config.json as fallback")
except Exception as e:
print(f"[ERROR] Failed to save configuration: {e}")
def get(self, key: str, default: Any = None) -> Any:
keys = key.split('.')
value = self.config_data
for k in keys:
if isinstance(value, dict) and k in value:
value = value[k]
else:
return default
return value
def set(self, key: str, value: Any):
keys = key.split('.')
config = self.config_data
for k in keys[:-1]:
if k not in config:
config[k] = {}
config = config[k]
config[keys[-1]] = value
self._save_config()
def get_spotify_config(self) -> Dict[str, str]:
return self.get('spotify', {})
def get_plex_config(self) -> Dict[str, str]:
return self.get('plex', {})
def get_jellyfin_config(self) -> Dict[str, str]:
return self.get('jellyfin', {})
@ -72,25 +260,25 @@ class ConfigManager:
def get_soulseek_config(self) -> Dict[str, str]:
return self.get('soulseek', {})
def get_settings(self) -> Dict[str, Any]:
return self.get('settings', {})
def get_database_config(self) -> Dict[str, str]:
return self.get('database', {})
def get_logging_config(self) -> Dict[str, str]:
return self.get('logging', {})
def get_active_media_server(self) -> str:
return self.get('active_media_server', 'plex')
def set_active_media_server(self, server: str):
"""Set the active media server (plex, jellyfin, or navidrome)"""
if server not in ['plex', 'jellyfin', 'navidrome']:
raise ValueError(f"Invalid media server: {server}")
self.set('active_media_server', server)
def get_active_media_server_config(self) -> Dict[str, str]:
"""Get configuration for the currently active media server"""
active_server = self.get_active_media_server()
@ -102,12 +290,12 @@ class ConfigManager:
return self.get_navidrome_config()
else:
return {}
def is_configured(self) -> bool:
spotify = self.get_spotify_config()
active_server = self.get_active_media_server()
soulseek = self.get_soulseek_config()
# Check active media server configuration
media_server_configured = False
if active_server == 'plex':
@ -119,22 +307,22 @@ class ConfigManager:
elif active_server == 'navidrome':
navidrome = self.get_navidrome_config()
media_server_configured = bool(navidrome.get('base_url')) and bool(navidrome.get('username')) and bool(navidrome.get('password'))
return (
bool(spotify.get('client_id')) and
bool(spotify.get('client_secret')) and
media_server_configured and
bool(soulseek.get('slskd_url'))
)
def validate_config(self) -> Dict[str, bool]:
active_server = self.get_active_media_server()
validation = {
'spotify': bool(self.get('spotify.client_id')) and bool(self.get('spotify.client_secret')),
'soulseek': bool(self.get('soulseek.slskd_url'))
}
# Validate all server types but mark active one
validation['plex'] = bool(self.get('plex.base_url')) and bool(self.get('plex.token'))
validation['jellyfin'] = bool(self.get('jellyfin.base_url')) and bool(self.get('jellyfin.api_key'))
@ -143,4 +331,4 @@ class ConfigManager:
return validation
config_manager = ConfigManager()
config_manager = ConfigManager()