dzgui/dzgui/managers/config.py

import logging
import traceback

from collections import deque
from typing import Any, TYPE_CHECKING

from dzgui.const.constants import (
    APP_NAME,
    FLATPAK_RUN_CMD,
    FLATPAK_SANDBOX,
    STEAM_CMD,
    WINDOW_DEFAULT_X,
    WINDOW_DEFAULT_Y,
)
from dzgui.api.probe import test_steam_api, test_bm_api
from dzgui.const.enum import Preferences
from dzgui.managers.threading import call_on_thread, StoredFunc, ThreadingManager
from dzgui.views.dialogs.generic import ExceptionDialog
from dzgui.strings import dialogs
from dzgui.util._json import read_json, write_json

import gi

gi.require_version("Gtk", "3.0")
from gi.repository.Gtk import main_quit  # noqa E402

if TYPE_CHECKING:
    from dzgui.config.userprefs import UserPrefs
    from dzgui.controllers.mc import Controller
    from dzgui.views.base import OuterWindow
    from dzgui.views.trees.tree_servers import ServerTreeView

logger = logging.getLogger(APP_NAME)


class ConfigManager:
    def __init__(self, prefs: "UserPrefs", controller: "Controller") -> None:
        self.prefs = prefs
        self.config = prefs.paths.config
        self.controller = controller
        self.emitter = controller.get_emitter()
        self.thread_man = ThreadingManager(controller)

    def lookup(self, enum: Preferences) -> Any:
        key = self.enum_to_key(enum)
        config = self.get_config()
        try:
            return config[key]
        except KeyError:
            return None

    def add_saved_server(self, record: str) -> None:
        ips = self.lookup(Preferences.IP_LIST)
        ips.append(record)
        self.update_config(Preferences.IP_LIST, ips)

    def update_history_file(self, fqip: str) -> None:
        try:
            with open(self.prefs.paths.history, "r") as f:
                ips = [line.rstrip() for line in f.readlines()]
        except FileNotFoundError:
            ips = []

        seen = set()
        ips.append(fqip)
        # NOTE: Preserve linear order of records while deduplicating
        unique = deque(maxlen=10)  # type: ignore
        for i in range(len(ips) - 1, -1, -1):
            if ips[i] not in seen:
                seen.add(ips[i])
                unique.appendleft(ips[i])
                # NOTE: appendleft will push 11th item off right edge
                if len(unique) == 10:
                    break

        with open(self.prefs.paths.history, "w") as f:
            for record in unique:
                f.write(f"{record}\n")

    def remove_saved_server(self, record: str) -> None:
        ips = self.lookup(Preferences.IP_LIST)
        ips.remove(record)
        self.update_config(Preferences.IP_LIST, ips)

    # TODO: drop/reroute
    def update_config(self, key: Preferences, value: str) -> None:
        self.write_config(key, value)
        # TODO: suppress signals
        # then reenable (or it spawns dialog twice)
        # TODO: do this on demand if/when in options page
        # self.mediator.grid.notebook.settings.populate_settings()

    def get_config(self) -> Any:
        # TODO: is this being called twice?
        try:
            return read_json(self.config)
        except Exception as e:
            logger.critical(e)
            raise e

    @call_on_thread(dialogs.checking_api)
    def update_api_key(self, key: Preferences, text: str) -> None:
        if key is Preferences.STEAM:
            res = test_steam_api(text)
        else:
            res = test_bm_api(text)
        if res is True:
            self.update_config(key, text)
        else:
            self.thread_man.set_cleanup_func(
                StoredFunc(lambda: self.emitter.emit("api_change_failed"))
            )
        return

    def get_start_tab(self) -> Any:
        return int(self.lookup(Preferences.START_TAB))

    def get_favorites(self) -> list[str]:
        return list(self.lookup(Preferences.IP_LIST))

    def get_favorite(self) -> tuple[str, str] | None:
        fav = str(self.lookup(Preferences.FAV_LBL))
        if len(fav) < 1:
            return None
        ip = str(self.lookup(Preferences.FAV_SRV))
        addr = ip.split(":")
        return fav, f"{addr[0]}:{addr[2]}"

    def is_in_favs(self, record: str) -> bool:
        favs = self.get_favorites()
        if record in favs:
            return True
        return False

    def enum_to_key(self, enum: Preferences) -> str:
        return str(enum.dict["key"])

    def get_client_index(self, client: str) -> int:
        if client == STEAM_CMD:
            return 0
        if client == FLATPAK_RUN_CMD:
            return 1
        if client == FLATPAK_SANDBOX:
            return 2
        return 0

    def toggle_config(self, key: Preferences) -> None:
        try:
            real_key = self.enum_to_key(key)
            _json = self.get_config()
            cur_val = not _json[real_key]
            _json[real_key] = cur_val
            write_json(_json, self.config)
        except Exception as e:
            logger.critical(e)
            raise e

    def set_fav(self, name: str, record: str) -> None:
        try:
            self.write_config(Preferences.FAV_LBL, name)
            self.write_config(Preferences.FAV_SRV, record)
        except Exception:
            return

    def write_config(self, key: Preferences, value: str) -> None:
        try:
            real_key = self.enum_to_key(key)
            conf = self.get_config()
            conf[real_key] = value
            write_json(conf, self.config)
        except Exception as e:
            logger.critical(e)
            trace = traceback.format_exc()
            dialog = ExceptionDialog(self.controller, trace)
            dialog.run()
            raise e

    def save_res_and_quit(self, tv: "ServerTreeView", window: "OuterWindow") -> None:
        columns = tv.get_columns()
        filters = self.controller.get_all_tree_filters()
        filters_file = self.prefs.paths.filters

        columns_file = self.prefs.paths.columns
        try:
            data = read_json(columns_file)
        except Exception as e:
            logger.critical(e)
            data = {"cols": {}}

        for column in columns:
            title = column.get_title()
            size = column.get_width()
            data["cols"][title] = size

        try:
            write_json(data, columns_file)
            write_json(filters, filters_file)
        except Exception as e:
            logger.critical(e)

        logger.info("Normal user exit")
        if window.props.is_maximized:
            main_quit()
            return

        w, h = window.get_size()
        data = {"res": {"width": w, "height": h}}

        res_path = self.prefs.paths.resolution
        try:
            write_json(data, res_path)
        except Exception as e:
            logger.critical(e)

        main_quit()

    def set_resolution(self, window: "OuterWindow") -> None:
        if self.prefs.is_game_mode:
            window.fullscreen()
            return
        elif self.lookup(Preferences.WINDOW) is True:
            window.fullscreen()

        try:
            data = read_json(self.prefs.paths.resolution)
            valid_json = True
        except Exception as e:
            valid_json = False
            logger.critical(e)

        if valid_json:
            res = data["res"]
            w, h = res["width"], res["height"]
            logger.info(f"Restoring window size to {w},{h}")
            window.set_default_size(w, h)
        else:
            w = WINDOW_DEFAULT_X
            h = WINDOW_DEFAULT_Y
            logger.info(f"Using default window size {w},{h}")
            window.set_default_size(w, h)

    def get_filters(self, tv: "ServerTreeView") -> Any | None:
        filters = self.prefs.paths.filters
        try:
            enum = tv.get_enum()
            filters = read_json(filters)
            return filters[str(enum)]
        except Exception:
            return None