diff --git a/EDMCLogging.py b/EDMCLogging.py index 59c8e312..201648a2 100644 --- a/EDMCLogging.py +++ b/EDMCLogging.py @@ -51,7 +51,7 @@ from threading import get_native_id as thread_native_id from time import gmtime from traceback import print_exc from typing import TYPE_CHECKING, cast -from config import appcmdname, appname, config, trace_on +from config import appcmdname, appname, config, trace_on, config_logger # TODO: Tests: # @@ -535,4 +535,6 @@ if not loglevel: base_logger_name = appcmdname if os.getenv('EDMC_NO_UI') else appname edmclogger = Logger(base_logger_name, loglevel=loglevel) +for h in config_logger.handlers[:]: + config_logger.removeHandler(h) logger: 'LoggerMixin' = edmclogger.get_logger() diff --git a/EDMCSystemProfiler.py b/EDMCSystemProfiler.py index a5d97ce4..123ca5fc 100644 --- a/EDMCSystemProfiler.py +++ b/EDMCSystemProfiler.py @@ -38,7 +38,7 @@ from monitor import monitor from EDMCLogging import get_main_logger -def get_sys_report(active_config: config.AbstractConfig) -> str: +def get_sys_report(active_config: config.Config) -> str: """Gather system information about Elite, the Host Computer, and EDMC.""" # Calculate Requested Information plt = platform.uname() @@ -108,7 +108,7 @@ def copy_sys_report(root: tk.Tk, report: str) -> None: messagebox.showinfo("System Profiler", "System Report copied to Clipboard", parent=root) -def main(active_config: config.AbstractConfig) -> None: +def main(active_config: config.Config) -> None: """Entry Point for the System Profiler.""" # Now Let's Begin root: tk.Tk = tk.Tk() diff --git a/config/__init__.py b/config/__init__.py index e58df7b6..e7505f1e 100644 --- a/config/__init__.py +++ b/config/__init__.py @@ -12,27 +12,27 @@ from __future__ import annotations __all__ = [ # defined in the order they appear in the file - 'GITVERSION_FILE', - 'appname', - 'applongname', - 'appcmdname', - 'copyright', - 'update_interval', - 'debug_senders', - 'trace_on', - 'capi_pretend_down', - 'capi_debug_access_token', - 'logger', - 'git_shorthash_from_head', - 'appversion', - 'user_agent', - 'appversion_nobuild', - 'AbstractConfig', - 'config', - 'get_update_feed', + "GITVERSION_FILE", + "appname", + "applongname", + "appcmdname", + "copyright", + "update_interval", + "debug_senders", + "trace_on", + "capi_pretend_down", + "capi_debug_access_token", + "logger", + "git_shorthash_from_head", + "appversion", + "user_agent", + "appversion_nobuild", + "Config", + "config", + "get_update_feed", + "config_logger", ] -import abc import contextlib import logging import os @@ -40,23 +40,25 @@ import pathlib import re import subprocess import sys -from abc import abstractmethod +import tomllib +import tomli_w +from time import gmtime from typing import Any, Callable, Type, TypeVar import semantic_version from constants import GITVERSION_FILE, applongname, appname # Any of these may be imported by plugins -appcmdname = 'EDMC' +appcmdname = "EDMC" # appversion **MUST** follow Semantic Versioning rules: # # Major.Minor.Patch(-prerelease)(+buildmetadata) # NB: Do *not* import this, use the functions appversion() and appversion_nobuild() -_static_appversion = '6.0.0-alpha0' +_static_appversion = "6.0.0-alpha0" _cached_version: semantic_version.Version | None = None -copyright = '© 2015-2019 Jonathan Harris, 2020-2025 EDCD' +copyright = "© 2015-2019 Jonathan Harris, 2020-2025 EDCD" -update_interval = 8*60*60 # 8 Hours +update_interval = 8 * 60 * 60 # 8 Hours # Providers marked to be in debug mode. Generally this is expected to switch to sending data to a log file debug_senders: list[str] = [] # TRACE logging code that should actually be used. Means not spamming it @@ -67,10 +69,14 @@ capi_pretend_down: bool = False capi_debug_access_token: str | None = None # This must be done here in order to avoid an import cycle with EDMCLogging. # Other code should use EDMCLogging.get_main_logger -logger = logging.getLogger(appcmdname) if os.getenv("EDMC_NO_UI") else logging.getLogger(appname) +logger = ( + logging.getLogger(appcmdname) + if os.getenv("EDMC_NO_UI") + else logging.getLogger(appname) +) -_T = TypeVar('_T') +_T = TypeVar("_T") def git_shorthash_from_head() -> str | None: @@ -94,14 +100,14 @@ def git_shorthash_from_head() -> str | None: return None if not re.fullmatch(r"[0-9a-f]{7,}", shorthash): - logger.error(f"'{shorthash}' doesn't look like a valid git short hash, forcing to None") + logger.error( + f"'{shorthash}' doesn't look like a valid git short hash, forcing to None" + ) return None with contextlib.suppress(Exception): diff_result = subprocess.run( - ["git", "diff", "--stat", "HEAD"], - capture_output=True, - check=True + ["git", "diff", "--stat", "HEAD"], capture_output=True, check=True ) if diff_result.stdout: shorthash += ".DIRTY" @@ -121,10 +127,12 @@ def appversion() -> semantic_version.Version: if _cached_version is not None: return _cached_version - if getattr(sys, 'frozen', False): + if getattr(sys, "frozen", False): # Running frozen, so we should have a .gitversion file # Yes, .parent because if frozen we're inside library.zip - with open(pathlib.Path(sys.path[0]).parent / GITVERSION_FILE, encoding='utf-8') as gitv: + with open( + pathlib.Path(sys.path[0]).parent / GITVERSION_FILE, encoding="utf-8" + ) as gitv: shorthash: str | None = gitv.read() else: @@ -134,16 +142,18 @@ def appversion() -> semantic_version.Version: shorthash = git_shorthash_from_head() if shorthash is None: if pathlib.Path(sys.path[0] + "/" + GITVERSION_FILE).exists(): - with open(pathlib.Path(sys.path[0] + "/" + GITVERSION_FILE), encoding='utf-8') as gitv: + with open( + pathlib.Path(sys.path[0] + "/" + GITVERSION_FILE), encoding="utf-8" + ) as gitv: shorthash = gitv.read() else: - shorthash = 'UNKNOWN' + shorthash = "UNKNOWN" - _cached_version = semantic_version.Version(f'{_static_appversion}+{shorthash}') + _cached_version = semantic_version.Version(f"{_static_appversion}+{shorthash}") return _cached_version -user_agent = f'EDCD-{appname}-{appversion()}' +user_agent = f"EDCD-{appname}-{appversion()}" def appversion_nobuild() -> semantic_version.Version: @@ -155,12 +165,12 @@ def appversion_nobuild() -> semantic_version.Version: :return: App version without any build meta data. """ - return appversion().truncate('prerelease') + return appversion().truncate("prerelease") -class AbstractConfig(abc.ABC): +class Config: """ - Abstract root class of all platform specific Config implementations. + Platform-unified Config class for 6.0+. Commented lines are no longer supported or replaced. """ @@ -197,13 +207,113 @@ class AbstractConfig(abc.ABC): __eddn_tracking_ui = False # Show EDDN tracking UI ? __skip_timecheck = False # Skip checking event timestamps? - def __init__(self) -> None: + def __init__(self, app_path) -> None: self.home_path = pathlib.Path.home() + # Set Needed Platform Var for app_dir_path + self.app_dir_path = app_path + + self.toml_path: pathlib.Path = self.app_dir_path / "config.toml" + self.generated: str | None = None + self.source: str | None = None + self.settings: dict[str, Any] = {} + self._load() + self.default_plugin_dir_path = self.app_dir_path / "plugins" + if (plugdir_str := self.get_str("plugin_dir")) is None or not pathlib.Path( + plugdir_str + ).is_dir(): + self.set("plugin_dir", str(self.default_plugin_dir_path)) + plugdir_str = self.default_plugin_dir + self.plugin_dir_path = pathlib.Path(plugdir_str) + self.plugin_dir_path.mkdir(exist_ok=True) + + # Call the rest of the platform var helpers + self._init_platform() + + def _init_platform(self): + if sys.platform == "win32": + from .windows import win_helper + + win_helper(self) + elif sys.platform == "linux": + from .linux import linux_helper + + linux_helper(self) + else: + raise RuntimeError(f"Unsupported platform: {sys.platform}") def set_shutdown(self): """Set flag denoting we're in the shutdown sequence.""" self.__in_shutdown = True + def _load(self): + """Load TOML from disk and store fields.""" + if not self.toml_path.exists(): + raise FileNotFoundError(f"Config file not found: {self.toml_path}") + + with self.toml_path.open("rb") as f: + data = tomllib.load(f) + + # Capture metadata + self.generated = data.get("generated") + self.source = data.get("source") + + # Settings dict created by write_registry_to_toml() + raw_settings = data.get("settings", {}) + self.settings = {} + + self.settings = dict(raw_settings) + + def get(self, key: str, default=None): + """Return raw stored value.""" + return self.settings.get(key, default) + + def get_str(self, key: str, default="") -> str: + """Return string value.""" + val = self.get(key, default) + return str(val) if val is not None else default + + def get_int(self, key: str, default=0) -> int: + """Adaptive int (handles booleans stored as ints).""" + val = self.get(key) + if isinstance(val, int): + return val + try: + return int(val) + except (ValueError, TypeError): + return default + + def get_bool(self, key: str, default=False) -> bool: + """ + Adaptive boolean reader. + + - Accepts ints 0/1 + - Accepts strings "true"/"false"/"1"/"0" + - Accepts real booleans + """ + val = self.get(key) + + if isinstance(val, bool): + return val + + if isinstance(val, int): + return val != 0 + + if isinstance(val, str): + v = val.strip().lower() + if v in ("1", "true", "yes", "on"): + return True + if v in ("0", "false", "no", "off"): + return False + + return default + + def get_list(self, key: str, default=None): + """Return the list referred to by the given key if it exists, or the default.""" + val = self.get(key) + return ( + val if isinstance(val, list) else (default if default is not None else []) + ) + @property def shutting_down(self) -> bool: """ @@ -315,8 +425,10 @@ class AbstractConfig(abc.ABC): @staticmethod def _suppress_call( - func: Callable[..., _T], exceptions: Type[BaseException] | list[Type[BaseException]] = Exception, - *args: Any, **kwargs: Any + func: Callable[..., _T], + exceptions: Type[BaseException] | list[Type[BaseException]] = Exception, + *args: Any, + **kwargs: Any, ) -> _T | None: if exceptions is None: exceptions = [Exception] @@ -329,96 +441,54 @@ class AbstractConfig(abc.ABC): return None - @abstractmethod - def get_list(self, key: str, *, default: list | None = None) -> list: - """ - Return the list referred to by the given key if it exists, or the default. - - Implements :meth:`AbstractConfig.get_list`. - """ - raise NotImplementedError - - @abstractmethod - def get_str(self, key: str, *, default: str | None = None) -> str: - """ - Return the string referred to by the given key if it exists, or the default. - - :param key: The key data is being requested for. - :param default: Default to return if the key does not exist, defaults to None. - :raises ValueError: If an internal error occurs getting or converting a value. - :raises OSError: On Windows, if a Registry error occurs. - :return: The requested data or the default. - """ - raise NotImplementedError - - @abstractmethod - def get_bool(self, key: str, *, default: bool | None = None) -> bool: - """ - Return the bool referred to by the given key if it exists, or the default. - - :param key: The key data is being requested for. - :param default: Default to return if the key does not exist, defaults to None - :raises ValueError: If an internal error occurs getting or converting a value - :raises OSError: On Windows, if a Registry error occurs. - :return: The requested data or the default - """ - raise NotImplementedError - - @abstractmethod - def get_int(self, key: str, *, default: int = 0) -> int: - """ - Return the int referred to by key if it exists in the config. - - For legacy reasons, the default is 0 and not None. - - :param key: The key data is being requested for. - :param default: Default to return if the key does not exist, defaults to 0. - :raises ValueError: If the internal representation of this key cannot be converted to an int. - :raises OSError: On Windows, if a Registry error occurs. - :return: The requested data or the default. - """ - raise NotImplementedError - - @abstractmethod - def set(self, key: str, val: int | str | list[str] | bool) -> None: - """ - Set the given key's data to the given value. - - :param key: The key to set the value on. - :param val: The value to set the key's data to. - :raises ValueError: On an invalid type. - :raises OSError: On Windows, if a Registry error occurs. - """ - raise NotImplementedError - - @abstractmethod def delete(self, key: str, *, suppress=False) -> None: - """ - Delete the given key from the config. + """Delete the given key from the config.""" + try: + self.settings.pop(key, None) + except (KeyError, IndexError): + if suppress: + return + raise + self.save() - :param key: The key to delete. - :param suppress: bool - Whether to suppress any errors. Useful in case - code to migrate settings is blindly removing an old key. - :raises OSError: On Windows, if a registry error occurs. - """ - raise NotImplementedError + def set(self, key: str, value: Any): + """Modify a setting and save to disk.""" + self.settings[key] = value + self.save() - @abstractmethod - def save(self) -> None: - """ - Save the current configuration. + def save(self): + """Write updated config back to TOML.""" + data = { + "generated": self.generated, + "source": self.source, + "settings": self.settings, + } - :raises OSError: On Windows, if a Registry error occurs. - """ - raise NotImplementedError + with self.toml_path.open("wb") as f: + tomli_w.dump(data, f) - @abstractmethod def close(self) -> None: - """Close this config and release any associated resources.""" - raise NotImplementedError + """Save config changes before closing.""" + self.save() -def get_config(*args, **kwargs) -> AbstractConfig: +def get_appdirpath(): + """Grab the Application Directory early.""" + app_dir_path = None + if sys.platform == "win32": + base = pathlib.Path(os.getenv("LOCALAPPDATA")) # type: ignore + app_dir_path = base / appname + if sys.platform == "linux": + xdg_data_home = pathlib.Path( + os.getenv("XDG_DATA_HOME", default="~/.local/share") + ).expanduser() + app_dir_path = xdg_data_home / appname + if app_dir_path is None: + raise ValueError + return app_dir_path + + +def get_config(*args, **kwargs) -> Config: """ Get the appropriate config class for the current platform. @@ -426,25 +496,45 @@ def get_config(*args, **kwargs) -> AbstractConfig: :param kwargs: Args to be passed through to implementation. :return: Instance of the implementation. """ + app_dir_path = get_appdirpath() + config = None + if pathlib.Path.exists(app_dir_path / "config.toml"): + return Config(app_path=app_dir_path) + if sys.platform == "win32": # pragma: sys-platform-win32 - from .windows import WinConfig - return WinConfig(*args, **kwargs) + from .windows import WinConfigMinimal + + config = WinConfigMinimal() if sys.platform == "linux": # pragma: sys-platform-linux - from .linux import LinuxConfig - return LinuxConfig(*args, **kwargs) + from .linux import LinuxConfigMinimal + config = LinuxConfigMinimal() - raise ValueError(f'Unknown platform: {sys.platform=}') + if config: + config.write_to_toml(f"{app_dir_path}/config.toml") # type: ignore + return Config(app_path=app_dir_path) + + raise ValueError(f"Unknown platform: {sys.platform=}") + + +# Set internal Config logger, because config is set up before main logger. +config_logger = logging.getLogger("pre_config") +config_logger.setLevel(logging.DEBUG) # Or INFO + +ch = logging.StreamHandler() +ch.setLevel(logging.DEBUG) +formatter = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s") +formatter.converter = gmtime # Optional: match your main logger's UTC timestamps +ch.setFormatter(formatter) +config_logger.addHandler(ch) config = get_config() -if sys.platform == "win32": - config.write_registry_to_toml(f"{config.app_dir_path}/config.toml") # type: ignore # Wiki: https://github.com/EDCD/EDMarketConnector/wiki/Participating-in-Open-Betas-of-EDMC def get_update_feed() -> str: """Select the proper update feed for the current update track.""" - if config.get_bool('beta_optin'): - return 'https://raw.githubusercontent.com/EDCD/EDMarketConnector/releases/edmarketconnector-beta.xml' - return 'https://raw.githubusercontent.com/EDCD/EDMarketConnector/releases/edmarketconnector.xml' + if config.get_bool("beta_optin"): + return "https://raw.githubusercontent.com/EDCD/EDMarketConnector/releases/edmarketconnector-beta.xml" + return "https://raw.githubusercontent.com/EDCD/EDMarketConnector/releases/edmarketconnector.xml" diff --git a/config/linux.py b/config/linux.py index 651fea09..9260cd74 100644 --- a/config/linux.py +++ b/config/linux.py @@ -5,252 +5,87 @@ Copyright (c) EDCD, All Rights Reserved Licensed under the GNU General Public License. See LICENSE file. """ + from __future__ import annotations import os -import pathlib import sys -from configparser import ConfigParser -from config import AbstractConfig, appname, logger +from config import config_logger -if sys.platform != 'linux': +if sys.platform != "linux": raise EnvironmentError("This file is for Linux only.") -class LinuxConfig(AbstractConfig): - """Linux implementation of AbstractConfig.""" +import pathlib +from configparser import ConfigParser +import datetime +import tomli_w +from typing import Any, TYPE_CHECKING - SECTION = 'config' - # TODO: I dislike this, would rather use a sane config file format. But here we are. - __unescape_lut = {'\\': '\\', 'n': '\n', ';': ';', 'r': '\r', '#': '#'} - __escape_lut = {'\\': '\\', '\n': 'n', ';': ';', '\r': 'r'} +if TYPE_CHECKING: + from . import Config + + +class LinuxConfigMinimal: + """Minimal Linux config loader that dumps existing config to TOML.""" + + SECTION = "config" def __init__(self, filename: str | None = None) -> None: - super().__init__() - # http://standards.freedesktop.org/basedir-spec/latest/ar01s03.html - xdg_data_home = pathlib.Path(os.getenv('XDG_DATA_HOME', default='~/.local/share')).expanduser() - self.app_dir_path = xdg_data_home / appname - self.app_dir_path.mkdir(exist_ok=True, parents=True) - - self.default_plugin_dir_path = self.app_dir_path / 'plugins' - self.respath_path = pathlib.Path(__file__).parent.parent - - self.internal_plugin_dir_path = self.respath_path / 'plugins' - self.default_journal_dir_path = None # type: ignore - self.identifier = f'uk.org.marginal.{appname.lower()}' # TODO: Unused? - - config_home = pathlib.Path(os.getenv('XDG_CONFIG_HOME', default='~/.config')).expanduser() - - self.filename = config_home / appname / f'{appname}.ini' - if filename is not None: - self.filename = pathlib.Path(filename) - + # Determine config path + xdg_config_home = pathlib.Path( + os.getenv("XDG_CONFIG_HOME", "~/.config") + ).expanduser() + self.filename = ( + pathlib.Path(filename) + if filename + else xdg_config_home / "EDMarketConnector" / "EDMarketConnector.ini" + ) self.filename.parent.mkdir(exist_ok=True, parents=True) - self.config: ConfigParser | None = ConfigParser(comment_prefixes=('#',), interpolation=None) - self.config.read(self.filename) # read() ignores files that dont exist - - # Ensure that our section exists. This is here because configparser will happily create files for us, but it - # does not magically create sections - try: - self.config[self.SECTION].get("this_does_not_exist", fallback=None) - except KeyError: - logger.info("Config section not found. Backing up existing file (if any) and readding a section header") - if self.filename.exists(): - (self.filename.parent / f'{appname}.ini.backup').write_bytes(self.filename.read_bytes()) + # Load INI + self.config = ConfigParser(interpolation=None) + self.config.read(self.filename) + # Ensure section exists + if self.SECTION not in self.config: self.config.add_section(self.SECTION) - if (plugdir_str := self.get_str('plugin_dir')) is None or not pathlib.Path(plugdir_str).is_dir(): - self.set("plugin_dir", str(self.default_plugin_dir_path)) - plugdir_str = self.default_plugin_dir - self.plugin_dir_path = pathlib.Path(plugdir_str) - self.plugin_dir_path.mkdir(exist_ok=True) + def _get_settings_dict(self) -> dict[str, Any]: + """Return all keys/values from SECTION as a dict of strings.""" + return { + key: self.config[self.SECTION][key] for key in self.config[self.SECTION] + } - if (outdir := self.get_str('outdir')) is None or not pathlib.Path(outdir).is_dir(): - self.set('outdir', self.home) + def write_to_toml(self, toml_path: str) -> None: + """Dump existing config to TOML file.""" + config_logger.debug("Generating New Config File") + config_data: dict[str, Any] = { + "generated": datetime.datetime.now(datetime.timezone.utc).isoformat(), + "source": "linux_ini", + "section": self.SECTION, + "settings": self._get_settings_dict(), + } - def __escape(self, s: str) -> str: - """ - Escape a string using self.__escape_lut. - - This does NOT support multi-character escapes. - - :param s: str - String to be escaped. - :return: str - The escaped string. - """ - out = "" - for c in s: - if c not in self.__escape_lut: - out += c - continue - - out += '\\' + self.__escape_lut[c] - - return out - - def __unescape(self, s: str) -> str: - """ - Unescape a string. - - :param s: str - The string to unescape. - :return: str - The unescaped string. - """ - out: list[str] = [] - i = 0 - while i < len(s): - c = s[i] - if c != '\\': - out.append(c) - i += 1 - continue - - # We have a backslash, check what its escaping - if i == len(s)-1: - raise ValueError('Escaped string has unescaped trailer') - - unescaped = self.__unescape_lut.get(s[i+1]) - if unescaped is None: - raise ValueError(f'Unknown escape: \\ {s[i+1]}') - - out.append(unescaped) - i += 2 - - return "".join(out) - - def __raw_get(self, key: str) -> str | None: - """ - Get a raw data value from the config file. - - :param key: str - The key data is being requested for. - :return: str - The raw data, if found. - """ - if self.config is None: - raise ValueError('Attempt to use a closed config') - - return self.config[self.SECTION].get(key) - - def get_str(self, key: str, *, default: str | None = None) -> str: - """ - Return the string referred to by the given key if it exists, or the default. - - Implements :meth:`AbstractConfig.get_str`. - """ - data = self.__raw_get(key) - if data is None: - return default # type: ignore # Yes it could be None, but we're _assuming_ that people gave us a default - - if '\n' in data: - raise ValueError('asked for string, got list') - - return self.__unescape(data) - - def get_list(self, key: str, *, default: list | None = None) -> list: - """ - Return the list referred to by the given key if it exists, or the default. - - Implements :meth:`AbstractConfig.get_list`. - """ - data = self.__raw_get(key) - - if data is None: - return default # type: ignore # Yes it could be None, but we're _assuming_ that people gave us a default - - split = data.split('\n') - if split[-1] != ';': - raise ValueError('Encoded list does not have trailer sentinel') - - return list(map(self.__unescape, split[:-1])) - - def get_int(self, key: str, *, default: int = 0) -> int: - """ - Return the int referred to by key if it exists in the config. - - Implements :meth:`AbstractConfig.get_int`. - """ - data = self.__raw_get(key) - - if data is None: - return default - - try: - return int(data) - - except ValueError as e: - raise ValueError(f'requested {key=} as int cannot be converted to int') from e - - def get_bool(self, key: str, *, default: bool | None = None) -> bool: - """ - Return the bool referred to by the given key if it exists, or the default. - - Implements :meth:`AbstractConfig.get_bool`. - """ - if self.config is None: - raise ValueError('attempt to use a closed config') - - data = self.__raw_get(key) - if data is None: - return default # type: ignore # Yes it could be None, but we're _assuming_ that people gave us a default - - return bool(int(data)) - - def set(self, key: str, val: int | str | list[str]) -> None: - """ - Set the given key's data to the given value. - - Implements :meth:`AbstractConfig.set`. - """ - if self.config is None: - raise ValueError('attempt to use a closed config') - - to_set: str | None = None - if isinstance(val, bool): - to_set = str(int(val)) - - elif isinstance(val, str): - to_set = self.__escape(val) - - elif isinstance(val, int): - to_set = str(val) - - elif isinstance(val, list): - to_set = '\n'.join([self.__escape(s) for s in val] + [';']) - - else: - raise ValueError(f'Unexpected type for value {type(val)=}') - - self.config.set(self.SECTION, key, to_set) - self.save() - - def delete(self, key: str, *, suppress=False) -> None: - """ - Delete the given key from the config. - - Implements :meth:`AbstractConfig.delete`. - """ - if self.config is None: - raise ValueError('attempt to use a closed config') - - self.config.remove_option(self.SECTION, key) - self.save() - - def save(self) -> None: - """ - Save the current configuration. - - Implements :meth:`AbstractConfig.save`. - """ - if self.config is None: - raise ValueError('attempt to use a closed config') - - with open(self.filename, 'w', encoding='utf-8') as f: - self.config.write(f) + path_obj = pathlib.Path(toml_path) + path_obj.parent.mkdir(parents=True, exist_ok=True) + config_logger.debug("Writing Config File") + with path_obj.open("wb") as f: + tomli_w.dump(config_data, f) def close(self) -> None: - """ - Close this config and release any associated resources. - - Implements :meth:`AbstractConfig.close`. - """ - self.save() + """Release resources (stub).""" self.config = None + + +def linux_helper(config: Config) -> Config: + """Set Environment Specific Variables for Linux Config.""" + config_logger.debug("Linux environment detected. Setting platform-specific variables.") + config.respath_path = pathlib.Path(__file__).parent.parent + config.internal_plugin_dir_path = config.respath_path / "plugins" + config.default_journal_dir_path = None # type: ignore + if (outdir := config.get_str("outdir")) is None or not pathlib.Path( + outdir + ).is_dir(): + config.set("outdir", config.home) + return config diff --git a/config/windows.py b/config/windows.py index 96d073c1..11875b66 100644 --- a/config/windows.py +++ b/config/windows.py @@ -5,21 +5,23 @@ Copyright (c) EDCD, All Rights Reserved Licensed under the GNU General Public License. See LICENSE file. """ + from __future__ import annotations -import functools import pathlib import sys import uuid import winreg import datetime import tomli_w -import base64 -from typing import Literal, Any -from config import AbstractConfig, applongname, appname, logger +from config import config_logger from win32comext.shell import shell +from typing import Any, TYPE_CHECKING -if sys.platform != 'win32': +if TYPE_CHECKING: + from . import Config + +if sys.platform != "win32": raise EnvironmentError("This file is for Windows only.") REG_RESERVED_ALWAYS_ZERO = 0 @@ -30,254 +32,84 @@ def known_folder_path(guid: uuid.UUID) -> str | None: return shell.SHGetKnownFolderPath(guid, 0, 0) -class WinConfig(AbstractConfig): - """Implementation of AbstractConfig for Windows.""" +class WinConfigMinimal: + """Minimal Windows config for exporting registry values to TOML.""" + + REGISTRY_SUBKEY = r"Software\Marginal\EDMarketConnector" def __init__(self) -> None: - super().__init__() - - self.REGISTRY_SUBKEY = r'Software\Marginal\EDMarketConnector' - create_key_defaults = functools.partial( - winreg.CreateKeyEx, - key=winreg.HKEY_CURRENT_USER, + # Open or create the registry key + self.__reg_handle = winreg.CreateKeyEx( + winreg.HKEY_CURRENT_USER, + self.REGISTRY_SUBKEY, access=winreg.KEY_ALL_ACCESS | winreg.KEY_WOW64_64KEY, ) - try: - self.__reg_handle: winreg.HKEYType = create_key_defaults(sub_key=self.REGISTRY_SUBKEY) - - except OSError: - logger.exception('Could not create required registry keys') - raise - - if local_appdata := known_folder_path(shell.FOLDERID_LocalAppData): - self.app_dir_path = pathlib.Path(local_appdata) / appname - self.app_dir_path.mkdir(exist_ok=True) - - self.default_plugin_dir_path = self.app_dir_path / 'plugins' - if (plugdir_str := self.get_str('plugin_dir')) is None or not pathlib.Path(plugdir_str).is_dir(): - self.set("plugin_dir", str(self.default_plugin_dir_path)) - plugdir_str = self.default_plugin_dir - self.plugin_dir_path = pathlib.Path(plugdir_str) - self.plugin_dir_path.mkdir(exist_ok=True) - - if getattr(sys, 'frozen', False): - self.respath_path = pathlib.Path(sys.executable).parent - self.internal_plugin_dir_path = self.respath_path / 'plugins' - else: - self.respath_path = pathlib.Path(__file__).parent.parent - self.internal_plugin_dir_path = self.respath_path / 'plugins' - - self.home_path = pathlib.Path.home() - - journal_dir_path = pathlib.Path( - known_folder_path(shell.FOLDERID_SavedGames)) / 'Frontier Developments' / 'Elite Dangerous' # type: ignore - self.default_journal_dir_path = journal_dir_path if journal_dir_path.is_dir() else None # type: ignore - - self.identifier = applongname - if (outdir_str := self.get_str('outdir')) is None or not pathlib.Path(outdir_str).is_dir(): - docs = known_folder_path(shell.FOLDERID_Documents) - self.set("outdir", docs if docs is not None else self.home) - - def _convert_reg_value_for_toml(self, data, data_type): - """Convert Windows registry values into TOML-native types for 6.0's global config.""" - # Plain string + def _convert_reg_value_for_toml(self, data, data_type) -> Any: + """Convert Windows registry values into TOML-native types.""" if data_type == winreg.REG_SZ: return str(data) - - # Expandable string: keep as string if data_type == winreg.REG_EXPAND_SZ: try: - expanded = winreg.ExpandEnvironmentStrings(data) + return str(winreg.ExpandEnvironmentStrings(data)) except Exception: - expanded = data - return str(expanded) - - # DWORD or QWORD: store as integer + return str(data) if data_type in (winreg.REG_DWORD, winreg.REG_QWORD): return int(data) - - # Multi-string: store as list if data_type == winreg.REG_MULTI_SZ: return list(data) - - # Binary: store as base64 string - if data_type == winreg.REG_BINARY: - return base64.b64encode(data).decode("ascii") - - # Else, force to string + # Fallback: force to string return str(data) - def write_registry_to_toml(self, toml_path: str) -> None: - """ - Export all registry values under self.REGISTRY_SUBKEY into a global TOML config. - - Values are normalized into TOML-native types (str, int, list, base64). - Booleans are left as ints (0/1) and should be interpreted adaptively by get_bool. - """ + def write_to_toml(self, toml_path: str) -> None: + """Export all registry values under REGISTRY_SUBKEY to TOML.""" + config_logger.debug("Generating New Config File") key = self.__reg_handle + _, num_values, _ = winreg.QueryInfoKey(key) - try: - _, num_values, _ = winreg.QueryInfoKey(key) + config_data: dict[str, Any] = { + "generated": datetime.datetime.now(datetime.timezone.utc).isoformat(), + "source": "windows_registry", + "settings": {}, + } - # Write config, including writing where and when we got this from. - config_data: dict[str, Any] = { - "generated": datetime.datetime.now(datetime.timezone.utc).isoformat(), - "source": "windows_registry", - "subkey": self.REGISTRY_SUBKEY, - "settings": {} - } + for i in range(num_values): + name, data, data_type = winreg.EnumValue(key, i) + config_data["settings"][name] = self._convert_reg_value_for_toml( + data, data_type + ) - settings = config_data["settings"] - - for i in range(num_values): - name, data, data_type = winreg.EnumValue(key, i) - settings[name] = self._convert_reg_value_for_toml(data, data_type) - - # Write TOML to file - path_obj = pathlib.Path(toml_path) - path_obj.parent.mkdir(parents=True, exist_ok=True) - with path_obj.open("wb") as f: - tomli_w.dump(config_data, f) - - logger.info(f"Registry exported to TOML: {toml_path}") - - except OSError: - logger.exception(f"Could not export registry key {self.REGISTRY_SUBKEY}") - raise - - def __get_regentry(self, key: str) -> None | list | str | int: - """Access the Registry for the raw entry.""" - try: - value, _type = winreg.QueryValueEx(self.__reg_handle, key) - except FileNotFoundError: - # Key doesn't exist - return None - - # For programmers who want to actually know what is going on - if _type == winreg.REG_SZ: - return str(value) - - if _type == winreg.REG_DWORD: - return int(value) - - if _type == winreg.REG_MULTI_SZ: - return list(value) - - logger.warning(f'Registry key {key=} returned unknown type {_type=} {value=}') - return None - - def get_str(self, key: str, *, default: str | None = None) -> str: - """ - Return the string referred to by the given key if it exists, or the default. - - Implements :meth:`AbstractConfig.get_str`. - """ - res = self.__get_regentry(key) - if res is None: - return default # type: ignore # Yes it could be None, but we're _assuming_ that people gave us a default - - if not isinstance(res, str): - raise ValueError(f'Data from registry is not a string: {type(res)=} {res=}') - - return res - - def get_list(self, key: str, *, default: list | None = None) -> list: - """ - Return the list referred to by the given key if it exists, or the default. - - Implements :meth:`AbstractConfig.get_list`. - """ - res = self.__get_regentry(key) - if res is None: - return default # type: ignore # Yes it could be None, but we're _assuming_ that people gave us a default - - if not isinstance(res, list): - raise ValueError(f'Data from registry is not a list: {type(res)=} {res}') - - return res - - def get_int(self, key: str, *, default: int = 0) -> int: - """ - Return the int referred to by key if it exists in the config. - - Implements :meth:`AbstractConfig.get_int`. - """ - res = self.__get_regentry(key) - if res is None: - return default - - if not isinstance(res, int): - raise ValueError(f'Data from registry is not an int: {type(res)=} {res}') - - return res - - def get_bool(self, key: str, *, default: bool | None = None) -> bool: - """ - Return the bool referred to by the given key if it exists, or the default. - - Implements :meth:`AbstractConfig.get_bool`. - """ - res = self.get_int(key, default=default) # type: ignore - if res is None: - return default # Yes it could be None, but we're _assuming_ that people gave us a default - - return bool(res) - - def set(self, key: str, val: int | str | list[str] | bool) -> None: - """ - Set the given key's data to the given value. - - Implements :meth:`AbstractConfig.set`. - """ - # These are the types that winreg.REG_* below resolve to. - reg_type: Literal[1] | Literal[4] | Literal[7] - if isinstance(val, str): - reg_type = winreg.REG_SZ - - elif isinstance(val, int): - reg_type = winreg.REG_DWORD - - elif isinstance(val, list): - reg_type = winreg.REG_MULTI_SZ - - elif isinstance(val, bool): - reg_type = winreg.REG_DWORD - val = int(val) - - else: - raise ValueError(f'Unexpected type for value {type(val)=}') - - winreg.SetValueEx(self.__reg_handle, key, REG_RESERVED_ALWAYS_ZERO, reg_type, val) # type: ignore - - def delete(self, key: str, *, suppress=False) -> None: - """ - Delete the given key from the config. - - 'key' is relative to the base Registry path we use. - - Implements :meth:`AbstractConfig.delete`. - """ - try: - winreg.DeleteValue(self.__reg_handle, key) - except OSError: - if suppress: - return - - raise - - def save(self) -> None: - """ - Save the configuration. - - Not required for WinConfig as Registry keys are flushed on write. - """ - pass + path_obj = pathlib.Path(toml_path) + path_obj.parent.mkdir(parents=True, exist_ok=True) + config_logger.debug("Writing Config File") + with path_obj.open("wb") as f: + tomli_w.dump(config_data, f) def close(self): - """ - Close this config and release any associated resources. - - Implements :meth:`AbstractConfig.close`. - """ + """Close registry handle.""" self.__reg_handle.Close() + + +def win_helper(config: Config) -> Config: + """Set Environment Specific Variables for Windows Config.""" + config_logger.debug("Windows environment detected. Setting platform-specific variables.") + if getattr(sys, "frozen", False): + config.respath_path = pathlib.Path(sys.executable).parent + config.internal_plugin_dir_path = config.respath_path / "plugins" + else: + config.respath_path = pathlib.Path(__file__).parent.parent + config.internal_plugin_dir_path = config.respath_path / "plugins" + + journal_dir_path = ( + pathlib.Path(known_folder_path(shell.FOLDERID_SavedGames)) # type: ignore + / "Frontier Developments" + / "Elite Dangerous" + ) + config.default_journal_dir_path = journal_dir_path if journal_dir_path.is_dir() else None # type: ignore + + if (outdir_str := config.get_str("outdir")) is None or not pathlib.Path( + outdir_str + ).is_dir(): + docs = known_folder_path(shell.FOLDERID_Documents) + config.set("outdir", docs if docs is not None else config.home) + return config diff --git a/requirements.txt b/requirements.txt index 10b4ac67..32001a17 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,3 +6,4 @@ semantic-version==2.10.0 # For manipulating folder permissions and the like. pywin32==311; sys_platform == 'win32' psutil==7.1.3 +tomli-w==1.2.0