2488 lines
104 KiB
Python
2488 lines
104 KiB
Python
# <AUTO_MAA:A MAA Multi Account Management and Automation Tool>
|
||
# Copyright © <2024> <DLmaster361>
|
||
|
||
# This file is part of AUTO_MAA.
|
||
|
||
# AUTO_MAA is free software: you can redistribute it and/or modify
|
||
# it under the terms of the GNU General Public License as published
|
||
# by the Free Software Foundation, either version 3 of the License,
|
||
# or (at your option) any later version.
|
||
|
||
# AUTO_MAA is distributed in the hope that it will be useful,
|
||
# but WITHOUT ANY WARRANTY; without even the implied warranty
|
||
# of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See
|
||
# the GNU General Public License for more details.
|
||
|
||
# You should have received a copy of the GNU General Public License
|
||
# along with AUTO_MAA. If not, see <https://www.gnu.org/licenses/>.
|
||
|
||
# DLmaster_361@163.com
|
||
|
||
"""
|
||
AUTO_MAA
|
||
AUTO_MAA主程序
|
||
v4.1
|
||
作者：DLmaster_361
|
||
"""
|
||
|
||
from PySide6.QtWidgets import (
|
||
QWidget,
|
||
QApplication,
|
||
QInputDialog,
|
||
QFileDialog,
|
||
QMessageBox,
|
||
QLineEdit,
|
||
QToolBox,
|
||
QTableWidget,
|
||
QTableWidgetItem,
|
||
QComboBox,
|
||
QPushButton,
|
||
QHeaderView,
|
||
QSpinBox,
|
||
QTimeEdit,
|
||
QCheckBox,
|
||
QTextBrowser,
|
||
)
|
||
from PySide6.QtUiTools import QUiLoader
|
||
from PySide6.QtGui import QIcon
|
||
from PySide6 import QtCore
|
||
from functools import partial
|
||
from plyer import notification
|
||
import sqlite3
|
||
import json
|
||
import datetime
|
||
import os
|
||
import sys
|
||
import ctypes
|
||
import hashlib
|
||
import subprocess
|
||
import shutil
|
||
import time
|
||
import random
|
||
import secrets
|
||
import winreg
|
||
import requests
|
||
from Crypto.Cipher import AES
|
||
from Crypto.PublicKey import RSA
|
||
from Crypto.Cipher import PKCS1_OAEP
|
||
from Crypto.Util.Padding import pad, unpad
|
||
|
||
import Updater
|
||
|
||
uiLoader = QUiLoader()
|
||
|
||
|
||
class MaaRunner(QtCore.QThread):
|
||
|
||
question = QtCore.Signal()
|
||
push_notification = QtCore.Signal(str, str, str, int)
|
||
update_gui = QtCore.Signal(str, str, str, str, str)
|
||
update_user_info = QtCore.Signal(list, list, list, list, list, list)
|
||
accomplish = QtCore.Signal()
|
||
get_json = QtCore.Signal(list)
|
||
app_path = os.path.dirname(os.path.realpath(sys.argv[0])).replace(
|
||
"\\", "/"
|
||
) # 获取软件自身的路径
|
||
if_run = False
|
||
|
||
def __init__(
|
||
self, set_path, log_path, maa_path, routine, annihilation, num, data, mode
|
||
):
|
||
super(MaaRunner, self).__init__()
|
||
self.set_path = set_path
|
||
self.log_path = log_path
|
||
self.maa_path = maa_path
|
||
self.json_path = self.app_path + "/data/MAAconfig"
|
||
self.routine = routine
|
||
self.annihilation = annihilation
|
||
self.num = num
|
||
self.data = data
|
||
self.mode = mode
|
||
self.get_json_path = [0, 0, 0]
|
||
|
||
def run(self):
|
||
"""主进程，运行MAA代理进程"""
|
||
self.if_run = True
|
||
curdate = server_date()
|
||
begin_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
self.data = sorted(self.data, key=lambda x: (-len(x[15]), x[16]))
|
||
wait_index = []
|
||
over_index = []
|
||
error_index = []
|
||
all_index = [
|
||
_
|
||
for _ in range(len(self.data))
|
||
if (self.data[_][3] != 0 and self.data[_][4] == "y")
|
||
]
|
||
# 日常代理模式
|
||
if self.mode == "日常代理":
|
||
# 执行情况预处理
|
||
for _ in all_index:
|
||
if self.data[_][5] != curdate:
|
||
self.data[_][5] = curdate
|
||
self.data[_][14] = 0
|
||
self.data[_][0] += "_第" + str(self.data[_][14] + 1) + "次代理"
|
||
# 开始代理
|
||
for index in all_index:
|
||
if not self.if_run:
|
||
break
|
||
# 初始化代理情况记录和模式替换记录
|
||
run_book = [False for _ in range(2)]
|
||
mode_book = ["日常代理_剿灭", "日常代理_日常"]
|
||
# 尝试次数循环
|
||
for i in range(self.num):
|
||
if not self.if_run:
|
||
break
|
||
# 剿灭-日常模式循环
|
||
for j in range(2):
|
||
if not self.if_run:
|
||
break
|
||
if j == 0 and self.data[index][10] == "n":
|
||
run_book[0] = True
|
||
continue
|
||
if run_book[j]:
|
||
continue
|
||
# 配置MAA
|
||
self.set_maa(mode_book[j], index)
|
||
# 记录当前时间
|
||
start_time = datetime.datetime.now()
|
||
# 创建MAA任务
|
||
maa = subprocess.Popen([self.maa_path])
|
||
# 记录是否超时的标记
|
||
self.if_time_out = False
|
||
# 更新运行信息
|
||
wait_index = [
|
||
_
|
||
for _ in all_index
|
||
if (not _ in over_index + error_index + [index])
|
||
]
|
||
# 监测MAA运行状态
|
||
while self.if_run:
|
||
# 获取MAA日志
|
||
logs = self.get_maa_log(start_time)
|
||
# 判断是否超时
|
||
if len(logs) > 0:
|
||
last_time = datetime.datetime.now()
|
||
for _ in range(-1, 0 - len(logs) - 1, -1):
|
||
try:
|
||
last_time = datetime.datetime.strptime(
|
||
logs[_][1:20], "%Y-%m-%d %H:%M:%S"
|
||
)
|
||
break
|
||
except ValueError:
|
||
pass
|
||
now_time = datetime.datetime.now()
|
||
if (
|
||
j == 0
|
||
and now_time - last_time
|
||
> datetime.timedelta(minutes=self.annihilation)
|
||
) or (
|
||
j == 1
|
||
and now_time - last_time
|
||
> datetime.timedelta(minutes=self.routine)
|
||
):
|
||
self.if_time_out = True
|
||
# 合并日志
|
||
log = "".join(logs)
|
||
# 更新MAA日志
|
||
self.update_gui.emit(
|
||
self.data[index][0]
|
||
+ "_第"
|
||
+ str(i + 1)
|
||
+ "次_"
|
||
+ mode_book[j][5:7],
|
||
"\n".join([self.data[_][0] for _ in wait_index]),
|
||
"\n".join([self.data[_][0] for _ in over_index]),
|
||
"\n".join([self.data[_][0] for _ in error_index]),
|
||
log,
|
||
)
|
||
# 判断MAA程序运行状态
|
||
result = self.if_maa_success(log, mode_book[j])
|
||
if result == "Success!":
|
||
run_book[j] = True
|
||
self.update_gui.emit(
|
||
self.data[index][0]
|
||
+ "_第"
|
||
+ str(i + 1)
|
||
+ "次_"
|
||
+ mode_book[j][5:7],
|
||
"\n".join([self.data[_][0] for _ in wait_index]),
|
||
"\n".join([self.data[_][0] for _ in over_index]),
|
||
"\n".join([self.data[_][0] for _ in error_index]),
|
||
"检测到MAA进程完成代理任务\n正在等待相关程序结束\n请等待10s",
|
||
)
|
||
time.sleep(10)
|
||
break
|
||
elif result == "Wait":
|
||
# 检测时间间隔
|
||
time.sleep(1)
|
||
else:
|
||
# 打印中止信息
|
||
# 此时，log变量内存储的就是出现异常的日志信息，可以保存或发送用于问题排查
|
||
self.update_gui.emit(
|
||
self.data[index][0]
|
||
+ "_第"
|
||
+ str(i + 1)
|
||
+ "次_"
|
||
+ mode_book[j][5:7],
|
||
"\n".join([self.data[_][0] for _ in wait_index]),
|
||
"\n".join([self.data[_][0] for _ in over_index]),
|
||
"\n".join([self.data[_][0] for _ in error_index]),
|
||
result,
|
||
)
|
||
killprocess = subprocess.Popen(
|
||
"taskkill /F /T /PID " + str(maa.pid),
|
||
shell=True,
|
||
creationflags=subprocess.CREATE_NO_WINDOW,
|
||
)
|
||
killprocess.wait()
|
||
self.push_notification.emit(
|
||
"用户日常代理出现异常！",
|
||
"用户 "
|
||
+ self.data[index][0].replace("_", " 今天的")
|
||
+ "的"
|
||
+ mode_book[j][5:7]
|
||
+ "部分出现一次异常",
|
||
self.data[index][0].replace("_", " ")
|
||
+ "的"
|
||
+ mode_book[j][5:7]
|
||
+ "出现异常",
|
||
1,
|
||
)
|
||
if self.if_run:
|
||
time.sleep(10)
|
||
break
|
||
if run_book[0] and run_book[1]:
|
||
if self.data[index][14] == 0 and self.data[index][3] != -1:
|
||
self.data[index][3] -= 1
|
||
self.data[index][14] += 1
|
||
over_index.append(index)
|
||
self.push_notification.emit(
|
||
"成功完成一个日常代理任务！",
|
||
"已完成用户 "
|
||
+ self.data[index][0].replace("_", " 今天的")
|
||
+ "任务",
|
||
"已完成 " + self.data[index][0].replace("_", " 的"),
|
||
3,
|
||
)
|
||
break
|
||
if not (run_book[0] and run_book[1]):
|
||
error_index.append(index)
|
||
# 人工排查模式
|
||
elif self.mode == "人工排查":
|
||
# 标记是否需要启动模拟器
|
||
if_strat_app = True
|
||
# 标识排查模式
|
||
for _ in all_index:
|
||
self.data[_][0] += "_排查模式"
|
||
# 开始排查
|
||
for index in all_index:
|
||
if not self.if_run:
|
||
break
|
||
if self.data[index][15] == "beta":
|
||
if_strat_app = True
|
||
run_book = [False for _ in range(2)]
|
||
# 启动重试循环
|
||
while self.if_run:
|
||
# 配置MAA
|
||
if if_strat_app:
|
||
self.set_maa("人工排查_启动模拟器", index)
|
||
if_strat_app = False
|
||
else:
|
||
self.set_maa("人工排查_仅切换账号", index)
|
||
# 记录当前时间
|
||
start_time = datetime.datetime.now()
|
||
# 创建MAA任务
|
||
maa = subprocess.Popen([self.maa_path])
|
||
# 更新运行信息
|
||
wait_index = [
|
||
_
|
||
for _ in all_index
|
||
if (not _ in over_index + error_index + [index])
|
||
]
|
||
# 监测MAA运行状态
|
||
while self.if_run:
|
||
# 获取MAA日志
|
||
logs = self.get_maa_log(start_time)
|
||
# 合并日志
|
||
log = "".join(logs)
|
||
# 更新MAA日志
|
||
self.update_gui.emit(
|
||
self.data[index][0],
|
||
"\n".join([self.data[_][0] for _ in wait_index]),
|
||
"\n".join([self.data[_][0] for _ in over_index]),
|
||
"\n".join([self.data[_][0] for _ in error_index]),
|
||
log,
|
||
)
|
||
# 判断MAA程序运行状态
|
||
result = self.if_maa_success(log, "人工排查")
|
||
if result == "Success!":
|
||
run_book[0] = True
|
||
self.update_gui.emit(
|
||
self.data[index][0],
|
||
"\n".join([self.data[_][0] for _ in wait_index]),
|
||
"\n".join([self.data[_][0] for _ in over_index]),
|
||
"\n".join([self.data[_][0] for _ in error_index]),
|
||
"检测到MAA进程成功登录PRTS",
|
||
)
|
||
break
|
||
elif result == "Wait":
|
||
# 检测时间间隔
|
||
time.sleep(1)
|
||
else:
|
||
self.update_gui.emit(
|
||
self.data[index][0],
|
||
"\n".join([self.data[_][0] for _ in wait_index]),
|
||
"\n".join([self.data[_][0] for _ in over_index]),
|
||
"\n".join([self.data[_][0] for _ in error_index]),
|
||
result,
|
||
)
|
||
killprocess = subprocess.Popen(
|
||
"taskkill /F /T /PID " + str(maa.pid),
|
||
shell=True,
|
||
creationflags=subprocess.CREATE_NO_WINDOW,
|
||
)
|
||
killprocess.wait()
|
||
if_strat_app = True
|
||
if self.if_run:
|
||
time.sleep(10)
|
||
break
|
||
if run_book[0]:
|
||
break
|
||
elif self.if_run:
|
||
self.question_title = "操作提示"
|
||
self.question_info = "MAA未能正确登录到PRTS，是否重试？"
|
||
self.question_choice = "wait"
|
||
self.question.emit()
|
||
while self.question_choice == "wait":
|
||
time.sleep(1)
|
||
if self.question_choice == "No":
|
||
break
|
||
if run_book[0] and self.if_run:
|
||
self.question_title = "操作提示"
|
||
self.question_info = "请检查用户代理情况，如无异常请按下确认键。"
|
||
self.question_choice = "wait"
|
||
self.question.emit()
|
||
while self.question_choice == "wait":
|
||
time.sleep(1)
|
||
if self.question_choice == "Yes":
|
||
run_book[1] = True
|
||
if run_book[0] and run_book[1]:
|
||
if "未通过人工排查" in self.data[index][13]:
|
||
self.data[index][13] = self.data[index][13].replace(
|
||
"未通过人工排查|", ""
|
||
)
|
||
over_index.append(index)
|
||
elif not (run_book[0] and run_book[1]):
|
||
if not "未通过人工排查" in self.data[index][13]:
|
||
self.data[index][13] = "未通过人工排查|" + self.data[index][13]
|
||
error_index.append(index)
|
||
# 设置MAA模式
|
||
elif "设置MAA" in self.mode:
|
||
# 配置MAA
|
||
self.set_maa(self.mode, "")
|
||
# 创建MAA任务
|
||
maa = subprocess.Popen([self.maa_path])
|
||
# 记录当前时间
|
||
start_time = datetime.datetime.now()
|
||
# 监测MAA运行状态
|
||
while self.if_run:
|
||
# 获取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)
|
||
if "全局" in self.mode:
|
||
self.get_json.emit(["Default"])
|
||
elif "用户" in self.mode:
|
||
self.get_json.emit(self.get_json_path)
|
||
self.accomplish.emit()
|
||
self.if_run = False
|
||
if self.mode in ["日常代理", "人工排查"]:
|
||
# 关闭可能未正常退出的MAA进程
|
||
if not self.if_run:
|
||
killprocess = subprocess.Popen(
|
||
"taskkill /F /T /PID " + str(maa.pid),
|
||
shell=True,
|
||
creationflags=subprocess.CREATE_NO_WINDOW,
|
||
)
|
||
killprocess.wait()
|
||
# 更新用户数据
|
||
modes = [self.data[_][15] for _ in all_index]
|
||
uids = [self.data[_][16] for _ in all_index]
|
||
days = [self.data[_][3] for _ in all_index]
|
||
lasts = [self.data[_][5] for _ in all_index]
|
||
notes = [self.data[_][13] for _ in all_index]
|
||
numbs = [self.data[_][14] for _ in all_index]
|
||
self.update_user_info.emit(modes, uids, days, lasts, notes, numbs)
|
||
# 保存运行日志
|
||
end_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
with open(self.app_path + "/log.txt", "w", encoding="utf-8") as f:
|
||
print("任务开始时间：" + begin_time + "，结束时间：" + end_time, file=f)
|
||
print(
|
||
"已完成数："
|
||
+ str(len(over_index))
|
||
+ "，未完成数："
|
||
+ str(len(error_index) + len(wait_index))
|
||
+ "\n",
|
||
file=f,
|
||
)
|
||
if len(error_index) != 0:
|
||
print(self.mode[2:4] + "未成功的用户：", file=f)
|
||
print("\n".join([self.data[_][0] for _ in error_index]), file=f)
|
||
wait_index = [
|
||
_ for _ in all_index if (not _ in over_index + error_index)
|
||
]
|
||
if len(wait_index) != 0:
|
||
print("\n未开始" + self.mode[2:4] + "的用户：", file=f)
|
||
print("\n".join([self.data[_][0] for _ in wait_index]), file=f)
|
||
# 恢复GUI运行面板
|
||
with open(self.app_path + "/log.txt", "r", encoding="utf-8") as f:
|
||
end_log = f.read()
|
||
self.update_gui.emit("", "", "", "", end_log)
|
||
# 推送windows通知
|
||
self.push_notification.emit(
|
||
self.mode[2:4] + "任务已完成！",
|
||
"已完成用户数："
|
||
+ str(len(over_index))
|
||
+ "，未完成用户数："
|
||
+ str(len(error_index) + len(wait_index)),
|
||
"已完成用户数："
|
||
+ str(len(over_index))
|
||
+ "，未完成用户数："
|
||
+ str(len(error_index) + len(wait_index)),
|
||
10,
|
||
)
|
||
self.accomplish.emit()
|
||
self.if_run = False
|
||
|
||
def get_maa_log(self, start_time):
|
||
"""获取MAA日志"""
|
||
logs = []
|
||
if_log_start = False
|
||
with open(self.log_path, "r", encoding="utf-8") as f:
|
||
for entry in f:
|
||
if not if_log_start:
|
||
try:
|
||
entry_time = datetime.datetime.strptime(
|
||
entry[1:20], "%Y-%m-%d %H:%M:%S"
|
||
)
|
||
if entry_time > start_time:
|
||
if_log_start = True
|
||
logs.append(entry)
|
||
except ValueError:
|
||
pass
|
||
else:
|
||
logs.append(entry)
|
||
return logs
|
||
|
||
def if_maa_success(self, log, mode):
|
||
"""判断MAA程序运行状态"""
|
||
if "日常代理" in mode:
|
||
if mode == "日常代理_日常" and "任务出错: Fight" in log:
|
||
return "检测到MAA未能实际执行任务\n正在中止相关程序\n请等待10s"
|
||
if "任务出错: StartUp" in log:
|
||
return "检测到MAA未能正确登录PRTS\n正在中止相关程序\n请等待10s"
|
||
elif "任务已全部完成！" in log:
|
||
return "Success!"
|
||
elif (
|
||
("请「检查连接设置」或「尝试重启模拟器与 ADB」或「重启电脑」" 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"
|
||
elif not self.if_run:
|
||
return "您中止了本次任务\n正在中止相关程序\n请等待"
|
||
else:
|
||
return "Wait"
|
||
elif mode == "人工排查":
|
||
if "完成任务: StartUp" in log:
|
||
return "Success!"
|
||
elif (
|
||
("请「检查连接设置」或「尝试重启模拟器与 ADB」或「重启电脑」" in log)
|
||
or ("已停止" in log)
|
||
or ("MaaAssistantArknights GUI exited" in log)
|
||
):
|
||
return "检测到MAA进程异常\n正在中止相关程序\n请等待10s"
|
||
elif not self.if_run:
|
||
return "您中止了本次任务\n正在中止相关程序\n请等待"
|
||
else:
|
||
return "Wait"
|
||
elif mode == "设置MAA":
|
||
if "MaaAssistantArknights GUI exited" in log:
|
||
return "Success!"
|
||
else:
|
||
return "Wait"
|
||
|
||
def set_maa(self, mode, index):
|
||
"""配置MAA运行参数"""
|
||
if mode == "设置MAA_用户":
|
||
set_book1 = ["/simple/", "/beta/"]
|
||
set_book2 = ["/routine/gui.json", "/annihilation/gui.json"]
|
||
shutil.copy(
|
||
self.json_path
|
||
+ set_book1[self.get_json_path[0]]
|
||
+ str(self.get_json_path[1])
|
||
+ set_book2[self.get_json_path[2]],
|
||
self.set_path,
|
||
)
|
||
elif (mode == "设置MAA_全局") or (
|
||
("日常代理" in mode or "人工排查" in mode)
|
||
and self.data[index][15] == "simple"
|
||
):
|
||
shutil.copy(
|
||
self.json_path + "/Default/gui.json",
|
||
self.set_path,
|
||
)
|
||
elif "日常代理" in mode and self.data[index][15] == "beta":
|
||
if mode == "日常代理_剿灭":
|
||
shutil.copy(
|
||
self.json_path
|
||
+ "/beta/"
|
||
+ str(self.data[index][16])
|
||
+ "/annihilation/gui.json",
|
||
self.set_path,
|
||
)
|
||
elif mode == "日常代理_日常":
|
||
shutil.copy(
|
||
self.json_path
|
||
+ "/beta/"
|
||
+ str(self.data[index][16])
|
||
+ "/routine/gui.json",
|
||
self.set_path,
|
||
)
|
||
elif "人工排查" in mode and self.data[index][15] == "beta":
|
||
shutil.copy(
|
||
self.json_path
|
||
+ "/beta/"
|
||
+ str(self.data[index][16])
|
||
+ "/routine/gui.json",
|
||
self.set_path,
|
||
)
|
||
with open(self.set_path, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
# 人工排查配置
|
||
if "人工排查" in mode:
|
||
data["Current"] = "Default" # 切换配置
|
||
for i in range(1, 9):
|
||
data["Global"]["Timer.Timer" + str(i)] = "False" # 时间设置
|
||
data["Configurations"]["Default"][
|
||
"MainFunction.PostActions"
|
||
] = "8" # 完成后退出MAA
|
||
data["Configurations"]["Default"][
|
||
"Start.RunDirectly"
|
||
] = "True" # 启动MAA后直接运行
|
||
data["Configurations"]["Default"][
|
||
"Start.MinimizeDirectly"
|
||
] = "True" # 启动MAA后直接最小化
|
||
# 启动MAA后自动开启模拟器
|
||
if "启动模拟器" in mode:
|
||
data["Configurations"]["Default"][
|
||
"Start.OpenEmulatorAfterLaunch"
|
||
] = "True"
|
||
elif "仅切换账号" in mode:
|
||
data["Configurations"]["Default"][
|
||
"Start.OpenEmulatorAfterLaunch"
|
||
] = "False"
|
||
data["Configurations"]["Default"][
|
||
"Start.MinimizingStartup"
|
||
] = "False" # 最小化启动模拟器
|
||
if self.data[index][15] == "simple":
|
||
data["Global"][
|
||
"VersionUpdate.ScheduledUpdateCheck"
|
||
] = "False" # 定时检查更新
|
||
data["Global"][
|
||
"VersionUpdate.AutoDownloadUpdatePackage"
|
||
] = "False" # 自动下载更新包
|
||
data["Global"][
|
||
"VersionUpdate.AutoInstallUpdatePackage"
|
||
] = "False" # 自动安装更新包
|
||
data["Configurations"]["Default"]["Start.ClientType"] = self.data[
|
||
index
|
||
][
|
||
2
|
||
] # 客户端类型
|
||
# 账号切换
|
||
if self.data[index][2] == "Official":
|
||
data["Configurations"]["Default"]["Start.AccountName"] = (
|
||
self.data[index][1][:3] + "****" + self.data[index][1][7:]
|
||
)
|
||
elif self.data[index][2] == "Bilibili":
|
||
data["Configurations"]["Default"]["Start.AccountName"] = self.data[
|
||
index
|
||
][1]
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.WakeUp.IsChecked"
|
||
] = "True" # 开始唤醒
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.Recruiting.IsChecked"
|
||
] = "False" # 自动公招
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.Base.IsChecked"
|
||
] = "False" # 基建换班
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.Combat.IsChecked"
|
||
] = "False" # 刷理智
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.Mission.IsChecked"
|
||
] = "False" # 领取奖励
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.Mall.IsChecked"
|
||
] = "False" # 获取信用及购物
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.AutoRoguelike.IsChecked"
|
||
] = "False" # 自动肉鸽
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.Reclamation.IsChecked"
|
||
] = "False" # 生息演算
|
||
# 设置MAA配置
|
||
elif "设置MAA" in mode:
|
||
data["Current"] = "Default" # 切换配置
|
||
for i in range(1, 9):
|
||
data["Global"]["Timer.Timer" + str(i)] = "False" # 时间设置
|
||
data["Configurations"]["Default"][
|
||
"MainFunction.PostActions"
|
||
] = "0" # 完成后无动作
|
||
data["Configurations"]["Default"][
|
||
"Start.RunDirectly"
|
||
] = "False" # 启动MAA后直接运行
|
||
data["Configurations"]["Default"][
|
||
"Start.OpenEmulatorAfterLaunch"
|
||
] = "False" # 启动MAA后自动开启模拟器
|
||
if "全局" in mode:
|
||
data["Global"][
|
||
"VersionUpdate.ScheduledUpdateCheck"
|
||
] = "False" # 定时检查更新
|
||
data["Global"][
|
||
"VersionUpdate.AutoDownloadUpdatePackage"
|
||
] = "False" # 自动下载更新包
|
||
data["Global"][
|
||
"VersionUpdate.AutoInstallUpdatePackage"
|
||
] = "False" # 自动安装更新包
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.WakeUp.IsChecked"
|
||
] = "False" # 开始唤醒
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.Recruiting.IsChecked"
|
||
] = "False" # 自动公招
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.Base.IsChecked"
|
||
] = "False" # 基建换班
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.Combat.IsChecked"
|
||
] = "False" # 刷理智
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.Mission.IsChecked"
|
||
] = "False" # 领取奖励
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.Mall.IsChecked"
|
||
] = "False" # 获取信用及购物
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.AutoRoguelike.IsChecked"
|
||
] = "False" # 自动肉鸽
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.Reclamation.IsChecked"
|
||
] = "False" # 生息演算
|
||
# 剿灭代理配置
|
||
elif mode == "日常代理_剿灭":
|
||
data["Current"] = "Default" # 切换配置
|
||
for i in range(1, 9):
|
||
data["Global"]["Timer.Timer" + str(i)] = "False" # 时间设置
|
||
data["Configurations"]["Default"][
|
||
"MainFunction.PostActions"
|
||
] = "12" # 完成后退出MAA和模拟器
|
||
data["Configurations"]["Default"][
|
||
"Start.RunDirectly"
|
||
] = "True" # 启动MAA后直接运行
|
||
data["Configurations"]["Default"][
|
||
"Start.OpenEmulatorAfterLaunch"
|
||
] = "True" # 启动MAA后自动开启模拟器
|
||
if self.data[index][15] == "simple":
|
||
data["Global"][
|
||
"VersionUpdate.ScheduledUpdateCheck"
|
||
] = "True" # 定时检查更新
|
||
data["Global"][
|
||
"VersionUpdate.AutoDownloadUpdatePackage"
|
||
] = "True" # 自动下载更新包
|
||
data["Global"][
|
||
"VersionUpdate.AutoInstallUpdatePackage"
|
||
] = "True" # 自动安装更新包
|
||
data["Configurations"]["Default"]["Start.ClientType"] = self.data[
|
||
index
|
||
][
|
||
2
|
||
] # 客户端类型
|
||
# 账号切换
|
||
if self.data[index][2] == "Official":
|
||
data["Configurations"]["Default"]["Start.AccountName"] = (
|
||
self.data[index][1][:3] + "****" + self.data[index][1][7:]
|
||
)
|
||
elif self.data[index][2] == "Bilibili":
|
||
data["Configurations"]["Default"]["Start.AccountName"] = self.data[
|
||
index
|
||
][1]
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.WakeUp.IsChecked"
|
||
] = "True" # 开始唤醒
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.Recruiting.IsChecked"
|
||
] = "False" # 自动公招
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.Base.IsChecked"
|
||
] = "False" # 基建换班
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.Combat.IsChecked"
|
||
] = "True" # 刷理智
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.Mission.IsChecked"
|
||
] = "False" # 领取奖励
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.Mall.IsChecked"
|
||
] = "False" # 获取信用及购物
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.AutoRoguelike.IsChecked"
|
||
] = "False" # 自动肉鸽
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.Reclamation.IsChecked"
|
||
] = "False" # 生息演算
|
||
data["Configurations"]["Default"][
|
||
"MainFunction.Stage1"
|
||
] = "Annihilation" # 主关卡
|
||
data["Configurations"]["Default"][
|
||
"MainFunction.Stage2"
|
||
] = "" # 备选关卡1
|
||
data["Configurations"]["Default"][
|
||
"MainFunction.Stage3"
|
||
] = "" # 备选关卡2
|
||
data["Configurations"]["Default"][
|
||
"Fight.RemainingSanityStage"
|
||
] = "" # 剩余理智关卡
|
||
data["Configurations"]["Default"][
|
||
"MainFunction.Series.Quantity"
|
||
] = "1" # 连战次数
|
||
data["Configurations"]["Default"][
|
||
"Penguin.IsDrGrandet"
|
||
] = "False" # 博朗台模式
|
||
data["Configurations"]["Default"][
|
||
"GUI.CustomStageCode"
|
||
] = "True" # 手动输入关卡名
|
||
data["Configurations"]["Default"][
|
||
"GUI.UseAlternateStage"
|
||
] = "False" # 使用备选关卡
|
||
data["Configurations"]["Default"][
|
||
"Fight.UseRemainingSanityStage"
|
||
] = "False" # 使用剩余理智
|
||
data["Configurations"]["Default"][
|
||
"Fight.UseExpiringMedicine"
|
||
] = "True" # 无限吃48小时内过期的理智药
|
||
# 日常代理配置
|
||
elif mode == "日常代理_日常":
|
||
data["Current"] = "Default" # 切换配置
|
||
for i in range(1, 9):
|
||
data["Global"]["Timer.Timer" + str(i)] = "False" # 时间设置
|
||
data["Configurations"]["Default"][
|
||
"MainFunction.PostActions"
|
||
] = "12" # 完成后退出MAA和模拟器
|
||
data["Configurations"]["Default"][
|
||
"Start.RunDirectly"
|
||
] = "True" # 启动MAA后直接运行
|
||
data["Configurations"]["Default"][
|
||
"Start.OpenEmulatorAfterLaunch"
|
||
] = "True" # 启动MAA后自动开启模拟器
|
||
if self.data[index][15] == "simple":
|
||
data["Global"][
|
||
"VersionUpdate.ScheduledUpdateCheck"
|
||
] = "True" # 定时检查更新
|
||
data["Global"][
|
||
"VersionUpdate.AutoDownloadUpdatePackage"
|
||
] = "True" # 自动下载更新包
|
||
data["Global"][
|
||
"VersionUpdate.AutoInstallUpdatePackage"
|
||
] = "True" # 自动安装更新包
|
||
data["Configurations"]["Default"]["Start.ClientType"] = self.data[
|
||
index
|
||
][
|
||
2
|
||
] # 客户端类型
|
||
# 账号切换
|
||
if self.data[index][2] == "Official":
|
||
data["Configurations"]["Default"]["Start.AccountName"] = (
|
||
self.data[index][1][:3] + "****" + self.data[index][1][7:]
|
||
)
|
||
elif self.data[index][2] == "Bilibili":
|
||
data["Configurations"]["Default"]["Start.AccountName"] = self.data[
|
||
index
|
||
][1]
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.WakeUp.IsChecked"
|
||
] = "True" # 开始唤醒
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.Recruiting.IsChecked"
|
||
] = "True" # 自动公招
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.Base.IsChecked"
|
||
] = "True" # 基建换班
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.Combat.IsChecked"
|
||
] = "True" # 刷理智
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.Mission.IsChecked"
|
||
] = "True" # 领取奖励
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.Mall.IsChecked"
|
||
] = "True" # 获取信用及购物
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.AutoRoguelike.IsChecked"
|
||
] = "False" # 自动肉鸽
|
||
data["Configurations"]["Default"][
|
||
"TaskQueue.Reclamation.IsChecked"
|
||
] = "False" # 生息演算
|
||
# 主关卡
|
||
if self.data[index][6] == "-":
|
||
data["Configurations"]["Default"]["MainFunction.Stage1"] = ""
|
||
else:
|
||
data["Configurations"]["Default"]["MainFunction.Stage1"] = (
|
||
self.data[index][6]
|
||
)
|
||
# 备选关卡1
|
||
if self.data[index][7] == "-":
|
||
data["Configurations"]["Default"]["MainFunction.Stage2"] = ""
|
||
else:
|
||
data["Configurations"]["Default"]["MainFunction.Stage2"] = (
|
||
self.data[index][7]
|
||
)
|
||
# 备选关卡2
|
||
if self.data[index][8] == "-":
|
||
data["Configurations"]["Default"]["MainFunction.Stage3"] = ""
|
||
else:
|
||
data["Configurations"]["Default"]["MainFunction.Stage3"] = (
|
||
self.data[index][8]
|
||
)
|
||
data["Configurations"]["Default"][
|
||
"Fight.RemainingSanityStage"
|
||
] = "" # 剩余理智关卡
|
||
# 连战次数
|
||
if self.data[index][6] == "1-7":
|
||
data["Configurations"]["Default"][
|
||
"MainFunction.Series.Quantity"
|
||
] = "6"
|
||
else:
|
||
data["Configurations"]["Default"][
|
||
"MainFunction.Series.Quantity"
|
||
] = "1"
|
||
data["Configurations"]["Default"][
|
||
"Penguin.IsDrGrandet"
|
||
] = "False" # 博朗台模式
|
||
data["Configurations"]["Default"][
|
||
"GUI.CustomStageCode"
|
||
] = "True" # 手动输入关卡名
|
||
# 备选关卡
|
||
if self.data[index][7] == "-" and self.data[index][8] == "-":
|
||
data["Configurations"]["Default"]["GUI.UseAlternateStage"] = "False"
|
||
else:
|
||
data["Configurations"]["Default"]["GUI.UseAlternateStage"] = "True"
|
||
data["Configurations"]["Default"][
|
||
"Fight.UseRemainingSanityStage"
|
||
] = "False" # 使用剩余理智
|
||
data["Configurations"]["Default"][
|
||
"Fight.UseExpiringMedicine"
|
||
] = "True" # 无限吃48小时内过期的理智药
|
||
# 自定义基建配置
|
||
if self.data[index][11] == "n":
|
||
data["Configurations"]["Default"][
|
||
"Infrast.CustomInfrastEnabled"
|
||
] = "False" # 禁用自定义基建配置
|
||
else:
|
||
data["Configurations"]["Default"][
|
||
"Infrast.CustomInfrastEnabled"
|
||
] = "True" # 启用自定义基建配置
|
||
data["Configurations"]["Default"][
|
||
"Infrast.DefaultInfrast"
|
||
] = "user_defined" # 内置配置
|
||
data["Configurations"]["Default"][
|
||
"Infrast.IsCustomInfrastFileReadOnly"
|
||
] = "False" # 自定义基建配置文件只读
|
||
data["Configurations"]["Default"]["Infrast.CustomInfrastFile"] = (
|
||
self.json_path
|
||
+ "/simple/"
|
||
+ str(self.data[index][16])
|
||
+ "/infrastructure/infrastructure.json"
|
||
) # 自定义基建配置文件地址
|
||
# 覆写配置文件
|
||
with open(self.set_path, "w", encoding="utf-8") as f:
|
||
json.dump(data, f, indent=4)
|
||
return True
|
||
|
||
|
||
class MainTimer(QtCore.QThread):
|
||
|
||
get_config = QtCore.Signal()
|
||
start_for_timer = QtCore.Signal()
|
||
app_path = os.path.realpath(sys.argv[0]) # 获取软件自身的路径
|
||
app_name = os.path.basename(app_path) # 获取软件自身的名称
|
||
ES_CONTINUOUS = 0x80000000
|
||
ES_SYSTEM_REQUIRED = 0x00000001
|
||
is_maa_run = False
|
||
|
||
def __init__(self, config):
|
||
super(MainTimer, self).__init__()
|
||
self.config = config
|
||
|
||
def run(self):
|
||
"""主功能代码，实现定时执行以及相关配置信息的实时同步"""
|
||
while True:
|
||
self.get_config.emit()
|
||
self.set_system()
|
||
time_set = [
|
||
self.config["Default"]["TimeSet.run" + str(_ + 1)]
|
||
for _ in range(10)
|
||
if self.config["Default"]["TimeSet.set" + str(_ + 1)] == "True"
|
||
]
|
||
curtime = datetime.datetime.now().strftime("%H:%M")
|
||
if (curtime in time_set) and not self.is_maa_run:
|
||
self.start_for_timer.emit()
|
||
time.sleep(1)
|
||
|
||
def set_system(self):
|
||
"""设置系统相关配置"""
|
||
# 同步系统休眠状态
|
||
if self.config["Default"]["SelfSet.IfSleep"] == "True":
|
||
# 设置系统电源状态
|
||
ctypes.windll.kernel32.SetThreadExecutionState(
|
||
self.ES_CONTINUOUS | self.ES_SYSTEM_REQUIRED
|
||
)
|
||
elif self.config["Default"]["SelfSet.IfSleep"] == "False":
|
||
# 恢复系统电源状态
|
||
ctypes.windll.kernel32.SetThreadExecutionState(self.ES_CONTINUOUS)
|
||
|
||
# 同步开机自启
|
||
if (
|
||
self.config["Default"]["SelfSet.IfSelfStart"] == "True"
|
||
and not self.is_startup()
|
||
):
|
||
key = winreg.OpenKey(
|
||
winreg.HKEY_CURRENT_USER,
|
||
r"Software\Microsoft\Windows\CurrentVersion\Run",
|
||
winreg.KEY_SET_VALUE,
|
||
winreg.KEY_ALL_ACCESS | winreg.KEY_WRITE | winreg.KEY_CREATE_SUB_KEY,
|
||
)
|
||
winreg.SetValueEx(key, self.app_name, 0, winreg.REG_SZ, self.app_path)
|
||
winreg.CloseKey(key)
|
||
elif (
|
||
self.config["Default"]["SelfSet.IfSelfStart"] == "False"
|
||
and self.is_startup()
|
||
):
|
||
key = winreg.OpenKey(
|
||
winreg.HKEY_CURRENT_USER,
|
||
r"Software\Microsoft\Windows\CurrentVersion\Run",
|
||
winreg.KEY_SET_VALUE,
|
||
winreg.KEY_ALL_ACCESS | winreg.KEY_WRITE | winreg.KEY_CREATE_SUB_KEY,
|
||
)
|
||
winreg.DeleteValue(key, self.app_name)
|
||
winreg.CloseKey(key)
|
||
|
||
def is_startup(self):
|
||
"""判断程序是否已经开机自启"""
|
||
key = winreg.OpenKey(
|
||
winreg.HKEY_CURRENT_USER,
|
||
r"Software\Microsoft\Windows\CurrentVersion\Run",
|
||
0,
|
||
winreg.KEY_READ,
|
||
)
|
||
try:
|
||
value, _ = winreg.QueryValueEx(key, self.app_name)
|
||
return True
|
||
except FileNotFoundError:
|
||
return False
|
||
finally:
|
||
winreg.CloseKey(key)
|
||
|
||
|
||
class Main(QWidget):
|
||
|
||
app_path = os.path.dirname(os.path.realpath(sys.argv[0])).replace(
|
||
"\\", "/"
|
||
) # 获取软件自身的路径
|
||
|
||
def __init__(self, PASSWARD=""):
|
||
super().__init__()
|
||
|
||
self.database_path = self.app_path + "/data/data.db"
|
||
self.config_path = self.app_path + "/config/gui.json"
|
||
self.key_path = self.app_path + "/data/key"
|
||
self.gameid_path = self.app_path + "/data/gameid.txt"
|
||
self.version_path = self.app_path + "/res/version.json"
|
||
self.PASSWORD = PASSWARD
|
||
self.if_user_list_editable = True
|
||
self.if_update_database = True
|
||
self.if_update_config = True
|
||
self.user_mode_list = ["simple", "beta"]
|
||
self.user_column = [
|
||
"admin",
|
||
"id",
|
||
"server",
|
||
"day",
|
||
"status",
|
||
"last",
|
||
"game",
|
||
"game_1",
|
||
"game_2",
|
||
"routines",
|
||
"annihilation",
|
||
"infrastructure",
|
||
"password",
|
||
"notes",
|
||
"numb",
|
||
"mode",
|
||
"uid",
|
||
]
|
||
self.userlist_simple_index = [
|
||
0,
|
||
1,
|
||
2,
|
||
3,
|
||
4,
|
||
5,
|
||
6,
|
||
7,
|
||
8,
|
||
"-",
|
||
9,
|
||
10,
|
||
11,
|
||
12,
|
||
"-",
|
||
"-",
|
||
"-",
|
||
]
|
||
self.userlist_beta_index = [
|
||
0,
|
||
"-",
|
||
"-",
|
||
1,
|
||
2,
|
||
3,
|
||
"-",
|
||
"-",
|
||
"-",
|
||
4,
|
||
5,
|
||
"-",
|
||
6,
|
||
7,
|
||
"-",
|
||
"-",
|
||
"-",
|
||
]
|
||
|
||
self.ui = uiLoader.load(self.app_path + "/gui/ui/main.ui")
|
||
self.ui.setWindowIcon(QIcon(self.app_path + "/gui/ico/AUTO_MAA.ico"))
|
||
# 检查文件完整性
|
||
self.initialize()
|
||
self.check_config()
|
||
self.check_database()
|
||
# 初始化数据库连接
|
||
self.db = sqlite3.connect(self.database_path)
|
||
self.cur = self.db.cursor()
|
||
# 初始化控件
|
||
self.user_set = self.ui.findChild(QToolBox, "toolBox_userset")
|
||
self.user_set.currentChanged.connect(self.change_userlist_method)
|
||
|
||
self.user_list_simple = self.ui.findChild(
|
||
QTableWidget, "tableWidget_userlist_simple"
|
||
)
|
||
self.user_list_simple.itemChanged.connect(
|
||
lambda item: self.change_user_Item(item, "simple")
|
||
)
|
||
self.user_list_simple.setStyleSheet("background-color: rgb(255, 255, 255);")
|
||
|
||
self.user_list_beta = self.ui.findChild(
|
||
QTableWidget, "tableWidget_userlist_beta"
|
||
)
|
||
self.user_list_beta.itemChanged.connect(
|
||
lambda item: self.change_user_Item(item, "beta")
|
||
)
|
||
self.user_list_beta.setStyleSheet("background-color: rgb(255, 255, 255);")
|
||
|
||
self.user_add = self.ui.findChild(QPushButton, "pushButton_new")
|
||
self.user_add.clicked.connect(self.add_user)
|
||
|
||
self.user_del = self.ui.findChild(QPushButton, "pushButton_del")
|
||
self.user_del.clicked.connect(self.del_user)
|
||
|
||
self.user_switch = self.ui.findChild(QPushButton, "pushButton_switch")
|
||
self.user_switch.clicked.connect(self.switch_user)
|
||
|
||
self.user_changeset = self.ui.findChild(QPushButton, "pushButton_changeset")
|
||
self.user_changeset.clicked.connect(self.change_user_set)
|
||
|
||
self.read_PASSWORD = self.ui.findChild(QPushButton, "pushButton_password")
|
||
self.read_PASSWORD.clicked.connect(lambda: self.read("key"))
|
||
|
||
self.refresh = self.ui.findChild(QPushButton, "pushButton_refresh")
|
||
self.refresh.clicked.connect(lambda: self.update_user_info("clear"))
|
||
|
||
self.run_now = self.ui.findChild(QPushButton, "pushButton_runnow")
|
||
self.run_now.clicked.connect(self.routine_starter)
|
||
|
||
self.check_start = self.ui.findChild(QPushButton, "pushButton_checkstart")
|
||
self.check_start.clicked.connect(self.check_starter)
|
||
|
||
self.maa_path = self.ui.findChild(QLineEdit, "lineEdit_MAApath")
|
||
self.maa_path.textChanged.connect(self.change_config)
|
||
self.maa_path.setReadOnly(True)
|
||
|
||
self.get_maa_path = self.ui.findChild(QPushButton, "pushButton_getMAApath")
|
||
self.get_maa_path.clicked.connect(lambda: self.read("file_path_maa"))
|
||
|
||
self.set_maa = self.ui.findChild(QPushButton, "pushButton_setMAA")
|
||
self.set_maa.clicked.connect(lambda: self.maa_set_starter("设置MAA_全局"))
|
||
|
||
self.routine = self.ui.findChild(QSpinBox, "spinBox_routine")
|
||
self.routine.valueChanged.connect(self.change_config)
|
||
|
||
self.annihilation = self.ui.findChild(QSpinBox, "spinBox_annihilation")
|
||
self.annihilation.valueChanged.connect(self.change_config)
|
||
|
||
self.num = self.ui.findChild(QSpinBox, "spinBox_numt")
|
||
self.num.valueChanged.connect(self.change_config)
|
||
|
||
self.if_self_start = self.ui.findChild(QCheckBox, "checkBox_ifselfstart")
|
||
self.if_self_start.stateChanged.connect(self.change_config)
|
||
|
||
self.if_sleep = self.ui.findChild(QCheckBox, "checkBox_ifsleep")
|
||
self.if_sleep.stateChanged.connect(self.change_config)
|
||
|
||
self.if_proxy_directly = self.ui.findChild(
|
||
QCheckBox, "checkBox_ifproxydirectly"
|
||
)
|
||
self.if_proxy_directly.stateChanged.connect(self.change_config)
|
||
|
||
self.check_update = self.ui.findChild(QPushButton, "pushButton_check_update")
|
||
self.check_update.clicked.connect(self.check_version)
|
||
|
||
self.run_text = self.ui.findChild(QTextBrowser, "textBrowser_run")
|
||
self.wait_text = self.ui.findChild(QTextBrowser, "textBrowser_wait")
|
||
self.over_text = self.ui.findChild(QTextBrowser, "textBrowser_over")
|
||
self.error_text = self.ui.findChild(QTextBrowser, "textBrowser_error")
|
||
self.log_text = self.ui.findChild(QTextBrowser, "textBrowser_log")
|
||
|
||
self.start_time = []
|
||
for i in range(10):
|
||
listx = []
|
||
listx.append(self.ui.findChild(QCheckBox, "checkBox_t" + str(i + 1)))
|
||
listx.append(self.ui.findChild(QTimeEdit, "timeEdit_" + str(i + 1)))
|
||
self.start_time.append(listx)
|
||
self.start_time[i][0].stateChanged.connect(self.change_config)
|
||
self.start_time[i][1].timeChanged.connect(self.change_config)
|
||
|
||
self.change_password = self.ui.findChild(
|
||
QPushButton, "pushButton_changePASSWORD"
|
||
)
|
||
self.change_password.clicked.connect(self.change_PASSWORD)
|
||
# 初始化线程
|
||
self.set_path_ = self.config["Default"]["MaaSet.path"] + "/config/gui.json"
|
||
self.log_path_ = self.config["Default"]["MaaSet.path"] + "/debug/gui.log"
|
||
self.maa_path_ = self.config["Default"]["MaaSet.path"] + "/MAA.exe"
|
||
self.routine_ = self.config["Default"]["TimeLimit.routine"]
|
||
self.annihilation_ = self.config["Default"]["TimeLimit.annihilation"]
|
||
self.num_ = self.config["Default"]["TimesLimit.run"]
|
||
self.cur.execute("SELECT * FROM adminx WHERE True")
|
||
self.data_ = self.cur.fetchall()
|
||
self.data_ = [list(row) for row in self.data_]
|
||
|
||
self.MaaRunner = MaaRunner(
|
||
self.set_path_,
|
||
self.log_path_,
|
||
self.maa_path_,
|
||
self.routine_,
|
||
self.annihilation_,
|
||
self.num_,
|
||
self.data_,
|
||
"未知",
|
||
)
|
||
self.MaaRunner.question.connect(lambda: self.read("question_runner"))
|
||
self.MaaRunner.update_gui.connect(self.update_board)
|
||
self.MaaRunner.update_user_info.connect(self.change_user_info)
|
||
self.MaaRunner.push_notification.connect(self.push_notification)
|
||
self.MaaRunner.accomplish.connect(lambda: self.maa_ender("日常代理_结束"))
|
||
self.MaaRunner.get_json.connect(self.get_maa_config)
|
||
|
||
self.MainTimer = MainTimer(self.config)
|
||
self.MainTimer.get_config.connect(self.give_config)
|
||
self.MainTimer.start_for_timer.connect(self.routine_starter)
|
||
self.MainTimer.start()
|
||
|
||
# 载入GUI数据
|
||
self.update_user_info("normal")
|
||
self.update_config()
|
||
self.change_userlist_method()
|
||
|
||
if self.config["Default"]["SelfSet.IfProxyDirectly"] == "True":
|
||
self.routine_starter()
|
||
|
||
def initialize(self):
|
||
"""初始化程序的配置文件"""
|
||
# 检查目录
|
||
os.makedirs(self.app_path + "/data", exist_ok=True)
|
||
os.makedirs(self.app_path + "/config", exist_ok=True)
|
||
os.makedirs(self.app_path + "/data/MAAconfig/simple", exist_ok=True)
|
||
os.makedirs(self.app_path + "/data/MAAconfig/beta", exist_ok=True)
|
||
os.makedirs(self.app_path + "/data/MAAconfig/Default", exist_ok=True)
|
||
# 生成版本信息文件
|
||
if not os.path.exists(self.version_path):
|
||
version = {
|
||
"main_version": "0.0.0.0",
|
||
"updater_version": "0.0.0.0",
|
||
}
|
||
with open(self.version_path, "w", encoding="utf-8") as f:
|
||
json.dump(version, f, indent=4)
|
||
# 生成配置文件
|
||
if not os.path.exists(self.config_path):
|
||
config = {"Default": {}}
|
||
with open(self.config_path, "w", encoding="utf-8") as f:
|
||
json.dump(config, f, indent=4)
|
||
# 生成预设gameid替换方案文件
|
||
if not os.path.exists(self.gameid_path):
|
||
with open(self.gameid_path, "w", encoding="utf-8") as f:
|
||
print(
|
||
"龙门币：CE-6\n技能：CA-5\n红票：AP-5\n经验：LS-6\n剿灭模式：Annihilation",
|
||
file=f,
|
||
)
|
||
# 生成管理密钥
|
||
if not os.path.exists(self.key_path):
|
||
while True:
|
||
self.PASSWORD, ok_pressed = QInputDialog.getText(
|
||
self.ui,
|
||
"请设置管理密钥",
|
||
"未检测到管理密钥，请设置您的管理密钥：",
|
||
QLineEdit.Password,
|
||
"",
|
||
)
|
||
if ok_pressed and self.PASSWORD != "":
|
||
self.get_PASSWORD()
|
||
break
|
||
else:
|
||
choice = QMessageBox.question(
|
||
self.ui, "确认", "您没有输入管理密钥，确定要暂时跳过这一步吗？"
|
||
)
|
||
if choice == QMessageBox.Yes:
|
||
break
|
||
|
||
def check_config(self):
|
||
"""检查配置文件字段完整性并补全"""
|
||
config_list = [
|
||
["TimeSet.set1", "False"],
|
||
["TimeSet.run1", "00:00"],
|
||
["TimeSet.set2", "False"],
|
||
["TimeSet.run2", "00:00"],
|
||
["TimeSet.set3", "False"],
|
||
["TimeSet.run3", "00:00"],
|
||
["TimeSet.set4", "False"],
|
||
["TimeSet.run4", "00:00"],
|
||
["TimeSet.set5", "False"],
|
||
["TimeSet.run5", "00:00"],
|
||
["TimeSet.set6", "False"],
|
||
["TimeSet.run6", "00:00"],
|
||
["TimeSet.set7", "False"],
|
||
["TimeSet.run7", "00:00"],
|
||
["TimeSet.set8", "False"],
|
||
["TimeSet.run8", "00:00"],
|
||
["TimeSet.set9", "False"],
|
||
["TimeSet.run9", "00:00"],
|
||
["TimeSet.set10", "False"],
|
||
["TimeSet.run10", "00:00"],
|
||
["MaaSet.path", ""],
|
||
["TimeLimit.routine", 10],
|
||
["TimeLimit.annihilation", 40],
|
||
["TimesLimit.run", 3],
|
||
["SelfSet.IfSelfStart", "False"],
|
||
["SelfSet.IfSleep", "False"],
|
||
["SelfSet.IfProxyDirectly", "False"],
|
||
]
|
||
# 导入配置文件
|
||
with open(self.config_path, "r", encoding="utf-8") as f:
|
||
config = json.load(f)
|
||
# 检查并补充缺失的字段
|
||
for i in range(len(config_list)):
|
||
if not config_list[i][0] in config["Default"]:
|
||
config["Default"][config_list[i][0]] = config_list[i][1]
|
||
self.config = config
|
||
# 导出配置文件
|
||
with open(self.config_path, "w", encoding="utf-8") as f:
|
||
json.dump(config, f, indent=4)
|
||
|
||
def check_database(self):
|
||
"""检查用户数据库文件并处理数据库版本更新"""
|
||
# 生成用户数据库
|
||
if not os.path.exists(self.database_path):
|
||
db = sqlite3.connect(self.database_path)
|
||
cur = db.cursor()
|
||
cur.execute(
|
||
"CREATE TABLE adminx(admin text,id text,server text,day int,status text,last date,game text,game_1 text,game_2 text,routines text,annihilation text,infrastructure text,password byte,notes text,numb int,mode text,uid int)"
|
||
)
|
||
cur.execute("CREATE TABLE version(v text)")
|
||
cur.execute("INSERT INTO version VALUES(?)", ("v1.2",))
|
||
db.commit()
|
||
cur.close()
|
||
db.close()
|
||
# 数据库版本更新
|
||
db = sqlite3.connect(self.database_path)
|
||
cur = db.cursor()
|
||
cur.execute("SELECT * FROM version WHERE True")
|
||
version = cur.fetchall()
|
||
# v1.0-->v1.1
|
||
if version[0][0] == "v1.0":
|
||
cur.execute("SELECT * FROM adminx WHERE True")
|
||
data = cur.fetchall()
|
||
cur.execute("DROP TABLE IF EXISTS adminx")
|
||
cur.execute(
|
||
"CREATE TABLE adminx(admin text,id text,server text,day int,status text,last date,game text,game_1 text,game_2 text,routines text,annihilation text,infrastructure text,password byte,notes text,numb int,mode text,uid int)"
|
||
)
|
||
for i in range(len(data)):
|
||
cur.execute(
|
||
"INSERT INTO adminx VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
|
||
(
|
||
data[i][0], # 0 0 0
|
||
data[i][1], # 1 1 -
|
||
"Official", # 2 2 -
|
||
data[i][2], # 3 3 1
|
||
data[i][3], # 4 4 2
|
||
data[i][4], # 5 5 3
|
||
data[i][5], # 6 6 -
|
||
data[i][6], # 7 7 -
|
||
data[i][7], # 8 8 -
|
||
"y", # 9 - 4
|
||
data[i][8], # 10 9 5
|
||
data[i][9], # 11 10 -
|
||
data[i][10], # 12 11 6
|
||
data[i][11], # 13 12 7
|
||
data[i][12], # 14 - -
|
||
"simple", # 15 - -
|
||
data[i][13], # 16 - -
|
||
),
|
||
)
|
||
self.get_maa_config([0, data[i][13], 0])
|
||
self.get_maa_config([0, data[i][13], 1])
|
||
cur.execute("DELETE FROM version WHERE v = ?", ("v1.0",))
|
||
cur.execute("INSERT INTO version VALUES(?)", ("v1.1",))
|
||
db.commit()
|
||
# v1.1-->v1.2
|
||
if version[0][0] == "v1.1":
|
||
cur.execute("SELECT * FROM adminx WHERE True")
|
||
data = cur.fetchall()
|
||
for i in range(len(data)):
|
||
cur.execute(
|
||
"UPDATE adminx SET infrastructure = 'n' WHERE mode = ? AND uid = ?",
|
||
(
|
||
data[i][15],
|
||
data[i][16],
|
||
),
|
||
)
|
||
cur.execute("DELETE FROM version WHERE v = ?", ("v1.1",))
|
||
cur.execute("INSERT INTO version VALUES(?)", ("v1.2",))
|
||
db.commit()
|
||
cur.close()
|
||
db.close()
|
||
|
||
def get_PASSWORD(self):
|
||
"""配置管理密钥"""
|
||
# 检查目录
|
||
os.makedirs(self.app_path + "/data/key", exist_ok=True)
|
||
# 生成RSA密钥对
|
||
key = RSA.generate(2048)
|
||
public_key_local = key.publickey()
|
||
private_key = key
|
||
# 保存RSA公钥
|
||
with open(self.app_path + "/data/key/public_key.pem", "wb") as f:
|
||
f.write(public_key_local.exportKey())
|
||
# 生成密钥转换与校验随机盐
|
||
PASSWORD_salt = secrets.token_hex(random.randint(32, 1024))
|
||
with open(
|
||
self.app_path + "/data/key/PASSWORDsalt.txt", "w", encoding="utf-8"
|
||
) as f:
|
||
print(PASSWORD_salt, file=f)
|
||
verify_salt = secrets.token_hex(random.randint(32, 1024))
|
||
with open(
|
||
self.app_path + "/data/key/verifysalt.txt", "w", encoding="utf-8"
|
||
) as f:
|
||
print(verify_salt, file=f)
|
||
# 将管理密钥转化为AES-256密钥
|
||
AES_password = hashlib.sha256(
|
||
(self.PASSWORD + PASSWORD_salt).encode("utf-8")
|
||
).digest()
|
||
# 生成AES-256密钥校验哈希值并保存
|
||
AES_password_verify = hashlib.sha256(
|
||
AES_password + verify_salt.encode("utf-8")
|
||
).digest()
|
||
with open(self.app_path + "/data/key/AES_password_verify.bin", "wb") as f:
|
||
f.write(AES_password_verify)
|
||
# AES-256加密RSA私钥并保存密文
|
||
AES_key = AES.new(AES_password, AES.MODE_ECB)
|
||
private_key_local = AES_key.encrypt(pad(private_key.exportKey(), 32))
|
||
with open(self.app_path + "/data/key/private_key.bin", "wb") as f:
|
||
f.write(private_key_local)
|
||
|
||
def encryptx(self, note):
|
||
"""加密数据"""
|
||
# 读取RSA公钥
|
||
with open(self.app_path + "/data/key/public_key.pem", "rb") as f:
|
||
public_key_local = RSA.import_key(f.read())
|
||
# 使用RSA公钥对数据进行加密
|
||
cipher = PKCS1_OAEP.new(public_key_local)
|
||
encrypted = cipher.encrypt(note.encode("utf-8"))
|
||
return encrypted
|
||
|
||
def decryptx(self, note):
|
||
"""解密数据"""
|
||
# 读入RSA私钥密文、盐与校验哈希值
|
||
with open(self.app_path + "/data/key/private_key.bin", "rb") as f:
|
||
private_key_local = f.read().strip()
|
||
with open(
|
||
self.app_path + "/data/key/PASSWORDsalt.txt", "r", encoding="utf-8"
|
||
) as f:
|
||
PASSWORD_salt = f.read().strip()
|
||
with open(
|
||
self.app_path + "/data/key/verifysalt.txt", "r", encoding="utf-8"
|
||
) as f:
|
||
verify_salt = f.read().strip()
|
||
with open(self.app_path + "/data/key/AES_password_verify.bin", "rb") as f:
|
||
AES_password_verify = f.read().strip()
|
||
# 将管理密钥转化为AES-256密钥并验证
|
||
AES_password = hashlib.sha256(
|
||
(self.PASSWORD + PASSWORD_salt).encode("utf-8")
|
||
).digest()
|
||
AES_password_SHA = hashlib.sha256(
|
||
AES_password + verify_salt.encode("utf-8")
|
||
).digest()
|
||
if AES_password_SHA != AES_password_verify:
|
||
return "管理密钥错误"
|
||
else:
|
||
# AES解密RSA私钥
|
||
AES_key = AES.new(AES_password, AES.MODE_ECB)
|
||
private_key_pem = unpad(AES_key.decrypt(private_key_local), 32)
|
||
private_key = RSA.import_key(private_key_pem)
|
||
# 使用RSA私钥解密数据
|
||
decrypter = PKCS1_OAEP.new(private_key)
|
||
note = decrypter.decrypt(note)
|
||
return note.decode("utf-8")
|
||
|
||
def change_PASSWORD(self):
|
||
"""修改管理密钥"""
|
||
# 获取用户信息
|
||
self.cur.execute("SELECT * FROM adminx WHERE True")
|
||
data = self.cur.fetchall()
|
||
if len(data) == 0:
|
||
QMessageBox.information(self.ui, "验证通过", "当前无用户，验证自动通过")
|
||
# 获取新的管理密钥
|
||
while True:
|
||
PASSWORD_new = self.read("newkey")
|
||
if PASSWORD_new == 0:
|
||
choice = QMessageBox.question(
|
||
self.ui,
|
||
"确认",
|
||
"您没有输入新的管理密钥，是否取消修改管理密钥？",
|
||
)
|
||
if choice == QMessageBox.Yes:
|
||
break
|
||
else:
|
||
# 修改管理密钥
|
||
self.PASSWORD = PASSWORD_new
|
||
self.get_PASSWORD()
|
||
QMessageBox.information(self.ui, "操作成功", "管理密钥修改成功")
|
||
break
|
||
else:
|
||
# 验证管理密钥
|
||
if_change = True
|
||
while if_change:
|
||
if self.read("oldkey"):
|
||
if self.decryptx(self.encryptx("")) == "管理密钥错误":
|
||
QMessageBox.critical(self.ui, "错误", "管理密钥错误")
|
||
else:
|
||
# 获取新的管理密钥
|
||
while True:
|
||
PASSWORD_new = self.read("newkey")
|
||
if PASSWORD_new == 0:
|
||
choice = QMessageBox.question(
|
||
self.ui,
|
||
"确认",
|
||
"您没有输入新的管理密钥，是否取消修改管理密钥？",
|
||
)
|
||
if choice == QMessageBox.Yes:
|
||
if_change = False
|
||
break
|
||
# 修改管理密钥
|
||
else:
|
||
# 使用旧管理密钥解密
|
||
new_data = []
|
||
for i in range(len(data)):
|
||
new_data.append(self.decryptx(data[i][12]))
|
||
# 使用新管理密钥重新加密
|
||
self.PASSWORD = PASSWORD_new
|
||
self.get_PASSWORD()
|
||
for i in range(len(data)):
|
||
self.cur.execute(
|
||
"UPDATE adminx SET password = ? WHERE mode = ? AND uid = ?",
|
||
(
|
||
self.encryptx(new_data[i]),
|
||
data[i][15],
|
||
data[i][16],
|
||
),
|
||
)
|
||
self.db.commit()
|
||
QMessageBox.information(
|
||
self.ui, "操作成功", "管理密钥修改成功"
|
||
)
|
||
if_change = False
|
||
break
|
||
else:
|
||
choice = QMessageBox.question(
|
||
self.ui, "确认", "您没有输入管理密钥，是否取消修改管理密钥？"
|
||
)
|
||
if choice == QMessageBox.Yes:
|
||
break
|
||
|
||
def update_user_info(self, operation):
|
||
"""将本地数据库中的用户配置同步至GUI的用户管理界面"""
|
||
|
||
self.cur.execute("SELECT * FROM adminx WHERE True")
|
||
data = self.cur.fetchall()
|
||
|
||
if operation == "clear":
|
||
self.PASSWORD = ""
|
||
elif operation == "read_only":
|
||
self.if_user_list_editable = False
|
||
elif operation == "editable":
|
||
self.if_user_list_editable = True
|
||
|
||
self.if_update_database = False
|
||
|
||
data_simple = [_ for _ in data if _[15] == "simple"]
|
||
self.user_list_simple.setRowCount(len(data_simple))
|
||
for i, row in enumerate(data_simple):
|
||
for j, value in enumerate(row):
|
||
if self.userlist_simple_index[j] == "-":
|
||
continue
|
||
if j == 2:
|
||
item = QComboBox()
|
||
item.addItems(["官服", "B服"])
|
||
if value == "Official":
|
||
item.setCurrentIndex(0)
|
||
elif value == "Bilibili":
|
||
item.setCurrentIndex(1)
|
||
item.currentIndexChanged.connect(
|
||
partial(
|
||
self.change_user_CellWidget,
|
||
data_simple[i][16],
|
||
self.user_column[j],
|
||
)
|
||
)
|
||
elif j in [4, 10, 11]:
|
||
item = QComboBox()
|
||
item.addItems(["启用", "禁用"])
|
||
if value == "y":
|
||
item.setCurrentIndex(0)
|
||
elif value == "n":
|
||
item.setCurrentIndex(1)
|
||
item.currentIndexChanged.connect(
|
||
partial(
|
||
self.change_user_CellWidget,
|
||
data_simple[i][16],
|
||
self.user_column[j],
|
||
)
|
||
)
|
||
elif j == 3 and value == -1:
|
||
item = QTableWidgetItem("无限")
|
||
elif j == 5:
|
||
curdate = server_date()
|
||
if curdate != value:
|
||
item = QTableWidgetItem("今日未代理")
|
||
else:
|
||
item = QTableWidgetItem(
|
||
"今日已代理" + str(data_simple[i][14]) + "次"
|
||
)
|
||
item.setFlags(QtCore.Qt.ItemIsSelectable | QtCore.Qt.ItemIsEnabled)
|
||
elif j == 12:
|
||
if self.PASSWORD == "":
|
||
item = QTableWidgetItem("******")
|
||
item.setFlags(
|
||
QtCore.Qt.ItemIsSelectable | QtCore.Qt.ItemIsEnabled
|
||
)
|
||
else:
|
||
result = self.decryptx(value)
|
||
item = QTableWidgetItem(result)
|
||
if result == "管理密钥错误":
|
||
item.setFlags(
|
||
QtCore.Qt.ItemIsSelectable | QtCore.Qt.ItemIsEnabled
|
||
)
|
||
else:
|
||
item = QTableWidgetItem(str(value))
|
||
if j in [2, 4, 10, 11]:
|
||
if not self.if_user_list_editable:
|
||
item.setEnabled(False)
|
||
self.user_list_simple.setCellWidget(
|
||
data_simple[i][16], self.userlist_simple_index[j], item
|
||
)
|
||
else:
|
||
self.user_list_simple.setItem(
|
||
data_simple[i][16], self.userlist_simple_index[j], item
|
||
)
|
||
|
||
data_beta = [_ for _ in data if _[15] == "beta"]
|
||
self.user_list_beta.setRowCount(len(data_beta))
|
||
for i, row in enumerate(data_beta):
|
||
for j, value in enumerate(row):
|
||
if self.userlist_beta_index[j] == "-":
|
||
continue
|
||
if j in [4, 9, 10]:
|
||
item = QComboBox()
|
||
item.addItems(["启用", "禁用"])
|
||
if value == "y":
|
||
item.setCurrentIndex(0)
|
||
elif value == "n":
|
||
item.setCurrentIndex(1)
|
||
item.currentIndexChanged.connect(
|
||
partial(
|
||
self.change_user_CellWidget,
|
||
data_beta[i][16],
|
||
self.user_column[j],
|
||
)
|
||
)
|
||
elif j == 3 and value == -1:
|
||
item = QTableWidgetItem("无限")
|
||
elif j == 5:
|
||
curdate = server_date()
|
||
if curdate != value:
|
||
item = QTableWidgetItem("今日未代理")
|
||
else:
|
||
item = QTableWidgetItem(
|
||
"今日已代理" + str(data_beta[i][14]) + "次"
|
||
)
|
||
item.setFlags(QtCore.Qt.ItemIsSelectable | QtCore.Qt.ItemIsEnabled)
|
||
elif j == 12:
|
||
if self.PASSWORD == "":
|
||
item = QTableWidgetItem("******")
|
||
item.setFlags(
|
||
QtCore.Qt.ItemIsSelectable | QtCore.Qt.ItemIsEnabled
|
||
)
|
||
else:
|
||
result = self.decryptx(value)
|
||
item = QTableWidgetItem(result)
|
||
if result == "管理密钥错误":
|
||
item.setFlags(
|
||
QtCore.Qt.ItemIsSelectable | QtCore.Qt.ItemIsEnabled
|
||
)
|
||
else:
|
||
item = QTableWidgetItem(str(value))
|
||
if j in [4, 9, 10]:
|
||
if not self.if_user_list_editable:
|
||
item.setEnabled(False)
|
||
self.user_list_beta.setCellWidget(
|
||
data_beta[i][16], self.userlist_beta_index[j], item
|
||
)
|
||
else:
|
||
self.user_list_beta.setItem(
|
||
data_beta[i][16], self.userlist_beta_index[j], item
|
||
)
|
||
if self.if_user_list_editable:
|
||
self.user_list_simple.setEditTriggers(QTableWidget.AllEditTriggers)
|
||
self.user_list_beta.setEditTriggers(QTableWidget.AllEditTriggers)
|
||
else:
|
||
self.user_list_simple.setEditTriggers(QTableWidget.NoEditTriggers)
|
||
self.user_list_beta.setEditTriggers(QTableWidget.NoEditTriggers)
|
||
# 设置QComboBox为可编辑
|
||
self.if_update_database = True
|
||
self.user_list_simple.horizontalHeader().setSectionResizeMode(
|
||
QHeaderView.Stretch
|
||
)
|
||
self.user_list_beta.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch)
|
||
|
||
def update_config(self):
|
||
"""将self.config中的程序配置同步至GUI界面"""
|
||
|
||
self.if_update_config = False
|
||
self.maa_path.setText(self.config["Default"]["MaaSet.path"].replace("\\", "/"))
|
||
self.routine.setValue(self.config["Default"]["TimeLimit.routine"])
|
||
self.annihilation.setValue(self.config["Default"]["TimeLimit.annihilation"])
|
||
self.num.setValue(self.config["Default"]["TimesLimit.run"])
|
||
|
||
self.if_self_start.setChecked(
|
||
bool(self.config["Default"]["SelfSet.IfSelfStart"] == "True")
|
||
)
|
||
|
||
self.if_sleep.setChecked(
|
||
bool(self.config["Default"]["SelfSet.IfSleep"] == "True")
|
||
)
|
||
|
||
self.if_proxy_directly.setChecked(
|
||
bool(self.config["Default"]["SelfSet.IfProxyDirectly"] == "True")
|
||
)
|
||
|
||
for i in range(10):
|
||
self.start_time[i][0].setChecked(
|
||
bool(self.config["Default"]["TimeSet.set" + str(i + 1)] == "True")
|
||
)
|
||
time = QtCore.QTime(
|
||
int(self.config["Default"]["TimeSet.run" + str(i + 1)][:2]),
|
||
int(self.config["Default"]["TimeSet.run" + str(i + 1)][3:]),
|
||
)
|
||
self.start_time[i][1].setTime(time)
|
||
self.if_update_config = True
|
||
|
||
def update_board(self, run_text, wait_text, over_text, error_text, log_text):
|
||
"""写入数据至GUI执行界面的调度台面板"""
|
||
self.run_text.setPlainText(run_text)
|
||
self.wait_text.setPlainText(wait_text)
|
||
self.over_text.setPlainText(over_text)
|
||
self.error_text.setPlainText(error_text)
|
||
self.log_text.setPlainText(log_text)
|
||
self.log_text.verticalScrollBar().setValue(
|
||
self.log_text.verticalScrollBar().maximum()
|
||
)
|
||
|
||
def add_user(self):
|
||
"""添加一位新用户"""
|
||
if not self.check_maa_path():
|
||
QMessageBox.critical(
|
||
self.ui,
|
||
"错误",
|
||
"请先正确配置MAA路径再执行添加用户操作",
|
||
)
|
||
return None
|
||
if self.user_set.currentIndex() == 0:
|
||
self.cur.execute(
|
||
"INSERT INTO adminx VALUES('新用户','手机号码（官服）/B站ID（B服）','Official',0,'y','2000-01-01','1-7','-','-','y','y','n',?,'无',0,'simple',?)",
|
||
(
|
||
self.encryptx("未设置"),
|
||
self.user_list_simple.rowCount(),
|
||
),
|
||
)
|
||
self.get_maa_config(
|
||
[self.user_set.currentIndex(), self.user_list_simple.rowCount(), 0]
|
||
)
|
||
self.get_maa_config(
|
||
[self.user_set.currentIndex(), self.user_list_simple.rowCount(), 1]
|
||
)
|
||
elif self.user_set.currentIndex() == 1:
|
||
self.cur.execute(
|
||
"INSERT INTO adminx VALUES('新用户','手机号码（官服）/B站ID（B服）','Official',0,'y','2000-01-01','1-7','-','-','y','y','-',?,'无',0,'beta',?)",
|
||
(
|
||
self.encryptx("未设置"),
|
||
self.user_list_beta.rowCount(),
|
||
),
|
||
)
|
||
self.get_maa_config(
|
||
[self.user_set.currentIndex(), self.user_list_beta.rowCount(), 0]
|
||
)
|
||
self.get_maa_config(
|
||
[self.user_set.currentIndex(), self.user_list_beta.rowCount(), 1]
|
||
)
|
||
self.db.commit()
|
||
self.update_user_info("normal")
|
||
|
||
def del_user(self):
|
||
"""删除选中的首位用户"""
|
||
if self.user_set.currentIndex() == 0:
|
||
row = self.user_list_simple.currentRow()
|
||
elif self.user_set.currentIndex() == 1:
|
||
row = self.user_list_beta.currentRow()
|
||
if row == -1:
|
||
QMessageBox.critical(self.ui, "错误", "请选中一个用户后再执行删除操作")
|
||
return None
|
||
self.cur.execute(
|
||
"SELECT * FROM adminx WHERE mode = ? AND uid = ?",
|
||
(
|
||
self.user_mode_list[self.user_set.currentIndex()],
|
||
row,
|
||
),
|
||
)
|
||
data = self.cur.fetchall()
|
||
choice = QMessageBox.question(
|
||
self.ui, "确认", "确定要删除用户 " + data[0][0] + " 吗？"
|
||
)
|
||
|
||
if choice == QMessageBox.Yes:
|
||
self.cur.execute(
|
||
"DELETE FROM adminx WHERE mode = ? AND uid = ?",
|
||
(
|
||
self.user_mode_list[self.user_set.currentIndex()],
|
||
row,
|
||
),
|
||
)
|
||
self.db.commit()
|
||
shutil.rmtree(
|
||
self.app_path
|
||
+ "/data/MAAconfig/"
|
||
+ self.user_mode_list[self.user_set.currentIndex()]
|
||
+ "/"
|
||
+ str(row)
|
||
)
|
||
if self.user_set.currentIndex() == 0:
|
||
current_numb = self.user_list_simple.rowCount()
|
||
elif self.user_set.currentIndex() == 1:
|
||
current_numb = self.user_list_beta.rowCount()
|
||
for i in range(row + 1, current_numb):
|
||
self.cur.execute(
|
||
"UPDATE adminx SET uid = ? WHERE mode = ? AND uid = ?",
|
||
(i - 1, self.user_mode_list[self.user_set.currentIndex()], i),
|
||
)
|
||
self.db.commit()
|
||
os.rename(
|
||
self.app_path
|
||
+ "/data/MAAconfig/"
|
||
+ self.user_mode_list[self.user_set.currentIndex()]
|
||
+ "/"
|
||
+ str(i),
|
||
self.app_path
|
||
+ "/data/MAAconfig/"
|
||
+ self.user_mode_list[self.user_set.currentIndex()]
|
||
+ "/"
|
||
+ str(i - 1),
|
||
)
|
||
self.update_user_info("normal")
|
||
|
||
def switch_user(self):
|
||
"""切换用户配置模式"""
|
||
if self.user_set.currentIndex() == 0:
|
||
row = self.user_list_simple.currentRow()
|
||
elif self.user_set.currentIndex() == 1:
|
||
row = self.user_list_beta.currentRow()
|
||
if row == -1:
|
||
QMessageBox.critical(self.ui, "错误", "请选中一个用户后再执行切换操作")
|
||
return None
|
||
self.cur.execute(
|
||
"SELECT * FROM adminx WHERE mode = ? AND uid = ?",
|
||
(
|
||
self.user_mode_list[self.user_set.currentIndex()],
|
||
row,
|
||
),
|
||
)
|
||
data = self.cur.fetchall()
|
||
mode_list = ["简洁", "高级"]
|
||
choice = QMessageBox.question(
|
||
self.ui,
|
||
"确认",
|
||
"确定要将用户 "
|
||
+ data[0][0]
|
||
+ " 转为"
|
||
+ mode_list[1 - self.user_set.currentIndex()]
|
||
+ "配置模式吗？",
|
||
)
|
||
|
||
if choice == QMessageBox.Yes:
|
||
self.cur.execute("SELECT * FROM adminx WHERE True")
|
||
data = self.cur.fetchall()
|
||
if self.user_set.currentIndex() == 0:
|
||
current_numb = self.user_list_simple.rowCount()
|
||
elif self.user_set.currentIndex() == 1:
|
||
current_numb = self.user_list_beta.rowCount()
|
||
other_numb = len(data) - current_numb
|
||
self.cur.execute(
|
||
"UPDATE adminx SET mode = ?, uid = ? WHERE mode = ? AND uid = ?",
|
||
(
|
||
self.user_mode_list[1 - self.user_set.currentIndex()],
|
||
other_numb,
|
||
self.user_mode_list[self.user_set.currentIndex()],
|
||
row,
|
||
),
|
||
)
|
||
self.db.commit()
|
||
shutil.move(
|
||
self.app_path
|
||
+ "/data/MAAconfig/"
|
||
+ self.user_mode_list[self.user_set.currentIndex()]
|
||
+ "/"
|
||
+ str(row),
|
||
self.app_path
|
||
+ "/data/MAAconfig/"
|
||
+ self.user_mode_list[1 - self.user_set.currentIndex()]
|
||
+ "/"
|
||
+ str(other_numb),
|
||
)
|
||
for i in range(row + 1, current_numb):
|
||
self.cur.execute(
|
||
"UPDATE adminx SET uid = ? WHERE mode = ? AND uid = ?",
|
||
(i - 1, self.user_mode_list[self.user_set.currentIndex()], i),
|
||
)
|
||
self.db.commit()
|
||
os.rename(
|
||
self.app_path
|
||
+ "/data/MAAconfig/"
|
||
+ self.user_mode_list[self.user_set.currentIndex()]
|
||
+ "/"
|
||
+ str(i),
|
||
self.app_path
|
||
+ "/data/MAAconfig/"
|
||
+ self.user_mode_list[self.user_set.currentIndex()]
|
||
+ "/"
|
||
+ str(i - 1),
|
||
)
|
||
self.update_user_info("normal")
|
||
|
||
def change_user_set(self):
|
||
"""执行用户配置列表的进一步配置"""
|
||
if self.user_set.currentIndex() == 0:
|
||
if self.user_list_simple.currentColumn() in [10]:
|
||
self.get_maa_config([0, self.user_list_simple.currentRow(), 2])
|
||
else:
|
||
QMessageBox.critical(self.ui, "错误", "该项目无法进一步配置")
|
||
elif self.user_set.currentIndex() == 1:
|
||
if self.user_list_beta.currentColumn() in [4, 5]:
|
||
self.MaaRunner.get_json_path = [
|
||
self.user_set.currentIndex(),
|
||
self.user_list_beta.currentRow(),
|
||
self.user_list_beta.currentColumn() - 4,
|
||
]
|
||
self.maa_set_starter("设置MAA_用户")
|
||
else:
|
||
QMessageBox.critical(self.ui, "错误", "该项目无法进一步配置")
|
||
|
||
def get_maa_config(self, info):
|
||
"""获取MAA配置文件"""
|
||
set_book1 = ["simple/", "beta/"]
|
||
set_book2 = ["/routine", "/annihilation"]
|
||
if info == ["infrastructure"]:
|
||
pass
|
||
elif info == ["Default"]:
|
||
os.makedirs(
|
||
self.app_path + "/data/MAAconfig/Default",
|
||
exist_ok=True,
|
||
)
|
||
shutil.copy(
|
||
self.config["Default"]["MaaSet.path"] + "/config/gui.json",
|
||
self.app_path + "/data/MAAconfig/Default",
|
||
)
|
||
elif info[2] == 2:
|
||
infrastructure_path = self.read("file_path_infrastructure")
|
||
if infrastructure_path:
|
||
os.makedirs(
|
||
self.app_path
|
||
+ "/data/MAAconfig/"
|
||
+ set_book1[info[0]]
|
||
+ str(info[1])
|
||
+ "/infrastructure",
|
||
exist_ok=True,
|
||
)
|
||
shutil.copy(
|
||
infrastructure_path,
|
||
self.app_path
|
||
+ "/data/MAAconfig/"
|
||
+ set_book1[info[0]]
|
||
+ str(info[1])
|
||
+ "/infrastructure/infrastructure.json",
|
||
)
|
||
return True
|
||
else:
|
||
QMessageBox.critical(
|
||
self.ui,
|
||
"错误",
|
||
"未选择自定义基建文件",
|
||
)
|
||
return False
|
||
else:
|
||
os.makedirs(
|
||
self.app_path
|
||
+ "/data/MAAconfig/"
|
||
+ set_book1[info[0]]
|
||
+ str(info[1])
|
||
+ set_book2[info[2]],
|
||
exist_ok=True,
|
||
)
|
||
shutil.copy(
|
||
self.config["Default"]["MaaSet.path"] + "/config/gui.json",
|
||
self.app_path
|
||
+ "/data/MAAconfig/"
|
||
+ set_book1[info[0]]
|
||
+ str(info[1])
|
||
+ set_book2[info[2]],
|
||
)
|
||
|
||
def change_user_Item(self, item, mode):
|
||
"""将GUI中发生修改的用户配置表中的一般信息同步至本地数据库"""
|
||
if not self.if_update_database:
|
||
return None
|
||
text = item.text()
|
||
if mode == "simple":
|
||
if item.column() == 3:
|
||
try:
|
||
text = max(int(text), -1)
|
||
except ValueError:
|
||
self.update_user_info("normal")
|
||
return None
|
||
if item.column() in [6, 7, 8]:
|
||
# 导入与应用特殊关卡规则
|
||
games = {}
|
||
with open(self.gameid_path, encoding="utf-8") as f:
|
||
gameids = f.readlines()
|
||
for line in gameids:
|
||
if "：" in line:
|
||
game_in, game_out = line.split("：", 1)
|
||
games[game_in.strip()] = game_out.strip()
|
||
text = games.get(text, text)
|
||
if item.column() == 11:
|
||
text = self.encryptx(text)
|
||
if text != "":
|
||
self.cur.execute(
|
||
f"UPDATE adminx SET {self.user_column[self.userlist_simple_index.index(item.column())]} = ? WHERE mode = 'simple' AND uid = ?",
|
||
(text, item.row()),
|
||
)
|
||
elif mode == "beta":
|
||
if item.column() == 1:
|
||
try:
|
||
text = max(int(text), -1)
|
||
except ValueError:
|
||
self.update_user_info("normal")
|
||
return None
|
||
if item.column() == 6:
|
||
text = self.encryptx(text)
|
||
if text != "":
|
||
self.cur.execute(
|
||
f"UPDATE adminx SET {self.user_column[self.userlist_beta_index.index(item.column())]} = ? WHERE mode = 'beta' AND uid = ?",
|
||
(text, item.row()),
|
||
)
|
||
self.db.commit()
|
||
self.update_user_info("normal")
|
||
|
||
def change_user_CellWidget(self, row, column, index):
|
||
"""将GUI中发生修改的用户配置表中的CellWidget类信息同步至本地数据库"""
|
||
if not self.if_update_database:
|
||
return None
|
||
if (
|
||
self.user_set.currentIndex() == 0
|
||
and column == "infrastructure"
|
||
and index == 0
|
||
):
|
||
if not os.path.exists(
|
||
self.app_path
|
||
+ "/data/MAAconfig/"
|
||
+ self.user_mode_list[self.user_set.currentIndex()]
|
||
+ "/"
|
||
+ str(row)
|
||
+ "/infrastructure/infrastructure.json",
|
||
):
|
||
result = self.get_maa_config([0, row, 2])
|
||
if not result:
|
||
index = 1
|
||
if self.user_set.currentIndex() == 0 and column == "server":
|
||
server_list = ["Official", "Bilibili"]
|
||
self.cur.execute(
|
||
f"UPDATE adminx SET server = ? WHERE mode = 'simple' AND uid = ?",
|
||
(server_list[index], row),
|
||
)
|
||
else:
|
||
index_list = ["y", "n"]
|
||
self.cur.execute(
|
||
f"UPDATE adminx SET {column} = ? WHERE mode = ? AND uid = ?",
|
||
(
|
||
index_list[index],
|
||
self.user_mode_list[self.user_set.currentIndex()],
|
||
row,
|
||
),
|
||
)
|
||
self.db.commit()
|
||
self.update_user_info("normal")
|
||
|
||
def change_user_info(self, modes, uids, days, lasts, notes, numbs):
|
||
"""将代理完成后发生改动的用户信息同步至本地数据库"""
|
||
for index in range(len(uids)):
|
||
self.cur.execute(
|
||
"UPDATE adminx SET day = ? WHERE mode = ? AND uid = ?",
|
||
(days[index], modes[index], uids[index]),
|
||
)
|
||
self.cur.execute(
|
||
"UPDATE adminx SET last = ? WHERE mode = ? AND uid = ?",
|
||
(lasts[index], modes[index], uids[index]),
|
||
)
|
||
self.cur.execute(
|
||
"UPDATE adminx SET notes = ? WHERE mode = ? AND uid = ?",
|
||
(notes[index], modes[index], uids[index]),
|
||
)
|
||
self.cur.execute(
|
||
"UPDATE adminx SET numb = ? WHERE mode = ? AND uid = ?",
|
||
(numbs[index], modes[index], uids[index]),
|
||
)
|
||
self.db.commit()
|
||
self.update_user_info("normal")
|
||
|
||
def change_config(self):
|
||
"""将GUI中发生修改的程序配置同步至self.config变量"""
|
||
if not self.if_update_config:
|
||
return None
|
||
|
||
self.config["Default"]["MaaSet.path"] = self.maa_path.text().replace("\\", "/")
|
||
if not self.check_maa_path():
|
||
self.config["Default"]["MaaSet.path"] = ""
|
||
with open(self.config_path, "w", encoding="utf-8") as f:
|
||
json.dump(self.config, f, indent=4)
|
||
self.update_config()
|
||
QMessageBox.critical(
|
||
self.ui, "错误", "未找到MAA.exe或MAA配置文件，请重新设置MAA路径！"
|
||
)
|
||
return None
|
||
|
||
self.config["Default"]["TimeLimit.routine"] = self.routine.value()
|
||
self.config["Default"]["TimeLimit.annihilation"] = self.annihilation.value()
|
||
self.config["Default"]["TimesLimit.run"] = self.num.value()
|
||
|
||
if self.if_sleep.isChecked():
|
||
self.config["Default"]["SelfSet.IfSleep"] = "True"
|
||
else:
|
||
self.config["Default"]["SelfSet.IfSleep"] = "False"
|
||
|
||
if self.if_self_start.isChecked():
|
||
self.config["Default"]["SelfSet.IfSelfStart"] = "True"
|
||
else:
|
||
self.config["Default"]["SelfSet.IfSelfStart"] = "False"
|
||
|
||
if self.if_proxy_directly.isChecked():
|
||
self.config["Default"]["SelfSet.IfProxyDirectly"] = "True"
|
||
else:
|
||
self.config["Default"]["SelfSet.IfProxyDirectly"] = "False"
|
||
|
||
for i in range(10):
|
||
if self.start_time[i][0].isChecked():
|
||
self.config["Default"]["TimeSet.set" + str(i + 1)] = "True"
|
||
else:
|
||
self.config["Default"]["TimeSet.set" + str(i + 1)] = "False"
|
||
time = self.start_time[i][1].time().toString("HH:mm")
|
||
self.config["Default"]["TimeSet.run" + str(i + 1)] = time
|
||
|
||
with open(self.config_path, "w", encoding="utf-8") as f:
|
||
json.dump(self.config, f, indent=4)
|
||
|
||
self.update_config()
|
||
|
||
def change_userlist_method(self):
|
||
"""更新GUI界面使之适配用户配置模式"""
|
||
user_switch_list = ["转为高级", "转为简洁"]
|
||
self.user_switch.setText(user_switch_list[self.user_set.currentIndex()])
|
||
self.update_user_info("normal")
|
||
|
||
def read(self, operation):
|
||
"""弹出对话框组件进行读入"""
|
||
# 读入PASSWORD
|
||
if operation == "key":
|
||
self.PASSWORD, ok_pressed = QInputDialog.getText(
|
||
self.ui, "请输入管理密钥", "管理密钥：", QLineEdit.Password, ""
|
||
)
|
||
if ok_pressed and self.PASSWORD != "":
|
||
self.update_user_info("normal")
|
||
elif operation == "oldkey":
|
||
self.PASSWORD, ok_pressed = QInputDialog.getText(
|
||
self.ui, "请输入旧的管理密钥", "旧管理密钥：", QLineEdit.Password, ""
|
||
)
|
||
if ok_pressed and self.PASSWORD != "":
|
||
return True
|
||
else:
|
||
return False
|
||
elif operation == "newkey":
|
||
new_PASSWORD, ok_pressed = QInputDialog.getText(
|
||
self.ui, "请输入新的管理密钥", "新管理密钥：", QLineEdit.Password, ""
|
||
)
|
||
if ok_pressed and new_PASSWORD != "":
|
||
return new_PASSWORD
|
||
else:
|
||
return None
|
||
# 读入选择
|
||
elif operation == "question_runner":
|
||
choice = QMessageBox.question(
|
||
self.ui,
|
||
self.MaaRunner.question_title,
|
||
self.MaaRunner.question_info,
|
||
)
|
||
if choice == QMessageBox.Yes:
|
||
self.MaaRunner.question_choice = "Yes"
|
||
elif choice == QMessageBox.No:
|
||
self.MaaRunner.question_choice = "No"
|
||
# 读入MAA文件目录
|
||
elif operation == "file_path_maa":
|
||
file_path = QFileDialog.getExistingDirectory(self.ui, "选择MAA文件夹")
|
||
if file_path:
|
||
self.maa_path.setText(file_path)
|
||
# 读入自定义基建文件目录
|
||
elif operation == "file_path_infrastructure":
|
||
file_path, _ = QFileDialog.getOpenFileName(
|
||
self.ui, "选择自定义基建文件", "", "JSON 文件 (*.json)"
|
||
)
|
||
return file_path
|
||
|
||
def check_maa_path(self):
|
||
if os.path.exists(
|
||
self.config["Default"]["MaaSet.path"] + "/MAA.exe"
|
||
) and os.path.exists(
|
||
self.config["Default"]["MaaSet.path"] + "/config/gui.json"
|
||
):
|
||
self.get_maa_config(["Default"])
|
||
return True
|
||
else:
|
||
return False
|
||
|
||
def routine_starter(self):
|
||
"""启动MaaRunner线程运行日常代理任务"""
|
||
if not self.check_maa_path():
|
||
QMessageBox.critical(self.ui, "错误", "您还未正确配置MAA路径！")
|
||
return None
|
||
self.maa_running_set("日常代理_开始")
|
||
# 配置参数
|
||
self.MaaRunner.set_path = (
|
||
self.config["Default"]["MaaSet.path"] + "/config/gui.json"
|
||
)
|
||
self.MaaRunner.log_path = (
|
||
self.config["Default"]["MaaSet.path"] + "/debug/gui.log"
|
||
)
|
||
self.MaaRunner.maa_path = self.config["Default"]["MaaSet.path"] + "/MAA.exe"
|
||
self.MaaRunner.routine = self.config["Default"]["TimeLimit.routine"]
|
||
self.MaaRunner.annihilation = self.config["Default"]["TimeLimit.annihilation"]
|
||
self.MaaRunner.num = self.config["Default"]["TimesLimit.run"]
|
||
self.cur.execute("SELECT * FROM adminx WHERE True")
|
||
self.data_ = self.cur.fetchall()
|
||
self.MaaRunner.data = [list(row) for row in self.data_]
|
||
self.MaaRunner.mode = "日常代理"
|
||
# 启动执行线程
|
||
self.MainTimer.is_maa_run = True
|
||
self.MaaRunner.start()
|
||
|
||
def check_starter(self):
|
||
"""启动MaaRunner线程运行人工排查任务"""
|
||
if not self.check_maa_path():
|
||
QMessageBox.critical(self.ui, "错误", "您还未正确配置MAA路径！")
|
||
return None
|
||
self.maa_running_set("人工排查_开始")
|
||
# 配置参数
|
||
self.MaaRunner.set_path = (
|
||
self.config["Default"]["MaaSet.path"] + "/config/gui.json"
|
||
)
|
||
self.MaaRunner.log_path = (
|
||
self.config["Default"]["MaaSet.path"] + "/debug/gui.log"
|
||
)
|
||
self.MaaRunner.maa_path = self.config["Default"]["MaaSet.path"] + "/MAA.exe"
|
||
self.cur.execute("SELECT * FROM adminx WHERE True")
|
||
self.data_ = self.cur.fetchall()
|
||
self.MaaRunner.data = [list(row) for row in self.data_]
|
||
self.MaaRunner.mode = "人工排查"
|
||
# 启动执行线程
|
||
self.MainTimer.is_maa_run = True
|
||
self.MaaRunner.start()
|
||
|
||
def maa_set_starter(self, mode):
|
||
"""启动MaaRunner线程进行MAA设置"""
|
||
if not self.check_maa_path():
|
||
QMessageBox.critical(self.ui, "错误", "您还未正确配置MAA路径！")
|
||
return None
|
||
self.maa_running_set("设置MAA_开始")
|
||
# 配置参数
|
||
self.MaaRunner.set_path = (
|
||
self.config["Default"]["MaaSet.path"] + "/config/gui.json"
|
||
)
|
||
self.MaaRunner.log_path = (
|
||
self.config["Default"]["MaaSet.path"] + "/debug/gui.log"
|
||
)
|
||
self.MaaRunner.maa_path = self.config["Default"]["MaaSet.path"] + "/MAA.exe"
|
||
self.MaaRunner.mode = mode
|
||
# 启动执行线程
|
||
self.MainTimer.is_maa_run = True
|
||
self.MaaRunner.start()
|
||
|
||
def maa_ender(self, mode):
|
||
"""中止MAA线程"""
|
||
self.MaaRunner.if_run = False
|
||
self.MaaRunner.wait()
|
||
self.MainTimer.is_maa_run = False
|
||
self.maa_running_set(mode)
|
||
|
||
def maa_running_set(self, mode):
|
||
"""处理MAA运行过程中的GUI组件变化"""
|
||
if "开始" in mode:
|
||
|
||
self.MaaRunner.accomplish.disconnect()
|
||
self.user_add.setEnabled(False)
|
||
self.user_del.setEnabled(False)
|
||
self.user_switch.setEnabled(False)
|
||
self.user_changeset.setEnabled(False)
|
||
self.set_maa.setEnabled(False)
|
||
# self.update_user_info("read_only")
|
||
|
||
if mode == "日常代理_开始":
|
||
self.MaaRunner.accomplish.connect(
|
||
lambda: self.maa_ender("日常代理_结束")
|
||
)
|
||
self.check_start.setEnabled(False)
|
||
self.run_now.clicked.disconnect()
|
||
self.run_now.setText("结束运行")
|
||
self.run_now.clicked.connect(lambda: self.maa_ender("日常代理_结束"))
|
||
|
||
elif mode == "人工排查_开始":
|
||
self.MaaRunner.accomplish.connect(
|
||
lambda: self.maa_ender("人工排查_结束")
|
||
)
|
||
self.run_now.setEnabled(False)
|
||
self.check_start.clicked.disconnect()
|
||
self.check_start.setText("中止排查")
|
||
self.check_start.clicked.connect(
|
||
lambda: self.maa_ender("人工排查_结束")
|
||
)
|
||
|
||
elif mode == "设置MAA_开始":
|
||
self.MaaRunner.accomplish.connect(
|
||
lambda: self.maa_ender("设置MAA_结束")
|
||
)
|
||
self.run_now.setEnabled(False)
|
||
self.check_start.setEnabled(False)
|
||
|
||
elif "结束" in mode:
|
||
|
||
shutil.copy(
|
||
self.app_path + "/data/MAAconfig/Default/gui.json",
|
||
self.config["Default"]["MaaSet.path"] + "/config",
|
||
)
|
||
self.user_add.setEnabled(True)
|
||
self.user_del.setEnabled(True)
|
||
self.user_switch.setEnabled(True)
|
||
self.user_changeset.setEnabled(True)
|
||
self.set_maa.setEnabled(True)
|
||
# self.update_user_info("editable")
|
||
|
||
if mode == "设置MAA_结束":
|
||
self.run_now.setEnabled(True)
|
||
self.check_start.setEnabled(True)
|
||
|
||
elif mode == "人工排查_结束":
|
||
|
||
self.run_now.setEnabled(True)
|
||
self.check_start.clicked.disconnect()
|
||
self.check_start.setText("开始排查")
|
||
self.check_start.clicked.connect(self.check_starter)
|
||
|
||
elif mode == "日常代理_结束":
|
||
self.check_start.setEnabled(True)
|
||
self.run_now.clicked.disconnect()
|
||
self.run_now.setText("立即执行")
|
||
self.run_now.clicked.connect(self.routine_starter)
|
||
|
||
def check_version(self):
|
||
"""检查版本更新，调起文件下载进程"""
|
||
# 从本地版本信息文件获取当前版本信息
|
||
with open(self.version_path, "r", encoding="utf-8") as f:
|
||
version_current = json.load(f)
|
||
main_version_current = list(
|
||
map(int, version_current["main_version"].split("."))
|
||
)
|
||
updater_version_current = list(
|
||
map(int, version_current["updater_version"].split("."))
|
||
)
|
||
if not os.path.exists(self.app_path + "/Updater.exe"):
|
||
updater_version_current = [0, 0, 0, 0]
|
||
# 从远程服务器获取最新版本信息
|
||
try:
|
||
response = requests.get(
|
||
"https://ghp.ci/https://github.com/DLmaster361/AUTO_MAA/blob/main/res/version.json"
|
||
)
|
||
except Exception as e:
|
||
QMessageBox.critical(
|
||
self.ui,
|
||
"错误",
|
||
f"获取版本信息时出错：\n{e}",
|
||
)
|
||
return None
|
||
version_remote = response.json()
|
||
main_version_remote = list(map(int, version_remote["main_version"].split(".")))
|
||
updater_version_remote = list(
|
||
map(int, version_remote["updater_version"].split("."))
|
||
)
|
||
|
||
if (main_version_remote > main_version_current) or (
|
||
updater_version_remote > updater_version_current
|
||
):
|
||
choice = QMessageBox.question(
|
||
self.ui,
|
||
"版本更新",
|
||
f"发现新版本：\n 主程序：{self.version_text(main_version_current)} --> {self.version_text(main_version_remote)}\n 更新器：{self.version_text(updater_version_current)} --> {self.version_text(updater_version_remote)}\n 更新说明：\n{version_remote['announcement'].replace("\n","\n ")}\n\n是否开始更新？\n\n 注意：主程序更新时AUTO_MAA将自动关闭",
|
||
)
|
||
if choice == QMessageBox.No:
|
||
return None
|
||
|
||
if updater_version_remote > updater_version_current:
|
||
self.updater = Updater.Updater(
|
||
self.app_path,
|
||
"AUTO_MAA更新器",
|
||
version_remote["updater_download_url"],
|
||
version_remote["updater_version"],
|
||
)
|
||
if main_version_remote > main_version_current:
|
||
self.updater.update_process.accomplish.connect(self.update_main)
|
||
self.updater.ui.show()
|
||
elif main_version_remote > main_version_current:
|
||
self.update_main()
|
||
else:
|
||
self.push_notification("已是最新版本~", " ", " ", 10)
|
||
|
||
def update_main(self):
|
||
subprocess.Popen(
|
||
self.app_path + "/Updater.exe",
|
||
shell=True,
|
||
creationflags=subprocess.CREATE_NO_WINDOW,
|
||
)
|
||
sys.exit()
|
||
|
||
def version_text(self, version_numb):
|
||
"""将版本号列表转为可读的文本信息"""
|
||
if version_numb[3] == 0:
|
||
version = f"v{'.'.join(str(_) for _ in version_numb[0:3])}"
|
||
else:
|
||
version = f"v{'.'.join(str(_) for _ in version_numb[0:3])}_beta"
|
||
return version
|
||
|
||
def push_notification(self, title, message, ticker, t):
|
||
"""推送系统通知"""
|
||
notification.notify(
|
||
title=title,
|
||
message=message,
|
||
app_name="AUTO_MAA",
|
||
app_icon=self.app_path + "/gui/ico/AUTO_MAA.ico",
|
||
timeout=t,
|
||
ticker=ticker,
|
||
toast=True,
|
||
)
|
||
|
||
def give_config(self):
|
||
"""同步配置文件到子线程"""
|
||
self.MainTimer.config = self.config
|
||
|
||
def closeEvent(self, event):
|
||
"""清理残余进程"""
|
||
self.MainTimer.quit()
|
||
self.MaaRunner.if_run = False
|
||
self.MaaRunner.wait()
|
||
self.cur.close()
|
||
self.db.close()
|
||
super().closeEvent(event)
|
||
|
||
|
||
class AUTO_MAA(QApplication):
|
||
def __init__(self):
|
||
super().__init__()
|
||
|
||
self.main = Main()
|
||
self.main.ui.show()
|
||
|
||
|
||
def server_date():
|
||
"""获取当前的服务器日期"""
|
||
dt = datetime.datetime.now()
|
||
if dt.time() < datetime.datetime.min.time().replace(hour=4):
|
||
dt = dt - datetime.timedelta(days=1)
|
||
return dt.strftime("%Y-%m-%d")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
app = AUTO_MAA()
|
||
app.exec()
|