MAAsConfig/ArknightsMaaConfig/maa.py

#!/usr/bin/env python3
import os
import sys
import asyncio
import re
import time
from datetime import datetime
# time record
def fprint(s:str)->None:
    print(s)
    sys.stdout.flush()

fprint("-------------------------------------------------------")
fprint(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
fprint("-------------------------------------------------------")
time.sleep(1)

class CommandProcessor:
    def Command(self,s:str) -> int:
        self._command = s
        fprint("_> " + s)
        if (t:=os.system(s))!=0:
            fprint("Fatal Error: Command Failed:" + s)
            fprint("Error Number: " + str(t))
            return t
        return 0
    def RetryCommand(self,s:str,retry:int=3) -> int:
        result = 0
        while retry > 0:
            result = cp.Command(s)
            if result == 0:
                break
            retry-=1
            time.sleep(1)
        return result


    async def AsyncCommand(self,command:str,regex:str):
        self._asyncCommand = command
        fprint("_> await! " + command)
        process = await asyncio.create_subprocess_shell(command,stderr=asyncio.subprocess.PIPE)
        while True:
            line = (await process.stderr.readline()).decode()
            fprint(line)
            if re.search(regex, line):
                fprint("done")
                break


    
        
# cp.Command("command")
# cp.RetryCommand("command",retry_time:int)
# asyncio.run(cp.AsyncCommand("command","stop track regex"))        
# script_dir = os.path.dirname((os.path.abspath(__file__)))
cp = CommandProcessor()
haveError = False
def TaskFailRetry(command:str)->int:
    retry = 3
    result = 0
    while retry > 0:
        retry -= 1
        result = cp.RetryCommand(command)
        if result == 0:
            return 0
        fprint(f"Task Command is {command}, ret code is {result}")
        fprint("Task Failed, retrying...")
        cp.RetryCommand("maa closedown")
        cp.RetryCommand("maa run startup")
    return result
# update
cp.Command("maa update") 
cp.Command("maa self update") 
# fire it up
cp.Command("docker stop redroid")
redroid_path = os.getenv('redroid_path', "/home/lichx/.config/redroid-rk3588/")
adb_address = os.getenv('adb_address', "localhost:5555")
log_dir = os.getenv('MAA_LOG', "/home/lichx/.config/redroid-rk3588/log/")
os.chdir(redroid_path)
cp.Command("docker compose up -d")
time.sleep(15)
cp.RetryCommand(f"adb connect {adb_address}")
cp.RetryCommand(f"adb -s {adb_address} shell am start --windowingMode 4 com.hypergryph.arknights/com.u8.sdk.U8UnityContext")
# run maa
haveError |= TaskFailRetry("maa run startup") != 0
haveError |= TaskFailRetry("maa run recruit") != 0
haveError |= TaskFailRetry("maa run infrast") != 0
haveError |= TaskFailRetry("maa run fight") != 0
haveError |= TaskFailRetry("maa run mall") != 0
haveError |= TaskFailRetry("maa run award") != 0
# storage error log
if haveError:
    cp.Command(f"cp {log_dir}/pyshell.log {log_dir}/pyshell.{datetime.now().strftime('%Y-%m-%d-%H-%M-%S')}.log")
# shutdown
cp.RetryCommand("maa closedown")
cp.Command("docker stop redroid")