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

Merge branch 'DLmaster361:dev' into dev

Browse Source
This commit is contained in:
Zrief
2025-05-16 21:55:16 +08:00
committed by GitHub
parent 5d966f98df 0037914db8
commit 061f205224
9 changed files with 911 additions and 290 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
app/core/__init__.py
View File

@@ -29,7 +29,7 @@ __version__ = "4.2.0"
__author__ = "DLmaster361 <DLmaster_361@163.com>"
__license__ = "GPL-3.0 license"
from .config import QueueConfig, MaaConfig, MaaUserConfig, Config
from .config import QueueConfig, MaaConfig, MaaUserConfig, MaaPlanConfig, Config
from .main_info_bar import MainInfoBar
from .network import Network
from .task_manager import Task, TaskManager
@@ -40,6 +40,7 @@ __all__ = [
"QueueConfig",
"MaaConfig",
"MaaUserConfig",
"MaaPlanConfig",
"MainInfoBar",
"Network",
"Task",

463
app/core/config.py
View File

@@ -78,7 +78,78 @@ class UrlListValidator(ConfigValidator):
return list(set([_ for _ in urls if self.validate(_)]))
class GlobalConfig(QConfig):
class LQConfig(QConfig):
"""局域配置类"""
def __init__(self) -> None:
super().__init__()
def toDict(self, serialize=True):
"""convert config items to `dict`"""
items = {}
for name in dir(self._cfg):
item = getattr(self._cfg, name)
if not isinstance(item, ConfigItem):
continue
value = item.serialize() if serialize else item.value
if not items.get(item.group):
if not item.name:
items[item.group] = value
else:
items[item.group] = {}
if item.name:
items[item.group][item.name] = value
return items
@exceptionHandler()
def load(self, file=None, config=None):
"""load config
Parameters
----------
file: str or Path
the path of json config file
config: Config
config object to be initialized
"""
if isinstance(config, QConfig):
self._cfg = config
self._cfg.themeChanged.connect(self.themeChanged)
if isinstance(file, (str, Path)):
self._cfg.file = Path(file)
try:
with open(self._cfg.file, encoding="utf-8") as f:
cfg = json.load(f)
except:
cfg = {}
# map config items'key to item
items = {}
for name in dir(self._cfg):
item = getattr(self._cfg, name)
if isinstance(item, ConfigItem):
items[item.key] = item
# update the value of config item
for k, v in cfg.items():
if not isinstance(v, dict) and items.get(k) is not None:
items[k].deserializeFrom(v)
elif isinstance(v, dict):
for key, value in v.items():
key = k + "." + key
if items.get(key) is not None:
items[key].deserializeFrom(value)
self.theme = self.get(self._cfg.themeMode)
class GlobalConfig(LQConfig):
"""全局配置"""
def __init__(self) -> None:
@@ -182,72 +253,8 @@ class GlobalConfig(QConfig):
)
self.update_MirrorChyanCDK = ConfigItem("Update", "MirrorChyanCDK", "")
def toDict(self, serialize=True):
"""convert config items to `dict`"""
items = {}
for name in dir(self._cfg):
item = getattr(self._cfg, name)
if not isinstance(item, ConfigItem):
continue
value = item.serialize() if serialize else item.value
if not items.get(item.group):
if not item.name:
items[item.group] = value
else:
items[item.group] = {}
if item.name:
items[item.group][item.name] = value
return items
@exceptionHandler()
def load(self, file=None, config=None):
"""load config
Parameters
----------
file: str or Path
the path of json config file
config: Config
config object to be initialized
"""
if isinstance(config, QConfig):
self._cfg = config
self._cfg.themeChanged.connect(self.themeChanged)
if isinstance(file, (str, Path)):
self._cfg.file = Path(file)
try:
with open(self._cfg.file, encoding="utf-8") as f:
cfg = json.load(f)
except:
cfg = {}
# map config items'key to item
items = {}
for name in dir(self._cfg):
item = getattr(self._cfg, name)
if isinstance(item, ConfigItem):
items[item.key] = item
# update the value of config item
for k, v in cfg.items():
if not isinstance(v, dict) and items.get(k) is not None:
items[k].deserializeFrom(v)
elif isinstance(v, dict):
for key, value in v.items():
key = k + "." + key
if items.get(key) is not None:
items[key].deserializeFrom(value)
self.theme = self.get(self._cfg.themeMode)
class QueueConfig(QConfig):
class QueueConfig(LQConfig):
"""队列配置"""
def __init__(self) -> None:
@@ -334,72 +341,8 @@ class QueueConfig(QConfig):
"Data", "LastProxyHistory", "暂无历史运行记录"
)
def toDict(self, serialize=True):
"""convert config items to `dict`"""
items = {}
for name in dir(self._cfg):
item = getattr(self._cfg, name)
if not isinstance(item, ConfigItem):
continue
value = item.serialize() if serialize else item.value
if not items.get(item.group):
if not item.name:
items[item.group] = value
else:
items[item.group] = {}
if item.name:
items[item.group][item.name] = value
return items
@exceptionHandler()
def load(self, file=None, config=None):
"""load config
Parameters
----------
file: str or Path
the path of json config file
config: Config
config object to be initialized
"""
if isinstance(config, QConfig):
self._cfg = config
self._cfg.themeChanged.connect(self.themeChanged)
if isinstance(file, (str, Path)):
self._cfg.file = Path(file)
try:
with open(self._cfg.file, encoding="utf-8") as f:
cfg = json.load(f)
except:
cfg = {}
# map config items'key to item
items = {}
for name in dir(self._cfg):
item = getattr(self._cfg, name)
if isinstance(item, ConfigItem):
items[item.key] = item
# update the value of config item
for k, v in cfg.items():
if not isinstance(v, dict) and items.get(k) is not None:
items[k].deserializeFrom(v)
elif isinstance(v, dict):
for key, value in v.items():
key = k + "." + key
if items.get(key) is not None:
items[key].deserializeFrom(value)
self.theme = self.get(self._cfg.themeMode)
class MaaConfig(QConfig):
class MaaConfig(LQConfig):
"""MAA配置"""
def __init__(self) -> None:
@@ -417,6 +360,9 @@ class MaaConfig(QConfig):
self.RunSet_ProxyTimesLimit = RangeConfigItem(
"RunSet", "ProxyTimesLimit", 0, RangeValidator(0, 1024)
)
self.RunSet_ADBSearchRange = RangeConfigItem(
"RunSet", "ADBSearchRange", 0, RangeValidator(0, 3)
)
self.RunSet_RunTimesLimit = RangeConfigItem(
"RunSet", "RunTimesLimit", 3, RangeValidator(1, 1024)
)
@@ -433,72 +379,8 @@ class MaaConfig(QConfig):
"RunSet", "AutoUpdateMaa", False, BoolValidator()
)
def toDict(self, serialize=True):
"""convert config items to `dict`"""
items = {}
for name in dir(self._cfg):
item = getattr(self._cfg, name)
if not isinstance(item, ConfigItem):
continue
value = item.serialize() if serialize else item.value
if not items.get(item.group):
if not item.name:
items[item.group] = value
else:
items[item.group] = {}
if item.name:
items[item.group][item.name] = value
return items
@exceptionHandler()
def load(self, file=None, config=None):
"""load config
Parameters
----------
file: str or Path
the path of json config file
config: Config
config object to be initialized
"""
if isinstance(config, QConfig):
self._cfg = config
self._cfg.themeChanged.connect(self.themeChanged)
if isinstance(file, (str, Path)):
self._cfg.file = Path(file)
try:
with open(self._cfg.file, encoding="utf-8") as f:
cfg = json.load(f)
except:
cfg = {}
# map config items'key to item
items = {}
for name in dir(self._cfg):
item = getattr(self._cfg, name)
if isinstance(item, ConfigItem):
items[item.key] = item
# update the value of config item
for k, v in cfg.items():
if not isinstance(v, dict) and items.get(k) is not None:
items[k].deserializeFrom(v)
elif isinstance(v, dict):
for key, value in v.items():
key = k + "." + key
if items.get(key) is not None:
items[key].deserializeFrom(value)
self.theme = self.get(self._cfg.themeMode)
class MaaUserConfig(QConfig):
class MaaUserConfig(LQConfig):
"""MAA用户配置"""
def __init__(self) -> None:
@@ -537,8 +419,8 @@ class MaaUserConfig(QConfig):
self.Info_SeriesNumb = OptionsConfigItem(
"Info",
"SeriesNumb",
"1",
OptionsValidator(["1000", "6", "5", "4", "3", "2", "1", "-1"]),
"0",
OptionsValidator(["0", "6", "5", "4", "3", "2", "1", "-1"]),
)
self.Info_GameId = ConfigItem("Info", "GameId", "-")
self.Info_GameId_1 = ConfigItem("Info", "GameId_1", "-")
@@ -557,69 +439,126 @@ class MaaUserConfig(QConfig):
"Data", "CustomInfrastPlanIndex", "0"
)
def toDict(self, serialize=True):
"""convert config items to `dict`"""
items = {}
for name in dir(self._cfg):
item = getattr(self._cfg, name)
if not isinstance(item, ConfigItem):
continue
value = item.serialize() if serialize else item.value
if not items.get(item.group):
if not item.name:
items[item.group] = value
else:
items[item.group] = {}
class MaaPlanConfig(LQConfig):
"""MAA计划表配置"""
if item.name:
items[item.group][item.name] = value
def __init__(self) -> None:
super().__init__()
return items
self.Info_Name = ConfigItem("Info", "Name", "新表格")
@exceptionHandler()
def load(self, file=None, config=None):
"""load config
self.Global_MedicineNumb = ConfigItem(
"Global", "MedicineNumb", 0, RangeValidator(0, 1024)
)
self.Global_SeriesNumb = OptionsConfigItem(
"Global",
"SeriesNumb",
"0",
OptionsValidator(["0", "6", "5", "4", "3", "2", "1", "-1"]),
)
self.Global_GameId = ConfigItem("Global", "GameId", "-")
self.Global_GameId_1 = ConfigItem("Global", "GameId_1", "-")
self.Global_GameId_2 = ConfigItem("Global", "GameId_2", "-")
self.Global_GameId_Remain = ConfigItem("Global", "GameId_Remain", "-")
Parameters
----------
file: str or Path
the path of json config file
self.Monday_MedicineNumb = ConfigItem(
"Monday", "MedicineNumb", 0, RangeValidator(0, 1024)
)
self.Monday_SeriesNumb = OptionsConfigItem(
"Monday",
"SeriesNumb",
"0",
OptionsValidator(["0", "6", "5", "4", "3", "2", "1", "-1"]),
)
self.Monday_GameId = ConfigItem("Monday", "GameId", "-")
self.Monday_GameId_1 = ConfigItem("Monday", "GameId_1", "-")
self.Monday_GameId_2 = ConfigItem("Monday", "GameId_2", "-")
self.Monday_GameId_Remain = ConfigItem("Monday", "GameId_Remain", "-")
config: Config
config object to be initialized
"""
if isinstance(config, QConfig):
self._cfg = config
self._cfg.themeChanged.connect(self.themeChanged)
self.Tuesday_MedicineNumb = ConfigItem(
"Tuesday", "MedicineNumb", 0, RangeValidator(0, 1024)
)
self.Tuesday_SeriesNumb = OptionsConfigItem(
"Tuesday",
"SeriesNumb",
"0",
OptionsValidator(["0", "6", "5", "4", "3", "2", "1", "-1"]),
)
self.Tuesday_GameId = ConfigItem("Tuesday", "GameId", "-")
self.Tuesday_GameId_1 = ConfigItem("Tuesday", "GameId_1", "-")
self.Tuesday_GameId_2 = ConfigItem("Tuesday", "GameId_2", "-")
self.Tuesday_GameId_Remain = ConfigItem("Tuesday", "GameId_Remain", "-")
if isinstance(file, (str, Path)):
self._cfg.file = Path(file)
self.Wednesday_MedicineNumb = ConfigItem(
"Wednesday", "MedicineNumb", 0, RangeValidator(0, 1024)
)
self.Wednesday_SeriesNumb = OptionsConfigItem(
"Wednesday",
"SeriesNumb",
"0",
OptionsValidator(["0", "6", "5", "4", "3", "2", "1", "-1"]),
)
self.Wednesday_GameId = ConfigItem("Wednesday", "GameId", "-")
self.Wednesday_GameId_1 = ConfigItem("Wednesday", "GameId_1", "-")
self.Wednesday_GameId_2 = ConfigItem("Wednesday", "GameId_2", "-")
self.Wednesday_GameId_Remain = ConfigItem("Wednesday", "GameId_Remain", "-")
try:
with open(self._cfg.file, encoding="utf-8") as f:
cfg = json.load(f)
except:
cfg = {}
self.Thursday_MedicineNumb = ConfigItem(
"Thursday", "MedicineNumb", 0, RangeValidator(0, 1024)
)
self.Thursday_SeriesNumb = OptionsConfigItem(
"Thursday",
"SeriesNumb",
"0",
OptionsValidator(["0", "6", "5", "4", "3", "2", "1", "-1"]),
)
self.Thursday_GameId = ConfigItem("Thursday", "GameId", "-")
self.Thursday_GameId_1 = ConfigItem("Thursday", "GameId_1", "-")
self.Thursday_GameId_2 = ConfigItem("Thursday", "GameId_2", "-")
self.Thursday_GameId_Remain = ConfigItem("Thursday", "GameId_Remain", "-")
# map config items'key to item
items = {}
for name in dir(self._cfg):
item = getattr(self._cfg, name)
if isinstance(item, ConfigItem):
items[item.key] = item
self.Friday_MedicineNumb = ConfigItem(
"Friday", "MedicineNumb", 0, RangeValidator(0, 1024)
)
self.Friday_SeriesNumb = OptionsConfigItem(
"Friday",
"SeriesNumb",
"0",
OptionsValidator(["0", "6", "5", "4", "3", "2", "1", "-1"]),
)
self.Friday_GameId = ConfigItem("Friday", "GameId", "-")
self.Friday_GameId_1 = ConfigItem("Friday", "GameId_1", "-")
self.Friday_GameId_2 = ConfigItem("Friday", "GameId_2", "-")
self.Friday_GameId_Remain = ConfigItem("Friday", "GameId_Remain", "-")
# update the value of config item
for k, v in cfg.items():
if not isinstance(v, dict) and items.get(k) is not None:
items[k].deserializeFrom(v)
elif isinstance(v, dict):
for key, value in v.items():
key = k + "." + key
if items.get(key) is not None:
items[key].deserializeFrom(value)
self.Saturday_MedicineNumb = ConfigItem(
"Saturday", "MedicineNumb", 0, RangeValidator(0, 1024)
)
self.Saturday_SeriesNumb = OptionsConfigItem(
"Saturday",
"SeriesNumb",
"0",
OptionsValidator(["0", "6", "5", "4", "3", "2", "1", "-1"]),
)
self.Saturday_GameId = ConfigItem("Saturday", "GameId", "-")
self.Saturday_GameId_1 = ConfigItem("Saturday", "GameId_1", "-")
self.Saturday_GameId_2 = ConfigItem("Saturday", "GameId_2", "-")
self.Saturday_GameId_Remain = ConfigItem("Saturday", "GameId_Remain", "-")
self.theme = self.get(self._cfg.themeMode)
self.Sunday_MedicineNumb = ConfigItem(
"Sunday", "MedicineNumb", 0, RangeValidator(0, 1024)
)
self.Sunday_SeriesNumb = OptionsConfigItem(
"Sunday",
"SeriesNumb",
"0",
OptionsValidator(["0", "6", "5", "4", "3", "2", "1", "-1"]),
)
self.Sunday_GameId = ConfigItem("Sunday", "GameId", "-")
self.Sunday_GameId_1 = ConfigItem("Sunday", "GameId_1", "-")
self.Sunday_GameId_2 = ConfigItem("Sunday", "GameId_2", "-")
self.Sunday_GameId_Remain = ConfigItem("Sunday", "GameId_Remain", "-")
class AppConfig(GlobalConfig):
@@ -704,7 +643,7 @@ class AppConfig(GlobalConfig):
# 从MAA服务器获取活动关卡信息
Network.set_info(
mode="get",
url="https://ota.maa.plus/MaaAssistantArknights/api/gui/StageActivity.json",
url="https://api.maa.plus/MaaAssistantArknights/api/gui/StageActivity.json",
)
Network.start()
Network.loop.exec()
@@ -1235,6 +1174,28 @@ class AppConfig(GlobalConfig):
sorted(user_dict.items(), key=lambda x: int(x[0][3:]))
)
def search_plan(self) -> None:
"""搜索所有计划表"""
self.plan_dict: Dict[str, Dict[str, Union[str, Path, MaaPlanConfig]]] = {}
if (self.app_path / "config/MaaPlanConfig").exists():
for maa_plan_dir in (self.app_path / "config/MaaPlanConfig").iterdir():
if maa_plan_dir.is_dir():
maa_plan_config = MaaPlanConfig()
maa_plan_config.load(maa_plan_dir / "config.json", maa_plan_config)
maa_plan_config.save()
self.member_dict[maa_plan_dir.name] = {
"Type": "Maa",
"Path": maa_plan_dir,
"Config": maa_plan_config,
}
self.plan_dict = dict(
sorted(self.plan_dict.items(), key=lambda x: int(x[0][3:]))
)
def search_queue(self):
"""搜索所有调度队列实例"""

