mirror/Huntarr-Sonarr
1
0
Fork 0
mirror of https://github.com/plexguide/Huntarr-Sonarr.git synced 2025-12-16 20:04:16 -06:00
Code Issues Packages Projects Releases Wiki Activity

Implement database integration for settings management and update .gitignore for local development data

Browse Source
This commit is contained in:
Admin9705
2025-06-12 17:58:26 -04:00
parent 94e4423ea5
commit 6420b9e93b
4 changed files with 431 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
.gitignore vendored
View File

@@ -227,4 +227,16 @@ cython_debug/
.ruff_cache/
# PyPI configuration file
.pypirc
.pypirc
# Huntarr local development data
data/
*.db
*.sqlite
*.sqlite3
huntarr.db
/config/
logs/
settings/
stateful/
tally/

10
main.py
View File

@@ -289,6 +289,16 @@ def main():
# Register cleanup handler
atexit.register(cleanup_handler)
# Initialize database with default configurations
try:
from primary.settings_manager import initialize_database
initialize_database()
huntarr_logger.info("Database initialization completed successfully")
except Exception as e:
huntarr_logger.error(f"Failed to initialize database: {e}")
huntarr_logger.error("Application may not function correctly without database")
# Continue anyway - the app might still work with defaults
background_thread = None
try:

258
src/primary/settings_manager.py
View File

