mirror of
https://github.com/KenanZhu/AutoLibrary.git
synced 2026-06-18 07:23:03 +08:00
refactor(config): 新增 ConfigUtils 工具类并优化配置管理逻辑
- 新增 ConfigUtils 工具类,提供配置路径获取等工具方法 - 将 ConfigManager.getValidateAutomationConfigPaths() 重构为 ConfigUtils.getAutomationConfigPaths() - 优化 MsgBase 中 LogManager 的导入方式,使用模块导入替代函数导入 - 规范化 TimerUtils.py 中 calculate_next_repeat_time() 的文档字符串格式
This commit is contained in:
+2
-2
@@ -11,7 +11,7 @@ import logging
|
|||||||
import queue
|
import queue
|
||||||
import datetime
|
import datetime
|
||||||
|
|
||||||
from managers.log.LogManager import getLogger
|
import managers.log.LogManager as LogManager
|
||||||
|
|
||||||
|
|
||||||
class MsgBase:
|
class MsgBase:
|
||||||
@@ -54,7 +54,7 @@ class MsgBase:
|
|||||||
self._input_queue = input_queue
|
self._input_queue = input_queue
|
||||||
self._output_queue = output_queue
|
self._output_queue = output_queue
|
||||||
try:
|
try:
|
||||||
self._logger = getLogger(self._class_name)
|
self._logger = LogManager.getLogger(self._class_name)
|
||||||
except RuntimeError:
|
except RuntimeError:
|
||||||
self._logger = None
|
self._logger = None
|
||||||
|
|
||||||
|
|||||||
@@ -24,6 +24,7 @@ import managers.config.ConfigManager as ConfigManager
|
|||||||
|
|
||||||
from utils.JSONReader import JSONReader
|
from utils.JSONReader import JSONReader
|
||||||
from utils.JSONWriter import JSONWriter
|
from utils.JSONWriter import JSONWriter
|
||||||
|
from utils.ConfigUtils import ConfigUtils
|
||||||
|
|
||||||
from gui.resources.ui.Ui_ALConfigWidget import Ui_ALConfigWidget
|
from gui.resources.ui.Ui_ALConfigWidget import Ui_ALConfigWidget
|
||||||
from gui.ALSeatMapSelectDialog import ALSeatMapSelectDialog
|
from gui.ALSeatMapSelectDialog import ALSeatMapSelectDialog
|
||||||
@@ -43,7 +44,7 @@ class ALConfigWidget(QWidget, Ui_ALConfigWidget):
|
|||||||
|
|
||||||
super().__init__(parent)
|
super().__init__(parent)
|
||||||
self.__cfg_mgr = ConfigManager.instance()
|
self.__cfg_mgr = ConfigManager.instance()
|
||||||
self.__config_paths = ConfigManager.getValidateAutomationConfigPaths()
|
self.__config_paths = ConfigUtils.getAutomationConfigPaths()
|
||||||
self.__config_data = {"run": {}, "user": {}}
|
self.__config_data = {"run": {}, "user": {}}
|
||||||
|
|
||||||
self.setupUi(self)
|
self.setupUi(self)
|
||||||
|
|||||||
@@ -19,9 +19,8 @@ from PySide6.QtGui import (
|
|||||||
QTextCursor, QCloseEvent, QFont, QIcon, QDesktopServices
|
QTextCursor, QCloseEvent, QFont, QIcon, QDesktopServices
|
||||||
)
|
)
|
||||||
|
|
||||||
import managers.config.ConfigManager as ConfigManager
|
|
||||||
|
|
||||||
from base.MsgBase import MsgBase
|
from base.MsgBase import MsgBase
|
||||||
|
from utils.ConfigUtils import ConfigUtils
|
||||||
|
|
||||||
from gui.resources.ui.Ui_ALMainWindow import Ui_ALMainWindow
|
from gui.resources.ui.Ui_ALMainWindow import Ui_ALMainWindow
|
||||||
from gui.resources import ALResource
|
from gui.resources import ALResource
|
||||||
@@ -44,9 +43,8 @@ class ALMainWindow(MsgBase, QMainWindow, Ui_ALMainWindow):
|
|||||||
|
|
||||||
MsgBase.__init__(self, queue.Queue(), queue.Queue())
|
MsgBase.__init__(self, queue.Queue(), queue.Queue())
|
||||||
QMainWindow.__init__(self)
|
QMainWindow.__init__(self)
|
||||||
self.__cfg_mgr = ConfigManager.instance()
|
|
||||||
self.__timer_task_queue = queue.Queue()
|
self.__timer_task_queue = queue.Queue()
|
||||||
self.__config_paths = ConfigManager.getValidateAutomationConfigPaths()
|
self.__config_paths = ConfigUtils.getAutomationConfigPaths()
|
||||||
self.__alTimerTaskManageWidget = None
|
self.__alTimerTaskManageWidget = None
|
||||||
self.__alConfigWidget = None
|
self.__alConfigWidget = None
|
||||||
self.__auto_lib_thread = None
|
self.__auto_lib_thread = None
|
||||||
@@ -300,7 +298,7 @@ class ALMainWindow(MsgBase, QMainWindow, Ui_ALMainWindow):
|
|||||||
self.__alConfigWidget.configWidgetIsClosed.disconnect(self.onConfigWidgetClosed)
|
self.__alConfigWidget.configWidgetIsClosed.disconnect(self.onConfigWidgetClosed)
|
||||||
self.__alConfigWidget.deleteLater()
|
self.__alConfigWidget.deleteLater()
|
||||||
self.__alConfigWidget = None
|
self.__alConfigWidget = None
|
||||||
self.__config_paths = ConfigManager.getValidateAutomationConfigPaths()
|
self.__config_paths = ConfigUtils.getAutomationConfigPaths()
|
||||||
self.setControlButtons(True, None, None)
|
self.setControlButtons(True, None, None)
|
||||||
self._showLog("配置窗口已关闭,配置文件路径已更新")
|
self._showLog("配置窗口已关闭,配置文件路径已更新")
|
||||||
|
|
||||||
|
|||||||
@@ -16,7 +16,7 @@ from PySide6.QtCore import Slot, QDateTime
|
|||||||
from PySide6.QtWidgets import QLabel, QDialog, QWidget, QSpinBox, QHBoxLayout, QGridLayout, QDateTimeEdit
|
from PySide6.QtWidgets import QLabel, QDialog, QWidget, QSpinBox, QHBoxLayout, QGridLayout, QDateTimeEdit
|
||||||
|
|
||||||
from gui.resources.ui.Ui_ALTimerTaskAddDialog import Ui_ALTimerTaskAddDialog
|
from gui.resources.ui.Ui_ALTimerTaskAddDialog import Ui_ALTimerTaskAddDialog
|
||||||
import utils.TimerUtils as TimerUtils
|
from utils.TimerUtils import TimerUtils
|
||||||
|
|
||||||
|
|
||||||
class ALTimerTaskStatus(Enum):
|
class ALTimerTaskStatus(Enum):
|
||||||
@@ -131,6 +131,7 @@ class ALTimerTaskAddDialog(QDialog, Ui_ALTimerTaskAddDialog):
|
|||||||
"repeat": self.RepeatCheckBox.isChecked(),
|
"repeat": self.RepeatCheckBox.isChecked(),
|
||||||
}
|
}
|
||||||
if task_data["repeat"]:
|
if task_data["repeat"]:
|
||||||
|
task_data["history"] = [] # repeat history
|
||||||
repeat_days = []
|
repeat_days = []
|
||||||
if self.MonCheckBox.isChecked():
|
if self.MonCheckBox.isChecked():
|
||||||
repeat_days.append(0)
|
repeat_days.append(0)
|
||||||
@@ -152,7 +153,7 @@ class ALTimerTaskAddDialog(QDialog, Ui_ALTimerTaskAddDialog):
|
|||||||
task_data["repeat_hour"] = execute_time.hour
|
task_data["repeat_hour"] = execute_time.hour
|
||||||
task_data["repeat_minute"] = execute_time.minute
|
task_data["repeat_minute"] = execute_time.minute
|
||||||
task_data["repeat_second"] = execute_time.second
|
task_data["repeat_second"] = execute_time.second
|
||||||
task_data["execute_time"] = TimerUtils.calculateNextRepeatTime(
|
task_data["execute_time"] = TimerUtils.getNextTimerRepeatTime(
|
||||||
task_data["repeat_days"],
|
task_data["repeat_days"],
|
||||||
task_data["repeat_hour"],
|
task_data["repeat_hour"],
|
||||||
task_data["repeat_minute"],
|
task_data["repeat_minute"],
|
||||||
|
|||||||
@@ -26,7 +26,7 @@ from PySide6.QtGui import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
import managers.config.ConfigManager as ConfigManager
|
import managers.config.ConfigManager as ConfigManager
|
||||||
import utils.TimerUtils as TimerUtils
|
from utils.TimerUtils import TimerUtils
|
||||||
|
|
||||||
from gui.resources.ui.Ui_ALTimerTaskManageWidget import Ui_ALTimerTaskManageWidget
|
from gui.resources.ui.Ui_ALTimerTaskManageWidget import Ui_ALTimerTaskManageWidget
|
||||||
from gui.ALTimerTaskAddDialog import ALTimerTaskAddDialog, ALTimerTaskStatus
|
from gui.ALTimerTaskAddDialog import ALTimerTaskAddDialog, ALTimerTaskStatus
|
||||||
@@ -610,7 +610,7 @@ class ALTimerTaskManageWidget(QWidget, Ui_ALTimerTaskManageWidget):
|
|||||||
"duration": 0,
|
"duration": 0,
|
||||||
"uuid": timer_task["uuid"]
|
"uuid": timer_task["uuid"]
|
||||||
})
|
})
|
||||||
next_time = TimerUtils.calculateNextRepeatTime(
|
next_time = TimerUtils.getNextTimerRepeatTime(
|
||||||
timer_task["repeat_days"],
|
timer_task["repeat_days"],
|
||||||
timer_task["repeat_hour"],
|
timer_task["repeat_hour"],
|
||||||
timer_task["repeat_minute"],
|
timer_task["repeat_minute"],
|
||||||
|
|||||||
@@ -176,49 +176,7 @@ class ConfigManager:
|
|||||||
|
|
||||||
|
|
||||||
# ConfigManager singleton instance.
|
# ConfigManager singleton instance.
|
||||||
_config_manager_instance = None
|
_config_manager_instance : ConfigManager | None = None
|
||||||
|
|
||||||
# Utility functions.
|
|
||||||
#
|
|
||||||
# Utility function to get validated automation config paths.
|
|
||||||
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.configDir(), 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
|
|
||||||
|
|
||||||
# Utility function to get base config directory.
|
|
||||||
def getBaseConfigDir(
|
|
||||||
) -> str:
|
|
||||||
"""
|
|
||||||
Get base config directory, on Windows, it is usually at :
|
|
||||||
'C:\\Users\\<username>\\AppData\\Local\\AutoLibrary\\config'.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
str: Base config directory.
|
|
||||||
"""
|
|
||||||
|
|
||||||
return _config_manager_instance.configDir()
|
|
||||||
|
|
||||||
# Singleton instance of ConfigManager.
|
# Singleton instance of ConfigManager.
|
||||||
_instance_lock = threading.Lock()
|
_instance_lock = threading.Lock()
|
||||||
@@ -240,6 +198,6 @@ def instance(
|
|||||||
else:
|
else:
|
||||||
if config_dir == "":
|
if config_dir == "":
|
||||||
return _config_manager_instance
|
return _config_manager_instance
|
||||||
if getBaseConfigDir() != config_dir:
|
if _config_manager_instance.configDir() != config_dir:
|
||||||
raise ValueError("ConfigManager 的实例已初始化,不能使用不同的配置目录。")
|
raise ValueError("ConfigManager 的实例已初始化,不能使用不同的配置目录。")
|
||||||
return _config_manager_instance
|
return _config_manager_instance
|
||||||
|
|||||||
@@ -186,7 +186,7 @@ def instance(
|
|||||||
raise ValueError("LogManager 的实例已初始化, 不能使用不同的日志目录")
|
raise ValueError("LogManager 的实例已初始化, 不能使用不同的日志目录")
|
||||||
return _log_manager_instance
|
return _log_manager_instance
|
||||||
|
|
||||||
|
# export function to get logger
|
||||||
def getLogger(
|
def getLogger(
|
||||||
name: Optional[str] = None
|
name: Optional[str] = None
|
||||||
) -> logging.Logger:
|
) -> logging.Logger:
|
||||||
|
|||||||
@@ -0,0 +1,44 @@
|
|||||||
|
# -*- 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 managers.config.ConfigManager as ConfigManager
|
||||||
|
|
||||||
|
class ConfigUtils:
|
||||||
|
"""
|
||||||
|
Config utilities class.
|
||||||
|
"""
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def getAutomationConfigPaths(
|
||||||
|
) -> dict[str]:
|
||||||
|
"""
|
||||||
|
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[str]: Validated automation config paths (include user and run config paths).
|
||||||
|
"""
|
||||||
|
cfg_mgr = ConfigManager.instance() # config manager instance
|
||||||
|
|
||||||
|
config_paths = {"run": "", "user": ""}
|
||||||
|
auto_config = cfg_mgr.get(ConfigManager.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(cfg_mgr.configDir(), 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
|
||||||
|
cfg_mgr.set(ConfigManager.ConfigType.GLOBAL, "automation", auto_config)
|
||||||
|
return config_paths
|
||||||
+41
-35
@@ -10,41 +10,47 @@ See the LICENSE file for details.
|
|||||||
from datetime import datetime, timedelta
|
from datetime import datetime, timedelta
|
||||||
|
|
||||||
|
|
||||||
def calculateNextRepeatTime(
|
class TimerUtils:
|
||||||
repeat_days: list,
|
|
||||||
hour: int,
|
|
||||||
minute: int,
|
|
||||||
second: int
|
|
||||||
) -> datetime:
|
|
||||||
"""
|
"""
|
||||||
Calculate the next repeat time based on repeat days and target time.
|
Timer utilities class.
|
||||||
|
|
||||||
This function calculates the next execution time for a repeatable task.
|
|
||||||
If the current day is in repeat_days and the target time has not passed,
|
|
||||||
it returns today's target time. Otherwise, it finds the next matching day.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
repeat_days (list): List of weekdays to repeat (0=Monday, 6=Sunday).
|
|
||||||
hour (int): Target hour (0-23).
|
|
||||||
minute (int): Target minute (0-59).
|
|
||||||
second (int): Target second (0-59).
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
datetime: The next repeat execution time.
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
current_time = datetime.now()
|
@staticmethod
|
||||||
current_weekday = current_time.weekday()
|
def getNextTimerRepeatTime(
|
||||||
target_time = current_time.replace(hour=hour, minute=minute, second=second, microsecond=0)
|
repeat_days: list[int],
|
||||||
if current_weekday in repeat_days:
|
hour: int,
|
||||||
if target_time > current_time:
|
minute: int,
|
||||||
return target_time
|
second: int
|
||||||
repeat_days_sorted = sorted(repeat_days)
|
) -> datetime:
|
||||||
for day in repeat_days_sorted:
|
"""
|
||||||
if day > current_weekday:
|
Calculate the next repeat time based on repeat days and target time.
|
||||||
days_until = day - current_weekday
|
|
||||||
next_time = target_time + timedelta(days=days_until)
|
This function calculates the next execution time for a repeatable task.
|
||||||
return next_time
|
If the current day is in repeat_days and the target time has not passed,
|
||||||
days_until = 7 - current_weekday + repeat_days_sorted[0]
|
it returns today's target time. Otherwise, it finds the next matching day.
|
||||||
next_time = target_time + timedelta(days=days_until)
|
|
||||||
return next_time
|
Args:
|
||||||
|
repeat_days (list[int]): List of weekdays to repeat (0=Monday, 6=Sunday).
|
||||||
|
hour (int): Target hour (0-23).
|
||||||
|
minute (int): Target minute (0-59).
|
||||||
|
second (int): Target second (0-59).
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
datetime: The next repeat execution time.
|
||||||
|
"""
|
||||||
|
|
||||||
|
current_time = datetime.now()
|
||||||
|
current_weekday = current_time.weekday()
|
||||||
|
target_time = current_time.replace(hour=hour, minute=minute, second=second, microsecond=0)
|
||||||
|
if current_weekday in repeat_days:
|
||||||
|
if target_time > current_time:
|
||||||
|
return target_time
|
||||||
|
repeat_days_sorted = sorted(repeat_days)
|
||||||
|
for day in repeat_days_sorted:
|
||||||
|
if day > current_weekday:
|
||||||
|
days_until = day - current_weekday
|
||||||
|
next_time = target_time + timedelta(days=days_until)
|
||||||
|
return next_time
|
||||||
|
days_until = 7 - current_weekday + repeat_days_sorted[0]
|
||||||
|
next_time = target_time + timedelta(days=days_until)
|
||||||
|
return next_time
|
||||||
|
|||||||
Reference in New Issue
Block a user