129f69fce9
Introduce an optional "Blasphemy Mode" that deletes the original FLAC after a verified MP3 copy is created. - config: add lossy_copy.delete_original (default: false). - webui/index.html & static script: add checkbox and warning in settings UI and persist the setting. - web_server.py: make _create_lossy_copy return the MP3 path when it deletes the FLAC (otherwise None); validate the MP3 using mutagen before removing the FLAC; rename associated .lrc files if present; update post-processing to use the final processed path in logs and wishlist checks and to consider .mp3 variants when FLAC may have been removed. Behavior is off by default and includes safety checks and logging to avoid accidental deletion of originals.
402 lines
14 KiB
Python
402 lines
14 KiB
Python
import json
|
||
import os
|
||
import sqlite3
|
||
from typing import Dict, Any, Optional
|
||
from cryptography.fernet import Fernet
|
||
from pathlib import Path
|
||
|
||
class ConfigManager:
|
||
def __init__(self, config_path: str = "config/config.json"):
|
||
# Determine strict absolute path to settings.py directory to help resolve config.json
|
||
# This handles cases where CWD is different (e.g. running from /Users vs /Users/project)
|
||
self.base_dir = Path(__file__).parent.parent.absolute()
|
||
|
||
# Check for environment variable override first (Unified logic with web_server.py)
|
||
env_config_path = os.environ.get('SOULSYNC_CONFIG_PATH')
|
||
if env_config_path:
|
||
config_path = env_config_path
|
||
|
||
# Resolve config path
|
||
if os.path.isabs(config_path):
|
||
self.config_path = Path(config_path)
|
||
else:
|
||
# Try to resolve relative to CWD first (legacy behavior), then relative to project root
|
||
cwd_path = Path(config_path)
|
||
project_path = self.base_dir / config_path
|
||
|
||
if cwd_path.exists():
|
||
self.config_path = cwd_path.absolute()
|
||
elif project_path.exists():
|
||
self.config_path = project_path
|
||
else:
|
||
# Default to project path even if it doesn't exist yet (for creation/fallback)
|
||
self.config_path = project_path
|
||
|
||
print(f"🔧 ConfigManager initialized with path: {self.config_path}")
|
||
|
||
self.config_data: Dict[str, Any] = {}
|
||
self.encryption_key: Optional[bytes] = None
|
||
|
||
# Use DATABASE_PATH env var, fallback to database/music_library.db
|
||
db_path_env = os.environ.get('DATABASE_PATH')
|
||
if db_path_env:
|
||
self.database_path = Path(db_path_env)
|
||
else:
|
||
self.database_path = self.base_dir / "database" / "music_library.db"
|
||
|
||
print(f"💾 Database path set to: {self.database_path}")
|
||
|
||
self.load_config(str(self.config_path))
|
||
|
||
def load_config(self, config_path: str = None):
|
||
"""
|
||
Load configuration from database or file.
|
||
Can be called to reload settings into the existing instance.
|
||
"""
|
||
if config_path:
|
||
self.config_path = Path(config_path)
|
||
|
||
self._load_config()
|
||
|
||
def _get_encryption_key(self) -> bytes:
|
||
key_file = self.config_path.parent / ".encryption_key"
|
||
if key_file.exists():
|
||
with open(key_file, 'rb') as f:
|
||
return f.read()
|
||
else:
|
||
key = Fernet.generate_key()
|
||
with open(key_file, 'wb') as f:
|
||
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"
|
||
},
|
||
"download_source": {
|
||
"mode": "soulseek", # Options: "soulseek", "youtube", "hybrid"
|
||
"hybrid_primary": "soulseek", # Which source to try first in hybrid mode
|
||
"youtube_min_confidence": 0.65 # Minimum confidence for YouTube matches
|
||
},
|
||
"listenbrainz": {
|
||
"token": ""
|
||
},
|
||
"acoustid": {
|
||
"api_key": "",
|
||
"enabled": False # Disabled by default - requires API key and fpcalc
|
||
},
|
||
"logging": {
|
||
"path": "logs/app.log",
|
||
"level": "INFO"
|
||
},
|
||
"database": {
|
||
"path": os.environ.get('DATABASE_PATH', 'database/music_library.db'),
|
||
"max_workers": 5
|
||
},
|
||
"metadata_enhancement": {
|
||
"enabled": True,
|
||
"embed_album_art": True
|
||
},
|
||
"musicbrainz": {
|
||
"embed_tags": True
|
||
},
|
||
"playlist_sync": {
|
||
"create_backup": True
|
||
},
|
||
"settings": {
|
||
"audio_quality": "flac"
|
||
},
|
||
"lossy_copy": {
|
||
"enabled": False,
|
||
"bitrate": "320",
|
||
"delete_original": False
|
||
},
|
||
"import": {
|
||
"staging_path": "./Staging"
|
||
}
|
||
}
|
||
|
||
def _load_config(self):
|
||
"""
|
||
Load configuration with priority:
|
||
1. Database (primary storage)
|
||
2. config.json (migration from file-based config)
|
||
3. Defaults (fresh install)
|
||
"""
|
||
print(f"📥 Loading configuration...")
|
||
|
||
# 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
|
||
print(f"⚠️ Configuration not found in database. Attempting migration from: {self.config_path}")
|
||
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 to database.")
|
||
self.config_data = config_data
|
||
return
|
||
else:
|
||
print("[WARN] ⚠️ Migration failed - using file-based config temporarily.")
|
||
self.config_data = config_data
|
||
return
|
||
|
||
# No config.json either - use defaults
|
||
print("[INFO] ℹ️ No existing configuration found (DB or File) - 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):
|
||
"""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', {})
|
||
|
||
def get_navidrome_config(self) -> Dict[str, str]:
|
||
return self.get('navidrome', {})
|
||
|
||
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()
|
||
if active_server == 'plex':
|
||
return self.get_plex_config()
|
||
elif active_server == 'jellyfin':
|
||
return self.get_jellyfin_config()
|
||
elif active_server == 'navidrome':
|
||
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':
|
||
plex = self.get_plex_config()
|
||
media_server_configured = bool(plex.get('base_url')) and bool(plex.get('token'))
|
||
elif active_server == 'jellyfin':
|
||
jellyfin = self.get_jellyfin_config()
|
||
media_server_configured = bool(jellyfin.get('base_url')) and bool(jellyfin.get('api_key'))
|
||
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'))
|
||
validation['navidrome'] = bool(self.get('navidrome.base_url')) and bool(self.get('navidrome.username')) and bool(self.get('navidrome.password'))
|
||
validation['active_media_server'] = active_server
|
||
|
||
return validation
|
||
|
||
config_manager = ConfigManager()
|