421 lines
14 KiB
Python
421 lines
14 KiB
Python
# <AUTO_MAA:A MAA Multi Account Management and Automation Tool>
|
||
# Copyright © <2024> <DLmaster361>
|
||
|
||
# This file is part of AUTO_MAA.
|
||
|
||
# AUTO_MAA is free software: you can redistribute it and/or modify
|
||
# it under the terms of the GNU General Public License as published
|
||
# by the Free Software Foundation, either version 3 of the License,
|
||
# or (at your option) any later version.
|
||
|
||
# AUTO_MAA is distributed in the hope that it will be useful,
|
||
# but WITHOUT ANY WARRANTY; without even the implied warranty
|
||
# of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See
|
||
# the GNU General Public License for more details.
|
||
|
||
# You should have received a copy of the GNU General Public License
|
||
# along with AUTO_MAA. If not, see <https://www.gnu.org/licenses/>.
|
||
|
||
# DLmaster_361@163.com
|
||
|
||
"""
|
||
AUTO_MAA
|
||
AUTO_MAA调度中枢界面
|
||
v4.2
|
||
作者：DLmaster_361
|
||
"""
|
||
|
||
from loguru import logger
|
||
from PySide6.QtWidgets import (
|
||
QWidget,
|
||
QVBoxLayout,
|
||
QStackedWidget,
|
||
QHBoxLayout,
|
||
)
|
||
from qfluentwidgets import (
|
||
CardWidget,
|
||
Pivot,
|
||
ScrollArea,
|
||
FluentIcon,
|
||
HeaderCardWidget,
|
||
FluentIcon,
|
||
TextBrowser,
|
||
ComboBox,
|
||
SubtitleLabel,
|
||
PushButton,
|
||
)
|
||
from PySide6.QtCore import Qt
|
||
from PySide6.QtGui import QTextCursor
|
||
from typing import List, Dict
|
||
import json
|
||
|
||
|
||
from app.core import Config, TaskManager, Task, MainInfoBar
|
||
from .Widget import StatefulItemCard
|
||
|
||
|
||
class DispatchCenter(QWidget):
|
||
|
||
def __init__(self, parent=None):
|
||
super().__init__(parent)
|
||
|
||
self.setObjectName("调度中枢")
|
||
|
||
self.pivot = Pivot(self)
|
||
self.stackedWidget = QStackedWidget(self)
|
||
self.Layout = QVBoxLayout(self)
|
||
|
||
self.script_list: Dict[str, DispatchBox] = {}
|
||
|
||
dispatch_box = DispatchBox("主调度台", self)
|
||
self.script_list["主调度台"] = dispatch_box
|
||
self.stackedWidget.addWidget(self.script_list["主调度台"])
|
||
self.pivot.addItem(
|
||
routeKey="主调度台",
|
||
text="主调度台",
|
||
onClick=self.update_top_bar,
|
||
icon=FluentIcon.CAFE,
|
||
)
|
||
|
||
self.Layout.addWidget(self.pivot, 0, Qt.AlignHCenter)
|
||
self.Layout.addWidget(self.stackedWidget)
|
||
self.Layout.setContentsMargins(0, 0, 0, 0)
|
||
|
||
self.pivot.currentItemChanged.connect(
|
||
lambda index: self.stackedWidget.setCurrentWidget(self.script_list[index])
|
||
)
|
||
|
||
def add_board(self, task: Task) -> None:
|
||
"""添加一个调度台界面"""
|
||
|
||
dispatch_box = DispatchBox(task.name, self)
|
||
|
||
dispatch_box.top_bar.button.clicked.connect(
|
||
lambda: TaskManager.stop_task(task.name)
|
||
)
|
||
|
||
task.create_task_list.connect(dispatch_box.info.task.create_task)
|
||
task.create_user_list.connect(dispatch_box.info.user.create_user)
|
||
task.update_task_list.connect(dispatch_box.info.task.update_task)
|
||
task.update_user_list.connect(dispatch_box.info.user.update_user)
|
||
task.update_log_text.connect(dispatch_box.info.log_text.text.setText)
|
||
task.accomplish.connect(lambda: self.del_board(f"调度台_{task.name}"))
|
||
|
||
self.script_list[f"调度台_{task.name}"] = dispatch_box
|
||
|
||
self.stackedWidget.addWidget(self.script_list[f"调度台_{task.name}"])
|
||
|
||
self.pivot.addItem(routeKey=f"调度台_{task.name}", text=f"调度台 {task.name}")
|
||
|
||
def del_board(self, name: str) -> None:
|
||
"""删除指定子界面"""
|
||
|
||
self.pivot.setCurrentItem("主调度台")
|
||
self.stackedWidget.removeWidget(self.script_list[name])
|
||
self.script_list[name].deleteLater()
|
||
self.pivot.removeWidget(name)
|
||
|
||
def connect_main_board(self, task: Task) -> None:
|
||
"""连接主调度台"""
|
||
|
||
self.script_list["主调度台"].top_bar.Lable.setText(
|
||
f"{task.name} - {task.mode.replace("_主调度台","")}模式"
|
||
)
|
||
self.script_list["主调度台"].top_bar.Lable.show()
|
||
self.script_list["主调度台"].top_bar.object.hide()
|
||
self.script_list["主调度台"].top_bar.mode.hide()
|
||
self.script_list["主调度台"].top_bar.button.clicked.disconnect()
|
||
self.script_list["主调度台"].top_bar.button.setText("中止任务")
|
||
self.script_list["主调度台"].top_bar.button.clicked.connect(
|
||
lambda: TaskManager.stop_task(task.name)
|
||
)
|
||
task.create_task_list.connect(
|
||
self.script_list["主调度台"].info.task.create_task
|
||
)
|
||
task.create_user_list.connect(
|
||
self.script_list["主调度台"].info.user.create_user
|
||
)
|
||
task.update_task_list.connect(
|
||
self.script_list["主调度台"].info.task.update_task
|
||
)
|
||
task.update_user_list.connect(
|
||
self.script_list["主调度台"].info.user.update_user
|
||
)
|
||
task.update_log_text.connect(
|
||
self.script_list["主调度台"].info.log_text.text.setText
|
||
)
|
||
task.accomplish.connect(lambda: self.disconnect_main_board(task.name))
|
||
|
||
def disconnect_main_board(self, name: str) -> None:
|
||
"""断开主调度台"""
|
||
|
||
self.script_list["主调度台"].top_bar.Lable.hide()
|
||
self.script_list["主调度台"].top_bar.object.show()
|
||
self.script_list["主调度台"].top_bar.mode.show()
|
||
self.script_list["主调度台"].top_bar.button.clicked.disconnect()
|
||
self.script_list["主调度台"].top_bar.button.setText("开始任务")
|
||
self.script_list["主调度台"].top_bar.button.clicked.connect(
|
||
self.script_list["主调度台"].top_bar.start_task
|
||
)
|
||
self.script_list["主调度台"].info.log_text.text.setText(
|
||
Config.get_history(name)["History"]
|
||
)
|
||
|
||
def update_top_bar(self):
|
||
"""更新顶栏"""
|
||
|
||
list = []
|
||
queue_numb, member_numb = 0, 0
|
||
|
||
if (Config.app_path / "config/QueueConfig").exists():
|
||
for json_file in (Config.app_path / "config/QueueConfig").glob("*.json"):
|
||
list.append(f"队列 - {json_file.stem}")
|
||
queue_numb += 1
|
||
|
||
if (Config.app_path / "config/MaaConfig").exists():
|
||
for subdir in (Config.app_path / "config/MaaConfig").iterdir():
|
||
if subdir.is_dir():
|
||
list.append(f"实例 - Maa - {subdir.name}")
|
||
member_numb += 1
|
||
|
||
self.script_list["主调度台"].top_bar.object.clear()
|
||
self.script_list["主调度台"].top_bar.object.addItems(list)
|
||
self.script_list["主调度台"].top_bar.mode.clear()
|
||
self.script_list["主调度台"].top_bar.mode.addItems(["自动代理", "人工排查"])
|
||
|
||
if queue_numb == 1:
|
||
self.script_list["主调度台"].top_bar.object.setCurrentIndex(0)
|
||
elif member_numb == 1:
|
||
self.script_list["主调度台"].top_bar.object.setCurrentIndex(queue_numb)
|
||
else:
|
||
self.script_list["主调度台"].top_bar.object.setCurrentIndex(-1)
|
||
self.script_list["主调度台"].top_bar.mode.setCurrentIndex(0)
|
||
|
||
|
||
class DispatchBox(QWidget):
|
||
|
||
def __init__(self, name: str, parent=None):
|
||
super().__init__(parent)
|
||
|
||
self.setObjectName(name)
|
||
|
||
layout = QVBoxLayout()
|
||
|
||
scrollArea = ScrollArea()
|
||
scrollArea.setWidgetResizable(True)
|
||
|
||
content_widget = QWidget()
|
||
content_layout = QVBoxLayout(content_widget)
|
||
|
||
self.top_bar = self.DispatchTopBar(self, name)
|
||
self.info = self.DispatchInfoCard(self)
|
||
|
||
content_layout.addWidget(self.top_bar)
|
||
content_layout.addWidget(self.info)
|
||
|
||
scrollArea.setWidget(content_widget)
|
||
|
||
layout.addWidget(scrollArea)
|
||
|
||
self.setLayout(layout)
|
||
|
||
class DispatchTopBar(CardWidget):
|
||
|
||
def __init__(self, parent=None, name: str = None):
|
||
super().__init__(parent)
|
||
|
||
Layout = QHBoxLayout(self)
|
||
|
||
if name == "主调度台":
|
||
|
||
self.Lable = SubtitleLabel("", self)
|
||
self.Lable.hide()
|
||
self.object = ComboBox()
|
||
self.object.setPlaceholderText("请选择调度对象")
|
||
self.mode = ComboBox()
|
||
self.mode.setPlaceholderText("请选择调度模式")
|
||
|
||
self.button = PushButton("开始任务")
|
||
self.button.clicked.connect(self.start_task)
|
||
|
||
Layout.addWidget(self.Lable)
|
||
Layout.addWidget(self.object)
|
||
Layout.addWidget(self.mode)
|
||
Layout.addStretch(1)
|
||
Layout.addWidget(self.button)
|
||
|
||
else:
|
||
|
||
self.Lable = SubtitleLabel(name, self)
|
||
self.button = PushButton("中止任务")
|
||
|
||
Layout.addWidget(self.Lable)
|
||
Layout.addStretch(1)
|
||
Layout.addWidget(self.button)
|
||
|
||
def start_task(self):
|
||
"""开始任务"""
|
||
|
||
if self.object.currentIndex() == -1:
|
||
logger.warning("未选择调度对象")
|
||
MainInfoBar.push_info_bar(
|
||
"warning", "未选择调度对象", "请选择后再开始任务", 5000
|
||
)
|
||
return None
|
||
|
||
if self.mode.currentIndex() == -1:
|
||
logger.warning("未选择调度模式")
|
||
MainInfoBar.push_info_bar(
|
||
"warning", "未选择调度模式", "请选择后再开始任务", 5000
|
||
)
|
||
return None
|
||
|
||
name = self.object.currentText().split(" - ")[-1]
|
||
|
||
if name in Config.running_list:
|
||
logger.warning(f"任务已存在：{name}")
|
||
MainInfoBar.push_info_bar("warning", "任务已存在", name, 5000)
|
||
return None
|
||
|
||
if self.object.currentText().split(" - ")[0] == "队列":
|
||
|
||
with (Config.app_path / f"config/QueueConfig/{name}.json").open(
|
||
mode="r", encoding="utf-8"
|
||
) as f:
|
||
info = json.load(f)
|
||
|
||
logger.info(f"用户添加任务：{name}")
|
||
TaskManager.add_task(f"{self.mode.currentText()}_主调度台", name, info)
|
||
|
||
elif self.object.currentText().split(" - ")[0] == "实例":
|
||
|
||
if self.object.currentText().split(" - ")[1] == "Maa":
|
||
|
||
info = {"Queue": {"Member_1": name}}
|
||
|
||
logger.info(f"用户添加任务：{name}")
|
||
TaskManager.add_task(
|
||
f"{self.mode.currentText()}_主调度台", "自定义队列", info
|
||
)
|
||
|
||
class DispatchInfoCard(HeaderCardWidget):
|
||
|
||
def __init__(self, parent=None):
|
||
super().__init__(parent)
|
||
|
||
self.setTitle("调度信息")
|
||
|
||
self.task = self.TaskInfoCard(self)
|
||
self.user = self.UserInfoCard(self)
|
||
self.log_text = self.LogCard(self)
|
||
|
||
self.viewLayout.addWidget(self.task)
|
||
self.viewLayout.addWidget(self.user)
|
||
self.viewLayout.addWidget(self.log_text)
|
||
|
||
self.viewLayout.setStretch(0, 1)
|
||
self.viewLayout.setStretch(1, 1)
|
||
self.viewLayout.setStretch(2, 5)
|
||
|
||
def update_board(self, task_list: list, user_list: list, log: str):
|
||
"""更新调度信息"""
|
||
|
||
self.task.update_task(task_list)
|
||
self.user.update_user(user_list)
|
||
self.log_text.text.setText(log)
|
||
|
||
class TaskInfoCard(HeaderCardWidget):
|
||
|
||
def __init__(self, parent=None):
|
||
super().__init__(parent)
|
||
self.setTitle("任务队列")
|
||
|
||
self.Layout = QVBoxLayout()
|
||
self.viewLayout.addLayout(self.Layout)
|
||
self.viewLayout.setContentsMargins(3, 0, 3, 3)
|
||
|
||
self.task_cards: List[StatefulItemCard] = []
|
||
|
||
def create_task(self, task_list: list):
|
||
"""创建任务队列"""
|
||
|
||
while self.Layout.count() > 0:
|
||
item = self.Layout.takeAt(0)
|
||
if item.spacerItem():
|
||
self.Layout.removeItem(item.spacerItem())
|
||
elif item.widget():
|
||
item.widget().deleteLater()
|
||
|
||
self.task_cards = []
|
||
|
||
for task in task_list:
|
||
|
||
self.task_cards.append(StatefulItemCard(task))
|
||
self.Layout.addWidget(self.task_cards[-1])
|
||
|
||
self.Layout.addStretch(1)
|
||
|
||
def update_task(self, task_list: list):
|
||
"""更新任务队列"""
|
||
|
||
for i in range(len(task_list)):
|
||
|
||
self.task_cards[i].update_status(task_list[i][1])
|
||
|
||
class UserInfoCard(HeaderCardWidget):
|
||
|
||
def __init__(self, parent=None):
|
||
super().__init__(parent)
|
||
self.setTitle("用户队列")
|
||
|
||
self.Layout = QVBoxLayout()
|
||
self.viewLayout.addLayout(self.Layout)
|
||
self.viewLayout.setContentsMargins(3, 0, 3, 3)
|
||
|
||
self.user_cards: List[StatefulItemCard] = []
|
||
|
||
def create_user(self, user_list: list):
|
||
"""创建用户队列"""
|
||
|
||
while self.Layout.count() > 0:
|
||
item = self.Layout.takeAt(0)
|
||
if item.spacerItem():
|
||
self.Layout.removeItem(item.spacerItem())
|
||
elif item.widget():
|
||
item.widget().deleteLater()
|
||
|
||
self.user_cards = []
|
||
|
||
for user in user_list:
|
||
|
||
self.user_cards.append(StatefulItemCard(user))
|
||
self.Layout.addWidget(self.user_cards[-1])
|
||
|
||
self.Layout.addStretch(1)
|
||
|
||
def update_user(self, user_list: list):
|
||
"""更新用户队列"""
|
||
|
||
for i in range(len(user_list)):
|
||
|
||
self.user_cards[i].Label.setText(user_list[i][0])
|
||
self.user_cards[i].update_status(user_list[i][1])
|
||
|
||
class LogCard(HeaderCardWidget):
|
||
|
||
def __init__(self, parent=None):
|
||
super().__init__(parent)
|
||
self.setTitle("日志")
|
||
|
||
self.text = TextBrowser()
|
||
self.viewLayout.setContentsMargins(3, 0, 3, 3)
|
||
self.viewLayout.addWidget(self.text)
|
||
|
||
self.text.textChanged.connect(self.to_end)
|
||
|
||
def to_end(self):
|
||
"""滚动到底部"""
|
||
|
||
self.text.moveCursor(QTextCursor.End)
|
||
self.text.ensureCursorVisible()
|