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

Merge branch 'DLMS_dev' into dev

Browse Source
This commit is contained in:
DLmaster361
2025-04-15 22:34:52 +08:00
parent 49ebd50077 fe1910d16f
commit 986061dc97
10 changed files with 208 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
app/core/config.py
View File

@@ -599,7 +599,7 @@ class MaaUserConfig(QConfig):
class AppConfig(GlobalConfig):
VERSION = "4.3.3.0"
VERSION = "4.3.4.1"
gameid_refreshed = Signal()
PASSWORD_refreshed = Signal()
@@ -678,7 +678,8 @@ class AppConfig(GlobalConfig):
for _ in range(3):
try:
response = requests.get(
"https://ota.maa.plus/MaaAssistantArknights/api/gui/StageActivity.json"
"https://ota.maa.plus/MaaAssistantArknights/api/gui/StageActivity.json",
timeout=10,
)
gameid_infos: List[
Dict[str, Union[str, Dict[str, Union[str, int]]]]

226
app/models/MAA.py
View File

@@ -186,7 +186,7 @@ class MaaManager(QObject):
logger.info(f"{self.name} | 开始代理用户: {user[0]}")
# 初始化代理情况记录和模式替换记录
# 初始化代理情况记录和模式替换表
run_book = {"Annihilation": False, "Routine": False}
mode_book = {
"Annihilation": "自动代理_剿灭",
@@ -196,92 +196,80 @@ class MaaManager(QObject):
# 简洁模式用户默认开启日常选项
if user_data["Info"]["Mode"] == "简洁":
user_data["Info"]["Routine"] = True
# 详细模式用户首次代理需打开模拟器
elif user_data["Info"]["Mode"] == "详细":
check_book = {
"Annihilation": True,
"Routine": True,
}
self.if_open_emulator = True
user_logs_list = []
user_start_time = datetime.now()
# 尝试次数循环
for i in range(self.set["RunSet"]["RunTimesLimit"]):
# 剿灭-日常模式循环
for mode in ["Annihilation", "Routine"]:
if self.isInterruptionRequested:
break
logger.info(
f"{self.name} | 用户: {user[0]} - 尝试次数: {i + 1}/{self.set["RunSet"]["RunTimesLimit"]}"
# 剿灭模式；满足条件跳过剿灭
if (
mode == "Annihilation"
and self.set["RunSet"]["AnnihilationWeeklyLimit"]
and datetime.strptime(
user_data["Data"]["LastAnnihilationDate"], "%Y-%m-%d"
).isocalendar()[:2]
== datetime.strptime(curdate, "%Y-%m-%d").isocalendar()[:2]
):
logger.info(
f"{self.name} | 用户: {user_data["Info"]["Name"]} - 本周剿灭模式已达上限，跳过执行剿灭任务"
)
run_book[mode] = True
continue
else:
self.weekly_annihilation_limit_reached = False
if not user_data["Info"][mode]:
run_book[mode] = True
continue
if user_data["Info"]["Mode"] == "详细":
if not (
self.data[user[2]]["Path"] / f"{mode}/gui.json"
).exists():
logger.error(
f"{self.name} | 用户: {user[0]} - 未找到{mode_book[mode][5:7]}配置文件"
)
self.push_info_bar.emit(
"error",
"启动MAA代理进程失败",
f"未找到{user[0]}的{mode_book[mode][5:7]}配置文件！",
-1,
)
run_book[mode] = False
continue
# 更新当前模式到界面
self.update_user_list.emit(
[
(
[f"{_[0]} - {mode_book[mode][5:7]}", _[1], _[2]]
if _[2] == user[2]
else _
)
for _ in self.user_list
]
)
# 剿灭-日常模式循环
for mode in ["Annihilation", "Routine"]:
# 尝试次数循环
for i in range(self.set["RunSet"]["RunTimesLimit"]):
if self.isInterruptionRequested:
break
# 剿灭模式；满足条件跳过剿灭
if (
mode == "Annihilation"
and self.set["RunSet"]["AnnihilationWeeklyLimit"]
and datetime.strptime(
user_data["Data"]["LastAnnihilationDate"], "%Y-%m-%d"
).isocalendar()[:2]
== datetime.strptime(curdate, "%Y-%m-%d").isocalendar()[:2]
):
logger.info(
f"{self.name} | 用户: {user_data["Info"]["Name"]} - 本周剿灭模式已达上限，跳过执行剿灭任务"
)
run_book[mode] = True
continue
else:
self.weekly_annihilation_limit_reached = False
if not user_data["Info"][mode]:
run_book[mode] = True
continue
if run_book[mode]:
continue
break
logger.info(
f"{self.name} | 用户: {user[0]} - 模式: {mode_book[mode]}"
)
if user_data["Info"]["Mode"] == "详细":
self.if_open_emulator = True
if (
check_book[mode]
and not (
self.data[user[2]]["Path"] / f"{mode}/gui.json"
).exists()
):
logger.error(
f"{self.name} | 用户: {user[0]} - 未找到{mode_book[mode][5:7]}配置文件"
)
self.push_info_bar.emit(
"error",
"启动MAA代理进程失败",
f"未找到{user[0]}的{mode_book[mode][5:7]}配置文件！",
-1,
)
check_book[mode] = False
continue
elif not check_book[mode]:
continue
# 更新当前模式到界面
self.update_user_list.emit(
[
(
[f"{_[0]} - {mode_book[mode][5:7]}", _[1], _[2]]
if _[2] == user[2]
else _
)
for _ in self.user_list
]
f"{self.name} | 用户: {user[0]} - 模式: {mode_book[mode]} - 尝试次数: {i + 1}/{self.set["RunSet"]["RunTimesLimit"]}"
)
# 配置MAA
@@ -293,24 +281,50 @@ class MaaManager(QObject):
self.emulator_path = Path(
set["Configurations"]["Default"]["Start.EmulatorPath"]
)
self.emulator_arguments = set["Configurations"]["Default"][
"Start.EmulatorAddCommand"
].split()
self.ADB_path = Path(
set["Configurations"]["Default"]["Connect.AdbPath"]
)
self.ADB_address = set["Configurations"]["Default"][
"Connect.Address"
]
self.if_kill_emulator = bool(
set["Configurations"]["Default"]["MainFunction.PostActions"]
== "12"
)
self.if_open_emulator_process = bool(
set["Configurations"]["Default"][
"Start.OpenEmulatorAfterLaunch"
]
== "True"
)
# 添加静默进程标记
Config.silence_list.append(self.emulator_path)
# 增强任务：任务开始前强杀ADB
if "ADB" in self.set["RunSet"]["EnhanceTask"]:
System.kill_process(self.ADB_path)
else:
try:
subprocess.run(
[self.ADB_path, "disconnect", self.ADB_address],
creationflags=subprocess.CREATE_NO_WINDOW,
)
except subprocess.CalledProcessError as e:
# 忽略错误,因为可能本来就没有连接
logger.warning(f"{self.name} | 释放ADB时出现异常：{e}")
if self.if_open_emulator_process:
self.emulator_process = subprocess.Popen(
[self.emulator_path, *self.emulator_arguments],
creationflags=subprocess.CREATE_NO_WINDOW,
)
# 创建MAA任务
maa = subprocess.Popen(
[self.maa_exe_path],
shell=True,
creationflags=subprocess.CREATE_NO_WINDOW,
)
# 监测MAA运行状态
@@ -339,8 +353,13 @@ class MaaManager(QObject):
)
# 无命令行中止MAA与其子程序
System.kill_process(self.maa_exe_path)
if "Emulator" in self.set["RunSet"]["EnhanceTask"]:
System.kill_process(self.emulator_path)
else:
self.emulator_process.terminate()
self.emulator_process.wait()
self.if_open_emulator = True
# 推送异常通知
Notify.push_plyer(
@@ -357,14 +376,25 @@ class MaaManager(QObject):
# 移除静默进程标记
Config.silence_list.remove(self.emulator_path)
# 增强任务：任务结束后强杀ADB和模拟器
# 增强任务：任务结束后强杀ADB和模拟器或释放进程
if "ADB" in self.set["RunSet"]["EnhanceTask"]:
System.kill_process(self.ADB_path)
if (
self.if_kill_emulator
and "Emulator" in self.set["RunSet"]["EnhanceTask"]
):
System.kill_process(self.emulator_path)
else:
try:
subprocess.run(
[self.ADB_path, "disconnect", self.ADB_address],
creationflags=subprocess.CREATE_NO_WINDOW,
)
except subprocess.CalledProcessError as e:
# 忽略错误,因为可能本来就没有连接
logger.warning(f"{self.name} | 释放ADB时出现异常：{e}")
if self.if_kill_emulator:
if "Emulator" in self.set["RunSet"]["EnhanceTask"]:
System.kill_process(self.emulator_path)
else:
self.emulator_process.terminate()
self.emulator_process.wait()
self.if_open_emulator = True
# 记录剿灭情况
if (
@@ -392,23 +422,6 @@ class MaaManager(QObject):
{"user_name": user_data["Info"]["Name"]},
)
# 成功完成代理的用户修改相关参数
if run_book["Annihilation"] and run_book["Routine"]:
if (
user_data["Data"]["ProxyTimes"] == 0
and user_data["Info"]["RemainedDay"] != -1
):
user_data["Info"]["RemainedDay"] -= 1
user_data["Data"]["ProxyTimes"] += 1
user[1] = "完成"
Notify.push_plyer(
"成功完成一个自动代理任务！",
f"已完成用户 {user[0].replace("_", " 今天的")}任务",
f"已完成 {user[0].replace("_", " 的")}",
3,
)
break
if Config.get(Config.notify_IfSendStatistic):
statistics = Config.merge_maa_logs("指定项", user_logs_list)
@@ -428,8 +441,23 @@ class MaaManager(QObject):
"统计信息", f"用户 {user[0]} 的自动代理统计报告", statistics
)
# 录入代理失败的用户
if not (run_book["Annihilation"] and run_book["Routine"]):
if run_book["Annihilation"] and run_book["Routine"]:
# 成功完成代理的用户修改相关参数
if (
user_data["Data"]["ProxyTimes"] == 0
and user_data["Info"]["RemainedDay"] != -1
):
user_data["Info"]["RemainedDay"] -= 1
user_data["Data"]["ProxyTimes"] += 1
user[1] = "完成"
Notify.push_plyer(
"成功完成一个自动代理任务！",
f"已完成用户 {user[0].replace("_", " 今天的")}任务",
f"已完成 {user[0].replace("_", " 的")}",
3,
)
else:
# 录入代理失败的用户
user[1] = "异常"
self.update_user_list.emit(self.user_list)
@@ -475,7 +503,6 @@ class MaaManager(QObject):
# 创建MAA任务
maa = subprocess.Popen(
[self.maa_exe_path],
shell=True,
creationflags=subprocess.CREATE_NO_WINDOW,
)
@@ -545,7 +572,6 @@ class MaaManager(QObject):
# 创建MAA任务
maa = subprocess.Popen(
[self.maa_exe_path],
shell=True,
creationflags=subprocess.CREATE_NO_WINDOW,
)
# 记录当前时间
@@ -1273,11 +1299,7 @@ class MaaManager(QObject):
] = "False" # 生息演算
# 启动模拟器仅生效一次
if (
"设置MAA" not in mode
and self.if_open_emulator
and self.set["RunSet"]["TaskTransitionMethod"] != "ExitEmulator"
):
if "设置MAA" not in mode and self.if_open_emulator:
self.if_open_emulator = False
# 覆写配置文件

41
app/services/notification.py
View File

@@ -28,6 +28,7 @@ v4.3
from PySide6.QtWidgets import QWidget
from PySide6.QtCore import Signal
import requests
import time
from loguru import logger
from plyer import notification
import re
@@ -229,25 +230,41 @@ class Notification(QWidget):
content = f"{title}\n{content}"
data = {"msgtype": "text", "text": {"content": content}}
response = requests.post(
url=Config.get(Config.notify_CompanyWebHookBotUrl),
json=data,
)
if response.json()["errcode"] == 0:
logger.info("企业微信群机器人推送通知成功")
return True
# 从远程服务器获取最新主题图像
for _ in range(3):
try:
response = requests.post(
url=Config.get(Config.notify_CompanyWebHookBotUrl),
json=data,
timeout=10,
)
info = response.json()
break
except Exception as e:
err = e
time.sleep(0.1)
else:
logger.info("企业微信群机器人推送通知失败")
logger.error(response.json())
logger.error(f"推送企业微信群机器人时出错：{err}")
self.push_info_bar.emit(
"error",
"企业微信群机器人通知推送失败",
f'使用企业微信群机器人推送通知时出错：\n{response.json()["errmsg"]}',
f'使用企业微信群机器人推送通知时出错：{info["errmsg"]}',
-1,
)
return (
f'使用企业微信群机器人推送通知时出错：\n{response.json()["errmsg"]}'
return None
if info["errcode"] == 0:
logger.info("企业微信群机器人推送通知成功")
return True
else:
logger.error(f"企业微信群机器人推送通知失败：{info}")
self.push_info_bar.emit(
"error",
"企业微信群机器人通知推送失败",
f'使用企业微信群机器人推送通知时出错：{info["errmsg"]}',
-1,
)
return f'使用企业微信群机器人推送通知时出错：{info["errmsg"]}'
def send_test_notification(self):
"""发送测试通知到所有已启用的通知渠道"""

5
app/ui/home.py
View File

@@ -202,7 +202,8 @@ class Home(QWidget):
for _ in range(3):
try:
response = requests.get(
"https://gitee.com/DLmaster_361/AUTO_MAA/raw/server/theme_image.json"
"https://gitee.com/DLmaster_361/AUTO_MAA/raw/server/theme_image.json",
timeout=10,
)
theme_image = response.json()
break
@@ -238,7 +239,7 @@ class Home(QWidget):
> time_local
):
response = requests.get(theme_image["url"])
response = requests.get(theme_image["url"], timeout=10)
if response.status_code == 200:
with open(

9
app/ui/member_manager.py
View File

@@ -340,7 +340,8 @@ class MemberManager(QWidget):
for _ in range(3):
try:
response = requests.get(
"https://mirrorchyan.com/api/resources/MAA/latest?user_agent=AutoMaaGui&os=win&arch=x64&channel=stable"
"https://mirrorchyan.com/api/resources/MAA/latest?user_agent=AutoMaaGui&os=win&arch=x64&channel=stable",
timeout=10,
)
maa_info = response.json()
break
@@ -611,7 +612,7 @@ class MemberManager(QWidget):
self.card_TaskTransitionMethod = ComboBoxSettingCard(
icon=FluentIcon.PAGE_RIGHT,
title="任务切换方式",
content="简洁用户列表下相邻两个任务间的切换方式",
content="相邻两个任务间的切换方式，使用“详细”配置的用户固定为“重启模拟器”",
texts=["直接切换账号", "重启明日方舟", "重启模拟器"],
qconfig=self.config,
configItem=self.config.RunSet_TaskTransitionMethod,
@@ -620,7 +621,7 @@ class MemberManager(QWidget):
self.card_EnhanceTask = ComboBoxSettingCard(
icon=FluentIcon.PAGE_RIGHT,
title="自动代理增效任务",
content="自动代理时的额外操作，此操作无法区分多开，可能会干扰其他任务，也可能关闭您正在使用的模拟器",
content="自动代理时的额外操作，此操作无法区分多开模拟器，可能会干扰其他任务，也可能关闭您正在使用的模拟器",
texts=[
"禁用",
"强制关闭ADB",
@@ -634,7 +635,7 @@ class MemberManager(QWidget):
self.ProxyTimesLimit = SpinBoxSettingCard(
icon=FluentIcon.PAGE_RIGHT,
title="用户单日代理次数上限",
content="当用户本日代理成功次数超过该阈值时跳过代理，阈值为“0”时视为无代理次数上限",
content="当用户本日代理成功次数达到该阈值时跳过代理，阈值为“0”时视为无代理次数上限",
range=(0, 1024),
qconfig=self.config,
configItem=self.config.RunSet_ProxyTimesLimit,

2
app/ui/queue_manager.py
View File

@@ -421,7 +421,7 @@ class QueueManager(QWidget):
self.card_Enable = SwitchSettingCard(
icon=FluentIcon.HOME,
title="状态",
content="调度队列状态",
content="调度队列状态，仅启用时会执行定时任务",
qconfig=self.config,
configItem=self.config.queueSet_Enabled,
parent=self,

20
app/ui/setting.py
View File

@@ -51,6 +51,7 @@ import shutil
import requests
import subprocess
from datetime import datetime
from packaging import version
from pathlib import Path
from typing import Dict, List, Union
@@ -271,7 +272,8 @@ class Setting(QWidget):
for _ in range(3):
try:
response = requests.get(
f"https://mirrorchyan.com/api/resources/AUTO_MAA/latest?user_agent=AutoMaaGui&current_version={version_text(current_version)}&cdk={Crypto.win_decryptor(Config.get(Config.update_MirrorChyanCDK))}&channel={Config.get(Config.update_UpdateType)}"
f"https://mirrorchyan.com/api/resources/AUTO_MAA/latest?user_agent=AutoMaaGui&current_version={version_text(current_version)}&cdk={Crypto.win_decryptor(Config.get(Config.update_MirrorChyanCDK))}&channel={Config.get(Config.update_UpdateType)}",
timeout=10,
)
version_info: Dict[str, Union[int, str, Dict[str, str]]] = (
response.json()
@@ -336,7 +338,9 @@ class Setting(QWidget):
)
# 有版本更新
if remote_version > current_version:
if version.parse(version_text(remote_version)) > version.parse(
version_text(current_version)
):
version_info_json: Dict[str, Dict[str, str]] = json.loads(
re.sub(
@@ -353,9 +357,11 @@ class Setting(QWidget):
all_version_info = {}
for v_i in [
info
for version, info in version_info_json.items()
if list(map(int, version.split("."))) > current_version
for ver, info in version_info_json.items()
if version.parse(version_text(list(map(int, ver.split(".")))))
> version.parse(version_text(current_version))
]:
for key, value in v_i.items():
if key in update_version_info:
update_version_info[key] += value.copy()
@@ -400,8 +406,7 @@ class Setting(QWidget):
return None
subprocess.Popen(
str(Config.app_path / "AUTO_Updater.active.exe"),
shell=True,
[Config.app_path / "AUTO_Updater.active.exe"],
creationflags=subprocess.CREATE_NO_WINDOW,
)
self.window().close()
@@ -417,7 +422,8 @@ class Setting(QWidget):
for _ in range(3):
try:
response = requests.get(
"https://gitee.com/DLmaster_361/AUTO_MAA/raw/server/notice.json"
"https://gitee.com/DLmaster_361/AUTO_MAA/raw/server/notice.json",
timeout=10,
)
notice = response.json()
break

24
app/utils/downloader.py
View File

@@ -33,6 +33,7 @@ import subprocess
import time
import win32crypt
import base64
from packaging import version
from functools import partial
from pathlib import Path
@@ -302,7 +303,10 @@ class DownloadManager(QDialog):
elif self.config["mode"] == "MirrorChyan":
with requests.get(
self.config["url"], allow_redirects=True, stream=True
self.config["url"],
allow_redirects=True,
timeout=10,
stream=True,
) as response:
if response.status_code == 200:
return response.url
@@ -399,7 +403,7 @@ class DownloadManager(QDialog):
url = self.get_download_url("下载")
self.downloaded_size_list: List[List[int, bool]] = []
response = requests.head(url)
response = requests.head(url, timeout=10)
self.file_size = int(response.headers.get("content-length", 0))
part_size = self.file_size // self.config["thread_numb"]
@@ -526,14 +530,12 @@ class DownloadManager(QDialog):
# 主程序更新完成后打开对应程序
if not self.isInterruptionRequested and self.name == "AUTO_MAA":
subprocess.Popen(
str(self.app_path / "AUTO_MAA.exe"),
shell=True,
[self.app_path / "AUTO_MAA.exe"],
creationflags=subprocess.CREATE_NO_WINDOW,
)
elif not self.isInterruptionRequested and self.name == "MAA":
subprocess.Popen(
str(self.app_path / "MAA.exe"),
shell=True,
[self.app_path / "MAA.exe"],
creationflags=subprocess.CREATE_NO_WINDOW,
)
@@ -661,7 +663,8 @@ if __name__ == "__main__":
for _ in range(3):
try:
response = requests.get(
f"https://mirrorchyan.com/api/resources/AUTO_MAA/latest?user_agent=AutoMaaDownloader&current_version={version_text(current_version)}&cdk={mirrorchyan_CDK}&channel={update_type}"
f"https://mirrorchyan.com/api/resources/AUTO_MAA/latest?user_agent=AutoMaaDownloader&current_version={version_text(current_version)}&cdk={mirrorchyan_CDK}&channel={update_type}",
timeout=10,
)
version_info: Dict[str, Union[int, str, Dict[str, str]]] = response.json()
break
@@ -696,7 +699,8 @@ if __name__ == "__main__":
for _ in range(3):
try:
response = requests.get(
"https://gitee.com/DLmaster_361/AUTO_MAA/raw/server/download_info.json"
"https://gitee.com/DLmaster_361/AUTO_MAA/raw/server/download_info.json",
timeout=10,
)
download_info = response.json()
@@ -715,7 +719,9 @@ if __name__ == "__main__":
(app_path / "changes.json").unlink()
# 启动更新线程
if remote_version > current_version:
if version.parse(version_text(remote_version)) > version.parse(
version_text(current_version)
):
app = AUTO_MAA_Downloader(
app_path,
"AUTO_MAA",