mirror of
https://github.com/KenanZhu/AutoLibrary.git
synced 2026-06-17 23:13:03 +08:00
260 lines
8.3 KiB
Python
260 lines
8.3 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
Copyright (c) 2025 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 time
|
|
import queue
|
|
|
|
from PySide6.QtCore import (
|
|
Slot, Signal, QThread
|
|
)
|
|
|
|
from base.MsgBase import MsgBase
|
|
from operators.AutoLib import AutoLib
|
|
from utils.JSONReader import JSONReader
|
|
from autoscript import execute, registerDefaultTargetVars
|
|
|
|
|
|
class AutoLibWorker(MsgBase, QThread):
|
|
|
|
autoLibWorkerIsFinished = Signal()
|
|
autoLibWorkerFinishedWithError = Signal()
|
|
|
|
def __init__(
|
|
self,
|
|
input_queue: queue.Queue,
|
|
output_queue: queue.Queue,
|
|
config_paths: dict
|
|
):
|
|
|
|
MsgBase.__init__(self, input_queue, output_queue)
|
|
QThread.__init__(self)
|
|
self.__config_paths = config_paths
|
|
|
|
|
|
def checkTimeAvailable(
|
|
self,
|
|
) -> bool:
|
|
|
|
current_time = time.strftime("%H:%M", time.localtime())
|
|
if current_time >= "23:30" or current_time <= "07:30":
|
|
self._showTrace(
|
|
"当前时间不在图书馆开放时间内, 请在 07:30 - 23:30 之间尝试",
|
|
self.TraceLevel.WARNING
|
|
)
|
|
return False
|
|
self._showLog(f"时间检查通过, 当前时间: {current_time}", self.TraceLevel.INFO)
|
|
return True
|
|
|
|
|
|
def checkConfigPaths(
|
|
self,
|
|
) -> bool:
|
|
|
|
if not all(
|
|
os.path.exists(path) for path in self.__config_paths.values()
|
|
):
|
|
self._showTrace(
|
|
"配置文件路径不存在, 请检查配置文件路径是否正确",
|
|
self.TraceLevel.ERROR
|
|
)
|
|
return False
|
|
self._showLog(f"配置文件路径检查通过, 路径: {self.__config_paths}", self.TraceLevel.INFO)
|
|
return True
|
|
|
|
|
|
def loadConfigs(
|
|
self
|
|
) -> bool:
|
|
|
|
self._showTrace(
|
|
f"正在加载配置文件, 运行配置文件路径: {self.__config_paths["run"]}",
|
|
no_log=True
|
|
)
|
|
self._run_config = JSONReader(self.__config_paths["run"]).data()
|
|
self._showTrace(
|
|
f"正在加载配置文件, 用户配置文件路径: {self.__config_paths["user"]}",
|
|
no_log=True
|
|
)
|
|
self._user_config = JSONReader(self.__config_paths["user"]).data()
|
|
if self._run_config is None or self._user_config is None:
|
|
self._showTrace(
|
|
"配置文件加载失败, 请检查配置文件是否正确",
|
|
self.TraceLevel.ERROR
|
|
)
|
|
return False
|
|
if not self._user_config.get("groups"):
|
|
self._showTrace(
|
|
"用户配置文件中无有效任务组, 请检查用户配置文件是否正确",
|
|
self.TraceLevel.WARNING
|
|
)
|
|
return False
|
|
self._showLog(
|
|
f"配置文件加载成功, 任务组数量: {len(self._user_config.get('groups', []))}",
|
|
self.TraceLevel.INFO
|
|
)
|
|
return True
|
|
|
|
|
|
def run(
|
|
self
|
|
):
|
|
|
|
auto_lib = None
|
|
self._showTrace("AutoLibrary 开始运行")
|
|
if not self.checkTimeAvailable()\
|
|
or not self.checkConfigPaths():
|
|
# time or config existence check failed, skip and finish
|
|
pass
|
|
else:
|
|
try:
|
|
if not self.loadConfigs():
|
|
raise Exception("配置文件加载失败")
|
|
auto_lib = AutoLib(
|
|
self._input_queue,
|
|
self._output_queue,
|
|
self._run_config
|
|
)
|
|
groups = self._user_config.get("groups")
|
|
for group in groups:
|
|
if not group["enabled"]:
|
|
self._showTrace(f"任务组 {group["name"]} 已跳过", no_log=True)
|
|
continue
|
|
self._showTrace(f"正在运行任务组 {group["name"]}", no_log=True)
|
|
auto_lib.run(
|
|
{ "users": group.get("users", []) }
|
|
)
|
|
except Exception as e:
|
|
self._showTrace(
|
|
f"AutoLibrary 运行时发生异常 : {e}",
|
|
self.TraceLevel.ERROR
|
|
)
|
|
self.autoLibWorkerFinishedWithError.emit()
|
|
return
|
|
if auto_lib:
|
|
auto_lib.close()
|
|
self._showTrace("AutoLibrary 运行结束")
|
|
self.autoLibWorkerIsFinished.emit()
|
|
|
|
|
|
class TimerTaskWorker(AutoLibWorker):
|
|
|
|
timerTaskWorkerIsFinished = Signal(bool, dict)
|
|
|
|
def __init__(
|
|
self,
|
|
timer_task: dict,
|
|
input_queue: queue.Queue,
|
|
output_queue: queue.Queue,
|
|
config_paths: dict
|
|
):
|
|
|
|
super().__init__(input_queue, output_queue, config_paths)
|
|
self.__timer_task = timer_task
|
|
|
|
self.autoLibWorkerIsFinished.connect(self.onTimerTaskIsFinished)
|
|
self.autoLibWorkerFinishedWithError.connect(self.onTimerTaskFinishedWithError)
|
|
|
|
|
|
def run(
|
|
self
|
|
):
|
|
|
|
self._showTrace(f"定时任务 {self.__timer_task['name']} 开始运行")
|
|
if not self.checkTimeAvailable() or not self.checkConfigPaths():
|
|
self._showTrace("定时任务跳过执行 (时间或配置文件检查未通过)")
|
|
self.timerTaskWorkerIsFinished.emit(False, self.__timer_task)
|
|
return
|
|
try:
|
|
if not self.loadConfigs():
|
|
raise Exception("配置文件加载失败")
|
|
self.applyRepeatAutoScript()
|
|
auto_lib = AutoLib(
|
|
self._input_queue,
|
|
self._output_queue,
|
|
self._run_config
|
|
)
|
|
groups = self._user_config.get("groups")
|
|
for group in groups:
|
|
if not group["enabled"]:
|
|
self._showTrace(
|
|
f"任务组 {group['name']} 已跳过",
|
|
no_log=True
|
|
)
|
|
continue
|
|
self._showTrace(
|
|
f"正在运行任务组 {group['name']}",
|
|
no_log=True
|
|
)
|
|
auto_lib.run(
|
|
{"users": group.get("users", [])}
|
|
)
|
|
auto_lib.close()
|
|
except Exception as e:
|
|
self._showTrace(
|
|
f"定时任务 {self.__timer_task['name']} 运行时发生异常: {e}",
|
|
self.TraceLevel.ERROR
|
|
)
|
|
self.timerTaskWorkerIsFinished.emit(True, self.__timer_task)
|
|
return
|
|
self._showTrace(f"定时任务 {self.__timer_task['name']} 运行结束")
|
|
self.timerTaskWorkerIsFinished.emit(False, self.__timer_task)
|
|
|
|
|
|
def applyRepeatAutoScript(
|
|
self
|
|
):
|
|
|
|
auto_script = self.__timer_task.get("repeat_auto_script", "")
|
|
if not auto_script or not auto_script.strip():
|
|
return
|
|
self._showTrace(
|
|
f"检测到重复定时任务 AutoScript, 开始执行...",
|
|
no_log=True
|
|
)
|
|
groups = self._user_config.get("groups", [])
|
|
affected_count = 0
|
|
for group in groups:
|
|
if not group.get("enabled", False):
|
|
continue
|
|
for user in group.get("users", []):
|
|
try:
|
|
registerDefaultTargetVars()
|
|
execute(auto_script, user)
|
|
affected_count += 1
|
|
except ValueError as e:
|
|
self._showTrace(
|
|
f"AutoScript 执行错误 (用户 {user['username']}): {e}",
|
|
self.TraceLevel.ERROR
|
|
)
|
|
self._showLog(
|
|
f"AutoScript 执行完毕, "
|
|
f"影响 {affected_count} 个用户",
|
|
self.TraceLevel.INFO
|
|
)
|
|
|
|
@Slot()
|
|
def onTimerTaskIsFinished(
|
|
self
|
|
):
|
|
|
|
self._showTrace(f"定时任务 {self.__timer_task['name']} 运行结束")
|
|
self.timerTaskWorkerIsFinished.emit(False, self.__timer_task)
|
|
|
|
@Slot()
|
|
def onTimerTaskFinishedWithError(
|
|
self
|
|
):
|
|
|
|
self._showTrace(
|
|
f"定时任务 {self.__timer_task['name']} 运行时发生异常",
|
|
self.TraceLevel.ERROR
|
|
)
|
|
self.timerTaskWorkerIsFinished.emit(True, self.__timer_task)
|