Compare commits

...

56 Commits

Author SHA1 Message Date
DLmaster
54b697f2ee Merge branch 'dev' 2025-03-07 11:56:54 +08:00
DLmaster
70df428825 fix(rescourse): 修复统计信息HTML模板公招匹配错误 2025-03-07 11:56:28 +08:00
DLmaster
8993d66056 Merge branch 'dev' 2025-03-07 01:19:18 +08:00
DLmaster
863e6fb25e Merge branch 'notice_dev' into dev 2025-03-06 23:43:10 +08:00
DLmaster
181856173e feat(core): 调整获取主题图像的时机 2025-03-06 23:42:59 +08:00
DLmaster
576fe59bbc fix(models): 修复任务被手动中止项目无法被正常录入的问题 2025-03-06 23:37:51 +08:00
DLmaster
c73aca71f7 feat(core): 优化通知服务
- 优化相关设置项排布
- 邮件添加HTML模板
2025-03-06 23:02:23 +08:00
DLmaster
ce264de963 feat(ui): 6星公招通知添加独立设置项 2025-03-05 16:00:48 +08:00
DLmaster
1feb0cf83f fix(serices): 修复通知服务使用InfoBar报错时程序崩溃问题 2025-03-05 15:11:57 +08:00
DLmaster
6292624d41 fix(services): 添加邮箱通知校验过程 2025-03-05 14:34:15 +08:00
4271a07f03 fix: 修复企业微信群机器人推送通知消息未正确换行 2025-03-05 14:07:03 +08:00
DLmaster
254fb6916f feat(core): 初步完成六星公招通知 2025-03-03 21:59:37 +08:00
DLmaster
21857325a2 Merge branch 'notice_dev' into dev 2025-03-03 20:41:17 +08:00
DLmaster
175d6860a3 feat(core): 添加统计信息通知功能 2025-03-03 20:40:28 +08:00
DLmaster
d1c8f98408 feat(ui): 主页添加主题活动信息 2025-03-02 17:14:09 +08:00
DLmaster
3499fa9067 feat(utils): 更新器拥有多网址测速功能 2025-03-02 15:32:56 +08:00
DLmaster
cca2cd774c chore(release): 发版 v4.2.4-beta.6 2025-03-01 22:53:08 +08:00
DLmaster
6d60f8adb8 Merge branch 'gui_dev' into dev 2025-03-01 22:51:30 +08:00
DLmaster
3b406a7974 feat(gui): 主页功能完善 2025-03-01 22:49:51 +08:00
a116b3359c fix: 修复掉落统计正则表达式错误统计时间的问题 2025-02-27 16:04:45 +08:00
928019390b chore(release): 发版 v4.2.4-beta.5 2025-02-26 21:58:29 +08:00
022b698f54 fix: 添加剿灭模式特殊适配 2025-02-26 16:13:46 +08:00
0228ac8393 fix: 修复 RMA70-12 匹配正则表达式 2025-02-26 16:12:42 +08:00
DLmaster
a99f381f7f fix(ui): 换个正常的主页 2025-02-22 00:22:57 +08:00
DLmaster
7c0af24bf5 fix(ui): 修正部分主页网址 2025-02-21 21:39:16 +08:00
DLmaster
d3aa45cfb9 Merge branch 'DLMS_dev' into dev 2025-02-21 18:36:32 +08:00
DLmaster
f5461deb81 feat(ui): 软件主页初步完成 2025-02-21 18:35:55 +08:00
DLmaster
c19068128f feat(core): 添加启动时直接最小化功能 2025-02-21 16:15:55 +08:00
DLmaster
1367daf1b7 feat(ui): 添加软件临时主页 2025-02-21 15:54:28 +08:00
DLmaster
5fc6e74cd6 Merge branch 'log_dev' into dev 2025-02-21 12:05:08 +08:00
DLmaster
5d7227c009 feat(ui): 初步完成历史记录前端适配 2025-02-21 12:04:52 +08:00
3a9c670172 feat(core): 新增日志管理功能
- 在配置文件中添加日志保存和保留天数设置项
- 实现日志保存功能,每次运行后保存日志到指定目录
- 添加日志分析功能,掉落信息并保存为 JSON 文件
- 在设置界面新增日志管理相关配置选项

todo: 日志清理可能有问题、多账号日志可能会保存为上一个账号的日志(加了time.sleep还没测)
2025-02-18 17:29:13 +08:00
DLmaster
2768faed53 fix(core): 修正更新器方法;取消MAA运行中自动更新;补充MAA监测字段 2025-02-18 16:17:03 +08:00
DLmaster
85f3d6f09f fix(core): 修改下载器默认下载目录为/script 2025-02-17 22:54:45 +08:00
DLmaster
c99707ecb4 Merge branch 'dev' 2025-02-17 18:09:42 +08:00
DLmaster
2b8e648fe6 Merge branch 'DLMS_dev' into dev 2025-02-17 18:08:45 +08:00
DLmaster
fcf61fd39a feat(core): 接入镜像源 2025-02-17 18:06:05 +08:00
DLmaster
8e3026f91e ci(build): 激活测试流程 2025-02-17 14:27:06 +08:00
DLmaster
8e00676faf ci(build): 测试服务器上传功能 2025-02-17 14:20:23 +08:00
DLmaster
ae293c4c20 Merge pull request #28 from ClozyA:dev
ci(build): 增加预发布版本服务器上传
2025-02-17 13:28:53 +08:00
df4a5f3318 ci(build): 增加预发布版本服务器上传 2025-02-17 12:02:03 +08:00
DLmaster
1da96c4d1d fix(ui): 矫正老板键文案 2025-02-17 11:06:07 +08:00
DLmaster
144c7f5db7 feat(modles): 配置MAA前关闭可能未正常退出的MAA进程 2025-02-16 18:44:46 +08:00
DLmaster
b3a3ccfea3 feat(core): 恢复启动后直接运行主任务功能以及相关托盘菜单 2025-02-16 14:56:25 +08:00
DLmaster
c3212f0ca1 Merge branch 'DLMS_dev' into dev 2025-02-15 17:32:47 +08:00
DLmaster
a946999782 fix: 修正版本号 2025-02-15 17:32:33 +08:00
DLmaster
284c41feb7 fix(ui): 适配深色模式 2025-02-15 17:28:54 +08:00
DLmaster
ddf905cb13 docs: 文档站试运行 2025-02-15 15:55:25 +08:00
DLmaster
d5082d18ef fix: 更正版本描述 2025-02-14 23:21:38 +08:00
DLmaster
af51831062 Merge branch 'DLMS_dev' into dev 2025-02-14 16:51:04 +08:00
DLmaster
fe4707df84 feat(ui): 优化主调度台默认选项 2025-02-14 16:50:45 +08:00
DLmaster
292e7f51e0 chore(core): 升级日志监看方法 2025-02-14 16:27:18 +08:00
DLmaster
8936b1c41d fix(core): 修复部分异常;添加高级代理文件校验过程 2025-02-13 13:03:44 +08:00
DLmaster
d45da439bd chore(models): 优化MAA关闭方法 2025-02-12 22:15:17 +08:00
DLmaster
7dc057e30f feat(models): 简洁用户列表下相邻两个任务间的切换方式 2025-02-12 22:06:04 +08:00
DLmaster
eb2f9d2cea fix(models): 修复静默代理标记移除异常情况 2025-02-11 13:47:12 +08:00
28 changed files with 2786 additions and 530 deletions

View File

