automas_admin/AUTO-MAS-test
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(core):初步完成主调度自动代理功能开发

Browse Source
This commit is contained in:
DLmaster
2025-01-26 07:58:33 +08:00
parent 7e08c88a3e
commit c625354dec
41 changed files with 1645 additions and 694 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
app/core/__init__.py
View File

@@ -29,17 +29,18 @@ __version__ = "4.2.0"
__author__ = "DLmaster361 <DLmaster_361@163.com>"
__license__ = "GPL-3.0 license"
from .config import AppConfig, QueueConfig, MaaConfig
from .config import AppConfig, QueueConfig, MaaConfig, Config
from .main_info_bar import MainInfoBar
from .task_manager import Task, TaskManager
from .timer import MainTimer
from .task_manager import Task, Task_manager
from .timer import Main_timer
__all__ = [
"AppConfig",
"Config",
"QueueConfig",
"MaaConfig",
"MainInfoBar",
"Task",
"TaskManager",
"MainTimer",
"Task_manager",
"Main_timer",
]

BIN
app/core/__pycache__/__init__.cpython-312.pyc
View File

Binary file not shown.

BIN
app/core/__pycache__/config.cpython-312.pyc
View File

Binary file not shown.

BIN
app/core/__pycache__/task_manager.cpython-312.pyc
View File

Binary file not shown.

BIN
app/core/__pycache__/timer.cpython-312.pyc
View File

Binary file not shown.

7
app/core/config.py
View File

@@ -84,7 +84,7 @@ class AppConfig:
"updater_version": "0.0.0.0",
}
with self.version_path.open(mode="w", encoding="utf-8") as f:
json.dump(version, f, indent=4)
json.dump(version, f, ensure_ascii=False, indent=4)
# 生成预设gameid替换方案文件
if not self.gameid_path.exists():
@@ -253,7 +253,7 @@ class AppConfig:
history = json.load(f)
history[key] = content
with self.history_path.open(mode="w", encoding="utf-8") as f:
json.dump(history, f, indent=4)
json.dump(history, f, ensure_ascii=False, indent=4)
def get_history(self, key: str) -> dict:
"""获取历史记录"""
@@ -409,3 +409,6 @@ class MaaConfig(QConfig):
RunSet_RunTimesLimit = RangeConfigItem(
"RunSet", "RunTimesLimit", 3, RangeValidator(1, 1024)
)
Config = AppConfig()

75
app/core/task_manager.py
View File