162
app/models/MAA.py
View File

@@ -114,6 +114,10 @@ class MaaManager(QObject):
self.maa_log_path = self.maa_root_path / "debug/gui.log"
self.maa_exe_path = self.maa_root_path / "MAA.exe"
self.maa_tasks_path = self.maa_root_path / "resource/tasks/tasks.json"
self.port_range = [0] + [
(i // 2 + 1) * (-1 if i % 2 else 1)
for i in range(0, 2 * self.set["RunSet"]["ADBSearchRange"])
]
def run(self):
"""主进程，运行MAA代理进程"""
@@ -386,6 +390,12 @@ class MaaManager(QObject):
self.if_open_emulator = True
break
self.wait_time = int(
set["Configurations"]["Default"][
"Start.EmulatorWaitSeconds"
]
)
self.ADB_path = Path(
set["Configurations"]["Default"]["Connect.AdbPath"]
)
@@ -414,6 +424,7 @@ class MaaManager(QObject):
[self.ADB_path, "disconnect", self.ADB_address],
creationflags=subprocess.CREATE_NO_WINDOW,
)
logger.info(f"{self.name} | 释放ADB：{self.ADB_address}")
except subprocess.CalledProcessError as e:
# 忽略错误,因为可能本来就没有连接
logger.warning(f"{self.name} | 释放ADB时出现异常：{e}")
@@ -432,6 +443,9 @@ class MaaManager(QObject):
[self.emulator_path, *self.emulator_arguments],
creationflags=subprocess.CREATE_NO_WINDOW,
)
logger.info(
f"{self.name} | 启动模拟器：{self.emulator_path}，参数：{self.emulator_arguments}"
)
except Exception as e:
logger.error(f"{self.name} | 启动模拟器时出现异常：{e}")
self.push_info_bar.emit(
@@ -446,6 +460,8 @@ class MaaManager(QObject):
# 添加静默进程标记
Config.silence_list.append(self.emulator_path)
self.search_ADB_address()
# 创建MAA任务
maa = subprocess.Popen(
[self.maa_exe_path],
@@ -517,6 +533,12 @@ class MaaManager(QObject):
) as f:
data = json.load(f)
# 记录自定义基建索引
if self.task_dict["Base"] == "False":
user_data["Data"]["CustomInfrastPlanIndex"] = data[
"Configurations"
]["Default"]["Infrast.CustomInfrastPlanIndex"]
# 记录更新包路径
if (
data["Global"]["VersionUpdate.package"]
@@ -550,6 +572,7 @@ class MaaManager(QObject):
[self.ADB_path, "disconnect", self.ADB_address],
creationflags=subprocess.CREATE_NO_WINDOW,
)
logger.info(f"{self.name} | 释放ADB：{self.ADB_address}")
except subprocess.CalledProcessError as e:
# 忽略错误,因为可能本来就没有连接
logger.warning(f"{self.name} | 释放ADB时出现异常：{e}")
@@ -567,29 +590,6 @@ class MaaManager(QObject):
self.emulator_process.wait()
self.if_open_emulator = True
# 执行MAA解压更新动作
if self.maa_update_package:
logger.info(
f"{self.name} | 检测到MAA更新，正在执行更新动作"
)
self.update_log_text.emit(
f"检测到MAA存在更新\nMAA正在执行更新动作\n请等待10s"
)
self.set_maa("更新MAA", None)
subprocess.Popen(
[self.maa_exe_path],
creationflags=subprocess.CREATE_NO_WINDOW,
)
for _ in range(10):
if self.isInterruptionRequested:
break
time.sleep(1)
System.kill_process(self.maa_exe_path)
logger.info(f"{self.name} | 更新动作结束")
# 记录剿灭情况
if (
mode == "Annihilation"
@@ -616,6 +616,29 @@ class MaaManager(QObject):
{"user_name": user_data["Info"]["Name"]},
)
# 执行MAA解压更新动作
if self.maa_update_package:
logger.info(
f"{self.name} | 检测到MAA更新，正在执行更新动作"
)
self.update_log_text.emit(
f"检测到MAA存在更新\nMAA正在执行更新动作\n请等待10s"
)
self.set_maa("更新MAA", None)
subprocess.Popen(
[self.maa_exe_path],
creationflags=subprocess.CREATE_NO_WINDOW,
)
for _ in range(10):
if self.isInterruptionRequested:
break
time.sleep(1)
System.kill_process(self.maa_exe_path)
logger.info(f"{self.name} | 更新动作结束")
if Config.get(Config.notify_IfSendStatistic):
statistics = Config.merge_maa_logs("指定项", user_logs_list)
@@ -867,6 +890,80 @@ class MaaManager(QObject):
def __capture_response(self, response: bool) -> None:
self.response = response
def search_ADB_address(self) -> None:
"""搜索ADB实际地址"""
self.update_log_text.emit(
f"即将搜索ADB实际地址\n正在等待模拟器完成启动\n请等待{self.wait_time}s"
)
time.sleep(self.wait_time)
if "-" in self.ADB_address:
ADB_ip = f"{self.ADB_address.split("-")[0]}-"
ADB_port = self.ADB_address.split("-")[1]
elif ":" in self.ADB_address:
ADB_ip = f"{self.ADB_address.split(':')[0]}:"
ADB_port = int(self.ADB_address.split(":")[1])
logger.info(
f"{self.name} | 正在搜索ADB实际地址，ADB前缀：{ADB_ip}，初始端口：{ADB_port}，搜索范围：{self.port_range}"
)
for port in self.port_range:
ADB_address = f"{ADB_ip}{ADB_port + port}"
# 尝试通过ADB连接到指定地址
connect_result = subprocess.run(
[self.ADB_path, "connect", ADB_address],
creationflags=subprocess.CREATE_NO_WINDOW,
capture_output=True,
text=True,
encoding="utf-8",
)
if "connected" in connect_result.stdout:
# 检查连接状态
devices_result = subprocess.run(
[self.ADB_path, "devices"],
creationflags=subprocess.CREATE_NO_WINDOW,
capture_output=True,
text=True,
encoding="utf-8",
)
if ADB_address in devices_result.stdout:
logger.info(f"{self.name} | ADB实际地址：{ADB_address}")
# 断开连接
subprocess.run(
[self.ADB_path, "disconnect", ADB_address],
creationflags=subprocess.CREATE_NO_WINDOW,
)
self.ADB_address = ADB_address
# 覆写当前ADB地址
System.kill_process(self.maa_exe_path)
with self.maa_set_path.open(mode="r", encoding="utf-8") as f:
data = json.load(f)
data["Configurations"]["Default"][
"Connect.Address"
] = self.ADB_address
data["Configurations"]["Default"]["Start.EmulatorWaitSeconds"] = "0"
with self.maa_set_path.open(mode="w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=4)
return None
else:
logger.info(f"{self.name} | 无法连接到ADB地址：{ADB_address}")
else:
logger.info(f"{self.name} | 无法连接到ADB地址：{ADB_address}")
def refresh_maa_log(self) -> None:
"""刷新MAA日志"""
@@ -1419,12 +1516,20 @@ class MaaManager(QObject):
"Start.RunDirectly"
] = "True" # 启动MAA后直接运行
data["Global"]["Start.MinimizeDirectly"] = "True" # 启动MAA后直接最小化
data["Global"]["GUI.UseTray"] = "True" # 显示托盘图标
data["Global"]["GUI.MinimizeToTray"] = "True" # 最小化时隐藏至托盘
data["Configurations"]["Default"]["Start.OpenEmulatorAfterLaunch"] = str(
self.if_open_emulator
) # 启动MAA后自动开启模拟器
data["Global"][
"VersionUpdate.ScheduledUpdateCheck"
] = "False" # 定时检查更新
data["Global"][
"VersionUpdate.AutoDownloadUpdatePackage"
] = "False" # 自动下载更新包
data["Global"][
"VersionUpdate.AutoInstallUpdatePackage"
] = "False" # 自动安装更新包
# 账号切换
if user_data["Info"]["Server"] == "Official":
@@ -1440,15 +1545,6 @@ class MaaManager(QObject):
if user_data["Info"]["Mode"] == "简洁":
data["Global"][
"VersionUpdate.ScheduledUpdateCheck"
] = "False" # 定时检查更新
data["Global"][
"VersionUpdate.AutoDownloadUpdatePackage"
] = "False" # 自动下载更新包
data["Global"][
"VersionUpdate.AutoInstallUpdatePackage"
] = "False" # 自动安装更新包
data["Configurations"]["Default"]["Start.ClientType"] = user_data[
"Info"
][

10
app/ui/main_window.py
View File

@@ -351,7 +351,9 @@ class AUTO_MAA(MSFluentWindow):
"warning", "启动主任务失败", "“调度队列_1”与“脚本_1”均不存在", -1
)
def show_ui(self, mode: str, if_quick: bool = False) -> None:
def show_ui(
self, mode: str, if_quick: bool = False, if_start: bool = False
) -> None:
"""配置窗口状态"""
self.switch_theme()
@@ -378,8 +380,12 @@ class AUTO_MAA(MSFluentWindow):
self.window().show()
if not if_quick:
if Config.get(Config.ui_maximized):
self.titleBar.maxBtn.click()
self.window().showMaximized()
self.show_ui("配置托盘")
elif if_start:
if Config.get(Config.ui_maximized):
self.window().showMaximized()
self.show_ui("配置托盘")
if not any(
self.window().geometry().intersects(screen.availableGeometry())

10
app/ui/member_manager.py
View File

@@ -699,6 +699,15 @@ class MemberManager(QWidget):
configItem=self.config.RunSet_ProxyTimesLimit,
parent=self,
)
self.card_ADBSearchRange = SpinBoxSettingCard(
icon=FluentIcon.PAGE_RIGHT,
title="ADB端口号搜索范围",
content="在【±端口号范围】内搜索实际ADB端口号",
range=(0, 3),
qconfig=self.config,
configItem=self.config.RunSet_ADBSearchRange,
parent=self,
)
self.card_RunTimesLimit = SpinBoxSettingCard(
icon=FluentIcon.PAGE_RIGHT,
title="代理重试次数限制",
@@ -747,6 +756,7 @@ class MemberManager(QWidget):
Layout = QVBoxLayout(widget)
Layout.addWidget(self.card_TaskTransitionMethod)
Layout.addWidget(self.card_ProxyTimesLimit)
Layout.addWidget(self.card_ADBSearchRange)
Layout.addWidget(self.card_RunTimesLimit)
Layout.addWidget(self.card_AnnihilationTimeLimit)
Layout.addWidget(self.card_RoutineTimeLimit)

541
app/ui/plan_manager.py Normal file
View File

@@ -0,0 +1,541 @@
# AUTO_MAA:A MAA Multi Account Management and Automation Tool
# Copyright © 2024-2025 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/>.
# Contact: DLmaster_361@163.com
"""
AUTO_MAA
AUTO_MAA计划管理界面
v4.3
作者：DLmaster_361
"""
from loguru import logger
from PySide6.QtWidgets import (
QWidget,
QFileDialog,
QHBoxLayout,
QVBoxLayout,
QStackedWidget,
QTableWidgetItem,
QHeaderView,
)
from PySide6.QtGui import QIcon
from qfluentwidgets import (
Action,
ScrollArea,
FluentIcon,
MessageBox,
HeaderCardWidget,
CommandBar,
ExpandGroupSettingCard,
PushSettingCard,
TableWidget,
PrimaryToolButton,
)
from PySide6.QtCore import Signal
from datetime import datetime
from functools import partial
from pathlib import Path
from typing import List
import shutil
import json
from app.core import Config, MainInfoBar, TaskManager, MaaPlanConfig, Network
from app.services import Crypto
from .downloader import DownloadManager
from .Widget import (
LineEditMessageBox,
LineEditSettingCard,
SpinBoxSettingCard,
ComboBoxMessageBox,
EditableComboBoxSettingCard,
PasswordLineEditSettingCard,
UserLableSettingCard,
ComboBoxSettingCard,
SwitchSettingCard,
PushAndSwitchButtonSettingCard,
PushAndComboBoxSettingCard,
PivotArea,
)
class PlanManager(QWidget):
"""计划管理父界面"""
def __init__(self, parent=None):
super().__init__(parent)
self.setObjectName("计划管理")
layout = QVBoxLayout(self)
self.tools = CommandBar()
self.plan_manager = self.PlanSettingBox(self)
# 逐个添加动作
self.tools.addActions(
[
Action(FluentIcon.ADD_TO, "新建计划表", triggered=self.add_setting_box),
Action(
FluentIcon.REMOVE_FROM, "删除计划表", triggered=self.del_setting_box
),
]
)
self.tools.addSeparator()
self.tools.addActions(
[
Action(
FluentIcon.LEFT_ARROW, "向左移动", triggered=self.left_setting_box
),
Action(
FluentIcon.RIGHT_ARROW, "向右移动", triggered=self.right_setting_box
),
]
)
self.tools.addSeparator()
layout.addWidget(self.tools)
layout.addWidget(self.plan_manager)
def add_setting_box(self):
"""添加一个计划表"""
choice = ComboBoxMessageBox(
self.window(),
"选择一个计划类型以添加相应计划表",
["选择计划类型"],
[["MAA"]],
)
if choice.exec() and choice.input[0].currentIndex() != -1:
if choice.input[0].currentText() == "MAA":
index = len(Config.plan_dict) + 1
maa_plan_config = MaaPlanConfig()
maa_plan_config.load(
Config.app_path / f"config/MaaPlanConfig/计划_{index}/config.json",
maa_plan_config,
)
maa_plan_config.save()
Config.plan_dict[f"计划_{index}"] = {
"Type": "Maa",
"Path": Config.app_path / f"config/MaaPlanConfig/计划_{index}",
"Config": maa_plan_config,
}
self.plan_manager.add_MaaSettingBox(index)
self.plan_manager.switch_SettingBox(index)
logger.success(f"计划管理 计划_{index} 添加成功")
MainInfoBar.push_info_bar(
"success", "操作成功", f"添加计划表 计划_{index}", 3000
)
def del_setting_box(self):
"""删除一个计划表"""
name = self.plan_manager.pivot.currentRouteKey()
if name == None:
logger.warning("删除计划表时未选择计划表")
MainInfoBar.push_info_bar(
"warning", "未选择计划表", "请选择一个计划表", 5000
)
return None
if len(Config.running_list) > 0:
logger.warning("删除计划表时调度队列未停止运行")
MainInfoBar.push_info_bar(
"warning", "调度中心正在执行任务", "请等待或手动中止任务", 5000
)
return None
choice = MessageBox(
"确认",
f"确定要删除 {name} 吗？",
self.window(),
)
if choice.exec():
self.plan_manager.clear_SettingBox()
shutil.rmtree(Config.plan_dict[name]["Path"])
Config.change_plan(name, "禁用")
for i in range(int(name[3:]) + 1, len(Config.plan_dict) + 1):
if Config.plan_dict[f"计划_{i}"]["Path"].exists():
Config.plan_dict[f"计划_{i}"]["Path"].rename(
Config.plan_dict[f"计划_{i}"]["Path"].with_name(f"计划_{i-1}")
)
Config.change_queue(f"计划_{i}", f"计划_{i-1}")
self.plan_manager.show_SettingBox(max(int(name[3:]) - 1, 1))
logger.success(f"计划表 {name} 删除成功")
MainInfoBar.push_info_bar("success", "操作成功", f"删除计划表 {name}", 3000)
def left_setting_box(self):
"""向左移动计划表"""
name = self.plan_manager.pivot.currentRouteKey()
if name == None:
logger.warning("向左移动计划表时未选择计划表")
MainInfoBar.push_info_bar(
"warning", "未选择计划表", "请选择一个计划表", 5000
)
return None
index = int(name[3:])
if index == 1:
logger.warning("向左移动计划表时已到达最左端")
MainInfoBar.push_info_bar(
"warning", "已经是第一个计划表", "无法向左移动", 5000
)
return None
if len(Config.running_list) > 0:
logger.warning("向左移动计划表时调度队列未停止运行")
MainInfoBar.push_info_bar(
"warning", "调度中心正在执行任务", "请等待或手动中止任务", 5000
)
return None
self.plan_manager.clear_SettingBox()
Config.plan_dict[name]["Path"].rename(
Config.plan_dict[name]["Path"].with_name("计划_0")
)
Config.change_queue(name, "计划_0")
Config.plan_dict[f"计划_{index-1}"]["Path"].rename(
Config.plan_dict[name]["Path"]
)
Config.change_queue(f"计划_{index-1}", name)
Config.plan_dict[name]["Path"].with_name("计划_0").rename(
Config.plan_dict[f"计划_{index-1}"]["Path"]
)
Config.change_queue("计划_0", f"计划_{index-1}")
self.plan_manager.show_SettingBox(index - 1)
logger.success(f"计划表 {name} 左移成功")
MainInfoBar.push_info_bar("success", "操作成功", f"左移计划表 {name}", 3000)
def right_setting_box(self):
"""向右移动计划表"""
name = self.plan_manager.pivot.currentRouteKey()
if name == None:
logger.warning("向右移动计划表时未选择计划表")
MainInfoBar.push_info_bar(
"warning", "未选择计划表", "请选择一个计划表", 5000
)
return None
index = int(name[3:])
if index == len(Config.plan_dict):
logger.warning("向右移动计划表时已到达最右端")
MainInfoBar.push_info_bar(
"warning", "已经是最后一个计划表", "无法向右移动", 5000
)
return None
if len(Config.running_list) > 0:
logger.warning("向右移动计划表时调度队列未停止运行")
MainInfoBar.push_info_bar(
"warning", "调度中心正在执行任务", "请等待或手动中止任务", 5000
)
return None
self.plan_manager.clear_SettingBox()
Config.plan_dict[name]["Path"].rename(
Config.plan_dict[name]["Path"].with_name("计划_0")
)
Config.change_queue(name, "计划_0")
Config.plan_dict[f"计划_{index+1}"]["Path"].rename(
Config.plan_dict[name]["Path"]
)
Config.change_queue(f"计划_{index+1}", name)
Config.plan_dict[name]["Path"].with_name("计划_0").rename(
Config.plan_dict[f"计划_{index+1}"]["Path"]
)
Config.change_queue("计划_0", f"计划_{index+1}")
self.plan_manager.show_SettingBox(index + 1)
logger.success(f"计划表 {name} 右移成功")
MainInfoBar.push_info_bar("success", "操作成功", f"右移计划表 {name}", 3000)
def refresh_dashboard(self):
"""刷新所有计划表的用户仪表盘"""
for script in self.plan_manager.script_list:
script.user_setting.user_manager.user_dashboard.load_info()
class PlanSettingBox(QWidget):
"""计划管理子页面组"""
def __init__(self, parent=None):
super().__init__(parent)
self.setObjectName("计划管理页面组")
self.pivotArea = PivotArea(self)
self.pivot = self.pivotArea.pivot
self.stackedWidget = QStackedWidget(self)
self.stackedWidget.setContentsMargins(0, 0, 0, 0)
self.stackedWidget.setStyleSheet("background: transparent; border: none;")
self.script_list: List[PlanManager.PlanSettingBox.MaaPlanSettingBox] = []
self.Layout = QVBoxLayout(self)
self.Layout.addWidget(self.pivotArea)
self.Layout.addWidget(self.stackedWidget)
self.Layout.setContentsMargins(0, 0, 0, 0)
self.pivot.currentItemChanged.connect(
lambda index: self.switch_SettingBox(
int(index[3:]), if_chang_pivot=False
)
)
self.show_SettingBox(1)
def show_SettingBox(self, index) -> None:
"""加载所有子界面"""
Config.search_plan()
for name, info in Config.plan_dict.items():
if info["Type"] == "Maa":
self.add_MaaSettingBox(int(name[3:]))
self.switch_SettingBox(index)
def switch_SettingBox(self, index: int, if_chang_pivot: bool = True) -> None:
"""切换到指定的子界面"""
if len(Config.plan_dict) == 0:
return None
if index > len(Config.plan_dict):
return None
if if_chang_pivot:
self.pivot.setCurrentItem(self.script_list[index - 1].objectName())
self.stackedWidget.setCurrentWidget(self.script_list[index - 1])
def clear_SettingBox(self) -> None:
"""清空所有子界面"""
for sub_interface in self.script_list:
self.stackedWidget.removeWidget(sub_interface)
sub_interface.deleteLater()
self.script_list.clear()
self.pivot.clear()
def add_MaaPlanSettingBox(self, uid: int) -> None:
"""添加一个MAA设置界面"""
maa_plan_setting_box = self.MaaPlanSettingBox(uid, self)
self.script_list.append(maa_plan_setting_box)
self.stackedWidget.addWidget(self.script_list[-1])
self.pivot.addItem(routeKey=f"计划_{uid}", text=f"计划 {uid}")
class MaaPlanSettingBox(HeaderCardWidget):
"""MAA类计划设置界面"""
def __init__(self, uid: int, parent=None):
super().__init__(parent)
self.setObjectName(f"计划_{uid}")
self.config = Config.plan_dict[f"计划_{uid}"]["Config"]
self.dashboard = TableWidget(self)
self.dashboard.setColumnCount(11)
self.dashboard.setHorizontalHeaderLabels(
[
"吃理智药",
"连战次数",
"关卡选择",
"备选关卡 - 1",
"备选关卡 - 2",
"剩余理智关卡",
]
)
self.dashboard.setEditTriggers(TableWidget.NoEditTriggers)
self.dashboard.verticalHeader().setVisible(False)
for col in range(6):
self.dashboard.horizontalHeader().setSectionResizeMode(
col, QHeaderView.ResizeMode.Stretch
)
self.viewLayout.addWidget(self.dashboard)
self.viewLayout.setContentsMargins(3, 0, 3, 3)
Config.PASSWORD_refreshed.connect(self.load_info)
def load_info(self):
self.user_data = Config.plan_dict[self.name]["UserData"]
self.dashboard.setRowCount(len(self.user_data))
for name, info in self.user_data.items():
config = info["Config"]
text_list = []
if not config.get(config.Data_IfPassCheck):
text_list.append("未通过人工排查")
text_list.append(
f"今日已代理{config.get(config.Data_ProxyTimes)}次"
if Config.server_date().strftime("%Y-%m-%d")
== config.get(config.Data_LastProxyDate)
else "今日未进行代理"
)
text_list.append(
"本周剿灭已完成"
if datetime.strptime(
config.get(config.Data_LastAnnihilationDate),
"%Y-%m-%d",
).isocalendar()[:2]
== Config.server_date().isocalendar()[:2]
else "本周剿灭未完成"
)
button = PrimaryToolButton(FluentIcon.CHEVRON_RIGHT, self)
button.setFixedSize(32, 32)
button.clicked.connect(partial(self.switch_to.emit, name))
self.dashboard.setItem(
int(name[3:]) - 1,
0,
QTableWidgetItem(config.get(config.Info_Name)),
)
self.dashboard.setItem(
int(name[3:]) - 1,
1,
QTableWidgetItem(config.get(config.Info_Id)),
)
self.dashboard.setItem(
int(name[3:]) - 1,
2,
QTableWidgetItem(
Crypto.AUTO_decryptor(
config.get(config.Info_Password),
Config.PASSWORD,
)
if Config.PASSWORD
else "******"
),
)
self.dashboard.setItem(
int(name[3:]) - 1,
3,
QTableWidgetItem(
"启用"
if config.get(config.Info_Status)
and config.get(config.Info_RemainedDay) != 0
else "禁用"
),
)
self.dashboard.setItem(
int(name[3:]) - 1,
4,
QTableWidgetItem(" | ".join(text_list)),
)
self.dashboard.setItem(
int(name[3:]) - 1,
5,
QTableWidgetItem(str(config.get(config.Info_MedicineNumb))),
)
self.dashboard.setItem(
int(name[3:]) - 1,
6,
QTableWidgetItem(
Config.gameid_dict["ALL"]["text"][
Config.gameid_dict["ALL"]["value"].index(
config.get(config.Info_GameId)
)
]
if config.get(config.Info_GameId)
in Config.gameid_dict["ALL"]["value"]
else config.get(config.Info_GameId)
),
)
self.dashboard.setItem(
int(name[3:]) - 1,
7,
QTableWidgetItem(
Config.gameid_dict["ALL"]["text"][
Config.gameid_dict["ALL"]["value"].index(
config.get(config.Info_GameId_1)
)
]
if config.get(config.Info_GameId_1)
in Config.gameid_dict["ALL"]["value"]
else config.get(config.Info_GameId_1)
),
)
self.dashboard.setItem(
int(name[3:]) - 1,
8,
QTableWidgetItem(
Config.gameid_dict["ALL"]["text"][
Config.gameid_dict["ALL"]["value"].index(
config.get(config.Info_GameId_2)
)
]
if config.get(config.Info_GameId_2)
in Config.gameid_dict["ALL"]["value"]
else config.get(config.Info_GameId_2)
),
)
self.dashboard.setItem(
int(name[3:]) - 1,
9,
QTableWidgetItem(
"不使用"
if config.get(config.Info_GameId_Remain) == "-"
else (
(
Config.gameid_dict["ALL"]["text"][
Config.gameid_dict["ALL"]["value"].index(
config.get(config.Info_GameId_Remain)
)
]
)
if config.get(config.Info_GameId_Remain)
in Config.gameid_dict["ALL"]["value"]
else config.get(config.Info_GameId_Remain)
)
),
)
self.dashboard.setCellWidget(int(name[3:]) - 1, 10, button)