AutoLibrary/src/gui/ALTimerTaskManageWidget.py

# -*- 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 os
import sys
import copy

from enum import Enum
from datetime import datetime, timedelta

from PySide6.QtCore import (
    Qt, Signal, Slot, QTimer
)
from PySide6.QtWidgets import (
    QDialog, QWidget, QListWidgetItem, QMessageBox,
    QHBoxLayout, QVBoxLayout, QLabel, QPushButton
)
from PySide6.QtGui import (
    QCloseEvent
)

import utils.ConfigManager as ConfigManager
import utils.TimerUtils as TimerUtils

from gui.resources.ui.Ui_ALTimerTaskManageWidget import Ui_ALTimerTaskManageWidget
from gui.ALTimerTaskAddDialog import ALTimerTaskAddDialog, ALTimerTaskStatus
from gui.ALTimerTaskHistoryDialog import ALTimerTaskHistoryDialog


class ALTimerTaskItemWidget(QWidget):

    def __init__(
        self,
        parent = None,
        timer_task: dict = None
    ):

        super().__init__(parent)
        self.__timer_task = timer_task

        self.modifyUi()


    def modifyUi(
        self
    ):

        self.ItemWidgetLayout = QHBoxLayout(self)
        self.ItemWidgetLayout.setSpacing(10)
        self.ItemWidgetLayout.setContentsMargins(10, 5, 10, 5)

        self.TaskInfoLayout = QVBoxLayout()
        self.TaskInfoLayout.setSpacing(5)
        TaskNameLabel = QLabel(self.__timer_task["name"])
        TaskNameLabelFont = TaskNameLabel.font()
        TaskNameLabelFont.setBold(True)
        TaskNameLabel.setFont(TaskNameLabelFont)
        TaskNameLabel.setFixedHeight(25)
        self.TaskInfoLayout.addWidget(TaskNameLabel)
        ExecuteTimeStr = self.__timer_task["execute_time"].strftime("%Y-%m-%d %H:%M:%S")
        if self.__timer_task.get("repeat", False):
            repeat_days = self.__timer_task.get("repeat_days", [])
            repeat_hour = self.__timer_task.get("repeat_hour", 0)
            repeat_minute = self.__timer_task.get("repeat_minute", 0)
            repeat_second = self.__timer_task.get("repeat_second", 0)
            if len(repeat_days) == 7:
                time_str = f"{repeat_hour:02d}:{repeat_minute:02d}:{repeat_second:02d}"
                ExecuteTimeLabel = QLabel(f"下次执行时间: {ExecuteTimeStr} (每日 {time_str})")
            else:
                day_names = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"]
                selected_days = [day_names[d] for d in repeat_days]
                time_str = f"{repeat_hour:02d}:{repeat_minute:02d}:{repeat_second:02d}"
                ExecuteTimeLabel = QLabel(f"下次执行时间: {ExecuteTimeStr} (每{','.join(selected_days)} {time_str})")
        else:
            ExecuteTimeLabel = QLabel(f"执行时间: {ExecuteTimeStr}")
        ExecuteTimeLabel.setStyleSheet("color: #969696;")
        ExecuteTimeLabel.setFixedHeight(20)
        self.TaskInfoLayout.addWidget(ExecuteTimeLabel)
        self.ItemWidgetLayout.addLayout(self.TaskInfoLayout)
        self.ItemWidgetLayout.addStretch()

        match self.__timer_task["status"]:
            case ALTimerTaskStatus.PENDING:
                TaskStatusText = "等待中"
                TaskStatusColor = "#FF9800"
            case ALTimerTaskStatus.READY:
                TaskStatusText = "已就绪"
                TaskStatusColor = "#316BFF"
            case ALTimerTaskStatus.RUNNING:
                TaskStatusText = "执行中"
                TaskStatusColor = "#2294FF"
            case ALTimerTaskStatus.EXECUTED:
                TaskStatusText = "已执行"
                TaskStatusColor = "#4CAF50"
            case ALTimerTaskStatus.ERROR:
                TaskStatusText = "执行失败"
                TaskStatusColor = "#DC0000"
            case ALTimerTaskStatus.OUTDATED:
                TaskStatusText = "已过期"
                TaskStatusColor = "#DC0000"
        TaskStatusLabel = QLabel(TaskStatusText)
        TaskStatusLabel.setStyleSheet(f"""
            QLabel {{
                background-color: {TaskStatusColor};
                color: #FFFFFF;
                border-radius: 5px;
                font-weight: bold;
            }}
        """)
        TaskStatusLabel.setAlignment(Qt.AlignmentFlag.AlignCenter)
        TaskStatusLabel.setFixedSize(80, 25)
        self.ItemWidgetLayout.addWidget(TaskStatusLabel)

        TaskModeText = "静默" if self.__timer_task["silent"] else "显示"
        TaskModeColor = "#6325FF" if self.__timer_task["silent"] else "#2294FF"
        TaskModeLabel = QLabel(TaskModeText)
        TaskModeLabel.setStyleSheet(f"""
            QLabel {{
                background-color: {TaskModeColor};
                color: #FFFFFF;
                border-radius: 5px;
                font-weight: bold;
            }}
        """)
        TaskModeLabel.setAlignment(Qt.AlignmentFlag.AlignCenter)
        TaskModeLabel.setFixedSize(60, 25)
        self.ItemWidgetLayout.addWidget(TaskModeLabel)

        if self.__timer_task.get("repeat", False):
            self.HistoryButton = QPushButton("历史")
            self.HistoryButton.setFixedSize(80, 25)
            self.ItemWidgetLayout.addWidget(self.HistoryButton)
        self.DeleteButton = QPushButton("删除")
        self.DeleteButton.setFixedSize(80, 25)
        self.DeleteButton.setStyleSheet("color: #DC0000;")
        self.ItemWidgetLayout.addWidget(self.DeleteButton)
        if self.__timer_task["status"] == ALTimerTaskStatus.READY\
        or self.__timer_task["status"] == ALTimerTaskStatus.RUNNING:
            self.DeleteButton.setEnabled(False)
        self.setFixedHeight(55)