@@ -26,15 +26,13 @@ v4.2
"""
from loguru import logger
from PySide6.QtWidgets import QApplication
from PySide6 import QtCore
import json
from typing import Dict, Union, List
from .config import AppConfig
from .config import Config
from .main_info_bar import MainInfoBar
from app.models import MaaManager
from app.services import Notification
from app.services import Notify
class Task(QtCore.QThread):
@@ -51,15 +49,11 @@ class Task(QtCore.QThread):
def __init__(
self,
config: AppConfig,
notify: Notification,
name: str,
info: Dict[str, Dict[str, Union[str, int, bool]]],
):
super(Task, self).__init__()
self.config = config
self.notify = notify
self.name = name
self.info = info
@@ -84,17 +78,15 @@ class Task(QtCore.QThread):
self.task_list[i][1] = "运行"
self.update_task_list.emit(self.task_list)
if self.task_list[i][0] not in self.config.running_list:
if self.task_list[i][0] not in Config.running_list:
self.config.running_list.append(self.task_list[i][0])
Config.running_list.append(self.task_list[i][0])
logger.info(f"任务开始：{self.task_list[i][0]}")
self.push_info_bar.emit("info", "任务开始", self.task_list[i][0], 5000)
if self.member_dict[self.task_list[i][0]][0] == "Maa":
self.task = MaaManager(
self.config,
self.notify,
"自动代理",
self.member_dict[self.task_list[i][0]][1],
)
@@ -110,7 +102,7 @@ class Task(QtCore.QThread):
self.task.run()
self.config.running_list.remove(self.task_list[i][0])
Config.running_list.remove(self.task_list[i][0])
self.task_list[i][1] = "完成"
logger.info(f"任务完成：{self.task_list[i][0]}")
@@ -129,8 +121,8 @@ class Task(QtCore.QThread):
member_dict = {}
if (self.config.app_path / "config/MaaConfig").exists():
for subdir in (self.config.app_path / "config/MaaConfig").iterdir():
if (Config.app_path / "config/MaaConfig").exists():
for subdir in (Config.app_path / "config/MaaConfig").iterdir():
if subdir.is_dir():
member_dict[subdir.name] = ["Maa", subdir]
@@ -147,14 +139,12 @@ class TaskManager(QtCore.QObject):
"""业务调度器"""
create_gui = QtCore.Signal(Task)
connect_gui = QtCore.Signal(Task)
push_info_bar = QtCore.Signal(str, str, str, int)
def __init__(self, config: AppConfig, notify: Notification):
def __init__(self):
super(TaskManager, self).__init__()
self.config = config
self.notify = notify
self.task_list: Dict[str, Task] = {}
def add_task(
@@ -162,29 +152,35 @@ class TaskManager(QtCore.QObject):
):
"""添加任务"""
if (
mode == "运行队列"
and name not in self.config.running_list
and name not in self.task_list
):
if name in Config.running_list or name in self.task_list:
logger.info(f"任务开始：{name}")
MainInfoBar.push_info_bar("info", "任务开始", name, 5000)
logger.warning(f"任务已存在：{name}")
MainInfoBar.push_info_bar("warning", "任务已存在", name, 5000)
return None
self.config.running_list.append(name)
self.task_list[name] = Task(self.config, self.notify, name, info)
self.task_list[name].push_info_bar.connect(MainInfoBar.push_info_bar)
self.task_list[name].accomplish.connect(
lambda logs: self.remove_task(name, logs)
)
logger.info(f"任务开始：{name}")
MainInfoBar.push_info_bar("info", "任务开始", name, 3000)
Config.running_list.append(name)
self.task_list[name] = Task(name, info)
self.task_list[name].push_info_bar.connect(MainInfoBar.push_info_bar)
self.task_list[name].accomplish.connect(
lambda logs: self.remove_task(name, logs)
)
if mode == "新窗口":
self.create_gui.emit(self.task_list[name])
self.task_list[name].start()
elif mode == "主窗口":
self.connect_gui.emit(self.task_list[name])
self.task_list[name].start()
def stop_task(self, name: str):
"""中止任务"""
logger.info(f"中止任务：{name}")
MainInfoBar.push_info_bar("info", "中止任务", name, 5000)
MainInfoBar.push_info_bar("info", "中止任务", name, 3000)
if name == "ALL":
@@ -206,17 +202,20 @@ class TaskManager(QtCore.QObject):
"""移除任务标记"""
logger.info(f"任务结束：{name}")
MainInfoBar.push_info_bar("info", "任务结束", name, 5000)
MainInfoBar.push_info_bar("info", "任务结束", name, 3000)
if len(logs) > 0:
time = logs[0][1]["Time"]
history = ""
for log in logs:
self.config.save_history(log[0], log[1])
Config.save_history(log[0], log[1])
history += (
f"任务名称：{log[0]}，{log[1]["History"].replace("\n","\n ")}\n"
)
self.config.save_history(name, {"Time": time, "History": history})
Config.save_history(name, {"Time": time, "History": history})
self.task_list.pop(name)
self.config.running_list.remove(name)
Config.running_list.remove(name)
Task_manager = TaskManager()

32
app/core/timer.py
View File

@@ -31,26 +31,19 @@ from PySide6 import QtCore
import json
import datetime
from .config import AppConfig
from .task_manager import TaskManager
from app.services import SystemHandler
from .config import Config
from .task_manager import Task_manager
from app.services import System
class MainTimer(QWidget):
def __init__(
self,
config: AppConfig,
system: SystemHandler,
task_manager: TaskManager,
parent=None,
):
super().__init__(parent)
self.config = config
self.system = system
self.task_manager = task_manager
self.Timer = QtCore.QTimer()
self.Timer.timeout.connect(self.timed_start)
self.Timer.timeout.connect(self.set_silence)
@@ -69,7 +62,7 @@ class MainTimer(QWidget):
if not info["QueueSet"]["Enabled"]:
continue
history = self.config.get_history(name)
history = Config.get_history(name)
time_set = [
info["Time"][f"TimeSet_{_}"]
@@ -81,22 +74,22 @@ class MainTimer(QWidget):
if (
curtime[11:16] in time_set
and curtime != history["Time"][:16]
and name not in self.config.running_list
and name not in Config.running_list
):
logger.info(f"按时间调起任务：{name}")
self.task_manager.add_task("运行队列", name, info)
Task_manager.add_task("新窗口", name, info)
def set_silence(self):
"""设置静默模式"""
# # 临时
# windows = self.system.get_window_info()
# windows = System.get_window_info()
# if any(emulator_path in _ for _ in windows):
# try:
# pyautogui.hotkey(*boss_key)
# except pyautogui.FailSafeException as e:
# 执行日志记录，暂时缺省
logger.debug(self.config.running_list)
logger.debug(Config.running_list)
def set_last_time(self):
"""设置上次运行时间"""
@@ -108,12 +101,13 @@ class MainTimer(QWidget):
queue_list = []
if (self.config.app_path / "config/QueueConfig").exists():
for json_file in (self.config.app_path / "config/QueueConfig").glob(
"*.json"
):
if (Config.app_path / "config/QueueConfig").exists():
for json_file in (Config.app_path / "config/QueueConfig").glob("*.json"):
with json_file.open("r", encoding="utf-8") as f:
info = json.load(f)
queue_list.append([json_file.stem, info])
return queue_list
Main_timer = MainTimer()