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

UI优化，判定优化，修复gameid错误，代理策略优化，组件独立性优化

Browse Source
This commit is contained in:
DLmaster
2024-03-09 12:00:14 +08:00
parent eac2d13fee
commit 1a9af53e4d
11 changed files with 184 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

122
run.py
View File

@@ -20,49 +20,81 @@
import os
import subprocess
import atexit
import sqlite3
import datetime
import time
import json
from termcolor import colored
#判断MAA程序运行状态
def ifoff():
while True:
time.sleep(10)
with open(logpath,'r',encoding='utf-8') as f:
logs=f.readlines()[-1:-10:-1]
log=''.join(logs)
print(colored('\n'.join(logs[::-1]),"green"))
if "任务已全部完成！" in log:
return True
elif ("请检查连接设置或尝试重启模拟器与 ADB 或重启电脑" in log) or ("已停止" in log):
return False
#执行MAA任务
def runmaa(tel,game,num=2):
def runmaa(id,tel,game,num=3):
#配置MAA运行参数
with open(setpath,"r",encoding="utf-8") as f:
data = json.load(f)
data=json.load(f)
data["Configurations"]["Default"]["Start.AccountName"]=tel[:3]+"****"+tel[7:]
week=str(datetime.datetime.now().strftime('%A'))
if week=="Monday":
data["Configurations"]["Default"]["MainFunction.Stage1"]="Annihilation"
else:
data["Configurations"]["Default"]["MainFunction.Stage1"]=game
data["Configurations"]["Default"]["MainFunction.Stage1"]="Annihilation"
data["Configurations"]["Default"]["Fight.RemainingSanityStage"]=game
data["Configurations"]["Default"]["Fight.UseRemainingSanityStage"]="True"
data["Configurations"]["Default"]["GUI.CustomStageCode"]="True"
with open(setpath,"w",encoding="utf-8") as f:
json.dump(data,f)
#开始运行
for i in range(num):
global idnew,idold,idfail,idall,logx,logi
#创建MAA任务
maa=subprocess.Popen([maapath])
maapid=maa.pid
time.sleep(60)
if ifoff():
return True
#等待MAA启动
idsuccess=idnew+idold
idwait=[idx for idx in idall if not idx in idsuccess+idfail+[id]]
os.system('cls')
if i==0:
print(colored("正在代理：",'white')+colored(id,'blue'))
else:
os.system('taskkill /F /T /PID '+str(maapid))
print(colored("正在代理：",'white')+colored(id,'light_blue'))
print(colored("等待代理：",'white')+colored('，'.join(idwait),'yellow'))
print(colored("代理成功：",'white')+colored('，'.join(idsuccess),'green'))
print(colored("代理失败：",'white')+colored('，'.join(idfail),'red'))
print(colored("运行日志：",'white')+colored("等待MAA初始化",'light_green'))
time.sleep(60)
#监测MAA运行状态
while True:
#打印基本信息
os.system('cls')
if i==0:
print(colored("正在代理：",'white')+colored(id,'blue'))
else:
print(colored("正在代理：",'white')+colored(id,'light_blue'))
print(colored("等待代理：",'white')+colored('，'.join(idwait),'yellow'))
print(colored("代理成功：",'white')+colored('，'.join(idsuccess),'green'))
print(colored("代理失败：",'white')+colored('，'.join(idfail),'red'))
print(colored("运行日志：",'white'))
#读取并保存MAA日志
with open(logpath,'r',encoding='utf-8') as f:
logs=f.readlines()[-1:-10:-1]
print(colored(''.join(logs[::-1]),'light_green'))
log=''.join(logs)
logx[logi]=log
logi=(logi+1) % len(logx)
#判断MAA程序运行状态
if ("任务已全部完成！" in log):
return True
elif ("请检查连接设置或尝试重启模拟器与 ADB 或重启电脑" in log) or ("已停止" in log) or ("MaaAssistantArknights GUI exited" in log) or timeout():
os.system('taskkill /F /T /PID '+str(maapid))
break
time.sleep(10)
return False
#检查是否超时
def timeout():
global logx
log0=logx[0]
for i in range(len(logx)):
if logx[i]!=log0:
return False
return True
#更新已完成用户的数据
def updata(id):
db=sqlite3.connect(DATABASE)
@@ -77,6 +109,19 @@ def updata(id):
db.close()
return 0
#资源回收
def cleanup():
if os.path.exists("state/RUNNING"):
os.remove("state/RUNNING")
#读取运行情况
if os.path.exists("state/RUNNING"):
exit()
#标记当前正在运行
with open("state/RUNNING","w",encoding="utf-8") as f:
print("RUNNING",file=f)
#设置回调函数
atexit.register(cleanup)
#获取PATH与用户数据
DATABASE="data/data.db"
db=sqlite3.connect(DATABASE)
@@ -93,32 +138,39 @@ data=[list(row) for row in data]
cur.close()
db.close()
#开始执行
curdate=datetime.date.today()
curdate=curdate.strftime('%Y-%m-%d')
curdate=datetime.date.today().strftime('%Y-%m-%d')
begintime=datetime.datetime.now().strftime("%H:%M")
idnew=[]
idold=[]
idfail=[]
idall=[data[i][0] for i in range(len(data))]
LOGXLEN=60
logx=['' for i in range(LOGXLEN)]
logi=0
for i in range(len(data)):
if data[i][3]=='y' and data[i][4]!=curdate and data[i][2]>0:
book=runmaa(data[i][1],data[i][5])
book=runmaa(data[i][0],data[i][1],data[i][5])
if book:
updata(data[i][0])
idnew.append(data[i][0])
print(colored("已完成"+data[i][0]+"今日的代理","yellow"))
else:
idfail.append(data[i][0])
print(colored("异常中止"+data[i][0]+"的代理","red"))
for i in range(len(data)):
if data[i][3]=='y' and data[i][4]==curdate and data[i][2]>0:
book=runmaa(data[i][1],data[i][5])
book=runmaa(data[i][0],data[i][1],data[i][5])
if book:
idold.append(data[i][0])
print(colored("已重复完成"+data[i][0]+"今日的代理","yellow"))
with open("log.txt","w", encoding="utf-8") as f:
print("任务结束，已完成数："+str(len(idnew))+"，未完成数："+str(len(idfail))+"，重复执行数："+str(len(idold)),file=f)
else:
idall.remove(data[i][0])
endtime=datetime.datetime.now().strftime("%H:%M")
with open("log.txt","w",encoding="utf-8") as f:
print("任务开始时间："+begintime+"，结束时间："+endtime,file=f)
print("已完成数："+str(len(idnew))+"，未完成数："+str(len(idfail))+"，重复执行数："+str(len(idold)),file=f)
if len(idfail)!=0:
print("代理未完成的用户：",file=f)
for i in range(len(idfail)):
print(idfail[i],file=f)
with open("OVER","w", encoding="utf-8") as f:
print("OVER",file=f)
if os.path.exists("state/BEGIN"):
with open("state/END","w",encoding="utf-8") as f:
print("END",file=f)
exit()