From 25aab588a831a61588ecec478ead2e4e97e55774 Mon Sep 17 00:00:00 2001 From: KenanZhu <3471685733@qq.com> Date: Thu, 26 Feb 2026 21:18:18 +0800 Subject: [PATCH] =?UTF-8?q?feat(utils):=20=E6=B7=BB=E5=8A=A0=20ConfigManag?= =?UTF-8?q?er=20=E4=B8=8E=20JSON=20=E9=85=8D=E7=BD=AE=E8=AF=BB=E5=86=99?= =?UTF-8?q?=EF=BC=8C=E6=9B=BF=E6=8D=A2=E6=97=A7=E5=AE=9E=E7=8E=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit add: - src/utils/ConfigManager.py - src/utils/JSONReader.py - src/utils/JSONWriter.py remove: - src/utils/ConfigReader.py - src/utils/ConfigWriter.py refactor: - 更新调用方以使用 ConfigManager / JSONReader / JSONWriter（见 ALConfigWidget.py、ALMainWindow.py、ALTimerTaskManageWidget.py、ALMainWorkers.py 等） - 统一方法命名（initlize* -> initialize*）、改进错误提示与配置路径管理 BREAKING CHANGE: 删除 ConfigReader/ConfigWriter，外部调用需改为 JSONReader/JSONWriter 或通过 ConfigManager 访问配置 --- src/Main.py | 15 +- src/gui/ALConfigWidget.py | 163 +++++++++++--------- src/gui/ALMainWindow.py | 28 ++-- src/gui/ALMainWorkers.py | 6 +- src/gui/ALTimerTaskManageWidget.py | 56 ++----- src/utils/ConfigManager.py | 233 +++++++++++++++++++++++++++++ src/utils/ConfigReader.py | 115 -------------- src/utils/ConfigWriter.py | 116 -------------- src/utils/JSONReader.py | 85 +++++++++++ src/utils/JSONWriter.py | 82 ++++++++++ src/utils/__init__.py | 5 +- 11 files changed, 541 insertions(+), 363 deletions(-) create mode 100644 src/utils/ConfigManager.py delete mode 100644 src/utils/ConfigReader.py delete mode 100644 src/utils/ConfigWriter.py create mode 100644 src/utils/JSONReader.py create mode 100644 src/utils/JSONWriter.py diff --git a/src/Main.py b/src/Main.py index 73b0c4a..298234a 100644 --- a/src/Main.py +++ b/src/Main.py @@ -7,14 +7,25 @@ This software is provided "as is", without any warranty of any kind. You may use, modify, and distribute this file under the terms of the MIT License. See the LICENSE file for details. """ +import os import sys -from PySide6.QtCore import QTranslator +from PySide6.QtCore import QTranslator, QStandardPaths, QDir from PySide6.QtWidgets import QApplication from gui.ALMainWindow import ALMainWindow from gui.resources import ALResource +from utils.ConfigManager import instance + + +def initializeConfigManager(): + + app_dir = QStandardPaths.writableLocation(QStandardPaths.StandardLocation.AppDataLocation) + config_dir = os.path.join(app_dir, "config") + if not QDir(config_dir).exists(): + QDir().mkdir(config_dir) + instance(config_dir) def main(): @@ -23,6 +34,8 @@ def main(): if translator.load(":/res/trans/translators/qtbase_zh_CN.ts"): app.installTranslator(translator) app.setStyle('Fusion') + app.setApplicationName("AutoLibrary") + initializeConfigManager() window = ALMainWindow() window.show() sys.exit(app.exec_()) diff --git a/src/gui/ALConfigWidget.py b/src/gui/ALConfigWidget.py index 9fdeca5..be2e594 100644 --- a/src/gui/ALConfigWidget.py +++ b/src/gui/ALConfigWidget.py @@ -21,8 +21,10 @@ from PySide6.QtGui import ( QCloseEvent, QAction ) -from utils.ConfigReader import ConfigReader -from utils.ConfigWriter import ConfigWriter +from utils.JSONReader import JSONReader +from utils.JSONWriter import JSONWriter +from utils.ConfigManager import ConfigType, instance +from utils.ConfigManager import getValidateAutomationConfigPaths from gui.resources.ui.Ui_ALConfigWidget import Ui_ALConfigWidget from gui.ALSeatMapSelectDialog import ALSeatMapSelectDialog @@ -32,27 +34,22 @@ from gui.ALUserTreeWidget import ALUserTreeWidget, ALUserTreeItemType class ALConfigWidget(QWidget, Ui_ALConfigWidget): - configWidgetIsClosed = Signal(dict) + configWidgetIsClosed = Signal() def __init__( self, parent = None, - config_paths = { - "run": "", - "user": "" - } ): super().__init__(parent) - self.__config_paths = config_paths + self.__cfg_mgr = instance() + self.__config_paths = getValidateAutomationConfigPaths() self.__config_data = {"run": {}, "user": {}} self.setupUi(self) self.modifyUi() self.connectSignals() - self.initlizeFloorRoomMap() - self.initlizeDefaultConfigPaths() - if not self.initlizeConfigs(): + if not self.initializeConfigs(): self.close() @@ -68,8 +65,8 @@ class ALConfigWidget(QWidget, Ui_ALConfigWidget): self.UserListLayout.insertWidget(0, self.UserTreeWidget) self.UserTreeWidget.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu) self.UserTreeWidget.customContextMenuRequested.connect(self.onUserTreeWidgetContextMenu) - self.initlizeFloorRoomMap() - self.initilizeUserInfoWidget() + self.initializeFloorRoomMap() + self.initializeUserInfoWidget() def connectSignals( @@ -124,11 +121,11 @@ class ALConfigWidget(QWidget, Ui_ALConfigWidget): event: QCloseEvent ): - self.configWidgetIsClosed.emit(self.__config_paths) + self.configWidgetIsClosed.emit() super().closeEvent(event) - def initlizeFloorRoomMap( + def initializeFloorRoomMap( self ): @@ -162,19 +159,7 @@ class ALConfigWidget(QWidget, Ui_ALConfigWidget): } - def initlizeDefaultConfigPaths( - self - ): - - executable_path = sys.executable - executable_dir = QFileInfo(executable_path).absoluteDir() - self.__default_config_paths = { - "user": QDir.toNativeSeparators(executable_dir.absoluteFilePath("user.json")), - "run": QDir.toNativeSeparators(executable_dir.absoluteFilePath("run.json")) - } - - - def initlizeConfigToWidget( + def initializeConfigToWidget( self, which: str, config_data: dict @@ -184,12 +169,12 @@ class ALConfigWidget(QWidget, Ui_ALConfigWidget): self.setRunConfigToWidget(config_data) self.CurrentRunConfigEdit.setText(self.__config_paths["run"]) elif which == "user": - self.initilizeUserInfoWidget() + self.initializeUserInfoWidget() self.setUsersToTreeWidget(config_data) self.CurrentUserConfigEdit.setText(self.__config_paths["user"]) - def initlizeConfig( + def initializeConfig( self, which: str ) -> bool: @@ -225,18 +210,16 @@ class ALConfigWidget(QWidget, Ui_ALConfigWidget): return is_success - def initlizeConfigs( + def initializeConfigs( self ) -> bool: is_success = True for which in ["run", "user"]: - if not self.__config_paths[which]: - self.__config_paths[which] = self.__default_config_paths[which] - if not self.initlizeConfig(which): + if not self.initializeConfig(which): is_success = False break - self.initlizeConfigToWidget(which, self.__config_data[which]) + self.initializeConfigToWidget(which, self.__config_data[which]) return is_success @@ -321,19 +304,21 @@ class ALConfigWidget(QWidget, Ui_ALConfigWidget): QMessageBox.warning( self, "警告 - AutoLibrary", - f"运行配置文件: {self.__config_paths['run']}\n" - f"读取时键 '{e}' 发生错误，文件可能被意外修改或已经损坏\n" + f"运行配置文件读取键 '{e}' 时发生错误 ! :\n" + f"文件路径: {self.__config_paths['run']}\n" + "文件可能被意外修改或已经损坏\n" ) except Exception as e: QMessageBox.warning( self, "警告 - AutoLibrary", - f"运行配置文件: {self.__config_paths['run']}\n" - f"读取时键 '{e}' 发生未知错误，文件可能被意外修改或已经损坏\n" + f"运行配置文件读取键 '{e}' 时发生未知错误 ! :\n" + f"文件路径: {self.__config_paths['run']}\n" + "文件可能被意外修改或已经损坏\n" ) - def initilizeUserInfoWidget( + def initializeUserInfoWidget( self ): @@ -442,15 +427,17 @@ class ALConfigWidget(QWidget, Ui_ALConfigWidget): QMessageBox.warning( self, "警告 - AutoLibrary", - f"用户配置文件: {self.__config_paths['user']}\n"\ - f"读取时键 '{e}' 发生错误，文件可能被意外修改或已经损坏\n" + f"用户配置文件读取键 '{e}' 时发生错误 ! :\n" + f"文件路径: {self.__config_paths['user']}\n" + "文件可能被意外修改或已经损坏\n" ) except Exception as e: QMessageBox.warning( self, "警告 - AutoLibrary", - f"用户配置文件: {self.__config_paths['user']}\n"\ - f"读取时发生未知错误 '{e}'，文件可能被意外修改或已经损坏\n" + f"用户配置文件读取键 '{e}' 时发生未知错误 ! :\n" + f"文件路径: {self.__config_paths['user']}\n" + "文件可能被意外修改或已经损坏\n" ) @@ -480,15 +467,17 @@ class ALConfigWidget(QWidget, Ui_ALConfigWidget): QMessageBox.warning( self, "警告 - AutoLibrary", - f"用户配置文件: {self.__config_paths['user']}\n"\ - f"读取时键 '{e}' 发生错误，文件可能被意外修改或已经损坏\n" + f"用户配置文件读取键 '{e}' 时发生错误 ! :\n" + f"文件路径: {self.__config_paths['user']}\n" + "文件可能被意外修改或已经损坏\n" ) except Exception as e: QMessageBox.warning( self, "警告 - AutoLibrary", - f"用户配置文件: {self.__config_paths['user']}\n"\ - f"读取时发生未知错误 '{e}'，文件可能被意外修改或已经损坏\n" + f"用户配置文件读取键 '{e}' 时发生未知错误 ! :\n" + f"文件路径: {self.__config_paths['user']}\n" + "文件可能被意外修改或已经损坏\n" ) finally: self.UserTreeWidget.itemChanged.connect(self.onUserTreeWidgetItemChanged) @@ -502,17 +491,18 @@ class ALConfigWidget(QWidget, Ui_ALConfigWidget): try: if not run_config_path or not os.path.exists(run_config_path): raise Exception("文件路径不存在") - run_config = ConfigReader(run_config_path).getConfigs() + run_config = JSONReader(run_config_path).data() if run_config and "library" in run_config\ and "web_driver" in run_config\ and "login" in run_config: return run_config - return None + else: + return None except Exception as e: QMessageBox.warning( self, "警告 - AutoLibrary", - f"运行配置文件读取发生错误 ! : \n{e}" + f"运行配置文件读取发生错误 ! :\n{e}" ) return None @@ -528,7 +518,7 @@ class ALConfigWidget(QWidget, Ui_ALConfigWidget): raise Exception("文件路径为空") if not run_config_data or not isinstance(run_config_data, dict): raise Exception("运行配置数据为空或类型错误") - ConfigWriter(run_config_path, run_config_data) + JSONWriter(run_config_path, run_config_data) return True except Exception as e: QMessageBox.warning( @@ -547,11 +537,11 @@ class ALConfigWidget(QWidget, Ui_ALConfigWidget): try: if not user_config_path or not os.path.exists(user_config_path): raise Exception("文件路径不存在") - user_config = ConfigReader(user_config_path).getConfigs() + user_config = JSONReader(user_config_path).data() if user_config and "groups" in user_config: return user_config # compatibility with old version config format - if user_config and "users" in user_config: + elif user_config and "users" in user_config: user_config = { "groups": [ { @@ -562,12 +552,13 @@ class ALConfigWidget(QWidget, Ui_ALConfigWidget): ] } return user_config - return None + else: + return None except Exception as e: QMessageBox.warning( self, "警告 - AutoLibrary", - f"用户配置文件读取发生错误 ! : \n{e}" + f"用户配置文件读取发生错误 ! :\n{e}" ) return None @@ -583,13 +574,13 @@ class ALConfigWidget(QWidget, Ui_ALConfigWidget): raise Exception("文件路径为空") if not user_config_data or not isinstance(user_config_data, dict): raise Exception("用户配置数据为空或类型错误") - ConfigWriter(user_config_path, user_config_data) + JSONWriter(user_config_path, user_config_data) return True except Exception as e: QMessageBox.warning( self, "警告 - AutoLibrary", - f"用户配置文件写入发生错误 ! : \n{e}" + f"用户配置文件写入发生错误 ! :\n{e}" ) return False @@ -838,7 +829,7 @@ class ALConfigWidget(QWidget, Ui_ALConfigWidget): previous.setText(1, "" if user.get("enabled", True) else "跳过") previous.setData(0, Qt.UserRole, user) if current is None: - self.initilizeUserInfoWidget() + self.initializeUserInfoWidget() return if current.type() == ALUserTreeItemType.USER.value: user = current.data(0, Qt.UserRole) @@ -846,7 +837,7 @@ class ALConfigWidget(QWidget, Ui_ALConfigWidget): self.setUserToWidget(user) self.UsernameEdit.textEdited.connect(lambda text: current.setText(0, text)) else: - self.initilizeUserInfoWidget() + self.initializeUserInfoWidget() @Slot() def onUserTreeWidgetItemChanged( @@ -973,9 +964,27 @@ class ALConfigWidget(QWidget, Ui_ALConfigWidget): )[0] if run_config_path: run_config_path = QDir.toNativeSeparators(run_config_path) - if self.loadConfig(run_config_path): + data = self.loadRunConfig(run_config_path) + if data is not None: + self.__config_data["run"].update(data) + self.setRunConfigToWidget(data) self.__config_paths["run"] = run_config_path self.CurrentRunConfigEdit.setText(run_config_path) + paths = self.__cfg_mgr.get(ConfigType.GLOBAL, "automation.run_path.paths", []) + if run_config_path not in paths: + paths.append(run_config_path) + index = len(paths) - 1 + else: + index = paths.index(run_config_path) + self.__cfg_mgr.set(ConfigType.GLOBAL, "automation.run_path", {"current": index, "paths": paths}) + else: + QMessageBox.warning( + self, + "警告 - AutoLibrary", + "运行配置文件读取发生错误 ! :\n"\ + "无法从选择的运行配置文件中加载数据 ! :\n"\ + "可能选择了错误的配置文件类型" + ) @Slot() def onBrowseCurrentUserConfigButtonClicked( @@ -990,9 +999,27 @@ class ALConfigWidget(QWidget, Ui_ALConfigWidget): )[0] if user_config_path: user_config_path = QDir.toNativeSeparators(user_config_path) - if self.loadConfig(user_config_path): + data = self.loadUserConfig(user_config_path) + if data is not None: + self.__config_data["user"].update(data) + self.setUsersToTreeWidget(data) self.__config_paths["user"] = user_config_path self.CurrentUserConfigEdit.setText(user_config_path) + paths = self.__cfg_mgr.get(ConfigType.GLOBAL, "automation.user_path.paths", []) + if user_config_path not in paths: + paths.append(user_config_path) + index = len(paths) - 1 + else: + index = paths.index(user_config_path) + self.__cfg_mgr.set(ConfigType.GLOBAL, "automation.user_path", {"current": index, "paths": paths}) + else: + QMessageBox.warning( + self, + "警告 - AutoLibrary", + "用户配置文件读取发生错误 ! :\n"\ + "无法从选择的用户配置文件中加载数据 ! :\n"\ + "可能选择了错误的配置文件类型" + ) @Slot() def onBrowseExportRunConfigButtonClicked( @@ -1079,9 +1106,9 @@ class ALConfigWidget(QWidget, Ui_ALConfigWidget): if run_exists or user_exists: exist_files = [] if run_exists: - exist_files.append(run_config_path) + exist_files.append(f"运行配置文件: \n{run_config_path}") if user_exists: - exist_files.append(user_config_path) + exist_files.append(f"用户配置文件: \n{user_config_path}") reply = QMessageBox.information( self, "提示 - AutoLibrary", @@ -1097,8 +1124,8 @@ class ALConfigWidget(QWidget, Ui_ALConfigWidget): "run": run_config_path, "user": user_config_path } - self.initlizeConfigToWidget("run", self.__config_data["run"]) - self.initlizeConfigToWidget("user", self.__config_data["user"]) + self.initializeConfigToWidget("run", self.__config_data["run"]) + self.initializeConfigToWidget("user", self.__config_data["user"]) @Slot() def onConfirmButtonClicked( @@ -1115,7 +1142,7 @@ class ALConfigWidget(QWidget, Ui_ALConfigWidget): QMessageBox.information( self, "提示 - AutoLibrary", - "配置文件保存成功 !\n" + "配置文件保存成功 ! :\n" f"运行配置文件路径: \n{self.__config_paths['run']}\n"\ f"用户配置文件路径: \n{self.__config_paths['user']}" ) @@ -1123,7 +1150,7 @@ class ALConfigWidget(QWidget, Ui_ALConfigWidget): QMessageBox.warning( self, "警告 - AutoLibrary", - "配置文件保存失败, 请检查文件路径权限" + "配置文件保存失败 !\n" ) self.close() diff --git a/src/gui/ALMainWindow.py b/src/gui/ALMainWindow.py index 538cb43..ebfad56 100644 --- a/src/gui/ALMainWindow.py +++ b/src/gui/ALMainWindow.py @@ -7,12 +7,11 @@ This software is provided "as is", without any warranty of any kind. You may use, modify, and distribute this file under the terms of the MIT License. See the LICENSE file for details. """ -import sys -import time +import os import queue from PySide6.QtCore import ( - Qt, Signal, Slot, QDir, QFileInfo, QTimer, QUrl, + Qt, Signal, Slot, QTimer, QDir, QUrl, ) from PySide6.QtWidgets import ( QMainWindow, QMenu, QSystemTrayIcon, QMessageBox @@ -23,6 +22,9 @@ from PySide6.QtGui import ( from base.MsgBase import MsgBase +from utils.ConfigManager import ConfigType, instance +from utils.ConfigManager import getValidateAutomationConfigPaths + from gui.resources.ui.Ui_ALMainWindow import Ui_ALMainWindow from gui.resources import ALResource from gui.ALConfigWidget import ALConfigWidget @@ -44,14 +46,9 @@ class ALMainWindow(MsgBase, QMainWindow, Ui_ALMainWindow): MsgBase.__init__(self, queue.Queue(), queue.Queue()) QMainWindow.__init__(self) + self.__cfg_mgr = instance() self.__timer_task_queue = queue.Queue() - executable_path = sys.executable - exectuable_dir = QFileInfo(executable_path).absoluteDir() - self.__config_paths = { - "run": QDir.toNativeSeparators(exectuable_dir.absoluteFilePath("run.json")), - "user": QDir.toNativeSeparators(exectuable_dir.absoluteFilePath("user.json")), - "timer_task": QDir.toNativeSeparators(exectuable_dir.absoluteFilePath("timer_task.json")), - } + self.__config_paths = getValidateAutomationConfigPaths() self.__alTimerTaskManageWidget = None self.__alConfigWidget = None self.__auto_lib_thread = None @@ -78,7 +75,7 @@ class ALMainWindow(MsgBase, QMainWindow, Ui_ALMainWindow): # initialize timer task widget, but not show it try: - self.__alTimerTaskManageWidget = ALTimerTaskManageWidget(self, self.__config_paths["timer_task"]) + self.__alTimerTaskManageWidget = ALTimerTaskManageWidget(self) except Exception as e: QMessageBox.critical( self, @@ -295,8 +292,7 @@ class ALMainWindow(MsgBase, QMainWindow, Ui_ALMainWindow): @Slot(dict) def onConfigWidgetClosed( - self, - config_paths: dict + self ): if self.__alConfigWidget: @@ -304,7 +300,6 @@ class ALMainWindow(MsgBase, QMainWindow, Ui_ALMainWindow): self.__alConfigWidget.deleteLater() self.__alConfigWidget = None self.setControlButtons(True, None, None) - self.__config_paths = config_paths @Slot(dict) def onTimerTaskIsReady( @@ -359,10 +354,7 @@ class ALMainWindow(MsgBase, QMainWindow, Ui_ALMainWindow): ): if self.__alConfigWidget is None: - self.__alConfigWidget = ALConfigWidget( - self, - self.__config_paths - ) + self.__alConfigWidget = ALConfigWidget(self) self.__alConfigWidget.configWidgetIsClosed.connect(self.onConfigWidgetClosed) self.__alConfigWidget.show() self.__alConfigWidget.raise_() diff --git a/src/gui/ALMainWorkers.py b/src/gui/ALMainWorkers.py index 3913407..4fee821 100644 --- a/src/gui/ALMainWorkers.py +++ b/src/gui/ALMainWorkers.py @@ -17,7 +17,7 @@ from PySide6.QtCore import ( from base.MsgBase import MsgBase from operators.AutoLib import AutoLib -from utils.ConfigReader import ConfigReader +from utils.JSONReader import JSONReader class AutoLibWorker(MsgBase, QThread): @@ -69,11 +69,11 @@ class AutoLibWorker(MsgBase, QThread): self._showTrace( f"正在加载配置文件, 运行配置文件路径: {self.__config_paths["run"]}" ) - self.__run_config = ConfigReader(self.__config_paths["run"]).getConfigs() + self.__run_config = JSONReader(self.__config_paths["run"]).data() self._showTrace( f"正在加载配置文件, 用户配置文件路径: {self.__config_paths["user"]}" ) - self.__user_config = ConfigReader(self.__config_paths["user"]).getConfigs() + self.__user_config = JSONReader(self.__config_paths["user"]).data() if self.__run_config is None or self.__user_config is None: self._showTrace("配置文件加载失败, 请检查配置文件是否正确") self._showTrace("配置文件加载失败, 请检查配置文件是否正确") diff --git a/src/gui/ALTimerTaskManageWidget.py b/src/gui/ALTimerTaskManageWidget.py index 00b0590..4aeff70 100644 --- a/src/gui/ALTimerTaskManageWidget.py +++ b/src/gui/ALTimerTaskManageWidget.py @@ -25,8 +25,7 @@ from PySide6.QtGui import ( QCloseEvent ) -from utils.ConfigReader import ConfigReader -from utils.ConfigWriter import ConfigWriter +from utils.ConfigManager import ConfigType, instance from gui.resources.ui.Ui_ALTimerTaskManageWidget import Ui_ALTimerTaskManageWidget from gui.ALTimerTaskAddDialog import ALTimerTaskAddDialog, ALTimerTaskStatus @@ -142,16 +141,15 @@ class ALTimerTaskManageWidget(QWidget, Ui_ALTimerTaskManageWidget): def __init__( self, - parent = None, - timer_tasks_config_path: str = "" + parent = None ): super().__init__(parent) + self.__cfg_mgr = instance() self.__timer_tasks = [] self.__check_timer = None self.__sort_policy = self.SortPolicy.BY_EXECUTE_TIME self.__sort_order = Qt.SortOrder.AscendingOrder - self.__timer_tasks_config_path = timer_tasks_config_path self.setupUi(self) self.connectSignals() @@ -180,44 +178,28 @@ class ALTimerTaskManageWidget(QWidget, Ui_ALTimerTaskManageWidget): self.__check_timer.start(500) - def initlizeDefaultConfigPaths( - self - ): - - executable_path = sys.executable - executable_dir = QFileInfo(executable_path).absoluteDir() - self.__default_timer_tasks_config_path = QDir.toNativeSeparators(executable_dir.absoluteFilePath("timer_task.json")) - - def initializeTimerTasks( self ) -> bool: - if not self.__timer_tasks_config_path: - self.__timer_tasks_config_path = self.__default_timer_tasks_config_path - if os.path.exists(self.__timer_tasks_config_path): - timer_tasks = self.loadTimerTasks(self.__timer_tasks_config_path) - if timer_tasks is not None: - self.__timer_tasks = timer_tasks - self.timerTasksChanged.emit() - return True - timer_tasks = [] - if self.saveTimerTasks(self.__timer_tasks_config_path, copy.deepcopy(timer_tasks)): + timer_tasks = self.getTimerTasks() + if timer_tasks is not None: + self.__timer_tasks = timer_tasks + self.timerTasksChanged.emit() + return True + timer_tasks = [] + if self.setTimerTasks(copy.deepcopy(timer_tasks)): self.__timer_tasks = timer_tasks - self.updateTimerTaskList() return True return False - def loadTimerTasks( - self, - timer_tasks_config_path: str + def getTimerTasks( + self ) -> list: try: - if not timer_tasks_config_path or not os.path.exists(timer_tasks_config_path): - raise Exception("定时任务配置文件不存在") - timer_tasks = ConfigReader(timer_tasks_config_path).getConfigs() + timer_tasks = self.__cfg_mgr.get(ConfigType.TIMERTASK) if timer_tasks and "timer_tasks" in timer_tasks: for task in timer_tasks["timer_tasks"]: task["add_time"] = datetime.strptime(task["add_time"], "%Y-%m-%d %H:%M:%S") @@ -234,23 +216,17 @@ class ALTimerTaskManageWidget(QWidget, Ui_ALTimerTaskManageWidget): return None - def saveTimerTasks( + def setTimerTasks( self, - timer_tasks_config_path: str, timer_tasks: list ) -> bool: try: - if not timer_tasks_config_path: - raise Exception("配置文件路径为空") for task in timer_tasks: task["add_time"] = task["add_time"].strftime("%Y-%m-%d %H:%M:%S") task["execute_time"] = task["execute_time"].strftime("%Y-%m-%d %H:%M:%S") task["status"] = task["status"].value - ConfigWriter( - timer_tasks_config_path, - { "timer_tasks": timer_tasks } - ) + self.__cfg_mgr.set(ConfigType.TIMERTASK, "", { "timer_tasks": timer_tasks }) return True except Exception as e: QMessageBox.warning( @@ -470,7 +446,7 @@ class ALTimerTaskManageWidget(QWidget, Ui_ALTimerTaskManageWidget): self ): - self.saveTimerTasks(self.__timer_tasks_config_path, copy.deepcopy(self.__timer_tasks)) + self.setTimerTasks(copy.deepcopy(self.__timer_tasks)) self.updateTimerTaskList() self.updateStat() diff --git a/src/utils/ConfigManager.py b/src/utils/ConfigManager.py new file mode 100644 index 0000000..db211b3 --- /dev/null +++ b/src/utils/ConfigManager.py @@ -0,0 +1,233 @@ +# -*- coding: utf-8 -*- +""" +Copyright (c) 2026 KenanZhu. +All rights reserved. + +This software is provided "as is", without any warranty of any kind. +You may use, modify, and distribute this file under the terms of the MIT License. +See the LICENSE file for details. +""" +import os +import threading + +from enum import Enum +from typing import Any, Optional + +from utils.JSONReader import JSONReader +from utils.JSONWriter import JSONWriter + + +# This config manager class only responsible for global and other +# unconfigurable config files. + + +class ConfigType(Enum): + """ + Config type class. Values represent the default filename. + """ + GLOBAL = "autolibrary.json" # Global config file. + BULLETIN = "bulletin.json" # Bulletin board config file. + TIMERTASK = "timer_task.json" # Timer task config file. + + +class ConfigTemplate: + """ + Config template class. + """ + + def __init__( + self, + config_type: ConfigType + ): + + self.__config_type = config_type + + + def template( + self + ) -> dict: + """ + Get config template. + + Returns: + dict: Config template. + """ + match self.__config_type: + case ConfigType.GLOBAL: + return { + "automation": { + "run_path": { + "current": 0, + "paths": [] + }, + "user_path": { + "current": 0, + "paths": [] + } + } + } + case ConfigType.BULLETIN: + return { + "bulletin": [], + "last_sync_time": None + } + case ConfigType.TIMERTASK: + return { + "timer_tasks": [] + } + case _: + return {} + + +class ConfigManager: + + def __init__( + self, + config_dir: str + ): + + self.__config_dir = os.path.abspath(config_dir) + self.__config_lock = threading.Lock() + self.__config_data = {} + + self.initialize() + + + def initialize( + self + ): + + for config_type in ConfigType: + self.load(config_type) + + + def load( + self, + config_type: ConfigType + ): + + config_path = os.path.join(self.__config_dir, config_type.value) + if os.path.exists(config_path): + try: + config_data = JSONReader(config_path).data() + self.__config_data[config_type.value] = config_data + return + except: + pass + self.__config_data[config_type.value] = ConfigTemplate(config_type).template() + JSONWriter(config_path, self.__config_data[config_type.value]) + + + def get( + self, + config_type: ConfigType, + key: str = "", + default: Optional[Any] = None + ) -> Any: + + with self.__config_lock: + config_data = self.__config_data[config_type.value] + if key == "": + return config_data + keys = key.split('.') + for k in keys[:-1]: + config_data = config_data.get(k, None) + if config_data is None: + return default + return config_data.get(keys[-1], default) + + + def set( + self, + config_type: ConfigType, + key: str = "", + value: Any = None + ): + + with self.__config_lock: + root_data = self.__config_data[config_type.value] + if key == "": + self.__config_data[config_type.value] = value + else: + keys = key.split('.') + config_data = root_data + for k in keys[:-1]: + if k not in config_data: + config_data[k] = {} + config_data = config_data[k] + config_data[keys[-1]] = value + self.save(config_type) + + + def save( + self, + config_type: ConfigType + ): + + config_path = os.path.join(self.__config_dir, config_type.value) + JSONWriter(config_path, self.__config_data[config_type.value]) + + + def appDir( + self + ) -> str: + + return self.__config_dir + + +_config_manager_instance = None + +# Utility function to get config data (thread-safe and validated) from ConfigManager instance. +def getValidateAutomationConfigPaths( +) -> dict: + """ + Get validated automation config paths from ConfigManager instance. + These function will validate the config paths and return the validated paths in a dict. + + Returns: + dict: Validated automation config paths. + """ + config_paths = {"run": "", "user": ""} + auto_config = _config_manager_instance.get(ConfigType.GLOBAL, "automation", {}) + for cfg_type in ["run", "user"]: + paths = auto_config.get(f"{cfg_type}_path", {}).get("paths", []) + index = auto_config.get(f"{cfg_type}_path", {}).get("current", 0) + if paths == []: + paths.append(os.path.join(_config_manager_instance.appDir(), f"{cfg_type}.json")) + if index < 0: + index = 0 + if index >= len(paths): + index = len(paths) - 1 + config_paths[cfg_type] = paths[index] + data = {"current": index, "paths": paths} + auto_config[f"{cfg_type}_path"] = data + _config_manager_instance.set(ConfigType.GLOBAL, "automation", auto_config) + return config_paths + +def getBaseConfigDir( +) -> str: + + return _config_manager_instance.appDir() + +# Singleton instance of ConfigManager. +_instance_lock = threading.Lock() +def instance( + config_dir: str = "" +) -> ConfigManager: + """ + Initialize ConfigManager singleton instance. + + Args: + config_dir (str): Config directory. + """ + global _config_manager_instance + with _instance_lock: + if _config_manager_instance is None: + _config_manager_instance = ConfigManager(config_dir) + else: + if config_dir == "": + return _config_manager_instance + if _config_manager_instance.appDir() != config_dir: + raise ValueError( + "ConfigManager 的实例已初始化，不能使用不同的配置目录。") + return _config_manager_instance diff --git a/src/utils/ConfigReader.py b/src/utils/ConfigReader.py deleted file mode 100644 index 99e5659..0000000 --- a/src/utils/ConfigReader.py +++ /dev/null @@ -1,115 +0,0 @@ -# -*- coding: utf-8 -*- -""" -Copyright (c) 2025 - 2026 KenanZhu. -All rights reserved. - -This software is provided "as is", without any warranty of any kind. -You may use, modify, and distribute this file under the terms of the MIT License. -See the LICENSE file for details. -""" -import json -import copy - -from typing import Any - - -class ConfigReader: - """ - Config reader class. - - This class is used to read config file in JSON format. - - Args: - config_path (str): The path of config file. - - Examples: - >>> print(open("config.json", "r", encoding="utf-8").read()) - { - "key1": { - "key2": "value1" - } - } - >>> config_reader = ConfigReader("config.json") - >>> config_reader.get("key1/key2") - "value1" - """ - - def __init__( - self, - config_path: str - ): - - self.__config_path = config_path - self.__config_data = None - self.__readConfig() - - - def __readConfig( - self - ): - - try: - with open(self.__config_path, 'r', encoding='utf-8') as file: - self.__config_data = json.load(file) - except FileNotFoundError as e: - raise Exception(f"配置文件不存在: {self.__config_path}") from e - except PermissionError as e: - raise Exception(f"没有足够的权限读取配置文件: {self.__config_path}") from e - except json.JSONDecodeError as e: - raise Exception(f"JSON 解析错误: {self.__config_path}") from e - except Exception as e: - raise Exception(f"读取配置文件时未知错误: {e}") from e - - - def getConfigs( - self - ) -> dict: - - return self.__config_data.copy() - - - def getConfig( - self, - key: str - ) -> Any: - - config = self.__config_data.get(key, {}) - return copy.deepcopy(config) - - - def get( - self, - key: str, - default: Any = None - ) -> Any: - - keys = key.split('/') - current = self.__config_data - for k in keys: - if isinstance(current, dict) and k in current: - current = current[k] - else: - return default - return copy.deepcopy(current) - - - def hasConfig( - self, - key: str - ) -> bool: - - return self.getConfig(key) != {} - - - def reReadConfig( - self - ) -> bool: - - return self.__readConfig() - - - def configPath( - self - ) -> str: - - return self.__config_path diff --git a/src/utils/ConfigWriter.py b/src/utils/ConfigWriter.py deleted file mode 100644 index a7092df..0000000 --- a/src/utils/ConfigWriter.py +++ /dev/null @@ -1,116 +0,0 @@ -# -*- coding: utf-8 -*- -""" -Copyright (c) 2025 - 2026 KenanZhu. -All rights reserved. - -This software is provided "as is", without any warranty of any kind. -You may use, modify, and distribute this file under the terms of the MIT License. -See the LICENSE file for details. -""" -import json - -from typing import Any - - -class ConfigWriter: - """ - Config writer class. - - This class is used to write config file in JSON format. - - Args: - config_path (str): The path of config file. - config_data (dict): The config data to be written. - - Examples: - >>> config_data = { - ... "key1": { - ... "key2": "value1" - ... } - ... } - >>> config_writer = ConfigWriter("config.json", config_data) - >>> config_writer.set("key1/key2", "value1") - True - >>> print(open("config.json", "r", encoding="utf-8").read()) - { - "key1": { - "key2": "value1" - } - } - """ - - def __init__( - self, - config_path: str, - config_data: dict - ): - - self.__config_path = config_path - self.__config_data = config_data.copy() if config_data is not None else {} - self.__writeConfig() - - - def __writeConfig( - self - ): - - try: - with open(self.__config_path, "w", encoding="utf-8") as f: - json.dump(self.__config_data, f, indent=4, sort_keys=False) - except PermissionError as e: - raise Exception(f"没有足够的权限写入配置文件: {self.__config_path}") from e - except IOError as e: - raise Exception(f"写入配置文件时发生 IO 错误: {self.__config_path}") from e - except TypeError as e: - raise Exception(f"配置数据包含无法 JSON 序列化的类型: {e}") from e - except Exception as e: - raise Exception(f"写入配置文件时未知错误: {e}") from e - - - def setConfigs( - self, - configs: dict - ) -> bool: - - self.__config_data = configs - return self.__writeConfig() - - - def setConfig( - self, - key: str, - value: dict - ) -> bool: - - self.__config_data[key] = value - return self.__writeConfig() - - - def set( - self, - key: str, - value: Any - ) -> bool: - - keys = key.replace("\\", "/").split("/") - current = self.__config_data - for k in keys[:-1]: - if k not in current or not isinstance(current[k], dict): - current[k] = {} - current = current[k] - current[keys[-1]] = value - return self.__writeConfig() - - - def reWriteConfig( - self - ) -> bool: - - return self.__writeConfig() - - - def configPath( - self - ) -> str: - - return self.__config_path \ No newline at end of file diff --git a/src/utils/JSONReader.py b/src/utils/JSONReader.py new file mode 100644 index 0000000..04677a3 --- /dev/null +++ b/src/utils/JSONReader.py @@ -0,0 +1,85 @@ +# -*- coding: utf-8 -*- +""" +Copyright (c) 2026 KenanZhu. +All rights reserved. + +This software is provided "as is", without any warranty of any kind. +You may use, modify, and distribute this file under the terms of the MIT License. +See the LICENSE file for details. +""" +import os +import json + + +class JSONReader: + """ + JSON reader class. + + This class is used to read JSON file. + + Args: + json_path (str): The path of JSON file. + + Examples: + >>> print(open("config.json", "r", encoding="utf-8").read()) + { + "key1": { + "key2": "value1" + } + } + >>> json_reader = JSONReader("config.json") + >>> data = json_reader.data() + >>> data["key1"]["key2"] + "value1" + """ + + def __init__( + self, + json_path: str + ): + + self.__json_path = os.path.abspath(json_path) + self.__json_data = None + self.__read() + + + def __read( + self + ): + + try: + with open(self.__json_path, 'r', encoding='utf-8') as file: + self.__json_data = json.load(file) + except FileNotFoundError as e: + raise Exception(f"文件不存在: {self.__json_path}") from e + except PermissionError as e: + raise Exception(f"没有足够的权限读取文件: {self.__json_path}") from e + except json.JSONDecodeError as e: + raise Exception(f"JSON 解析错误: {self.__json_path}") from e + except Exception as e: + raise Exception(f"读取文件时发生未知错误: {e}") from e + + + def read( + self + ) -> bool: + + try: + self.__read() + except: + return False + return True + + + def data( + self + ) -> dict: + + return self.__json_data.copy() + + + def path( + self + ) -> str: + + return self.__json_path diff --git a/src/utils/JSONWriter.py b/src/utils/JSONWriter.py new file mode 100644 index 0000000..baa4d27 --- /dev/null +++ b/src/utils/JSONWriter.py @@ -0,0 +1,82 @@ +# -*- coding: utf-8 -*- +""" +Copyright (c) 2026 KenanZhu. +All rights reserved. + +This software is provided "as is", without any warranty of any kind. +You may use, modify, and distribute this file under the terms of the MIT License. +See the LICENSE file for details. +""" +import os +import json + + +class JSONWriter: + """ + JSON writer class. + + This class is used to write JSON file. + + Args: + json_path (str): The path of JSON file. + json_data (dict): The JSON data to be written. + + Examples: + >>> json_data = { + ... "key1": { + ... "key2": "value1" + ... } + ... } + >>> json_writer = JSONWriter("config.json", json_data) + >>> print(open("config.json", "r", encoding="utf-8").read()) + { + "key1": { + "key2": "value1" + } + } + """ + + def __init__( + self, + json_path: str, + json_data: dict + ): + + self.__json_path = os.path.abspath(json_path) + self.__json_data = json_data.copy() if json_data is not None else {} + self.__write() + + + def __write( + self + ): + + try: + with open(self.__json_path, "w", encoding="utf-8") as f: + json.dump(self.__json_data, f, indent=4, sort_keys=False) + except PermissionError as e: + raise Exception(f"没有足够的权限写入文件: {self.__json_path}") from e + except IOError as e: + raise Exception(f"写入文件时发生 IO 错误: {self.__json_path}") from e + except TypeError as e: + raise Exception(f"JSON 数据包含无法 JSON 序列化的类型: {e}") from e + except Exception as e: + raise Exception(f"写入文件时发生未知错误: {e}") from e + + + def write( + self + ) -> bool: + + try: + self.__write() + except: + return False + return True + + + def path( + self + ) -> str: + + return self.__json_path \ No newline at end of file diff --git a/src/utils/__init__.py b/src/utils/__init__.py index d1a4a08..a05bee0 100644 --- a/src/utils/__init__.py +++ b/src/utils/__init__.py @@ -2,6 +2,7 @@ Utils module for the AutoLibrary project. Here are the classes and modules in this package: - - ConfigReader: Configuration reader class for the AutoLibrary project. - - ConfigWriter: Configuration writer class for the AutoLibrary project. + - ConfigManager: Configuration manager class for the AutoLibrary project. + - JSONReader: JSON reader class for the AutoLibrary project. + - JSONWriter: JSON writer class for the AutoLibrary project. """ \ No newline at end of file