SageTube/ytsage/utils/ytsage_config_manager.py

"""
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,
        "skipped_update_version": None,  # set by the update dialog's "Skip this version"
        "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.")