@@ -1,7 +1,7 @@
#!/usr/bin/env python3
"""
Settings manager for Huntarr
Handles loading, saving, and providing settings from individual JSON files per app
Handles loading, saving, and providing settings from SQLite database
Supports default configurations for different Arr applications
"""
@@ -9,8 +9,6 @@ import os
import json
import pathlib
import logging
import shutil
import subprocess
import time
from typing import Dict, Any, Optional, List
@@ -18,19 +16,16 @@ from typing import Dict, Any, Optional, List
logging.basicConfig(level=logging.INFO)
settings_logger = logging.getLogger("settings_manager")
# Settings directory setup - Root config directory
# Use the centralized path configuration
from src.primary.utils.config_paths import SETTINGS_DIR
# Database integration
from src.primary.utils.database import get_database
# Settings directory is already created by config_paths module
# Default configs location remains the same
# Default configs location
DEFAULT_CONFIGS_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), 'default_configs'))
# Update or add this as a class attribute or constant
# Known app types
KNOWN_APP_TYPES = ["sonarr", "radarr", "lidarr", "readarr", "whisparr", "eros", "swaparr", "general"]
# Add a settings cache with timestamps to avoid excessive disk reads
# Add a settings cache with timestamps to avoid excessive database reads
settings_cache = {} # Format: {app_name: {'timestamp': timestamp, 'data': settings_dict}}
CACHE_TTL = 5 # Cache time-to-live in seconds
@@ -45,18 +40,10 @@ def clear_cache(app_name=None):
settings_logger.debug("Clearing entire settings cache")
settings_cache = {}
def get_settings_file_path(app_name: str) -> pathlib.Path:
"""Get the path to the settings file for a specific app."""
if app_name not in KNOWN_APP_TYPES:
# Log a warning but allow for potential future app types
settings_logger.warning(f"Requested settings file for unknown app type: {app_name}")
return SETTINGS_DIR / f"{app_name}.json"
def get_default_config_path(app_name: str) -> pathlib.Path:
"""Get the path to the default config file for a specific app."""
return pathlib.Path(DEFAULT_CONFIGS_DIR) / f"{app_name}.json"
# Helper function to load default settings for a specific app
def load_default_app_settings(app_name: str) -> Dict[str, Any]:
"""Load default settings for a specific app from its JSON file."""
default_file = get_default_config_path(app_name)
@@ -72,29 +59,41 @@ def load_default_app_settings(app_name: str) -> Dict[str, Any]:
return {}
def _ensure_config_exists(app_name: str) -> None:
"""Ensure the config file exists for an app, copying from default if not."""
settings_file = get_settings_file_path(app_name)
if not settings_file.exists():
default_file = get_default_config_path(app_name)
if default_file.exists():
try:
shutil.copyfile(default_file, settings_file)
settings_logger.info(f"Created default settings file for {app_name} at {settings_file}")
except Exception as e:
settings_logger.error(f"Error copying default settings for {app_name}: {e}")
"""Ensure the config exists for an app in the database."""
try:
db = get_database()
if app_name == 'general':
# Check if general settings exist
existing_settings = db.get_general_settings()
if not existing_settings:
# Load defaults and store in database
default_settings = load_default_app_settings(app_name)
if default_settings:
db.save_general_settings(default_settings)
settings_logger.info(f"Created default general settings in database")
else:
settings_logger.warning(f"No default config found for general settings")
else:
# Create an empty file if no default exists
settings_logger.warning(f"No default config found for {app_name}. Creating empty settings file.")
try:
with open(settings_file, 'w') as f:
json.dump({}, f)
except Exception as e:
settings_logger.error(f"Error creating empty settings file for {app_name}: {e}")
# Check if app config exists
config = db.get_app_config(app_name)
if config is None:
# Load defaults and store in database
default_settings = load_default_app_settings(app_name)
if default_settings:
db.save_app_config(app_name, default_settings)
settings_logger.info(f"Created default settings in database for {app_name}")
else:
# Create empty config in database
db.save_app_config(app_name, {})
settings_logger.warning(f"No default config found for {app_name}. Created empty database entry.")
except Exception as e:
settings_logger.error(f"Database error for {app_name}: {e}")
raise
def load_settings(app_type, use_cache=True):
"""
Load settings for a specific app type
Load settings for a specific app type from database
Args:
app_type: The app type to load settings for
@@ -120,64 +119,63 @@ def load_settings(app_type, use_cache=True):
else:
settings_logger.debug(f"Cache expired for {app_type} (age: {cache_age:.1f}s)")
# No valid cache entry, load from disk
_ensure_config_exists(app_type)
settings_file = get_settings_file_path(app_type)
# No valid cache entry, load from database
current_settings = {}
try:
with open(settings_file, 'r') as f:
# Load existing settings
current_settings = json.load(f)
# Load defaults to check for missing keys
default_settings = load_default_app_settings(app_type)
# Add missing keys from defaults without overwriting existing values
updated = False
for key, value in default_settings.items():
if key not in current_settings:
current_settings[key] = value
updated = True
# Apply Lidarr migration (artist -> album) for Huntarr 7.5.0+
if app_type == "lidarr":
if current_settings.get("hunt_missing_mode") == "artist":
settings_logger.info("Migrating Lidarr hunt_missing_mode from 'artist' to 'album' (Huntarr 7.5.0+)")
current_settings["hunt_missing_mode"] = "album"
updated = True
# If keys were added, save the updated file
if updated:
settings_logger.info(f"Added missing default keys to {app_type}.json")
save_settings(app_type, current_settings) # Use save_settings to handle writing
# Update cache
settings_cache[app_type] = {
'timestamp': time.time(),
'data': current_settings
}
return current_settings
except json.JSONDecodeError:
settings_logger.error(f"Error decoding JSON from {settings_file}. Restoring from default.")
# Attempt to restore from default
default_settings = load_default_app_settings(app_type)
save_settings(app_type, default_settings) # Save the restored defaults
db = get_database()
# Update cache with defaults
settings_cache[app_type] = {
'timestamp': time.time(),
'data': default_settings
}
if app_type == 'general':
current_settings = db.get_general_settings()
if not current_settings:
# Config doesn't exist in database, create it
_ensure_config_exists(app_type)
current_settings = db.get_general_settings()
else:
current_settings = db.get_app_config(app_type)
if current_settings is None:
# Config doesn't exist in database, create it
_ensure_config_exists(app_type)
current_settings = db.get_app_config(app_type) or {}
settings_logger.debug(f"Loaded {app_type} settings from database")
return default_settings
except Exception as e:
settings_logger.error(f"Error loading settings for {app_type} from {settings_file}: {e}")
return {} # Return empty dict on other errors
settings_logger.error(f"Database error loading {app_type}: {e}")
raise
# Load defaults to check for missing keys
default_settings = load_default_app_settings(app_type)
# Add missing keys from defaults without overwriting existing values
updated = False
for key, value in default_settings.items():
if key not in current_settings:
current_settings[key] = value
updated = True
# Apply Lidarr migration (artist -> album) for Huntarr 7.5.0+
if app_type == "lidarr":
if current_settings.get("hunt_missing_mode") == "artist":
settings_logger.info("Migrating Lidarr hunt_missing_mode from 'artist' to 'album' (Huntarr 7.5.0+)")
current_settings["hunt_missing_mode"] = "album"
updated = True
# If keys were added, save the updated settings
if updated:
settings_logger.info(f"Added missing default keys to {app_type} settings")
save_settings(app_type, current_settings)
# Update cache
settings_cache[app_type] = {
'timestamp': time.time(),
'data': current_settings
}
return current_settings
def save_settings(app_name: str, settings_data: Dict[str, Any]) -> bool:
"""Save settings for a specific app."""
"""Save settings for a specific app to database."""
if app_name not in KNOWN_APP_TYPES:
settings_logger.error(f"Attempted to save settings for unknown app type: {app_name}")
return False
@@ -186,17 +184,23 @@ def save_settings(app_name: str, settings_data: Dict[str, Any]) -> bool:
if app_name == 'general':
settings_logger.info(f"Saving general settings: {settings_data}")
settings_logger.info(f"Apprise URLs being saved: {settings_data.get('apprise_urls', 'NOT_FOUND')}")
settings_file = get_settings_file_path(app_name)
try:
# Ensure the directory exists (though it should from the top-level check)
settings_file.parent.mkdir(parents=True, exist_ok=True)
db = get_database()
# Write the provided settings data directly
with open(settings_file, 'w') as f:
json.dump(settings_data, f, indent=2)
settings_logger.info(f"Settings saved successfully for {app_name} to {settings_file}")
if app_name == 'general':
db.save_general_settings(settings_data)
else:
db.save_app_config(app_name, settings_data)
settings_logger.info(f"Settings saved successfully for {app_name} to database")
success = True
except Exception as e:
settings_logger.error(f"Database error saving {app_name}: {e}")
return False
if success:
# Clear cache for this app to ensure fresh reads
clear_cache(app_name)
@@ -208,11 +212,8 @@ def save_settings(app_name: str, settings_data: Dict[str, Any]) -> bool:
settings_logger.debug("Timezone cache cleared after general settings save")
except Exception as e:
settings_logger.warning(f"Failed to clear timezone cache: {e}")
return True
except Exception as e:
settings_logger.error(f"Error saving settings for {app_name} to {settings_file}: {e}")
return False
return success
def get_setting(app_name: str, key: str, default: Optional[Any] = None) -> Any:
"""Get a specific setting value for an app."""
@@ -231,8 +232,7 @@ def get_all_settings() -> Dict[str, Dict[str, Any]]:
"""Load settings for all known apps."""
all_settings = {}
for app_name in KNOWN_APP_TYPES:
# Only include apps if their config file exists or can be created from defaults
# Effectively, load_settings ensures the file exists and loads it.
# Only include apps if their config exists or can be created from defaults
settings = load_settings(app_name)
if settings: # Only add if settings were successfully loaded
all_settings[app_name] = settings
@@ -242,6 +242,9 @@ def get_configured_apps() -> List[str]:
"""Return a list of app names that have basic configuration (API URL and Key)."""
configured = []
for app_name in KNOWN_APP_TYPES:
if app_name == 'general':
continue # Skip general settings
settings = load_settings(app_name)
# First check if there are valid instances configured (multi-instance mode)
@@ -362,46 +365,49 @@ def get_advanced_setting(setting_name, default_value=None):
default_value: The default value to return if the setting is not found
Returns:
The value of the setting or the default value if not found
The value of the advanced setting, or default_value if not found
"""
if setting_name not in ADVANCED_SETTINGS:
settings_logger.warning(f"Requested unknown advanced setting: {setting_name}")
settings_logger.warning(f"get_advanced_setting called with unknown setting: {setting_name}")
# Get from general settings
general_settings = load_settings('general', use_cache=True)
general_settings = load_settings("general")
return general_settings.get(setting_name, default_value)
def get_ssl_verify_setting():
"""
Get the SSL verification setting.
Get the SSL verification setting from general settings.
Returns:
bool: True if SSL verification should be enabled (default), False otherwise
bool: True if SSL verification is enabled, False otherwise
"""
return get_advanced_setting("ssl_verify", True)
return get_advanced_setting("ssl_verify", True) # Default to True for security
def get_custom_tag(app_name: str, tag_type: str, default: str) -> str:
"""
Get a custom tag for an app and tag type.
Get a custom tag for a specific app and tag type.
Args:
app_name: The app name (sonarr, radarr, etc.)
tag_type: The tag type (missing, upgrade, shows_missing)
default: Default tag if custom tag not found
app_name: The name of the app (e.g., 'sonarr', 'radarr')
tag_type: The type of tag (e.g., 'missing', 'upgrade')
default: The default tag to return if not found
Returns:
The custom tag string
str: The custom tag or the default if not found
"""
settings = load_settings(app_name)
custom_tags = settings.get('custom_tags', {})
tag = custom_tags.get(tag_type, default)
# Validate tag length (max 25 characters as per UI)
if len(tag) > 25:
settings_logger.warning(f"Custom tag '{tag}' for {app_name}.{tag_type} exceeds 25 characters, truncating")
tag = tag[:25]
return tag
custom_tags = settings.get("custom_tags", {})
return custom_tags.get(tag_type, default)
def initialize_database():
"""Initialize the database with default configurations if needed."""
try:
db = get_database()
defaults_dir = pathlib.Path(DEFAULT_CONFIGS_DIR)
db.initialize_from_defaults(defaults_dir)
settings_logger.info("Database initialized with default configurations")
except Exception as e:
settings_logger.error(f"Failed to initialize database: {e}")
raise
# Example usage (for testing purposes, remove later)
if __name__ == "__main__":

