Homer
089b144626
The package version had been left at the scaffolded 0.1.0 while the released tag was v5.4.0. Nothing in the code chose 0.1.0 -- it was npm-init's equivalent, never updated -- and it made every version comparison meaningless: the window title, the About dialog and the update check all read it. ConfigManager merged a stored config over the defaults with dict.update(), which is shallow. A config written by an older build carries partial "player" and "feed" objects, and a shallow update replaces the whole nested default with the partial one, so keys added since came back missing. The `or 15` and `or "auto"` fallbacks at the call sites were load-bearing because of it. The merge is now recursive, and keys present only in the stored file are kept so a downgrade cannot destroy settings. Configs now carry a config_version. A file written before 5.4.0 -- including one inherited from an upstream YTSage install -- has its stored check_app_updates cleared once, because that setting used to point at PyPI's `ytsage` package and oop7/YTSage's releases, neither of which is this program. The settings tab read a missing check_app_updates as enabled and persisted that reading on OK, so merely opening Custom Options turned the checker back on. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
314 lines
12 KiB
Python
314 lines
12 KiB
Python
"""
|
|
Config Manager Module
|
|
=====================
|
|
|
|
This module provides **thread-safe** centralized management for application
|
|
configuration in YTSage. It handles reading, writing, and managing settings
|
|
stored in a JSON file, with support for nested keys via dot notation.
|
|
|
|
Thread safety is ensured using a reentrant lock (`RLock`), so multiple threads
|
|
can safely access or modify settings concurrently.
|
|
|
|
Features
|
|
--------
|
|
- Thread-safe operations for getting, setting, and deleting configuration values.
|
|
- Loads settings from a JSON config file (`APP_CONFIG_FILE`).
|
|
- Creates the config file with default values if missing or corrupt.
|
|
- Retrieves, sets, and deletes settings using dot-separated keys.
|
|
- Provides safe error handling with logging instead of raising exceptions.
|
|
- Persists updates back to disk automatically.
|
|
|
|
Usage
|
|
-----
|
|
from .ytsage_config_manager import ConfigManager
|
|
|
|
# Load settings (auto-loads if not already loaded)
|
|
download_path = ConfigManager.get("download_path")
|
|
|
|
# Update a value
|
|
ConfigManager.set("download_path", "D:/Downloads")
|
|
|
|
# Retrieve nested value
|
|
last_check = ConfigManager.get("cached_versions.ytdlp.last_check")
|
|
|
|
# Delete a key
|
|
ConfigManager.delete("cached_versions.ffmpeg.path")
|
|
|
|
Design Notes
|
|
------------
|
|
- Settings are stored in `ConfigManager.settings` (a dict).
|
|
- Default values are defined in `ConfigManager.default_config`.
|
|
- All modifications trigger a save (`_save`) to keep JSON in sync.
|
|
- Logs actions and errors using the app's central logger.
|
|
- Uses `RLock` to allow safe concurrent access from multiple threads.
|
|
|
|
Exceptions
|
|
----------
|
|
- Any issues during file I/O (permissions, disk errors, JSON corruption)
|
|
are caught and logged. The application continues running with defaults
|
|
when possible.
|
|
"""
|
|
|
|
import copy
|
|
import json
|
|
import os
|
|
import threading
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Optional
|
|
|
|
from .ytsage_constants import APP_CONFIG_FILE, USER_HOME_DIR
|
|
from .ytsage_logger import logger
|
|
|
|
|
|
class ConfigManager:
|
|
"""
|
|
Thread-safe configuration manager for YTSage.
|
|
|
|
Provides methods to load, save, get, set, and delete settings stored in a JSON file.
|
|
Supports nested keys via dot notation and automatically persists changes.
|
|
"""
|
|
|
|
_lock: threading.RLock = threading.RLock()
|
|
_config_file: Path = APP_CONFIG_FILE
|
|
_settings: Dict[str, Any] = {}
|
|
_default_config: Dict[str, Any] = {
|
|
# Bumped whenever a stored config needs rewriting rather than merely
|
|
# merging; see _migrate(). Absent means "written before 5.4.0".
|
|
"config_version": 2,
|
|
"download_path": str(USER_HOME_DIR / "Downloads"),
|
|
"generic_mode": True,
|
|
"speed_limit_value": None,
|
|
"speed_limit_unit_index": 0,
|
|
"cookie_source": "browser", # "browser" or "file"
|
|
"cookie_browser": "chrome",
|
|
"cookie_browser_profile": "",
|
|
"cookie_file_path": None,
|
|
"cookie_active": False, # True only if user explicitly applied cookies
|
|
"last_used_cookie_file": None,
|
|
"proxy_url": None,
|
|
"geo_proxy_url": None,
|
|
"auto_update_ytdlp": True,
|
|
"auto_update_frequency": "daily",
|
|
"check_app_updates": False, # fork updates come from Gitea, not the upstream PyPI package
|
|
"check_beta_updates": False,
|
|
"last_update_check": 0,
|
|
"concurrent_fragments": 1,
|
|
"language": "en",
|
|
"ytdlp_channel": "stable",
|
|
"force_output_format": False,
|
|
"preferred_output_format": "mp4",
|
|
"force_audio_format": False,
|
|
"preferred_audio_format": "best",
|
|
"audio_normalization": False,
|
|
"filename_format": "%(title)s_%(resolution)s_[%(id)s].%(ext)s",
|
|
"window_geometry": None,
|
|
"window_state": None,
|
|
"cached_versions": {
|
|
"ytdlp": {"version": None, "path": None, "last_check": 0, "path_mtime": 0},
|
|
"ffmpeg": {"version": None, "path": None, "last_check": 0, "path_mtime": 0},
|
|
},
|
|
"advanced": {
|
|
# Allow falling back to a system-installed yt-dlp from PATH when
|
|
# the app-managed, SHA256-verified binary is absent
|
|
"allow_system_ytdlp": False,
|
|
},
|
|
"player": {
|
|
"default_quality": 1080, # max stream height; None = auto/best
|
|
"volume": 100,
|
|
"resume": "auto", # auto | off
|
|
"source_mode": "ytdl", # ytdl (mpv ytdl_hook) | direct (raw stream URLs)
|
|
},
|
|
"feed": {
|
|
"mode": "local", # local (per-channel aggregation) | account (cookies)
|
|
"per_channel_items": 15,
|
|
"auto_refresh_minutes": 0, # 0 = manual refresh only
|
|
},
|
|
}
|
|
|
|
@classmethod
|
|
def _deep_merge(cls, defaults: Dict[str, Any], stored: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Recursively layer `stored` on top of `defaults`.
|
|
|
|
A shallow `dict.update()` here would be a data-loss bug: a config file
|
|
written by an older build contains a *partial* "player" or "feed"
|
|
object, and updating shallowly replaces the whole nested default with
|
|
it -- so every key added since would come back missing.
|
|
|
|
Rules:
|
|
- dict + dict recurse
|
|
- anything else (lists, scalars, type mismatches) replaces wholesale
|
|
- keys present only in `stored` are kept, never pruned; a user who
|
|
downgrades must not have their settings destroyed by the older build
|
|
"""
|
|
merged = copy.deepcopy(defaults)
|
|
for key, value in stored.items():
|
|
existing = merged.get(key)
|
|
if isinstance(value, dict) and isinstance(existing, dict):
|
|
merged[key] = cls._deep_merge(existing, value)
|
|
else:
|
|
merged[key] = copy.deepcopy(value)
|
|
return merged
|
|
|
|
@classmethod
|
|
def _migrate(cls, stored: Dict[str, Any]) -> bool:
|
|
"""
|
|
Bring a stored config up to CONFIG_VERSION in place.
|
|
|
|
Returns True when something changed, so the caller knows to re-save.
|
|
Migrations must be idempotent and must never raise -- a config that
|
|
cannot be migrated is still a config the app has to start with.
|
|
"""
|
|
changed = False
|
|
version = stored.get("config_version")
|
|
|
|
if not isinstance(version, int) or version < 2:
|
|
# Pre-5.4.0, and possibly inherited from an upstream YTSage
|
|
# install. Such a file carries "check_app_updates": true, which
|
|
# used to point at PyPI's `ytsage` package and oop7/YTSage's
|
|
# releases -- neither of which is this application. Dropping the
|
|
# stored value lets SageTube's own default decide.
|
|
if stored.pop("check_app_updates", None) is not None:
|
|
logger.info("Config migration: cleared inherited 'check_app_updates'.")
|
|
stored["config_version"] = 2
|
|
changed = True
|
|
|
|
return changed
|
|
|
|
@classmethod
|
|
def _load(cls) -> None:
|
|
"""
|
|
Loads configuration settings from a JSON file if it exists and is valid.
|
|
If the file is missing or corrupt, loads default settings and creates or overwrites the config file as needed.
|
|
Logs actions and errors during the process.
|
|
"""
|
|
with cls._lock:
|
|
if cls._config_file.exists():
|
|
try:
|
|
with open(cls._config_file, "r", encoding="utf-8") as f:
|
|
stored = json.load(f)
|
|
if not isinstance(stored, dict):
|
|
raise json.JSONDecodeError("config root is not an object", "", 0)
|
|
|
|
migrated = False
|
|
try:
|
|
migrated = cls._migrate(stored)
|
|
except Exception as e:
|
|
logger.exception(f"Config migration failed, continuing unmigrated: {e}")
|
|
|
|
# Merge on top of defaults so keys added in newer versions
|
|
# exist without call sites needing `or <default>` fallbacks
|
|
cls._settings = cls._deep_merge(cls._default_config, stored)
|
|
logger.info("Config loaded from file.")
|
|
if migrated:
|
|
cls._save()
|
|
except json.JSONDecodeError:
|
|
cls._settings = copy.deepcopy(cls._default_config)
|
|
logger.warning("Config file corrupt, loaded defaults.")
|
|
else:
|
|
cls._settings = copy.deepcopy(cls._default_config)
|
|
cls._save()
|
|
logger.info("Config file not found, created default config.")
|
|
|
|
@classmethod
|
|
def _save(cls) -> None:
|
|
"""
|
|
Save current settings to JSON file.
|
|
|
|
Note:
|
|
May raise exceptions if the file cannot be written due to permission issues,
|
|
disk errors, or other I/O problems.
|
|
"""
|
|
with cls._lock:
|
|
try:
|
|
# Atomic write: a crash mid-save must not truncate the config
|
|
tmp_file = cls._config_file.with_suffix(".json.tmp")
|
|
with open(tmp_file, "w", encoding="utf-8") as f:
|
|
json.dump(cls._settings, f, indent=4)
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
os.replace(tmp_file, cls._config_file)
|
|
logger.debug("Config saved to file.")
|
|
except (OSError, PermissionError) as e:
|
|
logger.exception(f"Failed to save config: {e}")
|
|
except Exception as e:
|
|
logger.exception(f"Unexpected error while saving config: {e}")
|
|
|
|
@classmethod
|
|
def get(cls, key: str) -> Optional[Any]:
|
|
"""
|
|
Retrieve a configuration value using a dotted key notation.
|
|
Args:
|
|
key (str): The dotted key string representing the path to the desired setting (e.g., "database.host").
|
|
Optional[Any]: The value associated with the given key, or None if the key does not exist.
|
|
Notes:
|
|
- If the configuration settings are not loaded, this method will load them before attempting retrieval.
|
|
- If any part of the dotted key path is missing, None is returned and a debug message is logged.
|
|
"""
|
|
with cls._lock:
|
|
if not cls._settings:
|
|
cls._load()
|
|
parts: list[str] = key.split(".")
|
|
value: Any = cls._settings
|
|
for part in parts:
|
|
if isinstance(value, dict) and part in value:
|
|
value = value[part]
|
|
else:
|
|
logger.debug(f"Config key '{key}' not found.")
|
|
return None
|
|
return value
|
|
|
|
@classmethod
|
|
def set(cls, key: str, value: Any) -> None:
|
|
"""
|
|
Sets a configuration value for a given key.
|
|
If the configuration settings are not loaded, loads them first.
|
|
Supports nested keys using dot notation (e.g., "database.host").
|
|
Updates the configuration dictionary with the provided value,
|
|
saves the updated settings, and logs the change.
|
|
Args:
|
|
key (str): The configuration key, possibly nested using dots.
|
|
value (Any): The value to set for the specified key.
|
|
Returns:
|
|
None
|
|
"""
|
|
with cls._lock:
|
|
if not cls._settings:
|
|
cls._load()
|
|
parts: list[str] = key.split(".")
|
|
d: Dict[str, Any] = cls._settings
|
|
for part in parts[:-1]:
|
|
d = d.setdefault(part, {})
|
|
d[parts[-1]] = value
|
|
cls._save()
|
|
logger.info(f"Config key '{key}' set to '{value}'.")
|
|
|
|
@classmethod
|
|
def delete(cls, key: str) -> None:
|
|
"""
|
|
Deletes a configuration key from the settings.
|
|
If the key is nested (dot-separated), traverses the settings dictionary accordingly.
|
|
If the key exists, removes it and saves the updated settings.
|
|
Logs the deletion or if the key was not found.
|
|
Args:
|
|
key (str): The dot-separated configuration key to delete.
|
|
Returns:
|
|
None
|
|
"""
|
|
with cls._lock:
|
|
if not cls._settings:
|
|
cls._load()
|
|
parts: list[str] = key.split(".")
|
|
d: Any = cls._settings
|
|
for part in parts[:-1]:
|
|
if part not in d:
|
|
logger.debug(f"Config key '{key}' not found for deletion.")
|
|
return
|
|
d = d[part]
|
|
if parts[-1] in d:
|
|
d.pop(parts[-1], None)
|
|
cls._save()
|
|
logger.info(f"Config key '{key}' deleted.")
|
|
else:
|
|
logger.debug(f"Config key '{key}' not found for deletion.")
|