@@ -150,4 +150,13 @@ jobs:
gh release delete "$TAGNAME" --yes
gh release create "$TAGNAME" --target "main" --title "$NAME" --notes "$NOTES" artifacts/*
env:
GITHUB_TOKEN: ${{ secrets.WORKFLOW_TOKEN }}
GITHUB_TOKEN: ${{ secrets.WORKFLOW_TOKEN }}
- name: Setup SSH Key
run: |
mkdir -p ~/.ssh
echo "${{ secrets.SSH_PRIVATE_KEY }}" > ~/.ssh/id_rsa
chmod 600 ~/.ssh/id_rsa
ssh-keyscan -H ${{ secrets.SERVER_IP }} >> ~/.ssh/known_hosts
- name: Upload Release to Server
run: |
scp -r artifacts/* ${{ secrets.SERVER_USER }}@${{ secrets.SERVER_IP }}:/home/user/files/AUTO_MAA/

View File

@@ -150,4 +150,13 @@ jobs:
gh release delete "$TAGNAME" --yes
gh release create "$TAGNAME" --target "main" --title "$NAME" --notes "$NOTES" --prerelease artifacts/*
env:
GITHUB_TOKEN: ${{ secrets.WORKFLOW_TOKEN }}
GITHUB_TOKEN: ${{ secrets.WORKFLOW_TOKEN }}
- name: Setup SSH Key
run: |
mkdir -p ~/.ssh
echo "${{ secrets.SSH_PRIVATE_KEY }}" > ~/.ssh/id_rsa
chmod 600 ~/.ssh/id_rsa
ssh-keyscan -H ${{ secrets.SERVER_IP }} >> ~/.ssh/known_hosts
- name: Upload Release to Server
run: |
scp -r artifacts/* ${{ secrets.SERVER_USER }}@${{ secrets.SERVER_IP }}:/home/user/files/AUTO_MAA/

View File

@@ -58,9 +58,9 @@ MAA多账号管理与自动化软件
# 使用方法
本项目已改用腾讯文档展示使用方法
访问AUTO_MAA官方文档站以获取使用指南和项目相关信息
- [AUTO_MAA用户指南》](https://docs.qq.com/aio/DQ2NwUHRiWGtMWHBy)
- [AUTO_MAA官方文档站](https://clozya.github.io/AUTOMAA_docs)
---
@@ -82,6 +82,8 @@ MAA多账号管理与自动化软件
![Alt](https://repobeats.axiom.co/api/embed/6c2f834141eff1ac297db70d12bd11c6236a58a5.svg "Repobeats analytics image")
感谢 [AoXuan (@ClozyA)](https://github.com/ClozyA) 为本项目提供的下载服务器
## Star History
[![Star History Chart](https://api.star-history.com/svg?repos=DLmaster361/AUTO_MAA&type=Date)](https://star-history.com/#DLmaster361/AUTO_MAA&Date)

View File

@@ -29,7 +29,7 @@ __version__ = "4.2.0"
__author__ = "DLmaster361 <DLmaster_361@163.com>"
__license__ = "GPL-3.0 license"
from .core import AppConfig, QueueConfig, MaaConfig, Task, Task_manager, Main_timer
from .core import AppConfig, QueueConfig, MaaConfig, Task, TaskManager, MainTimer
from .models import MaaManager
from .services import Notify, Crypto, System
from .ui import AUTO_MAA
@@ -40,8 +40,8 @@ __all__ = [
"QueueConfig",
"MaaConfig",
"Task",
"Task_manager",
"Main_timer",
"TaskManager",
"MainTimer",
"MaaManager",
"Notify",
"Crypto",

View File

@@ -31,8 +31,8 @@ __license__ = "GPL-3.0 license"
from .config import AppConfig, QueueConfig, MaaConfig, Config
from .main_info_bar import MainInfoBar
from .task_manager import Task, Task_manager
from .timer import Main_timer
from .task_manager import Task, TaskManager
from .timer import MainTimer
__all__ = [
"AppConfig",
@@ -41,6 +41,6 @@ __all__ = [
"MaaConfig",
"MainInfoBar",
"Task",
"Task_manager",
"Main_timer",
"TaskManager",
"MainTimer",
]

View File

@@ -30,6 +30,9 @@ import sqlite3
import json
import sys
import shutil
import re
from datetime import datetime
from collections import defaultdict
from pathlib import Path
from qfluentwidgets import (
QConfig,
@@ -42,6 +45,7 @@ from qfluentwidgets import (
OptionsValidator,
qconfig,
)
from typing import Union, Dict, List, Tuple
class AppConfig:
@@ -54,7 +58,7 @@ class AppConfig:
self.log_path = self.app_path / "debug/AUTO_MAA.log"
self.database_path = self.app_path / "data/data.db"
self.config_path = self.app_path / "config/config.json"
self.history_path = self.app_path / "config/history.json"
self.history_path = self.app_path / "history/main.json"
self.key_path = self.app_path / "data/key"
self.gameid_path = self.app_path / "data/gameid.txt"
self.version_path = self.app_path / "resources/version.json"
@@ -74,6 +78,7 @@ class AppConfig:
(self.app_path / "config").mkdir(parents=True, exist_ok=True)
(self.app_path / "data").mkdir(parents=True, exist_ok=True)
(self.app_path / "debug").mkdir(parents=True, exist_ok=True)
(self.app_path / "history").mkdir(parents=True, exist_ok=True)
# 生成版本信息文件
if not self.version_path.exists():
@@ -461,6 +466,224 @@ class AppConfig:
cur.close()
db.close()
def save_maa_log(self, log_path: Path, logs: list, maa_result: str) -> bool:
"""保存MAA日志"""
data: Dict[str, Union[str, Dict[str, Union[int, dict]]]] = {
"recruit_statistics": defaultdict(int),
"drop_statistics": defaultdict(dict),
"maa_result": maa_result,
}
if_six_star = False
# 公招统计(仅统计招募到的)
confirmed_recruit = False
current_star_level = None
i = 0
while i < len(logs):
if "公招识别结果:" in logs[i]:
current_star_level = None # 每次识别公招时清空之前的星级
i += 1
while i < len(logs) and "Tags" not in logs[i]: # 读取所有公招标签
i += 1
if i < len(logs) and "Tags" in logs[i]: # 识别星级
star_match = re.search(r"(\d+)\s*★ Tags", logs[i])
if star_match:
current_star_level = f"{star_match.group(1)}"
if current_star_level == "6★":
if_six_star = True
if "已确认招募" in logs[i]: # 只有确认招募后才统计
confirmed_recruit = True
if confirmed_recruit and current_star_level:
data["recruit_statistics"][current_star_level] += 1
confirmed_recruit = False # 重置,等待下一次公招
current_star_level = None # 清空已处理的星级
i += 1
# 掉落统计
current_stage = None
stage_drops = {}
for i, line in enumerate(logs):
drop_match = re.search(r"([A-Za-z0-9\-]+) 掉落统计:", line)
if drop_match:
# 发现新关卡,保存前一个关卡数据
if current_stage and stage_drops:
data["drop_statistics"][current_stage] = stage_drops
current_stage = drop_match.group(1)
if current_stage == "WE":
current_stage = "剿灭模式"
stage_drops = {}
continue
if current_stage:
item_match: List[str] = re.findall(
r"^(?!\[)([\u4e00-\u9fa5A-Za-z0-9\-]+)\s*:\s*([\d,]+)(?:\s*\(\+[\d,]+\))?",
line,
re.M,
)
for item, total in item_match:
# 解析数值时去掉逗号 (如 2,160 -> 2160
total = int(total.replace(",", ""))
# 黑名单
if item not in [
"当前次数",
"理智",
"最快截图耗时",
"专精等级",
"剩余时间",
]:
stage_drops[item] = total
# 处理最后一个关卡的掉落数据
if current_stage and stage_drops:
data["drop_statistics"][current_stage] = stage_drops
# 保存日志
log_path.parent.mkdir(parents=True, exist_ok=True)
with log_path.open("w", encoding="utf-8") as f:
f.writelines(logs)
with log_path.with_suffix(".json").open("w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=4)
logger.info(f"处理完成:{log_path}")
self.merge_maa_logs("所有项", log_path.parent)
return if_six_star
def merge_maa_logs(self, mode: str, logs_path: Union[Path, List[Path]]) -> dict:
"""合并指定数据统计信息文件"""
data = {
"recruit_statistics": defaultdict(int),
"drop_statistics": defaultdict(dict),
"maa_result": defaultdict(str),
}
if mode == "所有项":
logs_path_list = list(logs_path.glob("*.json"))
elif mode == "指定项":
logs_path_list = logs_path
for json_file in logs_path_list:
with json_file.open("r", encoding="utf-8") as f:
single_data: Dict[str, Union[str, Dict[str, Union[int, dict]]]] = (
json.load(f)
)
# 合并公招统计
for star_level, count in single_data["recruit_statistics"].items():
data["recruit_statistics"][star_level] += count
# 合并掉落统计
for stage, drops in single_data["drop_statistics"].items():
if stage not in data["drop_statistics"]:
data["drop_statistics"][stage] = {} # 初始化关卡
for item, count in drops.items():
if item in data["drop_statistics"][stage]:
data["drop_statistics"][stage][item] += count
else:
data["drop_statistics"][stage][item] = count
# 合并MAA结果
data["maa_result"][
json_file.name.replace(".json", "").replace("-", ":")
] = single_data["maa_result"]
# 生成汇总 JSON 文件
if mode == "所有项":
with logs_path.with_suffix(".json").open("w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=4)
logger.info(f"统计完成:{logs_path.with_suffix(".json")}")
return data
def load_maa_logs(
self, mode: str, json_path: Path
) -> Dict[str, Union[str, list, Dict[str, list]]]:
"""加载MAA日志统计信息"""
if mode == "总览":
with json_path.open("r", encoding="utf-8") as f:
info: Dict[str, Dict[str, Union[int, dict]]] = json.load(f)
data = {}
data["条目索引"] = [
[k, "完成" if v == "Success!" else "异常"]
for k, v in info["maa_result"].items()
]
data["条目索引"].insert(0, ["数据总览", "运行"])
data["统计数据"] = {"公招统计": list(info["recruit_statistics"].items())}
for game_id, drops in info["drop_statistics"].items():
data["统计数据"][f"掉落统计:{game_id}"] = list(drops.items())
data["统计数据"]["报错汇总"] = [
[k, v] for k, v in info["maa_result"].items() if v != "Success!"
]
elif mode == "单项":
with json_path.open("r", encoding="utf-8") as f:
info: Dict[str, Union[str, Dict[str, Union[int, dict]]]] = json.load(f)
data = {}
data["统计数据"] = {"公招统计": list(info["recruit_statistics"].items())}
for game_id, drops in info["drop_statistics"].items():
data["统计数据"][f"掉落统计:{game_id}"] = list(drops.items())
with json_path.with_suffix(".log").open("r", encoding="utf-8") as f:
log = f.read()
data["日志信息"] = log
return data
def search_history(self) -> dict:
"""搜索所有历史记录"""
history_dict = {}
for date_folder in (Config.app_path / "history").iterdir():
if not date_folder.is_dir():
continue # 只处理日期文件夹
try:
date = datetime.strptime(date_folder.name, "%Y-%m-%d")
history_dict[date.strftime("%Y年 %m月 %d")] = list(
date_folder.glob("*.json")
)
except ValueError:
logger.warning(f"非日期格式的目录: {date_folder}")
return {
k: v
for k, v in sorted(
history_dict.items(),
key=lambda x: datetime.strptime(x[0], "%Y年 %m月 %d"),
reverse=True,
)
}
def save_history(self, key: str, content: dict) -> None:
"""保存历史记录"""
@@ -488,6 +711,7 @@ class AppConfig:
self.maa_config.set(self.maa_config.MaaSet_Name, "")
self.maa_config.set(self.maa_config.MaaSet_Path, ".")
self.maa_config.set(self.maa_config.RunSet_TaskTransitionMethod, "ExitEmulator")
self.maa_config.set(self.maa_config.RunSet_ProxyTimesLimit, 0)
self.maa_config.set(self.maa_config.RunSet_AnnihilationTimeLimit, 40)
self.maa_config.set(self.maa_config.RunSet_RoutineTimeLimit, 10)
@@ -539,6 +763,15 @@ class AppConfig:
class GlobalConfig(QConfig):
"""全局配置"""
function_HomeImageMode = OptionsConfigItem(
"Function",
"HomeImageMode",
"默认",
OptionsValidator(["默认", "自定义", "主题图像"]),
)
function_HistoryRetentionTime = OptionsConfigItem(
"Function", "HistoryRetentionTime", 0, OptionsValidator([7, 15, 30, 60, 0])
)
function_IfAllowSleep = ConfigItem(
"Function", "IfAllowSleep", False, BoolValidator()
)
@@ -550,6 +783,9 @@ class GlobalConfig(QConfig):
start_IfSelfStart = ConfigItem("Start", "IfSelfStart", False, BoolValidator())
start_IfRunDirectly = ConfigItem("Start", "IfRunDirectly", False, BoolValidator())
start_IfMinimizeDirectly = ConfigItem(
"Start", "IfMinimizeDirectly", False, BoolValidator()
)
ui_IfShowTray = ConfigItem("UI", "IfShowTray", False, BoolValidator())
ui_IfToTray = ConfigItem("UI", "IfToTray", False, BoolValidator())
@@ -557,11 +793,18 @@ class GlobalConfig(QConfig):
ui_location = ConfigItem("UI", "location", "100x100")
ui_maximized = ConfigItem("UI", "maximized", False, BoolValidator())
notify_SendTaskResultTime = OptionsConfigItem(
"Notify",
"SendTaskResultTime",
"不推送",
OptionsValidator(["不推送", "任何时刻", "仅失败时"]),
)
notify_IfSendStatistic = ConfigItem(
"Notify", "IfSendStatistic", False, BoolValidator()
)
notify_IfSendSixStar = ConfigItem("Notify", "IfSendSixStar", False, BoolValidator())
notify_IfPushPlyer = ConfigItem("Notify", "IfPushPlyer", False, BoolValidator())
notify_IfSendMail = ConfigItem("Notify", "IfSendMail", False, BoolValidator())
notify_IfSendErrorOnly = ConfigItem(
"Notify", "IfSendErrorOnly", False, BoolValidator()
)
notify_SMTPServerAddress = ConfigItem("Notify", "SMTPServerAddress", "")
notify_AuthorizationCode = ConfigItem("Notify", "AuthorizationCode", "")
notify_FromAddress = ConfigItem("Notify", "FromAddress", "")
@@ -643,6 +886,12 @@ class MaaConfig(QConfig):
MaaSet_Name = ConfigItem("MaaSet", "Name", "")
MaaSet_Path = ConfigItem("MaaSet", "Path", ".", FolderValidator())
RunSet_TaskTransitionMethod = OptionsConfigItem(
"RunSet",
"TaskTransitionMethod",
"ExitEmulator",
OptionsValidator(["NoAction", "ExitGame", "ExitEmulator"]),
)
RunSet_ProxyTimesLimit = RangeConfigItem(
"RunSet", "ProxyTimesLimit", 0, RangeValidator(0, 1024)
)

View File

@@ -147,7 +147,7 @@ class Task(QThread):
)
)
self.task.accomplish.connect(
lambda log: self.save_log(self.task_dict[i][0], log)
lambda log: self.task_accomplish(self.task_dict[i][0], log)
)
self.task.run()
@@ -173,13 +173,14 @@ class Task(QThread):
return member_dict
def save_log(self, name: str, log: dict):
def task_accomplish(self, name: str, log: dict):
"""保存保存任务结果"""
self.logs.append([name, log])
self.task.deleteLater()
class TaskManager(QObject):
class _TaskManager(QObject):
"""业务调度器"""
create_gui = Signal(Task)
@@ -187,7 +188,7 @@ class TaskManager(QObject):
push_info_bar = Signal(str, str, str, int)
def __init__(self):
super(TaskManager, self).__init__()
super(_TaskManager, self).__init__()
self.task_dict: Dict[str, Task] = {}
@@ -252,6 +253,8 @@ class TaskManager(QObject):
logger.info(f"任务结束:{name}")
MainInfoBar.push_info_bar("info", "任务结束", name, 3000)
self.task_dict[name].deleteLater()
if len(logs) > 0:
time = logs[0][1]["Time"]
history = ""
@@ -290,4 +293,4 @@ class TaskManager(QObject):
self.task_dict[name].question_response.emit(bool(choice.exec_()))
Task_manager = TaskManager()
TaskManager = _TaskManager()

View File

@@ -33,16 +33,13 @@ from datetime import datetime
import pyautogui
from .config import Config
from .task_manager import Task_manager
from .task_manager import TaskManager
from app.services import System
class MainTimer(QWidget):
class _MainTimer(QWidget):
def __init__(
self,
parent=None,
):
def __init__(self, parent=None):
super().__init__(parent)
self.if_FailSafeException = False
@@ -81,7 +78,7 @@ class MainTimer(QWidget):
):
logger.info(f"定时任务:{name}")
Task_manager.add_task("自动代理_新调度台", name, info)
TaskManager.add_task("自动代理_新调度台", name, info)
def set_silence(self):
"""设置静默模式"""
@@ -125,4 +122,4 @@ class MainTimer(QWidget):
return queue_list
Main_timer = MainTimer()
MainTimer = _MainTimer()

View File

@@ -26,7 +26,7 @@ v4.2
"""
from loguru import logger
from PySide6.QtCore import QObject, Signal, QEventLoop
from PySide6.QtCore import QObject, Signal, QEventLoop, QFileSystemWatcher, QTimer
import json
import sqlite3
from datetime import datetime, timedelta
@@ -34,10 +34,11 @@ import subprocess
import shutil
import time
from pathlib import Path
from typing import List
from jinja2 import Environment, FileSystemLoader
from typing import Union, List
from app.core import Config
from app.services import Notify
from app.services import Notify, System
class MaaManager(QObject):
@@ -50,6 +51,7 @@ class MaaManager(QObject):
create_user_list = Signal(list)
update_user_list = Signal(list)
update_log_text = Signal(str)
interrupt = Signal()
accomplish = Signal(dict)
isInterruptionRequested = False
@@ -65,6 +67,12 @@ class MaaManager(QObject):
self.mode = mode
self.config_path = config_path
self.user_config_path = user_config_path
self.log_monitor = QFileSystemWatcher()
self.log_monitor_timer = QTimer()
self.log_monitor_timer.timeout.connect(self.refresh_maa_log)
self.monitor_loop = QEventLoop()
self.interrupt.connect(self.quit_monitor)
with (self.config_path / "config.json").open("r", encoding="utf-8") as f:
self.set = json.load(f)
@@ -85,6 +93,7 @@ class MaaManager(QObject):
def configure(self):
"""提取配置信息"""
self.name = self.set["MaaSet"]["Name"]
self.maa_root_path = Path(self.set["MaaSet"]["Path"])
self.maa_set_path = self.maa_root_path / "config/gui.json"
self.maa_log_path = self.maa_root_path / "debug/gui.log"
@@ -117,25 +126,27 @@ class MaaManager(QObject):
if "设置MAA" not in self.mode:
self.data = sorted(self.data, key=lambda x: (-len(x[15]), x[16]))
user_list: List[List[str, str, int]] = [
self.user_list: List[List[str, str, int]] = [
[_[0], "等待", index]
for index, _ in enumerate(self.data)
if (_[3] != 0 and _[4] == "y")
]
self.create_user_list.emit(user_list)
self.create_user_list.emit(self.user_list)
# 自动代理模式
if self.mode == "自动代理":
# 标记是否需要重启模拟器
self.if_open_emulator = True
# 执行情况预处理
for _ in user_list:
for _ in self.user_list:
if self.data[_[2]][5] != curdate:
self.data[_[2]][5] = curdate
self.data[_[2]][14] = 0
_[0] += f" - 第{self.data[_[2]][14] + 1}次代理"
# 开始代理
for user in user_list:
for user in self.user_list:
if self.isInterruptionRequested:
break
@@ -145,12 +156,14 @@ class MaaManager(QObject):
or self.data[user[2]][14] < self.set["RunSet"]["ProxyTimesLimit"]
):
user[1] = "运行"
self.update_user_list.emit(user_list)
self.update_user_list.emit(self.user_list)
else:
user[1] = "跳过"
self.update_user_list.emit(user_list)
self.update_user_list.emit(self.user_list)
continue
logger.info(f"{self.name} | 开始代理用户: {user[0]}")
# 初始化代理情况记录和模式替换记录
run_book = [False for _ in range(2)]
mode_book = ["自动代理_剿灭", "自动代理_日常"]
@@ -158,6 +171,14 @@ class MaaManager(QObject):
# 简洁模式用户默认开启日常选项
if self.data[user[2]][15] == "simple":
self.data[user[2]][9] = "y"
elif self.data[user[2]][15] == "beta":
check_book = [
[True, "annihilation", "剿灭"],
[True, "routine", "日常"],
]
user_logs_list = []
user_start_time = datetime.now()
# 尝试次数循环
for i in range(self.set["RunSet"]["RunTimesLimit"]):
@@ -165,6 +186,10 @@ class MaaManager(QObject):
if self.isInterruptionRequested:
break
logger.info(
f"{self.name} | 用户: {user[0]} - 尝试次数: {i + 1}/{self.set["RunSet"]["RunTimesLimit"]}"
)
# 剿灭-日常模式循环
for j in range(2):
@@ -177,6 +202,35 @@ class MaaManager(QObject):
if run_book[j]:
continue
logger.info(
f"{self.name} | 用户: {user[0]} - 模式: {mode_book[j]}"
)
if self.data[user[2]][15] == "beta":
self.if_open_emulator = True
if (
check_book[j][0]
and not (
self.config_path
/ f"beta/{self.data[user[2]][16]}/{check_book[j][1]}/gui.json"
).exists()
):
logger.error(
f"{self.name} | 用户: {user[0]} - 未找到{check_book[j][2]}配置文件"
)
self.push_info_bar.emit(
"error",
"启动MAA代理进程失败",
f"未找到{user[0]}{check_book[j][2]}配置文件!",
-1,
)
check_book[j][0] = False
continue
elif not check_book[j][0]:
continue
# 配置MAA
self.set_maa(mode_book[j], user[2])
# 记录当前时间
@@ -194,95 +248,73 @@ class MaaManager(QObject):
set["Configurations"]["Default"]["Start.EmulatorPath"]
)
Config.silence_list.append(self.emulator_path)
# 记录是否超时的标记
self.if_time_out = False
# 监测MAA运行状态
while not self.isInterruptionRequested:
self.start_monitor(start_time, mode_book[j])
# 获取MAA日志
logs = self.get_maa_log(start_time)
# 判断是否超时
if len(logs) > 0:
latest_time = datetime.now()
for _ in range(-1, 0 - len(logs) - 1, -1):
try:
latest_time = datetime.strptime(
logs[_][1:20], "%Y-%m-%d %H:%M:%S"
)
break
except ValueError:
pass
now_time = datetime.now()
if (
j == 0
and now_time - latest_time
> timedelta(
minutes=self.set["RunSet"][
"AnnihilationTimeLimit"
]
)
) or (
j == 1
and now_time - latest_time
> timedelta(
minutes=self.set["RunSet"]["RoutineTimeLimit"]
)
):
self.if_time_out = True
# 合并日志
log = "".join(logs)
# 更新MAA日志
if len(logs) > 100:
self.update_log_text.emit("".join(logs[-100:]))
else:
self.update_log_text.emit("".join(logs))
# 判断MAA程序运行状态
result = self.if_maa_success(log, mode_book[j])
if result == "Success!":
run_book[j] = True
self.update_log_text.emit(
"检测到MAA进程完成代理任务\n正在等待相关程序结束\n请等待10s"
)
# 移除静默进程标记
Config.silence_list.remove(self.emulator_path)
for _ in range(10):
if self.isInterruptionRequested:
break
time.sleep(1)
break
elif result == "Wait":
# 检测时间间隔
if self.maa_result == "Success!":
logger.info(
f"{self.name} | 用户: {user[0]} - MAA进程完成代理任务"
)
run_book[j] = True
self.update_log_text.emit(
"检测到MAA进程完成代理任务\n正在等待相关程序结束\n请等待10s"
)
for _ in range(10):
if self.isInterruptionRequested:
break
time.sleep(1)
else:
# 打印中止信息
# 此时log变量内存储的就是出现异常的日志信息可以保存或发送用于问题排查
self.update_log_text.emit(result)
# 无命令行中止MAA与其子程序
killprocess = subprocess.Popen(
f"taskkill /F /T /PID {maa.pid}",
shell=True,
creationflags=subprocess.CREATE_NO_WINDOW,
)
killprocess.wait()
# 移除静默进程标记
Config.silence_list.remove(self.emulator_path)
# 推送异常通知
Notify.push_notification(
"用户自动代理出现异常",
f"用户 {user[0].replace("_", " 今天的")}{mode_book[j][5:7]}部分出现一次异常",
f"{user[0].replace("_", " ")}{mode_book[j][5:7]}出现异常",
1,
)
for _ in range(10):
if self.isInterruptionRequested:
break
time.sleep(1)
break
else:
logger.error(
f"{self.name} | 用户: {user[0]} - 代理任务异常: {self.maa_result}"
)
# 打印中止信息
# 此时log变量内存储的就是出现异常的日志信息可以保存或发送用于问题排查
self.update_log_text.emit(
f"{self.maa_result}\n正在中止相关程序\n请等待10s"
)
# 无命令行中止MAA与其子程序
System.kill_process(self.maa_exe_path)
self.if_open_emulator = True
# 推送异常通知
Notify.push_plyer(
"用户自动代理出现异常!",
f"用户 {user[0].replace("_", " 今天的")}{mode_book[j][5:7]}部分出现一次异常",
f"{user[0].replace("_", " ")}{mode_book[j][5:7]}出现异常",
1,
)
for _ in range(10):
if self.isInterruptionRequested:
break
time.sleep(1)
# 移除静默进程标记
Config.silence_list.remove(self.emulator_path)
# 保存运行日志以及统计信息
if_six_star = Config.save_maa_log(
Config.app_path
/ f"history/{curdate}/{self.data[user[2]][0]}/{start_time.strftime("%H-%M-%S")}.log",
self.check_maa_log(start_time, mode_book[j]),
self.maa_result,
)
user_logs_list.append(
Config.app_path
/ f"history/{curdate}/{self.data[user[2]][0]}/{start_time.strftime("%H-%M-%S")}.json",
)
if (
Config.global_config.get(
Config.global_config.notify_IfSendSixStar
)
and if_six_star
):
self.push_notification(
"公招六星",
f"喜报:用户 {user[0]} 公招出六星啦!",
{"user_name": self.data[user[2]][0]},
)
# 成功完成代理的用户修改相关参数
if run_book[0] and run_book[1]:
@@ -290,7 +322,7 @@ class MaaManager(QObject):
self.data[user[2]][3] -= 1
self.data[user[2]][14] += 1
user[1] = "完成"
Notify.push_notification(
Notify.push_plyer(
"成功完成一个自动代理任务!",
f"已完成用户 {user[0].replace("_", " 今天的")}任务",
f"已完成 {user[0].replace("_", "")}",
@@ -298,32 +330,55 @@ class MaaManager(QObject):
)
break
if Config.global_config.get(
Config.global_config.notify_IfSendStatistic
):
statistics = Config.merge_maa_logs("指定项", user_logs_list)
statistics["user_info"] = user[0]
statistics["start_time"] = user_start_time.strftime(
"%Y-%m-%d %H:%M:%S"
)
statistics["end_time"] = datetime.now().strftime(
"%Y-%m-%d %H:%M:%S"
)
statistics["maa_result"] = (
"代理任务全部完成"
if (run_book[0] and run_book[1])
else "代理任务未全部完成"
)
self.push_notification(
"统计信息", f"用户 {user[0]} 的自动代理统计报告", statistics
)
# 录入代理失败的用户
if not (run_book[0] and run_book[1]):
user[1] = "异常"
self.update_user_list.emit(user_list)
self.update_user_list.emit(self.user_list)
# 人工排查模式
elif self.mode == "人工排查":
# 标记是否需要启动模拟器
if_strat_app = True
self.if_open_emulator = True
# 标识排查模式
for _ in user_list:
for _ in self.user_list:
_[0] += "_排查模式"
# 开始排查
for user in user_list:
for user in self.user_list:
if self.isInterruptionRequested:
break
logger.info(f"{self.name} | 开始排查用户: {user[0]}")
user[1] = "运行"
self.update_user_list.emit(user_list)
self.update_user_list.emit(self.user_list)
if self.data[user[2]][15] == "beta":
if_strat_app = True
self.if_open_emulator = True
run_book = [False for _ in range(2)]
@@ -331,11 +386,7 @@ class MaaManager(QObject):
while not self.isInterruptionRequested:
# 配置MAA
if if_strat_app:
self.set_maa("人工排查_启动模拟器", user[2])
if_strat_app = False
else:
self.set_maa("人工排查_仅切换账号", user[2])
self.set_maa("人工排查", user[2])
# 记录当前时间
start_time = datetime.now()
@@ -347,43 +398,28 @@ class MaaManager(QObject):
)
# 监测MAA运行状态
while not self.isInterruptionRequested:
self.start_monitor(start_time, "人工排查")
# 获取MAA日志
logs = self.get_maa_log(start_time)
# 合并日志
log = "".join(logs)
# 更新MAA日志
if len(logs) > 100:
self.update_log_text.emit("".join(logs[-100:]))
else:
self.update_log_text.emit("".join(logs))
# 判断MAA程序运行状态
result = self.if_maa_success(log, "人工排查")
if result == "Success!":
run_book[0] = True
self.update_log_text.emit("检测到MAA进程成功登录PRTS")
break
elif result == "Wait":
# 检测时间间隔
if self.maa_result == "Success!":
logger.info(
f"{self.name} | 用户: {user[0]} - MAA进程成功登录PRTS"
)
run_book[0] = True
self.update_log_text.emit("检测到MAA进程成功登录PRTS")
else:
logger.error(
f"{self.name} | 用户: {user[0]} - MAA未能正确登录到PRTS: {self.maa_result}"
)
self.update_log_text.emit(
f"{self.maa_result}\n正在中止相关程序\n请等待10s"
)
# 无命令行中止MAA与其子程序
System.kill_process(self.maa_exe_path)
self.if_open_emulator = True
for _ in range(10):
if self.isInterruptionRequested:
break
time.sleep(1)
else:
self.update_log_text.emit(result)
# 无命令行中止MAA与其子程序
killprocess = subprocess.Popen(
f"taskkill /F /T /PID {maa.pid}",
shell=True,
creationflags=subprocess.CREATE_NO_WINDOW,
)
killprocess.wait()
if_strat_app = True
for _ in range(10):
if self.isInterruptionRequested:
break
time.sleep(1)
break
# 登录成功,结束循环
if run_book[0]:
@@ -406,19 +442,21 @@ class MaaManager(QObject):
# 结果录入用户备注栏
if run_book[0] and run_book[1]:
logger.info(f"{self.name} | 用户 {user[0]} 通过人工排查")
if "未通过人工排查" in self.data[user[2]][13]:
self.data[user[2]][13] = self.data[user[2]][13].replace(
"未通过人工排查|", ""
)
user[1] = "完成"
elif not (run_book[0] and run_book[1]):
else:
logger.info(f"{self.name} | 用户 {user[0]} 未通过人工排查")
if not "未通过人工排查" in self.data[user[2]][13]:
self.data[user[2]][
13
] = f"未通过人工排查|{self.data[user[2]][13]}"
user[1] = "异常"
self.update_user_list.emit(user_list)
self.update_user_list.emit(self.user_list)
# 设置MAA模式
elif "设置MAA" in self.mode:
@@ -435,20 +473,7 @@ class MaaManager(QObject):
start_time = datetime.now()
# 监测MAA运行状态
while not self.isInterruptionRequested:
# 获取MAA日志
logs = self.get_maa_log(start_time)
# 合并日志
log = "".join(logs)
# 判断MAA程序运行状态
result = self.if_maa_success(log, "设置MAA")
if result == "Success!":
break
elif result == "Wait":
# 检测时间间隔
time.sleep(1)
self.start_monitor(start_time, "设置MAA")
if "全局" in self.mode:
(self.config_path / "Default").mkdir(parents=True, exist_ok=True)
@@ -458,82 +483,80 @@ class MaaManager(QObject):
self.user_config_path.mkdir(parents=True, exist_ok=True)
shutil.copy(self.maa_set_path, self.user_config_path)
end_log = ""
result_text = ""
# 导出结果
if self.mode in ["自动代理", "人工排查"]:
# 关闭可能未正常退出的MAA进程
if self.isInterruptionRequested:
killprocess = subprocess.Popen(
f"taskkill /F /T /PID {maa.pid}",
shell=True,
creationflags=subprocess.CREATE_NO_WINDOW,
)
killprocess.wait()
System.kill_process(self.maa_exe_path)
# 更新用户数据
modes = [self.data[_[2]][15] for _ in user_list]
uids = [self.data[_[2]][16] for _ in user_list]
days = [self.data[_[2]][3] for _ in user_list]
lasts = [self.data[_[2]][5] for _ in user_list]
notes = [self.data[_[2]][13] for _ in user_list]
numbs = [self.data[_[2]][14] for _ in user_list]
modes = [self.data[_[2]][15] for _ in self.user_list]
uids = [self.data[_[2]][16] for _ in self.user_list]
days = [self.data[_[2]][3] for _ in self.user_list]
lasts = [self.data[_[2]][5] for _ in self.user_list]
notes = [self.data[_[2]][13] for _ in self.user_list]
numbs = [self.data[_[2]][14] for _ in self.user_list]
self.update_user_info.emit(modes, uids, days, lasts, notes, numbs)
error_index = [_[2] for _ in user_list if _[1] == "异常"]
over_index = [_[2] for _ in user_list if _[1] == "完成"]
wait_index = [_[2] for _ in user_list if _[1] == "等待"]
error_index = [_[2] for _ in self.user_list if _[1] == "异常"]
over_index = [_[2] for _ in self.user_list if _[1] == "完成"]
wait_index = [_[2] for _ in self.user_list if _[1] == "等待"]
# 保存运行日志
end_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
end_log = (
f"任务开始时间:{begin_time},结束时间:{end_time}\n"
f"已完成数:{len(over_index)},未完成数:{len(error_index) + len(wait_index)}\n\n"
)
if len(error_index) != 0:
end_log += (
f"{self.mode[2:4]}未成功的用户:\n"
f"{"\n".join([self.data[_][0] for _ in error_index])}\n"
)
if len(wait_index) != 0:
end_log += (
f"\n未开始{self.mode[2:4]}的用户:\n"
f"{"\n".join([self.data[_][0] for _ in wait_index])}\n"
)
title = (
f"{self.set["MaaSet"]["Name"]}{self.mode[:4]}任务报告"
if self.set["MaaSet"]["Name"] != ""
else f"{self.mode[:4]}任务报告"
)
result = {
"title": f"{self.mode[:4]}任务报告",
"script_name": (
self.set["MaaSet"]["Name"]
if self.set["MaaSet"]["Name"] != ""
else "空白"
),
"start_time": begin_time,
"end_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"completed_count": len(over_index),
"uncompleted_count": len(error_index) + len(wait_index),
"failed_user": [self.data[_][0] for _ in error_index],
"waiting_user": [self.data[_][0] for _ in wait_index],
}
# 推送代理结果通知
Notify.push_notification(
Notify.push_plyer(
title.replace("报告", "已完成!"),
f"已完成用户数:{len(over_index)},未完成用户数:{len(error_index) + len(wait_index)}",
f"已完成用户数:{len(over_index)},未完成用户数:{len(error_index) + len(wait_index)}",
10,
)
if not Config.global_config.get(
Config.global_config.notify_IfSendErrorOnly
) or (
Config.global_config.get(Config.global_config.notify_IfSendErrorOnly)
if Config.global_config.get(
Config.global_config.notify_SendTaskResultTime
) == "任何时刻" or (
Config.global_config.get(Config.global_config.notify_SendTaskResultTime)
== "仅失败时"
and len(error_index) + len(wait_index) != 0
):
Notify.send_mail(
title,
f"{end_log}\n\nAUTO_MAA 敬上\n\n我们根据您在 AUTO_MAA 中的设置发送了这封电子邮件,本邮件无需回复\n",
result_text = self.push_notification("代理结果", title, result)
else:
result_text = self.push_notification(
"代理结果", title, result, if_get_text_only=True
)
Notify.ServerChanPush(title, f"{end_log}\n\nAUTO_MAA 敬上")
Notify.CompanyWebHookBotPush(title, f"{end_log}AUTO_MAA 敬上")
self.agree_bilibili(False)
self.accomplish.emit({"Time": begin_time, "History": end_log})
self.log_monitor.deleteLater()
self.log_monitor_timer.deleteLater()
self.accomplish.emit({"Time": begin_time, "History": result_text})
def requestInterruption(self) -> None:
logger.info(f"{self.name} | 收到任务中止申请")
logger.info("申请中止本次任务")
if len(self.log_monitor.files()) != 0:
self.interrupt.emit()
self.maa_result = "任务被手动中止"
self.isInterruptionRequested = True
def push_question(self, title: str, message: str) -> bool:
@@ -548,9 +571,16 @@ class MaaManager(QObject):
def _capture_response(self, response: bool) -> None:
self.response = response
def get_maa_log(self, start_time):
"""获取MAA日志"""
def refresh_maa_log(self) -> None:
"""刷新MAA日志"""
with self.maa_log_path.open(mode="r", encoding="utf-8") as f:
pass
def check_maa_log(self, start_time: datetime, mode: str) -> list:
"""获取MAA日志并检查以判断MAA程序运行状态"""
# 获取日志
logs = []
if_log_start = False
with self.maa_log_path.open(mode="r", encoding="utf-8") as f:
@@ -565,53 +595,107 @@ class MaaManager(QObject):
pass
else:
logs.append(entry)
return logs
log = "".join(logs)
def if_maa_success(self, log, mode):
"""判断MAA程序运行状态"""
# 更新MAA日志
if len(logs) > 100:
self.update_log_text.emit("".join(logs[-100:]))
else:
self.update_log_text.emit("".join(logs))
if "自动代理" in mode:
# 获取最近一条日志的时间
latest_time = start_time
for _ in logs[::-1]:
try:
latest_time = datetime.strptime(_[1:20], "%Y-%m-%d %H:%M:%S")
break
except ValueError:
pass
time_book = {
"自动代理_剿灭": "AnnihilationTimeLimit",
"自动代理_日常": "RoutineTimeLimit",
}
if mode == "自动代理_日常" and "任务出错: Fight" in log:
return "检测到MAA未能实际执行任务\n正在中止相关程序\n请等待10s"
self.maa_result = "检测到MAA未能实际执行任务"
if "任务出错: StartUp" in log:
return "检测到MAA未能正确登录PRTS\n正在中止相关程序\n请等待10s"
self.maa_result = "检测到MAA未能正确登录PRTS"
elif "任务已全部完成!" in log:
return "Success!"
self.maa_result = "Success!"
elif (
("请「检查连接设置」或「尝试重启模拟器与 ADB」或「重启电脑」" in log)
or ("未检测到任何模拟器" in log)
or ("已停止" in log)
or ("MaaAssistantArknights GUI exited" in log)
):
return "检测到MAA进程异常\n正在中止相关程序\n请等待10s"
elif self.if_time_out:
return "检测到MAA进程超时\n正在中止相关程序\n请等待10s"
self.maa_result = "检测到MAA进程异常"
elif datetime.now() - latest_time > timedelta(
minutes=self.set["RunSet"][time_book[mode]]
):
self.maa_result = "检测到MAA进程超时"
elif self.isInterruptionRequested:
return "您中止了本次任务\n正在中止相关程序\n请等待"
self.maa_result = "任务被手动中止"
else:
return "Wait"
self.maa_result = "Wait"
elif mode == "人工排查":
if "完成任务: StartUp" in log:
return "Success!"
self.maa_result = "Success!"
elif (
("请「检查连接设置」或「尝试重启模拟器与 ADB」或「重启电脑」" in log)
or ("未检测到任何模拟器" in log)
or ("已停止" in log)
or ("MaaAssistantArknights GUI exited" in log)
):
return "检测到MAA进程异常\n正在中止相关程序\n请等待10s"
self.maa_result = "检测到MAA进程异常"
elif self.isInterruptionRequested:
return "您中止了本次任务\n正在中止相关程序\n请等待"
self.maa_result = "任务被手动中止"
else:
return "Wait"
self.maa_result = "Wait"
elif mode == "设置MAA":
if "MaaAssistantArknights GUI exited" in log:
return "Success!"
self.maa_result = "Success!"
else:
return "Wait"
self.maa_result = "Wait"
if self.maa_result != "Wait":
self.quit_monitor()
return logs
def start_monitor(self, start_time: datetime, mode: str) -> None:
"""开始监视MAA日志"""
logger.info(f"{self.name} | 开始监视MAA日志")
self.log_monitor.addPath(str(self.maa_log_path))
self.log_monitor.fileChanged.connect(
lambda: self.check_maa_log(start_time, mode)
)
self.log_monitor_timer.start(1000)
self.monitor_loop.exec()
def quit_monitor(self) -> None:
"""退出MAA日志监视进程"""
if len(self.log_monitor.files()) != 0:
logger.info(f"{self.name} | 退出MAA日志监视")
self.log_monitor.removePath(str(self.maa_log_path))
self.log_monitor.fileChanged.disconnect()
self.log_monitor_timer.stop()
self.monitor_loop.quit()
def set_maa(self, mode, index):
"""配置MAA运行参数"""
logger.info(f"{self.name} | 配置MAA运行参数: {mode}/{index}")
# 配置MAA前关闭可能未正常退出的MAA进程
System.kill_process(self.maa_exe_path)
# 预导入MAA配置文件
if mode == "设置MAA_用户":
@@ -667,15 +751,34 @@ class MaaManager(QObject):
data["Current"] = "Default" # 切换配置
for i in range(1, 9):
data["Global"][f"Timer.Timer{i}"] = "False" # 时间设置
data["Configurations"]["Default"][
"MainFunction.PostActions"
] = "12" # 完成后退出MAA和模拟器
if (
[i for i, _ in enumerate(self.user_list) if _[2] == index][0]
== len(self.user_list) - 1
) or (
self.data[
self.user_list[
[i for i, _ in enumerate(self.user_list) if _[2] == index][0]
+ 1
][2]
][15]
== "beta"
):
data["Configurations"]["Default"][
"MainFunction.PostActions"
] = "12" # 完成后退出MAA和模拟器
else:
method_dict = {"NoAction": "8", "ExitGame": "9", "ExitEmulator": "12"}
data["Configurations"]["Default"]["MainFunction.PostActions"] = (
method_dict[self.set["RunSet"]["TaskTransitionMethod"]]
) # 完成后行为
data["Configurations"]["Default"][
"Start.RunDirectly"
] = "True" # 启动MAA后直接运行
data["Configurations"]["Default"][
"Start.OpenEmulatorAfterLaunch"
] = "True" # 启动MAA后自动开启模拟器
data["Configurations"]["Default"]["Start.OpenEmulatorAfterLaunch"] = (
"True" if self.if_open_emulator else "False"
) # 启动MAA后自动开启模拟器
if Config.global_config.get(Config.global_config.function_IfSilence):
data["Global"]["Start.MinimizeDirectly"] = "True" # 启动MAA后直接最小化
@@ -686,13 +789,13 @@ class MaaManager(QObject):
data["Global"][
"VersionUpdate.ScheduledUpdateCheck"
] = "True" # 定时检查更新
] = "False" # 定时检查更新
data["Global"][
"VersionUpdate.AutoDownloadUpdatePackage"
] = "True" # 自动下载更新包
] = "False" # 自动下载更新包
data["Global"][
"VersionUpdate.AutoInstallUpdatePackage"
] = "True" # 自动安装更新包
] = "False" # 自动安装更新包
data["Configurations"]["Default"]["Start.ClientType"] = self.data[
index
][
@@ -888,7 +991,6 @@ class MaaManager(QObject):
"Start.RunDirectly"
] = "True" # 启动MAA后直接运行
data["Global"]["Start.MinimizeDirectly"] = "True" # 启动MAA后直接最小化
# v5.1.12版本对以下字段处理
# 启动MAA后直接运行
data["Configurations"]["Default"]["Start.OpenEmulatorAfterLaunch"] = "True"
# 启动MAA后自动开启模拟器
@@ -896,15 +998,9 @@ class MaaManager(QObject):
data["Global"]["GUI.UseTray"] = "True" # 显示托盘图标
data["Global"]["GUI.MinimizeToTray"] = "True" # 最小化时隐藏至托盘
# 启动MAA后自动开启模拟器
if "启动模拟器" in mode:
data["Configurations"]["Default"][
"Start.OpenEmulatorAfterLaunch"
] = "True"
elif "仅切换账号" in mode:
data["Configurations"]["Default"][
"Start.OpenEmulatorAfterLaunch"
] = "False"
data["Configurations"]["Default"]["Start.OpenEmulatorAfterLaunch"] = (
"True" if self.if_open_emulator else "False"
) # 启动MAA后自动开启模拟器
if self.data[index][15] == "simple":
@@ -1016,6 +1112,14 @@ class MaaManager(QObject):
"TaskQueue.Reclamation.IsChecked"
] = "False" # 生息演算
# 启动模拟器仅生效一次
if (
"设置MAA" not in mode
and self.if_open_emulator
and self.set["RunSet"]["TaskTransitionMethod"] != "ExitEmulator"
):
self.if_open_emulator = False
# 覆写配置文件
with self.maa_set_path.open(mode="w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=4)
@@ -1024,6 +1128,9 @@ class MaaManager(QObject):
def agree_bilibili(self, if_agree):
"""向MAA写入Bilibili协议相关任务"""
logger.info(
f"{self.name} | Bilibili协议相关任务状态: {"启用" if if_agree else "禁用"}"
)
with self.maa_tasks_path.open(mode="r", encoding="utf-8") as f:
data = json.load(f)
@@ -1066,3 +1173,86 @@ class MaaManager(QObject):
if dt.time() < datetime.min.time().replace(hour=4):
dt = dt - timedelta(days=1)
return dt.strftime("%Y-%m-%d")
def push_notification(
self,
mode: str,
title: str,
message: Union[str, dict],
if_get_text_only: bool = False,
) -> str:
"""通过所有渠道推送通知"""
env = Environment(
loader=FileSystemLoader(str(Config.app_path / "resources/html"))
)
if mode == "代理结果":
# 生成文本通知内容
message_text = (
f"任务开始时间:{message["start_time"]},结束时间:{message["end_time"]}\n"
f"已完成数:{message["completed_count"]},未完成数:{message["uncompleted_count"]}\n\n"
)
if len(message["failed_user"]) > 0:
message_text += f"{self.mode[2:4]}未成功的用户:\n{"\n".join(message["failed_user"])}\n"
if len(message["waiting_user"]) > 0:
message_text += f"\n未开始{self.mode[2:4]}的用户:\n{"\n".join(message["waiting_user"])}\n"
if if_get_text_only:
return message_text
# 生成HTML通知内容
message["failed_user"] = "".join(message["failed_user"])
message["waiting_user"] = "".join(message["waiting_user"])
template = env.get_template("MAA_result.html")
message_html = template.render(message)
Notify.send_mail("网页", title, message_html)
Notify.ServerChanPush(title, f"{message_text}\n\nAUTO_MAA 敬上")
Notify.CompanyWebHookBotPush(title, f"{message_text}\n\nAUTO_MAA 敬上")
return message_text
elif mode == "统计信息":
# 生成文本通知内容
formatted = []
for stage, items in message["drop_statistics"].items():
formatted.append(f"掉落统计({stage}:")
for item, quantity in items.items():
formatted.append(f" {item}: {quantity}")
drop_text = "\n".join(formatted)
formatted = ["招募统计:"]
for star, count in message["recruit_statistics"].items():
formatted.append(f" {star}: {count}")
recruit_text = "\n".join(formatted)
message_text = (
f"开始时间: {message['start_time']}\n"
f"结束时间: {message['end_time']}\n"
f"MAA执行结果: {message['maa_result']}\n\n"
f"{recruit_text}\n"
f"{drop_text}"
)
# 生成HTML通知内容
template = env.get_template("MAA_statistics.html")
message_html = template.render(message)
Notify.send_mail("网页", title, message_html)
Notify.ServerChanPush(title, f"{message_text}\n\nAUTO_MAA 敬上")
Notify.CompanyWebHookBotPush(title, f"{message_text}\n\nAUTO_MAA 敬上")
elif mode == "公招六星":
# 生成HTML通知内容
template = env.get_template("MAA_six_star.html")
message_html = template.render(message)
Notify.send_mail("网页", title, message_html)
Notify.ServerChanPush(title, "好羡慕~\n\nAUTO_MAA 敬上")
Notify.CompanyWebHookBotPush(title, "好羡慕~\n\nAUTO_MAA 敬上")

View File

@@ -24,23 +24,33 @@ AUTO_MAA通知服务
v4.2
作者DLmaster_361
"""
from PySide6.QtWidgets import QWidget
from PySide6.QtCore import Signal
import requests
from loguru import logger
from plyer import notification
import re
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.header import Header
from email.utils import formataddr
from serverchan_sdk import sc_send
from app.core import Config, MainInfoBar
from app.core import Config
from app.services.security import Crypto
class Notification:
class Notification(QWidget):
def push_notification(self, title, message, ticker, t):
push_info_bar = Signal(str, str, str, int)
def __init__(self, parent=None):
super().__init__(parent)
def push_plyer(self, title, message, ticker, t):
"""推送系统通知"""
if Config.global_config.get(Config.global_config.notify_IfPushPlyer):
@@ -57,14 +67,50 @@ class Notification:
return True
def send_mail(self, title, content):
def send_mail(self, mode, title, content) -> None:
"""推送邮件通知"""
if Config.global_config.get(Config.global_config.notify_IfSendMail):
if (
Config.global_config.get(Config.global_config.notify_SMTPServerAddress)
== ""
or Config.global_config.get(
Config.global_config.notify_AuthorizationCode
)
== ""
or not bool(
re.match(
r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$",
Config.global_config.get(
Config.global_config.notify_FromAddress
),
)
)
or not bool(
re.match(
r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$",
Config.global_config.get(Config.global_config.notify_ToAddress),
)
)
):
logger.error(
"请正确设置邮件通知的SMTP服务器地址、授权码、发件人地址和收件人地址"
)
self.push_info_bar.emit(
"error",
"邮件通知推送异常",
"请正确设置邮件通知的SMTP服务器地址、授权码、发件人地址和收件人地址",
-1,
)
return None
try:
# 定义邮件正文
message = MIMEText(content, "plain", "utf-8")
if mode == "文本":
message = MIMEText(content, "plain", "utf-8")
elif mode == "网页":
message = MIMEMultipart("alternative")
message["From"] = formataddr(
(
Header("AUTO_MAA通知服务", "utf-8").encode(),
@@ -81,6 +127,9 @@ class Notification:
) # 收件人显示的名字
message["Subject"] = Header(title, "utf-8")
if mode == "网页":
message.attach(MIMEText(content, "html", "utf-8"))
smtpObj = smtplib.SMTP_SSL(
Config.global_config.get(
Config.global_config.notify_SMTPServerAddress
@@ -104,7 +153,7 @@ class Notification:
logger.success("邮件发送成功")
except Exception as e:
logger.error(f"发送邮件时出错:\n{e}")
MainInfoBar.push_info_bar("error", "发送邮件时出错", f"{e}", -1)
self.push_info_bar.emit("error", "发送邮件时出错", f"{e}", -1)
def ServerChanPush(self, title, content):
"""使用Server酱推送通知"""
@@ -133,7 +182,7 @@ class Notification:
else:
option["tags"] = ""
logger.warning("请正确设置Auto_MAA中ServerChan的Tag。")
MainInfoBar.push_info_bar(
self.push_info_bar.emit(
"warning",
"Server酱通知推送异常",
"请正确设置Auto_MAA中ServerChan的Tag。",
@@ -145,7 +194,7 @@ class Notification:
else:
option["channel"] = ""
logger.warning("请正确设置Auto_MAA中ServerChan的Channel。")
MainInfoBar.push_info_bar(
self.push_info_bar.emit(
"warning",
"Server酱通知推送异常",
"请正确设置Auto_MAA中ServerChan的Channel。",
@@ -159,7 +208,7 @@ class Notification:
else:
logger.info("Server酱推送通知失败")
logger.error(response)
MainInfoBar.push_info_bar(
self.push_info_bar.emit(
"error",
"Server酱通知推送失败",
f'使用Server酱推送通知时出错\n{response["data"]['error']}',
@@ -184,7 +233,7 @@ class Notification:
else:
logger.info("企业微信群机器人推送通知失败")
logger.error(response.json())
MainInfoBar.push_info_bar(
self.push_info_bar.emit(
"error",
"企业微信群机器人通知推送失败",
f'使用企业微信群机器人推送通知时出错:\n{response.json()["errmsg"]}',

View File

@@ -34,6 +34,7 @@ import win32process
import winreg
import psutil
import subprocess
from pathlib import Path
from app.core import Config
@@ -50,7 +51,7 @@ class _SystemHandler:
self.set_Sleep()
self.set_SelfStart()
def set_Sleep(self):
def set_Sleep(self) -> None:
"""同步系统休眠状态"""
if Config.global_config.get(Config.global_config.function_IfAllowSleep):
@@ -62,7 +63,7 @@ class _SystemHandler:
# 恢复系统电源状态
ctypes.windll.kernel32.SetThreadExecutionState(self.ES_CONTINUOUS)
def set_SelfStart(self):
def set_SelfStart(self) -> None:
"""同步开机自启"""
if (
@@ -90,7 +91,7 @@ class _SystemHandler:
winreg.DeleteValue(key, "AUTO_MAA")
winreg.CloseKey(key)
def set_power(self, mode):
def set_power(self, mode) -> None:
if sys.platform.startswith("win"):
@@ -144,7 +145,7 @@ class _SystemHandler:
self.main_window.close()
def is_startup(self):
def is_startup(self) -> bool:
"""判断程序是否已经开机自启"""
key = winreg.OpenKey(
@@ -162,7 +163,7 @@ class _SystemHandler:
winreg.CloseKey(key)
return False
def get_window_info(self):
def get_window_info(self) -> list:
"""获取当前窗口信息"""
def callback(hwnd, window_info):
@@ -176,5 +177,29 @@ class _SystemHandler:
win32gui.EnumWindows(callback, window_info)
return window_info
def kill_process(self, path: Path) -> None:
"""根据路径中止进程"""
for pid in self.search_pids(path):
killprocess = subprocess.Popen(
f"taskkill /F /T /PID {pid}",
shell=True,
creationflags=subprocess.CREATE_NO_WINDOW,
)
killprocess.wait()
def search_pids(self, path: Path) -> list:
"""根据路径查找进程PID"""
pids = []
for proc in psutil.process_iter(["pid", "exe"]):
try:
if proc.info["exe"] and proc.info["exe"].lower() == str(path).lower():
pids.append(proc.info["pid"])
except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
# 进程可能在此期间已结束或无法访问,忽略这些异常
pass
return pids
System = _SystemHandler()

View File

@@ -25,9 +25,9 @@ v4.2
作者DLmaster_361
"""
from PySide6.QtCore import Qt, QTime
from PySide6.QtGui import QIcon
from PySide6.QtWidgets import QWidget, QHBoxLayout
from PySide6.QtCore import Qt, QTime, QEvent
from PySide6.QtGui import QIcon, QPixmap, QPainter, QPainterPath
from qfluentwidgets import (
LineEdit,
PasswordLineEdit,
@@ -39,12 +39,21 @@ from qfluentwidgets import (
Signal,
ComboBox,
CheckBox,
IconWidget,
FluentIcon,
CardWidget,
BodyLabel,
qconfig,
ConfigItem,
TimeEdit,
OptionsConfigItem,
TeachingTip,
TransparentToolButton,
TeachingTipTailPosition,
)
from typing import Union, List
from qfluentwidgets.common.overload import singledispatchmethod
import os
from typing import Optional, Union, List
from app.services import Crypto
@@ -319,3 +328,207 @@ class TimeEditSettingCard(SettingCard):
qconfig.set(self.configItem_time, value)
self.TimeEdit.setTime(QTime.fromString(value, "HH:mm"))
class StatefulItemCard(CardWidget):
def __init__(self, item: list, parent=None):
super().__init__(parent)
self.Layout = QHBoxLayout(self)
self.Label = BodyLabel(item[0], self)
self.icon = IconWidget(FluentIcon.MORE, self)
self.icon.setFixedSize(16, 16)
self.update_status(item[1])
self.Layout.addWidget(self.icon)
self.Layout.addWidget(self.Label)
self.Layout.addStretch(1)
def update_status(self, status: str):
if status == "完成":
self.icon.setIcon(FluentIcon.ACCEPT)
self.Label.setTextColor("#0eb840", "#0eb840")
elif status == "等待":
self.icon.setIcon(FluentIcon.MORE)
self.Label.setTextColor("#161823", "#e3f9fd")
elif status == "运行":
self.icon.setIcon(FluentIcon.PLAY)
self.Label.setTextColor("#177cb0", "#70f3ff")
elif status == "跳过":
self.icon.setIcon(FluentIcon.REMOVE)
self.Label.setTextColor("#75878a", "#7397ab")
elif status == "异常":
self.icon.setIcon(FluentIcon.CLOSE)
self.Label.setTextColor("#ff2121", "#ff2121")
class QuantifiedItemCard(CardWidget):
def __init__(self, item: list, parent=None):
super().__init__(parent)
self.Layout = QHBoxLayout(self)
self.Name = BodyLabel(item[0], self)
self.Numb = BodyLabel(str(item[1]), self)
self.Layout.addWidget(self.Name)
self.Layout.addStretch(1)
self.Layout.addWidget(self.Numb)
class IconButton(TransparentToolButton):
"""包含下拉框的自定义设置卡片类。"""
@singledispatchmethod
def __init__(self, parent: QWidget = None):
TransparentToolButton.__init__(self, parent)
self._tooltip: Optional[TeachingTip] = None
@__init__.register
def _(self, icon: Union[str, QIcon, FluentIconBase], parent: QWidget = None):
self.__init__(parent)
self.setIcon(icon)
@__init__.register
def _(
self,
icon: Union[str, QIcon, FluentIconBase],
isTooltip: bool,
tip_title: str,
tip_content: str,
parent: QWidget = None,
):
self.__init__(parent)
self.setIcon(icon)
# 处理工具提示
if isTooltip:
self.installEventFilter(self)
self.tip_title: str = tip_title
self.tip_content: str = tip_content
def eventFilter(self, obj, event: QEvent) -> bool:
"""处理鼠标事件。"""
if event.type() == QEvent.Type.Enter:
self._show_tooltip()
elif event.type() == QEvent.Type.Leave:
self._hide_tooltip()
return super().eventFilter(obj, event)
def _show_tooltip(self) -> None:
"""显示工具提示。"""
self._tooltip = TeachingTip.create(
target=self,
title=self.tip_title,
content=self.tip_content,
tailPosition=TeachingTipTailPosition.RIGHT,
isClosable=False,
duration=-1,
parent=self,
)
# 设置偏移
if self._tooltip:
tooltip_pos = self.mapToGlobal(self.rect().topRight())
tooltip_pos.setX(
tooltip_pos.x() - self._tooltip.size().width() - 40
) # 水平偏移
tooltip_pos.setY(
tooltip_pos.y() - self._tooltip.size().height() / 2 + 35
) # 垂直偏移
self._tooltip.move(tooltip_pos)
def _hide_tooltip(self) -> None:
"""隐藏工具提示。"""
if self._tooltip:
self._tooltip.close()
self._tooltip = None
def __hash__(self):
return id(self)
def __eq__(self, other):
if not isinstance(other, self.__class__):
return NotImplemented
return self is other
class Banner(QWidget):
"""展示带有圆角的固定大小横幅小部件"""
def __init__(self, image_path: str = None, parent=None):
QWidget.__init__(self, parent)
self.image_path = None
self.banner_image = None
self.scaled_image = None
if image_path:
self.set_banner_image(image_path)
def set_banner_image(self, image_path: str):
"""设置横幅图片"""
self.image_path = image_path
self.banner_image = self.load_banner_image(image_path)
self.update_scaled_image()
def load_banner_image(self, image_path: str) -> QPixmap:
"""加载横幅图片,或创建渐变备用图片"""
if os.path.isfile(image_path):
return QPixmap(image_path)
return self._create_fallback_image()
def _create_fallback_image(self):
"""创建渐变备用图片"""
fallback_image = QPixmap(2560, 1280) # 使用原始图片的大小
fallback_image.fill(Qt.GlobalColor.gray)
return fallback_image
def update_scaled_image(self):
"""按高度缩放图片,宽度保持比例,超出裁剪"""
if self.banner_image:
self.scaled_image = self.banner_image.scaled(
self.size(),
Qt.AspectRatioMode.KeepAspectRatioByExpanding,
Qt.TransformationMode.SmoothTransformation,
)
self.update()
def paintEvent(self, event):
"""重载 paintEvent 以绘制缩放后的图片"""
if self.scaled_image:
painter = QPainter(self)
painter.setRenderHint(QPainter.RenderHint.Antialiasing)
painter.setRenderHint(QPainter.RenderHint.SmoothPixmapTransform)
# 创建圆角路径
path = QPainterPath()
path.addRoundedRect(self.rect(), 20, 20)
painter.setClipPath(path)
# 计算绘制位置,使图片居中
x = (self.width() - self.scaled_image.width()) // 2
y = (self.height() - self.scaled_image.height()) // 2
# 绘制缩放后的图片
painter.drawPixmap(x, y, self.scaled_image)
def resizeEvent(self, event):
"""重载 resizeEvent 以更新缩放后的图片"""
self.update_scaled_image()
QWidget.resizeEvent(self, event)
def set_percentage_size(self, width_percentage, height_percentage):
"""设置 Banner 的大小为窗口大小的百分比"""
parent = self.parentWidget()
if parent:
new_width = int(parent.width() * width_percentage)
new_height = int(parent.height() * height_percentage)
self.setFixedSize(new_width, new_height)
self.update_scaled_image()

View File

@@ -34,8 +34,6 @@ from PySide6.QtWidgets import (
)
from qfluentwidgets import (
CardWidget,
IconWidget,
BodyLabel,
Pivot,
ScrollArea,
FluentIcon,
@@ -52,7 +50,8 @@ from typing import List, Dict
import json
from app.core import Config, Task_manager, Task, MainInfoBar
from app.core import Config, TaskManager, Task, MainInfoBar
from .Widget import StatefulItemCard
class DispatchCenter(QWidget):
@@ -92,7 +91,7 @@ class DispatchCenter(QWidget):
dispatch_box = DispatchBox(task.name, self)
dispatch_box.top_bar.button.clicked.connect(
lambda: Task_manager.stop_task(task.name)
lambda: TaskManager.stop_task(task.name)
)
task.create_task_list.connect(dispatch_box.info.task.create_task)
@@ -128,7 +127,7 @@ class DispatchCenter(QWidget):
self.script_list["主调度台"].top_bar.button.clicked.disconnect()
self.script_list["主调度台"].top_bar.button.setText("中止任务")
self.script_list["主调度台"].top_bar.button.clicked.connect(
lambda: Task_manager.stop_task(task.name)
lambda: TaskManager.stop_task(task.name)
)
task.create_task_list.connect(
self.script_list["主调度台"].info.task.create_task
@@ -166,20 +165,31 @@ class DispatchCenter(QWidget):
"""更新顶栏"""
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.object.setCurrentIndex(-1)
self.script_list["主调度台"].top_bar.mode.setCurrentIndex(-1)
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):
@@ -223,7 +233,6 @@ class DispatchBox(QWidget):
self.object = ComboBox()
self.object.setPlaceholderText("请选择调度对象")
self.mode = ComboBox()
self.mode.addItems(["自动代理", "人工排查"])
self.mode.setPlaceholderText("请选择调度模式")
self.button = PushButton("开始任务")
@@ -276,7 +285,7 @@ class DispatchBox(QWidget):
info = json.load(f)
logger.info(f"用户添加任务:{name}")
Task_manager.add_task(f"{self.mode.currentText()}_主调度台", name, info)
TaskManager.add_task(f"{self.mode.currentText()}_主调度台", name, info)
elif self.object.currentText().split(" - ")[0] == "实例":
@@ -285,7 +294,7 @@ class DispatchBox(QWidget):
info = {"Queue": {"Member_1": name}}
logger.info(f"用户添加任务:{name}")
Task_manager.add_task(
TaskManager.add_task(
f"{self.mode.currentText()}_主调度台", "自定义队列", info
)
@@ -325,7 +334,7 @@ class DispatchBox(QWidget):
self.viewLayout.addLayout(self.Layout)
self.viewLayout.setContentsMargins(3, 0, 3, 3)
self.task_cards: List[ItemCard] = []
self.task_cards: List[StatefulItemCard] = []
def create_task(self, task_list: list):
"""创建任务队列"""
@@ -341,7 +350,7 @@ class DispatchBox(QWidget):
for task in task_list:
self.task_cards.append(ItemCard(task))
self.task_cards.append(StatefulItemCard(task))
self.Layout.addWidget(self.task_cards[-1])
self.Layout.addStretch(1)
@@ -363,7 +372,7 @@ class DispatchBox(QWidget):
self.viewLayout.addLayout(self.Layout)
self.viewLayout.setContentsMargins(3, 0, 3, 3)
self.user_cards: List[ItemCard] = []
self.user_cards: List[StatefulItemCard] = []
def create_user(self, user_list: list):
"""创建用户队列"""
@@ -379,7 +388,7 @@ class DispatchBox(QWidget):
for user in user_list:
self.user_cards.append(ItemCard(user))
self.user_cards.append(StatefulItemCard(user))
self.Layout.addWidget(self.user_cards[-1])
self.Layout.addStretch(1)
@@ -409,38 +418,3 @@ class DispatchBox(QWidget):
self.text.moveCursor(QTextCursor.End)
self.text.ensureCursorVisible()
class ItemCard(CardWidget):
def __init__(self, task_item: list, parent=None):
super().__init__(parent)
self.Layout = QHBoxLayout(self)
self.Label = BodyLabel(task_item[0], self)
self.icon = IconWidget(FluentIcon.MORE, self)
self.icon.setFixedSize(16, 16)
self.update_status(task_item[1])
self.Layout.addWidget(self.icon)
self.Layout.addWidget(self.Label)
self.Layout.addStretch(1)
def update_status(self, status: str):
if status == "完成":
self.icon.setIcon(FluentIcon.ACCEPT)
self.Label.setTextColor("#0eb840", "#0eb840")
elif status == "等待":
self.icon.setIcon(FluentIcon.MORE)
self.Label.setTextColor("#7397ab", "#7397ab")
elif status == "运行":
self.icon.setIcon(FluentIcon.PLAY)
self.Label.setTextColor("#2e4e7e", "#2e4e7e")
elif status == "跳过":
self.icon.setIcon(FluentIcon.REMOVE)
self.Label.setTextColor("#606060", "#d2d2d2")
elif status == "异常":
self.icon.setIcon(FluentIcon.CLOSE)
self.Label.setTextColor("#ff2121", "#ff2121")

258
app/ui/history.py Normal file
View File

@@ -0,0 +1,258 @@
# <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,
QHBoxLayout,
)
from qfluentwidgets import (
ScrollArea,
FluentIcon,
HeaderCardWidget,
PushButton,
ExpandGroupSettingCard,
TextBrowser,
)
from PySide6.QtCore import Signal
import os
from functools import partial
from pathlib import Path
from typing import List
from app.core import Config
from .Widget import StatefulItemCard, QuantifiedItemCard
class History(QWidget):
def __init__(self, parent=None):
super().__init__(parent)
self.setObjectName("历史记录")
content_widget = QWidget()
self.content_layout = QVBoxLayout(content_widget)
scrollArea = ScrollArea()
scrollArea.setWidgetResizable(True)
scrollArea.setWidget(content_widget)
layout = QVBoxLayout()
layout.addWidget(scrollArea)
self.setLayout(layout)
self.history_card_list = []
self.refresh()
def refresh(self):
"""刷新脚本实例界面"""
while self.content_layout.count() > 0:
item = self.content_layout.takeAt(0)
if item.spacerItem():
self.content_layout.removeItem(item.spacerItem())
elif item.widget():
item.widget().deleteLater()
self.history_card_list = []
history_dict = Config.search_history()
for date, user_list in history_dict.items():
self.history_card_list.append(HistoryCard(date, user_list, self))
self.content_layout.addWidget(self.history_card_list[-1])
self.content_layout.addStretch(1)
class HistoryCard(ExpandGroupSettingCard):
def __init__(self, date: str, user_list: List[Path], parent=None):
super().__init__(
FluentIcon.HISTORY, date, f"{date}的历史运行记录与统计信息", parent
)
widget = QWidget()
Layout = QVBoxLayout(widget)
self.viewLayout.setContentsMargins(0, 0, 0, 0)
self.viewLayout.setSpacing(0)
self.addGroupWidget(widget)
self.user_history_card_list = []
for user_path in user_list:
self.user_history_card_list.append(self.UserHistoryCard(user_path, self))
Layout.addWidget(self.user_history_card_list[-1])
class UserHistoryCard(HeaderCardWidget):
def __init__(
self,
user_history_path: Path,
parent=None,
):
super().__init__(parent)
self.setTitle(user_history_path.name.replace(".json", ""))
self.user_history_path = user_history_path
self.main_history = Config.load_maa_logs("总览", user_history_path)
self.index_card = self.IndexCard(self.main_history["条目索引"], self)
self.statistics_card = QHBoxLayout()
self.log_card = self.LogCard(self)
self.index_card.index_changed.connect(self.update_info)
self.viewLayout.addWidget(self.index_card)
self.viewLayout.addLayout(self.statistics_card)
self.viewLayout.addWidget(self.log_card)
self.viewLayout.setContentsMargins(0, 0, 0, 0)
self.viewLayout.setSpacing(0)
self.viewLayout.setStretch(0, 1)
self.viewLayout.setStretch(2, 4)
self.update_info("数据总览")
def update_info(self, index: str) -> None:
"""更新信息"""
if index == "数据总览":
while self.statistics_card.count() > 0:
item = self.statistics_card.takeAt(0)
if item.spacerItem():
self.statistics_card.removeItem(item.spacerItem())
elif item.widget():
item.widget().deleteLater()
for name, item_list in self.main_history["统计数据"].items():
statistics_card = self.StatisticsCard(name, item_list, self)
self.statistics_card.addWidget(statistics_card)
self.log_card.hide()
else:
single_history = Config.load_maa_logs(
"单项",
self.user_history_path.with_suffix("")
/ f"{index.replace(":","-")}.json",
)
while self.statistics_card.count() > 0:
item = self.statistics_card.takeAt(0)
if item.spacerItem():
self.statistics_card.removeItem(item.spacerItem())
elif item.widget():
item.widget().deleteLater()
for name, item_list in single_history["统计数据"].items():
statistics_card = self.StatisticsCard(name, item_list, self)
self.statistics_card.addWidget(statistics_card)
self.log_card.text.setText(single_history["日志信息"])
self.log_card.button.clicked.disconnect()
self.log_card.button.clicked.connect(
lambda: os.startfile(
self.user_history_path.with_suffix("")
/ f"{index.replace(":","-")}.log"
)
)
self.log_card.show()
self.viewLayout.setStretch(1, self.statistics_card.count())
self.setMinimumHeight(300)
class IndexCard(HeaderCardWidget):
index_changed = Signal(str)
def __init__(self, index_list: list, parent=None):
super().__init__(parent)
self.setTitle("记录条目")
self.Layout = QVBoxLayout()
self.viewLayout.addLayout(self.Layout)
self.viewLayout.setContentsMargins(3, 0, 3, 3)
self.index_cards: List[StatefulItemCard] = []
for index in index_list:
self.index_cards.append(StatefulItemCard(index))
self.index_cards[-1].clicked.connect(
partial(self.index_changed.emit, index[0])
)
self.Layout.addWidget(self.index_cards[-1])
self.Layout.addStretch(1)
class StatisticsCard(HeaderCardWidget):
def __init__(self, name: str, item_list: list, parent=None):
super().__init__(parent)
self.setTitle(name)
self.Layout = QVBoxLayout()
self.viewLayout.addLayout(self.Layout)
self.viewLayout.setContentsMargins(3, 0, 3, 3)
self.item_cards: List[QuantifiedItemCard] = []
for item in item_list:
self.item_cards.append(QuantifiedItemCard(item))
self.Layout.addWidget(self.item_cards[-1])
if len(item_list) == 0:
self.Layout.addWidget(QuantifiedItemCard(["暂无记录", ""]))
self.Layout.addStretch(1)
class LogCard(HeaderCardWidget):
def __init__(self, parent=None):
super().__init__(parent)
self.setTitle("日志")
self.text = TextBrowser(self)
self.button = PushButton("打开日志文件", self)
self.button.clicked.connect(lambda: print("打开日志文件"))
Layout = QVBoxLayout()
Layout.addWidget(self.text)
Layout.addWidget(self.button)
self.viewLayout.setContentsMargins(3, 0, 3, 3)
self.viewLayout.addLayout(Layout)

423
app/ui/home.py Normal file
View File

@@ -0,0 +1,423 @@
# <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,
QHBoxLayout,
QSpacerItem,
QSizePolicy,
QFileDialog,
)
from PySide6.QtCore import Qt, QSize, QUrl
from PySide6.QtGui import QDesktopServices, QColor
from qfluentwidgets import (
FluentIcon,
ScrollArea,
SimpleCardWidget,
PrimaryToolButton,
TextBrowser,
)
import re
import shutil
import requests
import json
import time
from datetime import datetime
from pathlib import Path
from app.core import Config, MainInfoBar
from .Widget import Banner, IconButton
class Home(QWidget):
def __init__(self, parent=None):
super().__init__(parent)
self.setObjectName("主页")
self.banner = Banner()
self.banner_text = TextBrowser()
widget = QWidget()
Layout = QVBoxLayout(widget)
Layout.addWidget(self.banner)
Layout.addWidget(self.banner_text)
Layout.setStretch(0, 2)
Layout.setStretch(1, 3)
v_layout = QVBoxLayout(self.banner)
v_layout.setContentsMargins(0, 0, 0, 15)
v_layout.setSpacing(5)
v_layout.setAlignment(Qt.AlignmentFlag.AlignTop)
# 空白占位符
v_layout.addItem(
QSpacerItem(10, 20, QSizePolicy.Policy.Fixed, QSizePolicy.Policy.Minimum)
)
# 顶部部分 (按钮组)
h1_layout = QHBoxLayout()
h1_layout.setAlignment(Qt.AlignmentFlag.AlignTop)
# 左边留白区域
h1_layout.addStretch()
# 按钮组
buttonGroup = ButtonGroup()
buttonGroup.setMaximumHeight(320)
h1_layout.addWidget(buttonGroup)
# 空白占位符
h1_layout.addItem(
QSpacerItem(20, 10, QSizePolicy.Policy.Fixed, QSizePolicy.Policy.Minimum)
)
# 将顶部水平布局添加到垂直布局
v_layout.addLayout(h1_layout)
# 中间留白区域
v_layout.addItem(
QSpacerItem(10, 10, QSizePolicy.Policy.Fixed, QSizePolicy.Policy.Minimum)
)
v_layout.addStretch()
# 中间留白区域
v_layout.addItem(
QSpacerItem(10, 10, QSizePolicy.Policy.Fixed, QSizePolicy.Policy.Minimum)
)
v_layout.addStretch()
# 底部部分 (图片切换按钮)
h2_layout = QHBoxLayout()
h2_layout.setAlignment(Qt.AlignmentFlag.AlignTop)
# 左边留白区域
h2_layout.addItem(
QSpacerItem(20, 10, QSizePolicy.Policy.Fixed, QSizePolicy.Policy.Minimum)
)
# # 公告卡片
# noticeCard = NoticeCard()
# h2_layout.addWidget(noticeCard)
h2_layout.addStretch()
# 自定义图像按钮布局
self.imageButton = PrimaryToolButton(FluentIcon.IMAGE_EXPORT)
self.imageButton.setFixedSize(56, 56)
self.imageButton.setIconSize(QSize(32, 32))
self.imageButton.clicked.connect(self.get_home_image)
v1_layout = QVBoxLayout()
v1_layout.addWidget(self.imageButton, alignment=Qt.AlignmentFlag.AlignBottom)
h2_layout.addLayout(v1_layout)
# 空白占位符
h2_layout.addItem(
QSpacerItem(25, 10, QSizePolicy.Policy.Fixed, QSizePolicy.Policy.Minimum)
)
# 将底部水平布局添加到垂直布局
v_layout.addLayout(h2_layout)
layout = QVBoxLayout()
scrollArea = ScrollArea()
scrollArea.setWidgetResizable(True)
scrollArea.setWidget(widget)
layout.addWidget(scrollArea)
self.setLayout(layout)
self.set_banner()
def get_home_image(self) -> None:
"""获取主页图片"""
if (
Config.global_config.get(Config.global_config.function_HomeImageMode)
== "默认"
):
pass
elif (
Config.global_config.get(Config.global_config.function_HomeImageMode)
== "自定义"
):
file_path, _ = QFileDialog.getOpenFileName(
self, "打开自定义主页图片", "", "图片文件 (*.png *.jpg *.bmp)"
)
if file_path:
for file in Config.app_path.glob(
"resources/images/Home/BannerCustomize.*"
):
file.unlink()
shutil.copy(
file_path,
Config.app_path
/ f"resources/images/Home/BannerCustomize{Path(file_path).suffix}",
)
logger.info(f"自定义主页图片更换成功:{file_path}")
MainInfoBar.push_info_bar(
"success",
"主页图片更换成功",
"自定义主页图片更换成功!",
3000,
)
else:
logger.warning("自定义主页图片更换失败:未选择图片文件")
MainInfoBar.push_info_bar(
"warning",
"主页图片更换失败",
"未选择图片文件!",
5000,
)
elif (
Config.global_config.get(Config.global_config.function_HomeImageMode)
== "主题图像"
):
# 从远程服务器获取最新主题图像
for _ in range(3):
try:
response = requests.get(
"https://gitee.com/DLmaster_361/AUTO_MAA/raw/server/theme_image.json"
)
theme_image = response.json()
break
except Exception as e:
err = e
time.sleep(0.1)
else:
logger.error(f"获取最新主题图像时出错:\n{err}")
MainInfoBar.push_info_bar(
"error",
"主题图像获取失败",
f"获取最新主题图像信息时出错:\n{err}",
-1,
)
return None
if (Config.app_path / "resources/theme_image.json").exists():
with (Config.app_path / "resources/theme_image.json").open(
mode="r", encoding="utf-8"
) as f:
theme_image_local = json.load(f)
time_local = datetime.strptime(
theme_image_local["time"], "%Y-%m-%d %H:%M"
)
else:
time_local = datetime.strptime("2000-01-01 00:00", "%Y-%m-%d %H:%M")
if not (
Config.app_path / "resources/images/Home/BannerTheme.jpg"
).exists() or (
datetime.now()
> datetime.strptime(theme_image["time"], "%Y-%m-%d %H:%M")
and datetime.strptime(theme_image["time"], "%Y-%m-%d %H:%M")
> time_local
):
response = requests.get(theme_image["url"])
if response.status_code == 200:
with open(
Config.app_path / "resources/images/Home/BannerTheme.jpg", "wb"
) as file:
file.write(response.content)
logger.info(f"主题图像「{theme_image["name"]}」下载成功")
MainInfoBar.push_info_bar(
"success",
"主题图像下载成功",
f"{theme_image["name"]}」下载成功!",
3000,
)
else:
logger.error("主题图像下载失败")
MainInfoBar.push_info_bar(
"error",
"主题图像下载失败",
f"主题图像下载失败:{response.status_code}",
-1,
)
with (Config.app_path / "resources/theme_image.json").open(
mode="w", encoding="utf-8"
) as f:
json.dump(theme_image, f, ensure_ascii=False, indent=4)
else:
logger.info("主题图像已是最新")
MainInfoBar.push_info_bar(
"info",
"主题图像已是最新",
"主题图像已是最新!",
3000,
)
self.set_banner()
def set_banner(self):
"""设置主页图像"""
if (
Config.global_config.get(Config.global_config.function_HomeImageMode)
== "默认"
):
self.banner.set_banner_image(
str(Config.app_path / "resources/images/Home/BannerDefault.png")
)
self.imageButton.hide()
self.banner_text.setVisible(False)
elif (
Config.global_config.get(Config.global_config.function_HomeImageMode)
== "自定义"
):
for file in Config.app_path.glob("resources/images/Home/BannerCustomize.*"):
self.banner.set_banner_image(str(file))
break
self.imageButton.show()
self.banner_text.setVisible(False)
elif (
Config.global_config.get(Config.global_config.function_HomeImageMode)
== "主题图像"
):
self.banner.set_banner_image(
str(Config.app_path / "resources/images/Home/BannerTheme.jpg")
)
self.imageButton.show()
self.banner_text.setVisible(True)
if (Config.app_path / "resources/theme_image.json").exists():
with (Config.app_path / "resources/theme_image.json").open(
mode="r", encoding="utf-8"
) as f:
theme_image = json.load(f)
html_content = theme_image["html"]
else:
html_content = "<h1>主题图像</h1><p>主题图像信息未知</p>"
self.banner_text.setHtml(re.sub(r"<img[^>]*>", "", html_content))
class ButtonGroup(SimpleCardWidget):
"""显示主页和 GitHub 按钮的竖直按钮组"""
def __init__(self, parent=None):
super().__init__(parent=parent)
self.setFixedSize(56, 180)
layout = QVBoxLayout(self)
layout.setAlignment(Qt.AlignmentFlag.AlignTop)
# 创建主页按钮
home_button = IconButton(
FluentIcon.HOME.icon(color=QColor("#fff")),
tip_title="AUTO_MAA官网",
tip_content="AUTO_MAA官方文档站",
isTooltip=True,
)
home_button.setIconSize(QSize(32, 32))
home_button.clicked.connect(self.open_home)
layout.addWidget(home_button)
# 创建 GitHub 按钮
github_button = IconButton(
FluentIcon.GITHUB.icon(color=QColor("#fff")),
tip_title="Github仓库",
tip_content="如果本项目有帮助到您~\n不妨给项目点一个Star⭐",
isTooltip=True,
)
github_button.setIconSize(QSize(32, 32))
github_button.clicked.connect(self.open_github)
layout.addWidget(github_button)
# # 创建 文档 按钮
# doc_button = IconButton(
# FluentIcon.DICTIONARY.icon(color=QColor("#fff")),
# tip_title="自助排障文档",
# tip_content="点击打开自助排障文档,好孩子都能看懂",
# isTooltip=True,
# )
# doc_button.setIconSize(QSize(32, 32))
# doc_button.clicked.connect(self.open_doc)
# layout.addWidget(doc_button)
# 创建 Q群 按钮
doc_button = IconButton(
FluentIcon.CHAT.icon(color=QColor("#fff")),
tip_title="官方社群",
tip_content="加入官方群聊【AUTO_MAA绝赞DeBug中",
isTooltip=True,
)
doc_button.setIconSize(QSize(32, 32))
doc_button.clicked.connect(self.open_chat)
layout.addWidget(doc_button)
# 创建 官方店铺 按钮 (当然没有)
doc_button = IconButton(
FluentIcon.SHOPPING_CART.icon(color=QColor("#fff")),
tip_title="官方店铺",
tip_content="暂时没有官方店铺,但是可以加入官方群聊哦~",
isTooltip=True,
)
doc_button.setIconSize(QSize(32, 32))
doc_button.clicked.connect(self.open_sales)
layout.addWidget(doc_button)
def _normalBackgroundColor(self):
return QColor(0, 0, 0, 96)
def open_home(self):
"""打开主页链接"""
QDesktopServices.openUrl(QUrl("https://clozya.github.io/AUTOMAA_docs"))
def open_github(self):
"""打开 GitHub 链接"""
QDesktopServices.openUrl(QUrl("https://github.com/DLmaster361/AUTO_MAA"))
def open_chat(self):
"""打开 Q群 链接"""
QDesktopServices.openUrl(QUrl("https://qm.qq.com/q/bd9fISNoME"))
def open_doc(self):
"""打开 文档 链接"""
QDesktopServices.openUrl(QUrl("https://clozya.github.io/AUTOMAA_docs"))
def open_sales(self):
"""其实还是打开 Q群 链接"""
QDesktopServices.openUrl(QUrl("https://qm.qq.com/q/bd9fISNoME"))

View File

@@ -26,10 +26,7 @@ v4.2
"""
from loguru import logger
from PySide6.QtWidgets import (
QApplication,
QSystemTrayIcon,
)
from PySide6.QtWidgets import QSystemTrayIcon
from qfluentwidgets import (
Action,
PushButton,
@@ -39,20 +36,27 @@ from qfluentwidgets import (
InfoBar,
InfoBarPosition,
setTheme,
isDarkTheme,
SystemThemeListener,
Theme,
MSFluentWindow,
NavigationItemPosition,
qconfig,
)
from PySide6.QtGui import QIcon, QCloseEvent
from PySide6.QtCore import Qt
from PySide6.QtCore import Qt, QTimer
import json
from datetime import datetime, timedelta
import shutil
from app.core import Config, Task_manager, Main_timer, MainInfoBar
from app.core import Config, TaskManager, MainTimer, MainInfoBar
from app.services import Notify, Crypto, System
from .setting import Setting
from .home import Home
from .member_manager import MemberManager
from .queue_manager import QueueManager
from .dispatch_center import DispatchCenter
from .history import History
from .setting import Setting
class AUTO_MAA(MSFluentWindow):
@@ -63,7 +67,7 @@ class AUTO_MAA(MSFluentWindow):
self.setWindowIcon(QIcon(str(Config.app_path / "resources/icons/AUTO_MAA.ico")))
self.setWindowTitle("AUTO_MAA")
setTheme(Theme.AUTO)
setTheme(Theme.AUTO, lazy=True)
self.splashScreen = SplashScreen(self.windowIcon(), self)
self.show_ui("显示主窗口", if_quick=True)
@@ -72,17 +76,19 @@ class AUTO_MAA(MSFluentWindow):
System.main_window = self.window()
# 创建主窗口
self.setting = Setting(self)
self.home = Home(self)
self.member_manager = MemberManager(self)
self.queue_manager = QueueManager(self)
self.dispatch_center = DispatchCenter(self)
self.history = History(self)
self.setting = Setting(self)
self.addSubInterface(
self.setting,
FluentIcon.SETTING,
"设置",
FluentIcon.SETTING,
NavigationItemPosition.BOTTOM,
self.home,
FluentIcon.HOME,
"主页",
FluentIcon.HOME,
NavigationItemPosition.TOP,
)
self.addSubInterface(
self.member_manager,
@@ -105,6 +111,20 @@ class AUTO_MAA(MSFluentWindow):
FluentIcon.IOT,
NavigationItemPosition.TOP,
)
self.addSubInterface(
self.history,
FluentIcon.HISTORY,
"历史记录",
FluentIcon.HISTORY,
NavigationItemPosition.BOTTOM,
)
self.addSubInterface(
self.setting,
FluentIcon.SETTING,
"设置",
FluentIcon.SETTING,
NavigationItemPosition.BOTTOM,
)
self.stackedWidget.currentChanged.connect(
lambda index: (self.member_manager.refresh() if index == 1 else None)
)
@@ -123,11 +143,13 @@ class AUTO_MAA(MSFluentWindow):
self.dispatch_center.update_top_bar() if index == 3 else None
)
)
self.stackedWidget.currentChanged.connect(
lambda index: (self.history.refresh() if index == 4 else None)
)
# 创建系统托盘及其菜单
self.tray = QSystemTrayIcon(
QIcon(str(Config.app_path / "resources/icons/AUTO_MAA.ico")),
self,
QIcon(str(Config.app_path / "resources/icons/AUTO_MAA.ico")), self
)
self.tray.setToolTip("AUTO_MAA")
self.tray_menu = SystemTrayMenu("AUTO_MAA", self)
@@ -143,22 +165,17 @@ class AUTO_MAA(MSFluentWindow):
self.tray_menu.addSeparator()
# 开始任务菜单项
# self.tray_menu.addActions(
# [
# Action(
# FluentIcon.PLAY,
# "运行自动代理",
# triggered=lambda: self.start_task("自动代理"),
# ),
# Action(
# FluentIcon.PLAY,
# "运行人工排查",
# triggered=lambda: self.start_task("人工排查"),
# ),
# Action(FluentIcon.PAUSE, "中止当前任务", triggered=self.stop_task),
# ]
# )
# self.tray_menu.addSeparator()
self.tray_menu.addActions(
[
Action(FluentIcon.PLAY, "运行自动代理", triggered=self.start_main_task),
Action(
FluentIcon.PAUSE,
"中止所有任务",
triggered=lambda: TaskManager.stop_task("ALL"),
),
]
)
self.tray_menu.addSeparator()
# 退出主程序菜单项
self.tray_menu.addAction(
@@ -169,15 +186,38 @@ class AUTO_MAA(MSFluentWindow):
self.tray.setContextMenu(self.tray_menu)
self.tray.activated.connect(self.on_tray_activated)
Task_manager.create_gui.connect(self.dispatch_center.add_board)
Task_manager.connect_gui.connect(self.dispatch_center.connect_main_board)
TaskManager.create_gui.connect(self.dispatch_center.add_board)
TaskManager.connect_gui.connect(self.dispatch_center.connect_main_board)
Notify.push_info_bar.connect(MainInfoBar.push_info_bar)
self.setting.ui.card_IfShowTray.checkedChanged.connect(
lambda: self.show_ui("配置托盘")
)
self.setting.ui.card_IfToTray.checkedChanged.connect(self.set_min_method)
self.setting.function.card_HomeImageMode.comboBox.currentIndexChanged.connect(
lambda index: (
self.home.get_home_image() if index == 2 else self.home.set_banner()
)
)
self.splashScreen.finish()
self.themeListener = SystemThemeListener(self)
self.themeListener.systemThemeChanged.connect(self.switch_theme)
self.themeListener.start()
def switch_theme(self):
"""切换主题"""
setTheme(Theme.AUTO, lazy=True)
QTimer.singleShot(100, lambda: setTheme(Theme.AUTO, lazy=True))
# 云母特效启用时需要增加重试机制
if self.isMicaEffectEnabled():
QTimer.singleShot(
100,
lambda: self.windowEffect.setMicaEffect(self.winId(), isDarkTheme()),
)
def start_up_task(self) -> None:
"""启动时任务"""
@@ -185,9 +225,19 @@ class AUTO_MAA(MSFluentWindow):
qconfig.load(Config.config_path, Config.global_config)
Config.global_config.save()
# 清理旧日志
self.clean_old_logs()
# 检查密码
self.setting.check_PASSWORD()
# 获取主题图像
if (
Config.global_config.get(Config.global_config.function_HomeImageMode)
== "主题图像"
):
self.home.get_home_image()
# 获取公告
self.setting.show_notice(if_show=False)
@@ -212,6 +262,16 @@ class AUTO_MAA(MSFluentWindow):
info.addWidget(Up)
info.show()
# 直接运行主任务
if Config.global_config.get(Config.global_config.start_IfRunDirectly):
self.start_main_task()
# 直接最小化
if Config.global_config.get(Config.global_config.start_IfMinimizeDirectly):
self.titleBar.minBtn.click()
def set_min_method(self) -> None:
"""设置最小化方法"""
@@ -230,40 +290,66 @@ class AUTO_MAA(MSFluentWindow):
if reason == QSystemTrayIcon.DoubleClick:
self.show_ui("显示主窗口")
# def start_task(self, mode):
# """调起对应任务"""
# if self.main.MaaManager.isRunning():
# Notify.push_notification(
# f"无法运行{mode}",
# "当前已有任务正在运行,请在该任务结束后重试",
# "当前已有任务正在运行,请在该任务结束后重试",
# 3,
# )
# else:
# self.main.maa_starter(mode)
def clean_old_logs(self):
"""
删除超过用户设定天数的日志文件(基于目录日期)
"""
# def stop_task(self):
# """中止当前任务"""
# if self.main.MaaManager.isRunning():
# if (
# self.main.MaaManager.mode == "自动代理"
# or self.main.MaaManager.mode == "人工排查"
# ):
# self.main.maa_ender(f"{self.main.MaaManager.mode}_结束")
# elif "设置MAA" in self.main.MaaManager.mode:
# Notify.push_notification(
# "正在设置MAA",
# "正在运行设置MAA任务无法中止",
# "正在运行设置MAA任务无法中止",
# 3,
# )
# else:
# Notify.push_notification(
# "无任务运行!",
# "当前无任务正在运行,无需中止",
# "当前无任务正在运行,无需中止",
# 3,
# )
if (
Config.global_config.get(Config.global_config.function_HistoryRetentionTime)
== 0
):
logger.info("由于用户设置日志永久保留,跳过日志清理")
return
deleted_count = 0
for date_folder in (Config.app_path / "history").iterdir():
if not date_folder.is_dir():
continue # 只处理日期文件夹
try:
# 只检查 `YYYY-MM-DD` 格式的文件夹
folder_date = datetime.strptime(date_folder.name, "%Y-%m-%d")
if datetime.now() - folder_date > timedelta(
days=Config.global_config.get(
Config.global_config.function_HistoryRetentionTime
)
):
shutil.rmtree(date_folder, ignore_errors=True)
deleted_count += 1
logger.info(f"已删除超期日志目录: {date_folder}")
except ValueError:
logger.warning(f"非日期格式的目录: {date_folder}")
logger.info(f"清理完成: {deleted_count} 个日期目录")
def start_main_task(self) -> None:
"""启动主任务"""
if (Config.app_path / "config/QueueConfig/调度队列_1.json").exists():
with (Config.app_path / "config/QueueConfig/调度队列_1.json").open(
mode="r", encoding="utf-8"
) as f:
info = json.load(f)
logger.info("自动添加任务调度队列_1")
TaskManager.add_task("自动代理_主调度台", "主任务队列", info)
elif (Config.app_path / "config/MaaConfig/脚本_1").exists():
info = {"Queue": {"Member_1": "脚本_1"}}
logger.info("自动添加任务脚本_1")
TaskManager.add_task("自动代理_主调度台", "主任务队列", info)
else:
logger.worning("启动主任务失败:未找到有效的主任务配置文件")
MainInfoBar.push_info_bar(
"warning", "启动主任务失败", "“调度队列_1”与“脚本_1”均不存在", -1
)
def show_ui(self, mode: str, if_quick: bool = False) -> None:
"""配置窗口状态"""
@@ -287,6 +373,8 @@ class AUTO_MAA(MSFluentWindow):
)
self.window().setGeometry(location[0], location[1], size[0], size[1])
self.window().show()
self.window().raise_()
self.window().activateWindow()
if not if_quick:
if Config.global_config.get(Config.global_config.ui_maximized):
self.window().showMaximized()
@@ -330,13 +418,17 @@ class AUTO_MAA(MSFluentWindow):
self.show_ui("隐藏到托盘", if_quick=True)
# 清理各功能线程
Main_timer.Timer.stop()
Main_timer.Timer.deleteLater()
Task_manager.stop_task("ALL")
MainTimer.Timer.stop()
MainTimer.Timer.deleteLater()
TaskManager.stop_task("ALL")
# 关闭数据库连接
Config.close_database()
# 关闭主题监听
self.themeListener.terminate()
self.themeListener.deleteLater()
logger.info("AUTO_MAA主程序关闭")
logger.info("----------------END----------------")

View File

@@ -46,9 +46,12 @@ from qfluentwidgets import (
HeaderCardWidget,
CommandBar,
ExpandGroupSettingCard,
ComboBoxSettingCard,
PushSettingCard,
)
from PySide6.QtCore import Qt
import requests
import time
from functools import partial
from pathlib import Path
from typing import List
@@ -56,8 +59,9 @@ from datetime import datetime, timedelta
import json
import shutil
from app.core import Config, MainInfoBar, Task_manager
from app.core import Config, MainInfoBar, TaskManager
from app.services import Crypto
from app.utils import Updater
from .Widget import (
LineEditMessageBox,
LineEditSettingCard,
@@ -68,10 +72,7 @@ from .Widget import (
class MemberManager(QWidget):
def __init__(
self,
parent=None,
):
def __init__(self, parent=None):
super().__init__(parent)
self.setObjectName("脚本管理")
@@ -107,14 +108,21 @@ class MemberManager(QWidget):
]
)
self.tools.addSeparator()
self.key = Action(
FluentIcon.HIDE,
"显示/隐藏密码",
checkable=True,
triggered=self.show_password,
)
self.tools.addAction(
self.key,
Action(
FluentIcon.DOWNLOAD,
"脚本下载器",
triggered=self.member_downloader,
)
)
self.tools.addSeparator()
self.tools.addAction(
Action(
FluentIcon.HIDE,
"显示/隐藏密码",
checkable=True,
triggered=self.show_password,
)
)
layout.addWidget(self.tools)
@@ -291,6 +299,65 @@ class MemberManager(QWidget):
self.member_manager.show_SettingBox(index + 1)
def member_downloader(self):
"""脚本下载器"""
choice = ComboBoxMessageBox(
self.window(),
"选择一个脚本类型以下载相应脚本",
["选择脚本类型"],
[["MAA"]],
)
if choice.exec() and choice.input[0].currentIndex() != -1:
if choice.input[0].currentText() == "MAA":
(Config.app_path / "script/MAA").mkdir(parents=True, exist_ok=True)
folder = QFileDialog.getExistingDirectory(
self, "选择MAA下载目录", str(Config.app_path / "script/MAA")
)
if not folder:
logger.warning("选择MAA下载目录时未选择文件夹")
MainInfoBar.push_info_bar(
"warning", "警告", "未选择MAA下载目录", 5000
)
return None
# 从mirrorc服务器获取最新版本信息
for _ in range(3):
try:
response = requests.get(
"https://mirrorc.top/api/resources/MAA/latest?user_agent=MaaWpfGui&os=win&arch=x64&channel=stable"
)
maa_info = response.json()
break
except Exception as e:
err = e
time.sleep(0.1)
else:
choice = MessageBox(
"错误",
f"获取版本信息时出错:\n{err}",
self.window(),
)
choice.cancelButton.hide()
choice.buttonLayout.insertStretch(1)
if choice.exec():
return None
maa_version = list(
map(
int,
maa_info["data"]["version_name"][1:]
.replace("-beta", "")
.split("."),
)
)
while len(maa_version) < 4:
maa_version.append(0)
self.downloader = Updater(Path(folder), "MAA", maa_version, [])
self.downloader.show()
def show_password(self):
if Config.PASSWORD == "":
@@ -522,7 +589,7 @@ class MaaSettingBox(QWidget):
)
)
self.card_Set.clicked.connect(
lambda: Task_manager.add_task("设置MAA_全局", self.name, None)
lambda: TaskManager.add_task("设置MAA_全局", self.name, None)
)
Layout.addWidget(self.card_Name)
@@ -571,16 +638,15 @@ class MaaSettingBox(QWidget):
class RunSetSettingCard(ExpandGroupSettingCard):
def __init__(self, parent=None):
super().__init__(
FluentIcon.SETTING,
"运行",
"MAA运行调控选项",
parent,
super().__init__(FluentIcon.SETTING, "运行", "MAA运行调控选项", parent)
self.card_TaskTransitionMethod = ComboBoxSettingCard(
configItem=Config.maa_config.RunSet_TaskTransitionMethod,
icon=FluentIcon.PAGE_RIGHT,
title="任务切换方式",
content="简洁用户列表下相邻两个任务间的切换方式",
texts=["直接切换账号", "重启明日方舟", "重启模拟器"],
)
widget = QWidget()
Layout = QVBoxLayout(widget)
self.ProxyTimesLimit = SpinBoxSettingCard(
(0, 1024),
FluentIcon.PAGE_RIGHT,
@@ -588,7 +654,6 @@ class MaaSettingBox(QWidget):
"当用户本日代理成功次数超过该阈值时跳过代理阈值为“0”时视为无代理次数上限",
Config.maa_config.RunSet_ProxyTimesLimit,
)
self.AnnihilationTimeLimit = SpinBoxSettingCard(
(1, 1024),
FluentIcon.PAGE_RIGHT,
@@ -596,7 +661,6 @@ class MaaSettingBox(QWidget):
"MAA日志无变化时间超过该阈值视为超时单位为分钟",
Config.maa_config.RunSet_AnnihilationTimeLimit,
)
self.RoutineTimeLimit = SpinBoxSettingCard(
(1, 1024),
FluentIcon.PAGE_RIGHT,
@@ -604,7 +668,6 @@ class MaaSettingBox(QWidget):
"MAA日志无变化时间超过该阈值视为超时单位为分钟",
Config.maa_config.RunSet_RoutineTimeLimit,
)
self.RunTimesLimit = SpinBoxSettingCard(
(1, 1024),
FluentIcon.PAGE_RIGHT,
@@ -613,14 +676,15 @@ class MaaSettingBox(QWidget):
Config.maa_config.RunSet_RunTimesLimit,
)
widget = QWidget()
Layout = QVBoxLayout(widget)
Layout.addWidget(self.card_TaskTransitionMethod)
Layout.addWidget(self.ProxyTimesLimit)
Layout.addWidget(self.AnnihilationTimeLimit)
Layout.addWidget(self.RoutineTimeLimit)
Layout.addWidget(self.RunTimesLimit)
self.viewLayout.setContentsMargins(0, 0, 0, 0)
self.viewLayout.setSpacing(0)
self.addGroupWidget(widget)
class UserSettingCard(HeaderCardWidget):
@@ -746,7 +810,7 @@ class MaaSettingBox(QWidget):
):
set_book = ["routine", "annihilation"]
Task_manager.add_task(
TaskManager.add_task(
"设置MAA_用户",
self.name,
{

View File

@@ -60,10 +60,7 @@ from .Widget import (
class QueueManager(QWidget):
def __init__(
self,
parent=None,
):
def __init__(self, parent=None):
super().__init__(parent)
self.setObjectName("调度队列")

View File

@@ -58,19 +58,10 @@ from .Widget import LineEditMessageBox, LineEditSettingCard, PasswordLineEditSet
class Setting(QWidget):
def __init__(
self,
parent=None,
):
def __init__(self, parent=None):
super().__init__(parent)
self.setObjectName("设置")
layout = QVBoxLayout()
scrollArea = ScrollArea()
scrollArea.setWidgetResizable(True)
content_widget = QWidget()
content_layout = QVBoxLayout(content_widget)
@@ -97,10 +88,11 @@ class Setting(QWidget):
content_layout.addWidget(self.updater)
content_layout.addWidget(self.other)
scrollArea = ScrollArea()
scrollArea.setWidgetResizable(True)
scrollArea.setWidget(content_widget)
layout = QVBoxLayout()
layout.addWidget(scrollArea)
self.setLayout(layout)
def agree_bilibili(self) -> None:
@@ -339,7 +331,7 @@ class Setting(QWidget):
if main_version_remote > main_version_current:
self.updater.update_process.accomplish.connect(self.update_main)
# 显示更新页面
self.updater.ui.show()
self.updater.show()
# 更新主程序
elif main_version_remote > main_version_current:
@@ -363,7 +355,7 @@ class Setting(QWidget):
def show_notice(self, if_show: bool = True):
"""显示公告"""
# 从远程服务器获取最新版本信息
# 从远程服务器获取最新公告
for _ in range(3):
try:
response = requests.get(
@@ -417,6 +409,20 @@ class FunctionSettingCard(HeaderCardWidget):
super().__init__(parent)
self.setTitle("功能")
self.card_HomeImageMode = ComboBoxSettingCard(
configItem=Config.global_config.function_HomeImageMode,
icon=FluentIcon.PAGE_RIGHT,
title="主页背景图模式",
content="选择主页背景图的来源",
texts=["默认", "自定义", "主题图像"],
)
self.card_HistoryRetentionTime = ComboBoxSettingCard(
configItem=Config.global_config.function_HistoryRetentionTime,
icon=FluentIcon.PAGE_RIGHT,
title="历史记录保留时间",
content="选择历史记录的保留时间,超期自动清理",
texts=["7 天", "15 天", "30 天", "60 天", "永久"],
)
self.card_IfAllowSleep = SwitchSettingCard(
icon=FluentIcon.PAGE_RIGHT,
title="启动时阻止系统休眠",
@@ -432,6 +438,8 @@ class FunctionSettingCard(HeaderCardWidget):
)
Layout = QVBoxLayout()
Layout.addWidget(self.card_HomeImageMode)
Layout.addWidget(self.card_HistoryRetentionTime)
Layout.addWidget(self.card_IfAllowSleep)
Layout.addWidget(self.card_IfSilence)
Layout.addWidget(self.card_IfAgreeBilibili)
@@ -454,10 +462,10 @@ class FunctionSettingCard(HeaderCardWidget):
configItem=Config.global_config.function_IfSilence,
)
self.card_BossKey = LineEditSettingCard(
text="请输入安卓模拟器老",
text="请输入安卓模拟器老",
icon=FluentIcon.PAGE_RIGHT,
title="模拟器老",
content="输入模拟器老快捷键,以“+”分隔",
title="模拟器老",
content="输入模拟器老快捷键,以“+”分隔",
configItem=Config.global_config.function_BossKey,
)
@@ -484,14 +492,21 @@ class StartSettingCard(HeaderCardWidget):
)
self.card_IfRunDirectly = SwitchSettingCard(
icon=FluentIcon.PAGE_RIGHT,
title="启动后直接运行",
content="启动AUTO_MAA后自动运行任务(暂不可用)",
title="启动后直接运行主任务",
content="启动AUTO_MAA后自动运行自动代理任务,优先级:调度队列 1 > 脚本 1",
configItem=Config.global_config.start_IfRunDirectly,
)
self.card_IfMinimizeDirectly = SwitchSettingCard(
icon=FluentIcon.PAGE_RIGHT,
title="启动后直接最小化",
content="启动AUTO_MAA后直接最小化",
configItem=Config.global_config.start_IfMinimizeDirectly,
)
Layout = QVBoxLayout()
Layout.addWidget(self.card_IfSelfStart)
Layout.addWidget(self.card_IfRunDirectly)
Layout.addWidget(self.card_IfMinimizeDirectly)
self.viewLayout.addLayout(Layout)
@@ -527,38 +542,82 @@ class NotifySettingCard(HeaderCardWidget):
self.setTitle("通知")
self.card_IfSendErrorOnly = SwitchSettingCard(
icon=FluentIcon.PAGE_RIGHT,
title="仅推送异常信息",
content="仅在任务出现异常时推送通知",
configItem=Config.global_config.notify_IfSendErrorOnly,
)
self.card_IfPushPlyer = SwitchSettingCard(
icon=FluentIcon.PAGE_RIGHT,
title="推送系统通知",
content="推送系统级通知,不会在通知中心停留",
configItem=Config.global_config.notify_IfPushPlyer,
)
self.card_SendMail = self.SendMailSettingCard(self)
self.card_NotifyContent = self.NotifyContentSettingCard(self)
self.card_Plyer = self.PlyerSettingCard(self)
self.card_EMail = self.EMailSettingCard(self)
self.card_ServerChan = self.ServerChanSettingCard(self)
self.card_CompanyWebhookBot = self.CompanyWechatPushSettingCard(self)
Layout = QVBoxLayout()
Layout.addWidget(self.card_IfSendErrorOnly)
Layout.addWidget(self.card_IfPushPlyer)
Layout.addWidget(self.card_SendMail)
Layout.addWidget(self.card_NotifyContent)
Layout.addWidget(self.card_Plyer)
Layout.addWidget(self.card_EMail)
Layout.addWidget(self.card_ServerChan)
Layout.addWidget(self.card_CompanyWebhookBot)
self.viewLayout.addLayout(Layout)
class SendMailSettingCard(ExpandGroupSettingCard):
class NotifyContentSettingCard(ExpandGroupSettingCard):
def __init__(self, parent=None):
super().__init__(
FluentIcon.SETTING,
"推送邮件通知",
"通过电子邮箱推送任务结果",
parent,
FluentIcon.SETTING, "通知内容选项", "选择需要推送的通知内容", parent
)
self.card_SendTaskResultTime = ComboBoxSettingCard(
configItem=Config.global_config.notify_SendTaskResultTime,
icon=FluentIcon.PAGE_RIGHT,
title="推送任务结果选项",
content="选择推送自动代理与人工排查任务结果的时机",
texts=["不推送", "任何时刻", "仅失败时"],
)
self.card_IfSendStatistic = SwitchSettingCard(
icon=FluentIcon.PAGE_RIGHT,
title="推送统计信息",
content="推送自动代理统计信息的通知",
configItem=Config.global_config.notify_IfSendStatistic,
)
self.card_IfSendSixStar = SwitchSettingCard(
icon=FluentIcon.PAGE_RIGHT,
title="推送公招高资喜报",
content="公招出现六星词条时推送喜报",
configItem=Config.global_config.notify_IfSendSixStar,
)
widget = QWidget()
Layout = QVBoxLayout(widget)
Layout.addWidget(self.card_SendTaskResultTime)
Layout.addWidget(self.card_IfSendStatistic)
Layout.addWidget(self.card_IfSendSixStar)
self.viewLayout.setContentsMargins(0, 0, 0, 0)
self.viewLayout.setSpacing(0)
self.addGroupWidget(widget)
class PlyerSettingCard(ExpandGroupSettingCard):
def __init__(self, parent=None):
super().__init__(
FluentIcon.SETTING, "推送系统通知", "Plyer系统通知推送渠道", parent
)
self.card_IfPushPlyer = SwitchSettingCard(
icon=FluentIcon.PAGE_RIGHT,
title="推送系统通知",
content="使用Plyer推送系统级通知不会在通知中心停留",
configItem=Config.global_config.notify_IfPushPlyer,
)
widget = QWidget()
Layout = QVBoxLayout(widget)
Layout.addWidget(self.card_IfPushPlyer)
self.viewLayout.setContentsMargins(0, 0, 0, 0)
self.viewLayout.setSpacing(0)
self.addGroupWidget(widget)
class EMailSettingCard(ExpandGroupSettingCard):
def __init__(self, parent=None):
super().__init__(
FluentIcon.SETTING, "推送邮件通知", "电子邮箱通知推送渠道", parent
)
self.card_IfSendMail = SwitchSettingCard(
@@ -612,7 +671,7 @@ class NotifySettingCard(HeaderCardWidget):
super().__init__(
FluentIcon.SETTING,
"推送ServerChan通知",
"通过ServerChan通知推送任务结果",
"ServerChan通知推送渠道",
parent,
)
@@ -659,7 +718,7 @@ class NotifySettingCard(HeaderCardWidget):
super().__init__(
FluentIcon.SETTING,
"推送企业微信机器人通知",
"通过企业微信机器人Webhook通知推送任务结果",
"企业微信机器人Webhook通知推送渠道",
parent,
)
@@ -750,11 +809,11 @@ class OtherSettingCard(HeaderCardWidget):
content="查看AUTO_MAA的最新公告",
)
self.card_UserDocs = HyperlinkCard(
url="https://docs.qq.com/aio/DQ2NwUHRiWGtMWHBy",
text="查看使用指南",
url="https://clozya.github.io/AUTOMAA_docs",
text="访问",
icon=FluentIcon.PAGE_RIGHT,
title="使用指南",
content="查看AUTO_MAA的使用教程和文档",
title="AUTO_MAA官方文档站",
content="访问AUTO_MAA的官方文档站,获取使用指南和项目相关信息",
)
self.card_Association = self.AssociationSettingCard()

View File

@@ -33,14 +33,16 @@ import subprocess
import time
from pathlib import Path
from PySide6.QtWidgets import (
QApplication,
QDialog,
QVBoxLayout,
from PySide6.QtWidgets import QApplication, QDialog, QVBoxLayout, QHBoxLayout
from qfluentwidgets import (
ProgressBar,
IndeterminateProgressBar,
BodyLabel,
PushButton,
EditableComboBox,
)
from qfluentwidgets import ProgressBar, IndeterminateProgressBar, BodyLabel
from PySide6.QtGui import QIcon
from PySide6.QtCore import QObject, QThread, Signal
from PySide6.QtGui import QIcon, QCloseEvent
from PySide6.QtCore import QThread, Signal, QEventLoop
def version_text(version_numb: list) -> str:
@@ -59,6 +61,8 @@ class UpdateProcess(QThread):
info = Signal(str)
progress = Signal(int, int, int)
question = Signal(dict)
question_response = Signal(str)
accomplish = Signal()
def __init__(
@@ -70,8 +74,11 @@ class UpdateProcess(QThread):
self.name = name
self.main_version = main_version
self.updater_version = updater_version
self.download_path = app_path / "AUTO_MAA_Update.zip" # 临时下载文件的路径
self.download_path = app_path / "DOWNLOAD_TEMP.zip" # 临时下载文件的路径
self.version_path = app_path / "resources/version.json"
self.response = None
self.question_response.connect(self._capture_response)
def run(self) -> None:
@@ -81,25 +88,41 @@ class UpdateProcess(QThread):
self.info.emit("正在获取下载链接")
url_list = self.get_download_url()
url_dict = {}
# 验证下载地址
for i, url in enumerate(url_list):
if self.isInterruptionRequested():
return None
self.progress.emit(0, len(url_list), i)
# 验证下载地址并获取文件大小
for i in range(len(url_list)):
try:
self.info.emit(f"正在验证下载地址:{url_list[i]}")
response = requests.get(url_list[i], stream=True)
self.info.emit(f"正在验证下载地址:{url}")
response = requests.get(url, stream=True)
if response.status_code != 200:
self.info.emit(
f"连接失败,错误代码 {response.status_code} ,正在切换代理({i+1}/{len(url_list)}"
)
self.info.emit(f"连接失败,错误代码 {response.status_code}")
time.sleep(1)
continue
file_size = response.headers.get("Content-Length")
break
url_dict[url] = response.elapsed.total_seconds()
except requests.RequestException:
self.info.emit(f"请求超时,正在切换代理({i+1}/{len(url_list)}")
self.info.emit(f"请求超时")
time.sleep(1)
else:
self.info.emit(f"连接失败,已尝试所有{len(url_list)}个代理")
download_url = self.push_question(url_dict)
# 获取文件大小
try:
self.info.emit(f"正在连接下载地址:{download_url}")
self.progress.emit(0, 0, 0)
response = requests.get(download_url, stream=True)
if response.status_code != 200:
self.info.emit(f"连接失败,错误代码 {response.status_code}")
return None
file_size = response.headers.get("Content-Length")
except requests.RequestException:
self.info.emit(f"请求超时")
return None
if file_size is None:
@@ -118,6 +141,9 @@ class UpdateProcess(QThread):
for chunk in response.iter_content(chunk_size=8192):
if self.isInterruptionRequested():
break
# 写入已下载数据
f.write(chunk)
downloaded_size += len(chunk)
@@ -143,6 +169,10 @@ class UpdateProcess(QThread):
)
self.progress.emit(0, 100, int(downloaded_size / file_size * 100))
if self.isInterruptionRequested() and self.download_path.exists():
self.download_path.unlink()
return None
except Exception as e:
e = str(e)
e = "\n".join([e[_ : _ + 75] for _ in range(0, len(e), 75)])
@@ -153,6 +183,9 @@ class UpdateProcess(QThread):
try:
while True:
if self.isInterruptionRequested():
self.download_path.unlink()
return None
try:
self.info.emit("正在解压更新文件")
self.progress.emit(0, 0, 0)
@@ -160,7 +193,7 @@ class UpdateProcess(QThread):
zip_ref.extractall(self.app_path)
break
except PermissionError:
self.info.emit("解压出错:AUTO_MAA正在运行,正在等待其关闭")
self.info.emit(f"解压出错:{self.name}正在运行,正在等待其关闭")
time.sleep(1)
self.info.emit("正在删除临时文件")
@@ -178,22 +211,34 @@ class UpdateProcess(QThread):
return None
# 更新version文件
with open(self.version_path, "r", encoding="utf-8") as f:
version_info = json.load(f)
if self.name == "AUTO_MAA更新器":
version_info["updater_version"] = ".".join(map(str, self.updater_version))
elif self.name == "AUTO_MAA主程序":
version_info["main_version"] = ".".join(map(str, self.main_version))
with open(self.version_path, "w", encoding="utf-8") as f:
json.dump(version_info, f, ensure_ascii=False, indent=4)
if not self.isInterruptionRequested and self.name in [
"AUTO_MAA主程序",
"AUTO_MAA更新器",
]:
with open(self.version_path, "r", encoding="utf-8") as f:
version_info = json.load(f)
if self.name == "AUTO_MAA主程序":
version_info["main_version"] = ".".join(map(str, self.main_version))
elif self.name == "AUTO_MAA更新器":
version_info["updater_version"] = ".".join(
map(str, self.updater_version)
)
with open(self.version_path, "w", encoding="utf-8") as f:
json.dump(version_info, f, ensure_ascii=False, indent=4)
# 主程序更新完成后打开AUTO_MAA
if self.name == "AUTO_MAA主程序":
if not self.isInterruptionRequested and self.name == "AUTO_MAA主程序":
subprocess.Popen(
str(self.app_path / "AUTO_MAA.exe"),
shell=True,
creationflags=subprocess.CREATE_NO_WINDOW,
)
elif not self.isInterruptionRequested and self.name == "MAA":
subprocess.Popen(
str(self.app_path / "MAA.exe"),
shell=True,
creationflags=subprocess.CREATE_NO_WINDOW,
)
self.accomplish.emit()
@@ -239,6 +284,9 @@ class UpdateProcess(QThread):
url_list.append(
f"https://gitee.com/DLmaster_361/AUTO_MAA/releases/download/{version_text(self.main_version)}/AUTO_MAA_{version_text(self.main_version)}.zip"
)
url_list.append(
f"https://jp-download.fearr.xyz/AUTO_MAA/AUTO_MAA_{version_text(self.main_version)}.zip"
)
for i in range(len(PROXY_list)):
url_list.append(
f"{PROXY_list[i]}https://github.com/DLmaster361/AUTO_MAA/releases/download/{version_text(self.main_version)}/AUTO_MAA_{version_text(self.main_version)}.zip"
@@ -247,33 +295,63 @@ class UpdateProcess(QThread):
url_list.append(
f"https://gitee.com/DLmaster_361/AUTO_MAA/releases/download/{version_text(self.main_version)}/Updater_{version_text(self.updater_version)}.zip"
)
url_list.append(
f"https://jp-download.fearr.xyz/AUTO_MAA/Updater_{version_text(self.updater_version)}.zip"
)
for i in range(len(PROXY_list)):
url_list.append(
f"{PROXY_list[i]}https://github.com/DLmaster361/AUTO_MAA/releases/download/{version_text(self.main_version)}/Updater_{version_text(self.updater_version)}.zip"
)
elif self.name == "MAA":
url_list.append(
f"https://jp-download.fearr.xyz/MAA/MAA-{version_text(self.main_version)}-win-x64.zip"
)
for i in range(len(PROXY_list)):
url_list.append(
f"{PROXY_list[i]}https://github.com/MaaAssistantArknights/MaaAssistantArknights/releases/download/{version_text(self.main_version)}/MAA-{version_text(self.main_version)}-win-x64.zip"
)
return url_list
def push_question(self, url_dict: dict) -> str:
self.question.emit(url_dict)
loop = QEventLoop()
self.question_response.connect(loop.quit)
loop.exec()
return self.response
class Updater(QObject):
def _capture_response(self, response: str) -> None:
self.response = response
class Updater(QDialog):
def __init__(
self, app_path: Path, name: str, main_version: list, updater_version: list
) -> None:
super().__init__()
self.ui = QDialog()
self.ui.setWindowTitle("AUTO_MAA更新器")
self.ui.resize(700, 70)
self.ui.setWindowIcon(
QIcon(str(app_path / "resources/icons/AUTO_MAA_Updater.ico"))
self.setWindowTitle("AUTO_MAA更新器")
self.setWindowIcon(
QIcon(
str(
Path(sys.argv[0]).resolve().parent
/ "resources/icons/AUTO_MAA_Updater.ico"
)
)
)
# 创建垂直布局
self.Layout = QVBoxLayout(self.ui)
self.Layout = QVBoxLayout(self)
self.info = BodyLabel("正在初始化", self.ui)
self.progress_1 = IndeterminateProgressBar(self.ui)
self.progress_2 = ProgressBar(self.ui)
self.info = BodyLabel("正在初始化", self)
self.progress_1 = IndeterminateProgressBar(self)
self.progress_2 = ProgressBar(self)
self.combo_box = EditableComboBox(self)
self.button = PushButton("继续", self)
self.h_layout = QHBoxLayout()
self.h_layout.addStretch(1)
self.h_layout.addWidget(self.button)
self.update_progress(0, 0, 0)
@@ -281,6 +359,8 @@ class Updater(QObject):
self.Layout.addStretch(1)
self.Layout.addWidget(self.progress_1)
self.Layout.addWidget(self.progress_2)
self.Layout.addWidget(self.combo_box)
self.Layout.addLayout(self.h_layout)
self.Layout.addStretch(1)
self.update_process = UpdateProcess(
@@ -289,21 +369,59 @@ class Updater(QObject):
self.update_process.info.connect(self.update_info)
self.update_process.progress.connect(self.update_progress)
self.update_process.question.connect(self.question)
self.update_process.start()
def update_info(self, text: str) -> None:
self.info.setText(text)
def update_progress(self, begin: int, end: int, current: int) -> None:
if begin == 0 and end == 0:
def update_progress(
self, begin: int, end: int, current: int, if_show_combo_box: bool = False
) -> None:
self.combo_box.setVisible(if_show_combo_box)
self.button.setVisible(if_show_combo_box)
if if_show_combo_box:
self.progress_1.setVisible(False)
self.progress_2.setVisible(False)
self.resize(1000, 90)
elif begin == 0 and end == 0:
self.progress_2.setVisible(False)
self.progress_1.setVisible(True)
self.resize(700, 70)
else:
self.progress_1.setVisible(False)
self.progress_2.setVisible(True)
self.progress_2.setRange(begin, end)
self.progress_2.setValue(current)
self.resize(700, 70)
def question(self, url_dict: dict) -> None:
self.update_info("测速完成,请选择或自行输入一个合适下载地址:")
self.update_progress(0, 0, 0, True)
url_dict = dict(sorted(url_dict.items(), key=lambda item: item[1]))
for url, time in url_dict.items():
self.combo_box.addItem(f"{url} | 响应时间:{time:.3f}")
self.button.clicked.connect(
lambda: self.update_process.question_response.emit(
self.combo_box.currentText().split(" | ")[0]
)
)
def closeEvent(self, event: QCloseEvent):
"""清理残余进程"""
self.update_process.requestInterruption()
self.update_process.quit()
self.update_process.wait()
event.accept()
class AUTO_MAA_Updater(QApplication):
@@ -313,7 +431,7 @@ class AUTO_MAA_Updater(QApplication):
super().__init__()
self.main = Updater(app_path, name, main_version, updater_version)
self.main.ui.show()
self.main.show()
if __name__ == "__main__":

View File

@@ -8,5 +8,6 @@ pywin32
pyautogui
pycryptodome
requests
Jinja2
serverchan_sdk
nuitka==2.6

View File

@@ -1,5 +1,7 @@
#主界面
"MainFunction.PostActions": "12" #完成后
"MainFunction.PostActions": "8" #完成后退出MAA
"MainFunction.PostActions": "9" #完成后退出MAA和游戏
"MainFunction.PostActions": "12" #完成后退出MAA和模拟器
"TaskQueue.WakeUp.IsChecked": "True" #开始唤醒
"TaskQueue.Recruiting.IsChecked": "True" #自动公招
"TaskQueue.Base.IsChecked": "True" #基建换班

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

BIN
resources/images/Home.png Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 2.8 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 2.8 MiB

View File

@@ -1,7 +1,7 @@
{
"main_version": "4.2.3.0",
"updater_version": "1.1.1.3",
"announcement": "\n## 新增功能\n- 添加用户每日代理次数上限功能 #15\n- 新增代理成功消息推送渠道Server酱与企业微信群机器人推送\n- 添加更新类别可选项\n- 添加调度队列完成任务后行为选项\n- 初步完成`托管bilibili游戏隐私政策`功能\n## 修复BUG\n- 修复自定义基建无法正常使用的问题\n- 修正人工排查文案\n- 修复高级MAA配置序号错位\n- 修复高级用户列表无法配置问题\n- 修复主调度台选项乱动问题\n- 修复更新器文件夹定位问题\n- 修复窗口记忆功能失效问题\n## 程序优化\n- 优化弹窗逻辑\n- 优化静默判定逻辑\n- 调整MAA设置目录时打开当前已配置的目录位置\n- 邮箱推送功能调整,改由用户提供发信邮箱",
"main_version": "4.2.5.1",
"updater_version": "1.1.2.1",
"announcement": "\n## 新增功能\n- 暂无\n## 修复BUG\n- 修复统计信息HTML模板公招匹配错误\n## 程序优化\n- 暂无",
"proxy_list": [
"",
"https://gitproxy.click/",