class ALTimerTaskManageWidget(QWidget, Ui_ALTimerTaskManageWidget):

    class SortPolicy(Enum):

        BY_NAME = "按名称"
        BY_ADD_TIME = "按添加时间"
        BY_EXECUTE_TIME = "按执行时间"

    timerTaskIsReady = Signal(dict)
    timerTasksChanged = Signal()
    timerTaskManageWidgetIsClosed = Signal()

    def __init__(
        self,
        parent = None
    ):

        super().__init__(parent)
        self.__cfg_mgr = ConfigManager.instance()
        self.__timer_tasks = []
        self.__check_timer = None
        self.__sort_policy = self.SortPolicy.BY_EXECUTE_TIME
        self.__sort_order = Qt.SortOrder.AscendingOrder

        self.setupUi(self)
        self.connectSignals()
        self.setupTimer()
        if not self.initializeTimerTasks():
            raise Exception("定时任务配置文件初始化失败 !")


    def connectSignals(
        self
    ):

        self.AddTimerTaskButton.clicked.connect(self.addTask)
        self.ClearAllTimerTasksButton.clicked.connect(self.clearAllTasks)
        self.TimerTaskSortTypeComboBox.currentIndexChanged.connect(self.onSortPolicyComboBoxChanged)
        self.TimerTaskSortOrderToggleButton.clicked.connect(self.onSortOrderToggleButtonClicked)
        self.timerTasksChanged.connect(self.onTimerTasksChanged)


    def setupTimer(
        self
    ):

        self.__check_timer = QTimer(self)
        self.__check_timer.timeout.connect(self.checkTasks)
        self.__check_timer.start(500)


    def initializeTimerTasks(
        self
    ) -> bool:

        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
            return True
        return False


    def getTimerTasks(
        self
    ) -> list:

        try:
            timer_tasks = self.__cfg_mgr.get(ConfigManager.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")
                    task["execute_time"] = datetime.strptime(task["execute_time"], "%Y-%m-%d %H:%M:%S")
                    task["status"] = ALTimerTaskStatus(task["status"])
                    if "history" in task:
                        for item in task["history"]:
                            item["result"] = ALTimerTaskStatus(item["result"])
                return timer_tasks["timer_tasks"]
            raise Exception("定时任务配置文件格式错误")
        except Exception as e:
            QMessageBox.warning(
                self,
                "警告 - AutoLibrary",
                f"加载定时任务配置发生错误 ! : \n{e}"
            )
            return None


    def setTimerTasks(
        self,
        timer_tasks: list
    ) -> bool:

        try:
            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
                if "history" in task:
                    for item in task["history"]:
                        item["result"] = item["result"].value
            self.__cfg_mgr.set(ConfigManager.ConfigType.TIMERTASK, "", { "timer_tasks": timer_tasks })
            return True
        except Exception as e:
            QMessageBox.warning(
                self,
                "警告 - AutoLibrary",
                f"保存定时任务配置发生错误 ! : \n{e}"
            )
            return False


    def showEvent(
        self,
        event
    ):

        result = super().showEvent(event)

        screen_rect = self.screen().geometry()
        target_pos = self.parent().geometry().center()
        target_pos.setX(target_pos.x() - self.width()//2)
        target_pos.setY(target_pos.y() - self.height()//2)
        if target_pos.x() < 0:
            target_pos.setX(0)
        if target_pos.x() + self.width() > screen_rect.width():
            target_pos.setX(screen_rect.width() - self.width())
        if target_pos.y() < 0:
            target_pos.setY(0)
        if target_pos.y() + self.height() > screen_rect.height():
            target_pos.setY(screen_rect.height() - self.height())
        self.move(target_pos)

        return result


    def closeEvent(
        self,
        event: QCloseEvent
    ):

        self.hide()
        self.timerTaskManageWidgetIsClosed.emit()
        event.ignore()


    def sortTimerTasks(
        self,
        policy: SortPolicy = SortPolicy.BY_EXECUTE_TIME,
        order: Qt.SortOrder = Qt.SortOrder.AscendingOrder
    ):

        if policy == self.SortPolicy.BY_NAME:
            self.__timer_tasks.sort(
                key = lambda x: x["name"],
                reverse = order is Qt.SortOrder.DescendingOrder
            )
        elif policy == self.SortPolicy.BY_ADD_TIME:
            self.__timer_tasks.sort(
                key = lambda x: x["add_time"],
                reverse = order is Qt.SortOrder.DescendingOrder
            )
        elif policy == self.SortPolicy.BY_EXECUTE_TIME:
            self.__timer_tasks.sort(
                key = lambda x: x["execute_time"],
                reverse = order is Qt.SortOrder.DescendingOrder
            )


    def updateStat(
        self
    ):

        pending = 0
        in_queue = 0
        executed = 0
        invalid = 0
        total = len(self.__timer_tasks)
        for timer_task in self.__timer_tasks:
            if timer_task["status"] == ALTimerTaskStatus.PENDING:
                pending += 1
            elif timer_task["status"] == ALTimerTaskStatus.READY\
            or timer_task["status"] == ALTimerTaskStatus.RUNNING:
                in_queue += 1
            elif timer_task["status"] == ALTimerTaskStatus.EXECUTED:
                executed += 1
            elif timer_task["status"] == ALTimerTaskStatus.ERROR\
            or timer_task["status"] == ALTimerTaskStatus.OUTDATED:
                invalid += 1
        self.TotalTaskLabel.setText(f"总任务：{total}")
        self.PendingTaskLabel.setText(f"待执行：{pending}")
        self.InQueueTaskLabel.setText(f"队列中：{in_queue}")
        self.ExecutedTaskLabel.setText(f"已执行：{executed}")
        self.InvalidTaskLabel.setText(f"无效的：{invalid}")


    def updateTimerTaskList(
        self
    ):

        self.TimerTasksListWidget.clear()
        self.sortTimerTasks(self.__sort_policy, self.__sort_order)
        for timer_task in self.__timer_tasks:
            item = QListWidgetItem()
            item.setData(Qt.UserRole, timer_task)
            widget = ALTimerTaskItemWidget(self, timer_task)
            widget.DeleteButton.clicked.connect(
                lambda _, task = timer_task: self.deleteTask(task)
            )
            if timer_task.get("repeat", False) and hasattr(widget, "HistoryButton"):
                widget.HistoryButton.clicked.connect(
                    lambda _, task = timer_task: self.showTaskHistory(task)
                )
            item.setSizeHint(widget.size())
            self.TimerTasksListWidget.addItem(item)
            self.TimerTasksListWidget.setItemWidget(item, widget)


    def addTask(
        self
    ):

        dialog = ALTimerTaskAddDialog(self)
        if dialog.exec() == QDialog.DialogCode.Accepted:
            timer_task = dialog.getTimerTask()
            self.__timer_tasks.append(timer_task)
            self.timerTasksChanged.emit()


    @staticmethod
    def getTimerTaskDetailMessage(
        timer_task: dict
    ):

        return (
            f"任务名称：{timer_task["name"]}\n"
            f"添加时间：{timer_task["add_time"]}\n"
            f"当前状态：{timer_task["status"].value}\n"
            f"下次执行时间：{datetime.strftime(timer_task["execute_time"], "%Y-%m-%d %H:%M:%S")}\n"
            f"已执行次数：{len(timer_task['history'] if 'history' in timer_task else 0)}"
        )


    def deleteTask(
        self,
        timer_task: dict
    ):

        if timer_task["repeat"]: # when delete a repeat task
            msgbox = QMessageBox(self)
            msgbox.setIcon(QMessageBox.Icon.Question)
            msgbox.setWindowTitle("警告 - AutoLibrary")
            msgbox.setStandardButtons(
                QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No
            )
            msgbox.setText("删除可重复性任务将同时删除所有已执行的记录 !\n是否继续 ?")
            msgbox.setDetailedText(
                "以下可重复性任务将被删除：\n"\
                "\n"
                f"{self.getTimerTaskDetailMessage(timer_task)}"
            )
            result = msgbox.exec()
            if result != QMessageBox.StandardButton.Yes:
                return
        task_uuid = timer_task["task_uuid"]
        self.__timer_tasks = [
            x for x in self.__timer_tasks
            if x["task_uuid"] != task_uuid
        ]
        self.timerTasksChanged.emit()


    def clearAllTasks(
        self
    ):

        if not self.__timer_tasks:
            return
        result = QMessageBox.question(
            self,
            "确认 - AutoLibrary",
            "是否要清除所有定时任务 ?",
            QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No
        )
        if result == QMessageBox.StandardButton.No:
            return
        # READY and RUNNING tasks cannot be cleared
        in_queue_tasks = [
            x for x in self.__timer_tasks
            if x["status"] == ALTimerTaskStatus.READY
            or x["status"] == ALTimerTaskStatus.RUNNING
        ]
        in_queue_count = len(in_queue_tasks)
        if in_queue_count > 0:
            QMessageBox.warning(
                self,
                "警告 - AutoLibrary",
                f"存在 {in_queue_count} 个正在执行或已就绪的队列任务，无法清除所有定时任务 !"
            )
            return
        # repeat tasks ask before clear
        repeat_tasks = [
            x for x in self.__timer_tasks
            if x.get("repeat", False)
        ]
        repeat_tasks_count = len(repeat_tasks)
        if repeat_tasks_count > 0:
            msgbox = QMessageBox(self)
            msgbox.setIcon(QMessageBox.Icon.Question)
            msgbox.setWindowTitle("警告 - AutoLibrary")
            msgbox.setStandardButtons(
                QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No
            )
            msgbox.setText(
                f"存在 {repeat_tasks_count} 个可重复性任务，\n"
                "删除可重复性任务将同时删除所有已执行的记录 !\n"
                "是否继续 ?"
            )
            delete_msgs = [
                self.getTimerTaskDetailMessage(x) for x in repeat_tasks
            ]
            msgbox.setDetailedText(
                "以下可重复性任务将被删除：\n"\
                "\n"
                f"{"\n\n".join(delete_msgs)}"
            )
            result = msgbox.exec()
            if result != QMessageBox.StandardButton.Yes:
                return
        # clear all tasks
        self.__timer_tasks.clear()
        self.timerTasksChanged.emit()


    def showTaskHistory(
        self,
        task: dict
    ):

        dialog = ALTimerTaskHistoryDialog(self, task)
        if dialog.exec() == QDialog.DialogCode.Accepted:
            self.timerTasksChanged.emit()


    def checkTasks(
        self
    ):

        need_update = False

        now = datetime.now()
        for timer_task in self.__timer_tasks:
            if timer_task["execute_time"] > now:
                continue
            if timer_task["status"] is not ALTimerTaskStatus.PENDING:
                continue
            if timer_task["execute_time"] <= now + timedelta(seconds = -5):
                if timer_task.get("repeat", False):
                    self.onRepeatTimerTaskIs(ALTimerTaskStatus.OUTDATED, timer_task)
                else:
                    timer_task["status"] = ALTimerTaskStatus.OUTDATED
                need_update = True
            else:
                timer_task["status"] = ALTimerTaskStatus.READY
                self.timerTaskIsReady.emit(timer_task)
                need_update = True
        if need_update:
            self.timerTasksChanged.emit()

    @Slot(int)
    def onSortPolicyComboBoxChanged(
        self,
        policy: int
    ):

        mapping = {
            0: self.SortPolicy.BY_NAME,
            1: self.SortPolicy.BY_ADD_TIME,
            2: self.SortPolicy.BY_EXECUTE_TIME
        }
        self.__sort_policy = mapping[policy]
        self.updateTimerTaskList()

    @Slot()
    def onSortOrderToggleButtonClicked(
        self
    ):

        self.__sort_order = Qt.SortOrder.AscendingOrder\
            if self.__sort_order is Qt.SortOrder.DescendingOrder\
            else Qt.SortOrder.DescendingOrder
        self.TimerTaskSortOrderToggleButton.setText(
            "↑" if self.__sort_order is Qt.SortOrder.AscendingOrder else "↓"
        )
        self.updateTimerTaskList()

    @Slot()
    def onTimerTasksChanged(
        self
    ):

        self.setTimerTasks(copy.deepcopy(self.__timer_tasks))
        self.updateTimerTaskList()
        self.updateStat()


    @Slot(dict)
    def onTimerTaskIsRunning(
        self,
        timer_task: dict
    ):

        for task in self.__timer_tasks:
            if task["task_uuid"] == timer_task["task_uuid"]:
                task["status"] = ALTimerTaskStatus.RUNNING
                break
        self.timerTasksChanged.emit()


    def onRepeatTimerTaskIs(
        self,
        status: ALTimerTaskStatus,
        timer_task: dict
    ) -> dict:

        if "history" not in timer_task:
            timer_task["history"] = []
        executed_time = datetime.now()
        duration = (executed_time - timer_task["execute_time"]).total_seconds()
        timer_task["history"].append({
            "execute_time": timer_task["execute_time"].strftime("%Y-%m-%d %H:%M:%S"),
            "executed_time": executed_time.strftime("%Y-%m-%d %H:%M:%S"),
            "result": status,
            "duration": duration if status is ALTimerTaskStatus.EXECUTED else 0,
            "uuid": timer_task["task_uuid"]
        })
        next_time = TimerUtils.calculateNextRepeatTime(
            timer_task["repeat_days"],
            timer_task["repeat_hour"],
            timer_task["repeat_minute"],
            timer_task["repeat_second"]
        )
        if next_time:
            timer_task["execute_time"] = next_time
            timer_task["status"] = ALTimerTaskStatus.PENDING
            timer_task["executed"] = False
        else:
            timer_task["status"] = status

    @Slot(dict)
    def onTimerTaskIsExecuted(
        self,
        timer_task: dict
    ):

        for task in self.__timer_tasks:
            if task["task_uuid"] == timer_task["task_uuid"]:
                if task.get("repeat", False):
                    self.onRepeatTimerTaskIs(ALTimerTaskStatus.EXECUTED, task)
                else:
                    task["status"] = ALTimerTaskStatus.EXECUTED
                break
        self.timerTasksChanged.emit()

    @Slot(dict)
    def onTimerTaskIsError(
        self,
        timer_task: dict
    ):

        for task in self.__timer_tasks:
            if task["task_uuid"] == timer_task["task_uuid"]:
                if task.get("repeat", False):
                    self.onRepeatTimerTaskIs(ALTimerTaskStatus.ERROR, task)
                else:
                    task["status"] = ALTimerTaskStatus.ERROR
                break
        self.timerTasksChanged.emit()