276
src/primary/utils/database.py Normal file
View File

@@ -0,0 +1,276 @@
"""
SQLite Database Manager for Huntarr
Replaces all JSON file operations with SQLite database for better performance and reliability.
Handles both app configurations and general settings.
"""
import os
import json
import sqlite3
from pathlib import Path
from typing import Dict, List, Any, Optional
from datetime import datetime
import logging
logger = logging.getLogger(__name__)
class HuntarrDatabase:
"""Database manager for all Huntarr configurations and settings"""
def __init__(self):
self.db_path = self._get_database_path()
self.ensure_database_exists()
def _get_database_path(self) -> Path:
"""Get database path - use local data directory for development"""
# For local development, use data directory in project root
project_root = Path(__file__).parent.parent.parent.parent
data_dir = project_root / "data"
# Ensure directory exists
data_dir.mkdir(parents=True, exist_ok=True)
return data_dir / "huntarr.db"
def ensure_database_exists(self):
"""Create database and all tables if they don't exist"""
with sqlite3.connect(self.db_path) as conn:
conn.execute('PRAGMA foreign_keys = ON')
# Create app_configs table for all app settings
conn.execute('''
CREATE TABLE IF NOT EXISTS app_configs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
app_type TEXT NOT NULL UNIQUE,
config_data TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# Create general_settings table for general/global settings
conn.execute('''
CREATE TABLE IF NOT EXISTS general_settings (
id INTEGER PRIMARY KEY AUTOINCREMENT,
setting_key TEXT NOT NULL UNIQUE,
setting_value TEXT NOT NULL,
setting_type TEXT DEFAULT 'string',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# Create indexes for better performance
conn.execute('CREATE INDEX IF NOT EXISTS idx_app_configs_type ON app_configs(app_type)')
conn.execute('CREATE INDEX IF NOT EXISTS idx_general_settings_key ON general_settings(setting_key)')
conn.commit()
logger.info(f"Database initialized at: {self.db_path}")
def get_app_config(self, app_type: str) -> Optional[Dict[str, Any]]:
"""Get app configuration from database"""
with sqlite3.connect(self.db_path) as conn:
cursor = conn.execute(
'SELECT config_data FROM app_configs WHERE app_type = ?',
(app_type,)
)
row = cursor.fetchone()
if row:
try:
return json.loads(row[0])
except json.JSONDecodeError as e:
logger.error(f"Failed to parse JSON for {app_type}: {e}")
return None
return None
def save_app_config(self, app_type: str, config_data: Dict[str, Any]):
"""Save app configuration to database"""
config_json = json.dumps(config_data, indent=2)
with sqlite3.connect(self.db_path) as conn:
conn.execute('''
INSERT OR REPLACE INTO app_configs (app_type, config_data, updated_at)
VALUES (?, ?, CURRENT_TIMESTAMP)
''', (app_type, config_json))
conn.commit()
logger.info(f"Saved {app_type} configuration to database")
def get_general_settings(self) -> Dict[str, Any]:
"""Get all general settings as a dictionary"""
with sqlite3.connect(self.db_path) as conn:
conn.row_factory = sqlite3.Row
cursor = conn.execute(
'SELECT setting_key, setting_value, setting_type FROM general_settings'
)
settings = {}
for row in cursor.fetchall():
key = row['setting_key']
value = row['setting_value']
setting_type = row['setting_type']
# Convert value based on type
if setting_type == 'boolean':
settings[key] = value.lower() == 'true'
elif setting_type == 'integer':
settings[key] = int(value)
elif setting_type == 'float':
settings[key] = float(value)
elif setting_type == 'json':
try:
settings[key] = json.loads(value)
except json.JSONDecodeError:
settings[key] = value
else: # string
settings[key] = value
return settings
def save_general_settings(self, settings: Dict[str, Any]):
"""Save general settings to database"""
with sqlite3.connect(self.db_path) as conn:
for key, value in settings.items():
# Determine type and convert value
if isinstance(value, bool):
setting_type = 'boolean'
setting_value = str(value).lower()
elif isinstance(value, int):
setting_type = 'integer'
setting_value = str(value)
elif isinstance(value, float):
setting_type = 'float'
setting_value = str(value)
elif isinstance(value, (list, dict)):
setting_type = 'json'
setting_value = json.dumps(value)
else:
setting_type = 'string'
setting_value = str(value)
conn.execute('''
INSERT OR REPLACE INTO general_settings
(setting_key, setting_value, setting_type, updated_at)
VALUES (?, ?, ?, CURRENT_TIMESTAMP)
''', (key, setting_value, setting_type))
conn.commit()
logger.info("Saved general settings to database")
def get_general_setting(self, key: str, default: Any = None) -> Any:
"""Get a specific general setting"""
with sqlite3.connect(self.db_path) as conn:
cursor = conn.execute(
'SELECT setting_value, setting_type FROM general_settings WHERE setting_key = ?',
(key,)
)
row = cursor.fetchone()
if row:
value, setting_type = row
# Convert value based on type
if setting_type == 'boolean':
return value.lower() == 'true'
elif setting_type == 'integer':
return int(value)
elif setting_type == 'float':
return float(value)
elif setting_type == 'json':
try:
return json.loads(value)
except json.JSONDecodeError:
return value
else: # string
return value
return default
def set_general_setting(self, key: str, value: Any):
"""Set a specific general setting"""
# Determine type and convert value
if isinstance(value, bool):
setting_type = 'boolean'
setting_value = str(value).lower()
elif isinstance(value, int):
setting_type = 'integer'
setting_value = str(value)
elif isinstance(value, float):
setting_type = 'float'
setting_value = str(value)
elif isinstance(value, (list, dict)):
setting_type = 'json'
setting_value = json.dumps(value)
else:
setting_type = 'string'
setting_value = str(value)
with sqlite3.connect(self.db_path) as conn:
conn.execute('''
INSERT OR REPLACE INTO general_settings
(setting_key, setting_value, setting_type, updated_at)
VALUES (?, ?, ?, CURRENT_TIMESTAMP)
''', (key, setting_value, setting_type))
conn.commit()
logger.debug(f"Set general setting {key} = {value}")
def get_all_app_types(self) -> List[str]:
"""Get list of all app types in database"""
with sqlite3.connect(self.db_path) as conn:
cursor = conn.execute('SELECT app_type FROM app_configs ORDER BY app_type')
return [row[0] for row in cursor.fetchall()]
def initialize_from_defaults(self, defaults_dir: Path):
"""Initialize database with default configurations if empty"""
app_types = ['sonarr', 'radarr', 'lidarr', 'readarr', 'whisparr', 'eros', 'swaparr', 'general']
for app_type in app_types:
# Check if config already exists
existing_config = self.get_app_config(app_type) if app_type != 'general' else self.get_general_settings()
if not existing_config:
# Load default config
default_file = defaults_dir / f"{app_type}.json"
if default_file.exists():
try:
with open(default_file, 'r') as f:
default_config = json.load(f)
if app_type == 'general':
self.save_general_settings(default_config)
else:
self.save_app_config(app_type, default_config)
logger.info(f"Initialized {app_type} with default configuration")
except Exception as e:
logger.error(f"Failed to initialize {app_type} from defaults: {e}")
def backup_to_json(self, backup_dir: Path):
"""Backup database configurations to JSON files"""
backup_dir.mkdir(parents=True, exist_ok=True)
# Backup app configs
for app_type in self.get_all_app_types():
config = self.get_app_config(app_type)
if config:
backup_file = backup_dir / f"{app_type}.json"
with open(backup_file, 'w') as f:
json.dump(config, f, indent=2)
logger.info(f"Backed up {app_type} to {backup_file}")
# Backup general settings
general_settings = self.get_general_settings()
if general_settings:
backup_file = backup_dir / "general.json"
with open(backup_file, 'w') as f:
json.dump(general_settings, f, indent=2)
logger.info(f"Backed up general settings to {backup_file}")
# Global database instance
_database_instance = None
def get_database() -> HuntarrDatabase:
"""Get the global database instance"""
global _database_instance
if _database_instance is None:
_database_instance = HuntarrDatabase()
return _database_instance