import os import re import threading import time from datetime import datetime, timedelta import inflection from cached_property import cached_property from module.base.decorator import del_cached_property from module.config.config import AzurLaneConfig, TaskEnd from module.config.deep import deep_get, deep_set from module.exception import * from module.logger import logger from module.notify import handle_notify class AzurLaneAutoScript: stop_event: threading.Event = None def __init__(self, config_name='alas'): logger.hr('Start', level=0) self.config_name = config_name # Skip first restart self.is_first_task = True # Failure count of tasks # Key: str, task name, value: int, failure count self.failure_record = {} @cached_property def config(self): try: config = AzurLaneConfig(config_name=self.config_name) return config except RequestHumanTakeover: logger.critical('Request human takeover') exit(1) except Exception as e: logger.exception(e) exit(1) @cached_property def device(self): try: from module.device.device import Device device = Device(config=self.config) return device except RequestHumanTakeover: logger.critical('Request human takeover') exit(1) except Exception as e: logger.exception(e) exit(1) @cached_property def checker(self): try: from module.server_checker import ServerChecker checker = ServerChecker(server=self.config.Emulator_ServerName) return checker except Exception as e: logger.exception(e) exit(1) def run(self, command, skip_first_screenshot=False): try: if not skip_first_screenshot: self.device.screenshot() self.__getattribute__(command)() return True except TaskEnd: return True except GameNotRunningError as e: logger.warning(e) self.config.task_call('Restart') return False except (GameStuckError, GameTooManyClickError) as e: logger.error(e) self.save_error_log() logger.warning(f'Game stuck, {self.device.package} will be restarted in 10 seconds') logger.warning('If you are playing by hand, please stop Alas') self.config.task_call('Restart') self.device.sleep(10) return False except GameBugError as e: logger.warning(e) self.save_error_log() logger.warning('An error has occurred in Azur Lane game client, Alas is unable to handle') logger.warning(f'Restarting {self.device.package} to fix it') self.config.task_call('Restart') self.device.sleep(10) return False except GamePageUnknownError: logger.info('Game server may be under maintenance or network may be broken, check server status now') self.checker.check_now() if self.checker.is_available(): logger.critical('Game page unknown') self.save_error_log() handle_notify( self.config.Error_OnePushConfig, title=f"Alas <{self.config_name}> crashed", content=f"<{self.config_name}> GamePageUnknownError", ) exit(1) else: self.checker.wait_until_available() return False except ScriptError as e: logger.exception(e) logger.critical('This is likely to be a mistake of developers, but sometimes just random issues') handle_notify( self.config.Error_OnePushConfig, title=f"Alas <{self.config_name}> crashed", content=f"<{self.config_name}> ScriptError", ) exit(1) except RequestHumanTakeover: logger.critical('Request human takeover') handle_notify( self.config.Error_OnePushConfig, title=f"Alas <{self.config_name}> crashed", content=f"<{self.config_name}> RequestHumanTakeover", ) exit(1) except Exception as e: logger.exception(e) self.save_error_log() handle_notify( self.config.Error_OnePushConfig, title=f"Alas <{self.config_name}> crashed", content=f"<{self.config_name}> Exception occured", ) exit(1) def save_error_log(self): """ Save last 60 screenshots in ./log/error/ Save logs to ./log/error//log.txt """ from module.base.utils import save_image from module.handler.sensitive_info import (handle_sensitive_image, handle_sensitive_logs) if self.config.Error_SaveError: if not os.path.exists('./log/error'): os.mkdir('./log/error') folder = f'./log/error/{int(time.time() * 1000)}' logger.warning(f'Saving error: {folder}') os.mkdir(folder) for data in self.device.screenshot_deque: image_time = datetime.strftime(data['time'], '%Y-%m-%d_%H-%M-%S-%f') image = handle_sensitive_image(data['image']) save_image(image, f'{folder}/{image_time}.png') with open(logger.log_file, 'r', encoding='utf-8') as f: lines = f.readlines() start = 0 for index, line in enumerate(lines): line = line.strip(' \r\t\n') if re.match('^═{15,}$', line): start = index lines = lines[start - 2:] lines = handle_sensitive_logs(lines) with open(f'{folder}/log.txt', 'w', encoding='utf-8') as f: f.writelines(lines) def restart(self): from module.handler.login import LoginHandler LoginHandler(self.config, device=self.device).app_restart() def start(self): from module.handler.login import LoginHandler LoginHandler(self.config, device=self.device).app_start() def goto_main(self): from module.handler.login import LoginHandler from module.ui.ui import UI if self.device.app_is_running(): logger.info('App is already running, goto main page') UI(self.config, device=self.device).ui_goto_main() else: logger.info('App is not running, start app and goto main page') LoginHandler(self.config, device=self.device).app_start() UI(self.config, device=self.device).ui_goto_main() def research(self): from module.research.research import RewardResearch RewardResearch(config=self.config, device=self.device).run() def commission(self): from module.commission.commission import RewardCommission RewardCommission(config=self.config, device=self.device).run() def tactical(self): from module.tactical.tactical_class import RewardTacticalClass RewardTacticalClass(config=self.config, device=self.device).run() def dorm(self): from module.dorm.dorm import RewardDorm RewardDorm(config=self.config, device=self.device).run() def meowfficer(self): from module.meowfficer.meowfficer import RewardMeowfficer RewardMeowfficer(config=self.config, device=self.device).run() def guild(self): from module.guild.guild_reward import RewardGuild RewardGuild(config=self.config, device=self.device).run() def reward(self): from module.reward.reward import Reward Reward(config=self.config, device=self.device).run() def awaken(self): from module.awaken.awaken import Awaken Awaken(config=self.config, device=self.device).run() def shop_frequent(self): from module.shop.shop_reward import RewardShop RewardShop(config=self.config, device=self.device).run_frequent() def shop_once(self): from module.shop.shop_reward import RewardShop RewardShop(config=self.config, device=self.device).run_once() def shipyard(self): from module.shipyard.shipyard_reward import RewardShipyard RewardShipyard(config=self.config, device=self.device).run() def gacha(self): from module.gacha.gacha_reward import RewardGacha RewardGacha(config=self.config, device=self.device).run() def freebies(self): from module.freebies.freebies import Freebies Freebies(config=self.config, device=self.device).run() def minigame(self): from module.minigame.minigame import Minigame Minigame(config=self.config, device=self.device).run() def private_quarters(self): from module.private_quarters.private_quarters import PrivateQuarters PrivateQuarters(config=self.config, device=self.device).run() def daily(self): from module.daily.daily import Daily Daily(config=self.config, device=self.device).run() def hard(self): from module.hard.hard import CampaignHard CampaignHard(config=self.config, device=self.device).run() def exercise(self): from module.exercise.exercise import Exercise Exercise(config=self.config, device=self.device).run() def sos(self): from module.sos.sos import CampaignSos CampaignSos(config=self.config, device=self.device).run() def war_archives(self): from module.war_archives.war_archives import CampaignWarArchives CampaignWarArchives(config=self.config, device=self.device).run( name=self.config.Campaign_Name, folder=self.config.Campaign_Event, mode=self.config.Campaign_Mode) def raid_daily(self): from module.raid.daily import RaidDaily RaidDaily(config=self.config, device=self.device).run() def event_a(self): from module.event.campaign_abcd import CampaignABCD CampaignABCD(config=self.config, device=self.device).run() def event_b(self): from module.event.campaign_abcd import CampaignABCD CampaignABCD(config=self.config, device=self.device).run() def event_c(self): from module.event.campaign_abcd import CampaignABCD CampaignABCD(config=self.config, device=self.device).run() def event_d(self): from module.event.campaign_abcd import CampaignABCD CampaignABCD(config=self.config, device=self.device).run() def event_sp(self): from module.event.campaign_sp import CampaignSP CampaignSP(config=self.config, device=self.device).run() def maritime_escort(self): from module.event.maritime_escort import MaritimeEscort MaritimeEscort(config=self.config, device=self.device).run() def opsi_ash_assist(self): from module.os_ash.meta import AshBeaconAssist AshBeaconAssist(config=self.config, device=self.device).run() def opsi_ash_beacon(self): from module.os_ash.meta import OpsiAshBeacon OpsiAshBeacon(config=self.config, device=self.device).run() def opsi_explore(self): from module.campaign.os_run import OSCampaignRun OSCampaignRun(config=self.config, device=self.device).opsi_explore() def opsi_shop(self): from module.campaign.os_run import OSCampaignRun OSCampaignRun(config=self.config, device=self.device).opsi_shop() def opsi_voucher(self): from module.campaign.os_run import OSCampaignRun OSCampaignRun(config=self.config, device=self.device).opsi_voucher() def opsi_daily(self): from module.campaign.os_run import OSCampaignRun OSCampaignRun(config=self.config, device=self.device).opsi_daily() def opsi_obscure(self): from module.campaign.os_run import OSCampaignRun OSCampaignRun(config=self.config, device=self.device).opsi_obscure() def opsi_month_boss(self): from module.campaign.os_run import OSCampaignRun OSCampaignRun(config=self.config, device=self.device).opsi_month_boss() def opsi_abyssal(self): from module.campaign.os_run import OSCampaignRun OSCampaignRun(config=self.config, device=self.device).opsi_abyssal() def opsi_archive(self): from module.campaign.os_run import OSCampaignRun OSCampaignRun(config=self.config, device=self.device).opsi_archive() def opsi_stronghold(self): from module.campaign.os_run import OSCampaignRun OSCampaignRun(config=self.config, device=self.device).opsi_stronghold() def opsi_meowfficer_farming(self): from module.campaign.os_run import OSCampaignRun OSCampaignRun(config=self.config, device=self.device).opsi_meowfficer_farming() def opsi_hazard1_leveling(self): from module.campaign.os_run import OSCampaignRun OSCampaignRun(config=self.config, device=self.device).opsi_hazard1_leveling() def opsi_cross_month(self): from module.campaign.os_run import OSCampaignRun OSCampaignRun(config=self.config, device=self.device).opsi_cross_month() def main(self): from module.campaign.run import CampaignRun CampaignRun(config=self.config, device=self.device).run( name=self.config.Campaign_Name, folder=self.config.Campaign_Event, mode=self.config.Campaign_Mode) def main2(self): from module.campaign.run import CampaignRun CampaignRun(config=self.config, device=self.device).run( name=self.config.Campaign_Name, folder=self.config.Campaign_Event, mode=self.config.Campaign_Mode) def main3(self): from module.campaign.run import CampaignRun CampaignRun(config=self.config, device=self.device).run( name=self.config.Campaign_Name, folder=self.config.Campaign_Event, mode=self.config.Campaign_Mode) def event(self): from module.campaign.run import CampaignRun CampaignRun(config=self.config, device=self.device).run( name=self.config.Campaign_Name, folder=self.config.Campaign_Event, mode=self.config.Campaign_Mode) def event2(self): from module.campaign.run import CampaignRun CampaignRun(config=self.config, device=self.device).run( name=self.config.Campaign_Name, folder=self.config.Campaign_Event, mode=self.config.Campaign_Mode) def raid(self): from module.raid.run import RaidRun RaidRun(config=self.config, device=self.device).run() def hospital(self): from module.event_hospital.hospital import Hospital Hospital(config=self.config, device=self.device).run() def coalition(self): from module.coalition.coalition import Coalition Coalition(config=self.config, device=self.device).run() def coalition_sp(self): from module.coalition.coalition_sp import CoalitionSP CoalitionSP(config=self.config, device=self.device).run() def c72_mystery_farming(self): from module.campaign.run import CampaignRun CampaignRun(config=self.config, device=self.device).run( name=self.config.Campaign_Name, folder=self.config.Campaign_Event, mode=self.config.Campaign_Mode) def c122_medium_leveling(self): from module.campaign.run import CampaignRun CampaignRun(config=self.config, device=self.device).run( name=self.config.Campaign_Name, folder=self.config.Campaign_Event, mode=self.config.Campaign_Mode) def c124_large_leveling(self): from module.campaign.run import CampaignRun CampaignRun(config=self.config, device=self.device).run( name=self.config.Campaign_Name, folder=self.config.Campaign_Event, mode=self.config.Campaign_Mode) def gems_farming(self): from module.campaign.gems_farming import GemsFarming GemsFarming(config=self.config, device=self.device).run( name=self.config.Campaign_Name, folder=self.config.Campaign_Event, mode=self.config.Campaign_Mode) def daemon(self): from module.daemon.daemon import AzurLaneDaemon AzurLaneDaemon(config=self.config, device=self.device, task="Daemon").run() def opsi_daemon(self): from module.daemon.os_daemon import AzurLaneDaemon AzurLaneDaemon(config=self.config, device=self.device, task="OpsiDaemon").run() def event_story(self): from module.eventstory.eventstory import EventStory EventStory(config=self.config, device=self.device, task="EventStory").run() def azur_lane_uncensored(self): from module.daemon.uncensored import AzurLaneUncensored AzurLaneUncensored(config=self.config, device=self.device, task="AzurLaneUncensored").run() def benchmark(self): from module.daemon.benchmark import run_benchmark run_benchmark(config=self.config) def game_manager(self): from module.daemon.game_manager import GameManager GameManager(config=self.config, device=self.device, task="GameManager").run() def wait_until(self, future): """ Wait until a specific time. Args: future (datetime): Returns: bool: True if wait finished, False if config changed. """ future = future + timedelta(seconds=1) self.config.start_watching() while 1: if datetime.now() > future: return True if self.stop_event is not None: if self.stop_event.is_set(): logger.info("Update event detected") logger.info(f"[{self.config_name}] exited. Reason: Update") exit(0) time.sleep(5) if self.config.should_reload(): return False def get_next_task(self): """ Returns: str: Name of the next task. """ while 1: task = self.config.get_next() self.config.task = task self.config.bind(task) from module.base.resource import release_resources if self.config.task.command != 'Alas': release_resources(next_task=task.command) if task.next_run > datetime.now(): logger.info(f'Wait until {task.next_run} for task `{task.command}`') self.is_first_task = False method = self.config.Optimization_WhenTaskQueueEmpty if method == 'close_game': logger.info('Close game during wait') self.device.app_stop() release_resources() self.device.release_during_wait() if not self.wait_until(task.next_run): del_cached_property(self, 'config') continue if task.command != 'Restart': self.config.task_call('Restart') del_cached_property(self, 'config') continue elif method == 'goto_main': logger.info('Goto main page during wait') self.run('goto_main') release_resources() self.device.release_during_wait() if not self.wait_until(task.next_run): del_cached_property(self, 'config') continue elif method == 'stay_there': logger.info('Stay there during wait') release_resources() self.device.release_during_wait() if not self.wait_until(task.next_run): del_cached_property(self, 'config') continue else: logger.warning(f'Invalid Optimization_WhenTaskQueueEmpty: {method}, fallback to stay_there') release_resources() self.device.release_during_wait() if not self.wait_until(task.next_run): del_cached_property(self, 'config') continue break AzurLaneConfig.is_hoarding_task = False return task.command def loop(self): logger.set_file_logger(self.config_name) logger.info(f'Start scheduler loop: {self.config_name}') while 1: # Check update event from GUI if self.stop_event is not None: if self.stop_event.is_set(): logger.info("Update event detected") logger.info(f"Alas [{self.config_name}] exited.") break # Check game server maintenance self.checker.wait_until_available() if self.checker.is_recovered(): # There is an accidental bug hard to reproduce # Sometimes, config won't be updated due to blocking # even though it has been changed # So update it once recovered del_cached_property(self, 'config') logger.info('Server or network is recovered. Restart game client') self.config.task_call('Restart') # Get task task = self.get_next_task() # Init device and change server _ = self.device self.device.config = self.config # Skip first restart if self.is_first_task and task == 'Restart': logger.info('Skip task `Restart` at scheduler start') self.config.task_delay(server_update=True) del_cached_property(self, 'config') continue # Run logger.info(f'Scheduler: Start task `{task}`') self.device.stuck_record_clear() self.device.click_record_clear() logger.hr(task, level=0) success = self.run(inflection.underscore(task)) logger.info(f'Scheduler: End task `{task}`') self.is_first_task = False # Check failures failed = deep_get(self.failure_record, keys=task, default=0) failed = 0 if success else failed + 1 deep_set(self.failure_record, keys=task, value=failed) if failed >= 3: logger.critical(f"Task `{task}` failed 3 or more times.") logger.critical("Possible reason #1: You haven't used it correctly. " "Please read the help text of the options.") logger.critical("Possible reason #2: There is a problem with this task. " "Please contact developers or try to fix it yourself.") logger.critical('Request human takeover') handle_notify( self.config.Error_OnePushConfig, title=f"Alas <{self.config_name}> crashed", content=f"<{self.config_name}> RequestHumanTakeover\nTask `{task}` failed 3 or more times.", ) exit(1) if success: del_cached_property(self, 'config') continue elif self.config.Error_HandleError: # self.config.task_delay(success=False) del_cached_property(self, 'config') self.checker.check_now() continue else: break from module.luahook.crack import * from module.scheduler_watcher import * from datetime import datetime, timedelta from module.counter import MaxCounter from typing import Dict from module.device.my_assets import my_OTHERS_LOGIN_CONFIRM from module.config.utils import filepath_i18n, read_file from module.luahook.exception import CrackerError class FailedTaskCounter: def __init__(self, max_retries): self.max_retries = max_retries self.tasks: Dict[str, MaxCounter] = {} def count_task(self, task) -> bool: if task not in self.tasks: self.tasks[task] = MaxCounter(self.max_retries) return self.tasks[task].count_once(throw=False) def reset_task(self, task): if task in self.tasks: self.tasks[task].reset() class RepeatedFailedTaskCounter: def __init__(self, max_retries): self.max_retries = max_retries if max_retries is None: self.counter = None elif max_retries <= 0: self.counter = None else: self.counter = MaxCounter(self.max_retries) def count_once(self) -> bool: if self.counter is None: return False return self.counter.count_once(throw=False) def reset(self): if self.counter is not None: self.counter.reset() class AzurLaneAutoScript(AzurLaneAutoScript): def __init__(self, config_name='alas'): self.pre_init() super().__init__(config_name) self.class_name = self.__class__.__name__ self.is_azur = False self.is_ark = False if self.class_name == "AzurLaneAutoScript": self.is_azur = True elif self.class_name == "ArknightsAutoScript": self.is_ark = True full_config = self.config.full_config self.is_fatal_error_restart = full_config.Restart_GameRestart_FatalErrorRestart self.max_retry_times_for_same_task_failed = full_config.Restart_GameRestart_MaxRetryTimesForSameTaskFailed self.failed_task_counter = FailedTaskCounter(self.max_retry_times_for_same_task_failed) self.is_enabled_disable_task = full_config.Restart_CloseTask_Enable self.max_error_occurs = full_config.Restart_CloseTask_MaxErrorOccur if self.is_enabled_disable_task else None self.repeated_failed_task_counter = RepeatedFailedTaskCounter(self.max_error_occurs) def pre_init(self): from PIL import Image Image.init() # Manually register PngImagePlugin from PIL import PngImagePlugin Image.register_open(PngImagePlugin.PngImageFile.format, PngImagePlugin.PngImageFile, PngImagePlugin._accept) Image.register_save(PngImagePlugin.PngImageFile.format, PngImagePlugin._save) Image.register_save_all(PngImagePlugin.PngImageFile.format, PngImagePlugin._save_all) Image.register_extensions(PngImagePlugin.PngImageFile.format, [".png", ".apng"]) Image.register_mime(PngImagePlugin.PngImageFile.format, "image/png") def handle_TaskEnd(self, e) -> bool: return True def handle_GameNotRunningError(self, e) -> bool: logger.warning(e) self.config.task_call('Restart') return False def handle_GameStuckError(self, e) -> bool: logger.error(e) self.save_error_log() logger.warning(f'Game stuck, {self.device.package} will be restarted in 10 seconds') logger.warning('If you are playing by hand, please stop Alas') self.config.task_call('Restart') self.device.sleep(10) return False def handle_GameTooManyClickError(self, e) -> bool: logger.error(e) self.save_error_log() logger.warning(f'Game stuck, {self.device.package} will be restarted in 10 seconds') logger.warning('If you are playing by hand, please stop Alas') self.config.task_call('Restart') self.device.sleep(10) return False def handle_GameBugError(self, e) -> bool: logger.warning(e) self.save_error_log() logger.warning('An error has occurred in Azur Lane game client, Alas is unable to handle') logger.warning(f'Restarting {self.device.package} to fix it') self.config.task_call('Restart') self.device.sleep(10) return False def handle_GamePageUnknownError(self, e) -> bool: logger.info('Game server may be under maintenance or network may be broken, check server status now') self.checker.check_now() if self.checker.is_available(): logger.critical('Game page unknown') self.save_error_log() if self.is_fatal_error_restart: logger.info("Fatal error occured, call restart") self.config.task_call('Restart') self.device.sleep(10) return False else: handle_notify( self.config.Error_OnePushConfig, title=f"Alas <{self.config_name}> crashed", content=f"<{self.config_name}> GamePageUnknownError", ) exit(1) else: self.checker.wait_until_available() return False def handle_ScriptError(self, e) -> bool: logger.exception(e) logger.critical('This is likely to be a mistake of developers, but sometimes just random issues') if self.is_fatal_error_restart: logger.info("Fatal error occured, call restart") self.config.task_call('Restart') self.device.sleep(10) return False else: handle_notify( self.config.Error_OnePushConfig, title=f"Alas <{self.config_name}> crashed", content=f"<{self.config_name}> ScriptError", ) exit(1) def handle_RequestHumanTakeover(self, e) -> bool: logger.critical('Request human takeover') if self.is_fatal_error_restart: logger.info("Fatal error occured, call restart") self.config.task_call('Restart') self.device.sleep(10) return False else: handle_notify( self.config.Error_OnePushConfig, title=f"Alas <{self.config_name}> crashed", content=f"<{self.config_name}> RequestHumanTakeover", ) exit(1) def handle_Exception(self, e) -> bool: logger.exception(e) self.save_error_log() if self.is_fatal_error_restart: logger.info("Fatal error occured, call restart") self.config.task_call('Restart') self.device.sleep(10) return False else: handle_notify( self.config.Error_OnePushConfig, title=f"Alas <{self.config_name}> crashed", content=f"<{self.config_name}> Exception occured", ) exit(1) def delay_all_tasks(self): target = datetime.now() + timedelta(minutes=self.config.others_login.interval) self.config.task_delay(target=target, task=self.config.task.command) for i in self.config.pending_task: self.config.task_delay(target=target, task=i.command) for i in self.config.waiting_task: if i.next_run < target: self.config.task_delay(target=target, task=i.command) def handle_CrackerError(self, e) -> bool: logger.critical(e) self.save_error_log() if self.is_fatal_error_restart: logger.info("Fatal error occured, call restart") self.config.task_call('Restart') self.device.sleep(10) return False else: handle_notify( self.config.Error_OnePushConfig, title=f"Alas <{self.config_name}> crashed", content=f"<{self.config_name}> Exception occured", ) exit(1) def handle_OthersLogin(self, e) -> bool: logger.warning("Detected others' login") self.delay_all_tasks() method = self.config.Optimization_WhenTaskQueueEmpty if method == "goto_main": self.config.Optimization_WhenTaskQueueEmpty = "stay_there" logger.warning("You should never goto main when you need to delay tasks as others login, change to stay there instead.") self.device.click(my_OTHERS_LOGIN_CONFIRM) if method == "stay_there": self.device.click(my_OTHERS_LOGIN_CONFIRM) if self.config.others_login.need_notify: handle_notify( self.config.Error_OnePushConfig, title=f"Alas <{self.config_name}>: account's owner logs in", content=f"Delay some tasks to {datetime.now() + timedelta(minutes=self.config.others_login.interval, seconds=-1)}", ) self.config.save() self.config.update() return True def run(self, command, skip_first_screenshot=False): try: if self.is_azur: if command != "restart" and command != "hook_op": luahook_disable_all(self.config, self.device) luahook_crack_all(self.config, self.device) if not skip_first_screenshot: self.device.screenshot() self.__getattribute__(command)() return True except TaskEnd as e: return self.handle_TaskEnd(e) except GameNotRunningError as e: return self.handle_GameNotRunningError(e) except GameStuckError as e: return self.handle_GameStuckError(e) except GameTooManyClickError as e: return self.handle_GameTooManyClickError(e) except GameBugError as e: return self.handle_GameBugError(e) except GamePageUnknownError as e: return self.handle_GamePageUnknownError(e) except ScriptError as e: return self.handle_ScriptError(e) except RequestHumanTakeover as e: return self.handle_RequestHumanTakeover(e) except OthersLogin as e: return self.handle_OthersLogin(e) except CrackerError as e: return self.handle_CrackerError(e) except Exception as e: return self.handle_Exception(e) def main4(self): from module.campaign.run import CampaignRun CampaignRun(config=self.config, device=self.device).run( name=self.config.Campaign_Name, folder=self.config.Campaign_Event, mode=self.config.Campaign_Mode) def main5(self): from module.campaign.run import CampaignRun CampaignRun(config=self.config, device=self.device).run( name=self.config.Campaign_Name, folder=self.config.Campaign_Event, mode=self.config.Campaign_Mode) def main6(self): from module.campaign.run import CampaignRun CampaignRun(config=self.config, device=self.device).run( name=self.config.Campaign_Name, folder=self.config.Campaign_Event, mode=self.config.Campaign_Mode) def event3(self): from module.campaign.run import CampaignRun CampaignRun(config=self.config, device=self.device).run( name=self.config.Campaign_Name, folder=self.config.Campaign_Event, mode=self.config.Campaign_Mode) def event4(self): from module.campaign.run import CampaignRun CampaignRun(config=self.config, device=self.device).run( name=self.config.Campaign_Name, folder=self.config.Campaign_Event, mode=self.config.Campaign_Mode) def event5(self): from module.campaign.run import CampaignRun CampaignRun(config=self.config, device=self.device).run( name=self.config.Campaign_Name, folder=self.config.Campaign_Event, mode=self.config.Campaign_Mode) def event6(self): from module.campaign.run import CampaignRun CampaignRun(config=self.config, device=self.device).run( name=self.config.Campaign_Name, folder=self.config.Campaign_Event, mode=self.config.Campaign_Mode) @disable_all_crack def minigame(self): super().minigame() @disable_all_crack @crack_op( before_call=[CrackOp.EnableExerciseGodMod, CrackOp.EnableExerciseMorePower], after_call=[CrackOp.DisableExerciseGodMod, CrackOp.DisableExerciseMorePower] ) def exercise(self): super().exercise() @disable_all_crack def raid_daily(self): super().raid_daily() @disable_all_crack def opsi_ash_assist(self): super().opsi_ash_assist() @disable_all_crack def opsi_ash_beacon(self): super().opsi_ash_beacon() @disable_all_crack def raid(self): super().raid() @disable_all_crack def coalition(self): super().coalition() @disable_all_crack def coalition_sp(self): super().coalition_sp() def event_shop(self): from module.shop_event.shop_event import EventShop EventShop(config=self.config, device=self.device).run() @disable_all_crack def raid_scuttle(self): from module.raid.scuttle import RaidScuttleRun RaidScuttleRun(config=self.config, device=self.device).run() def hook_op(self): op = self.config.full_config.HookOp_HookOp_Operation if op == "detect": CrackResource(self.config, self.device).auto_detect() elif op == "inject": CrackResource(self.config, self.device).ensure() elif op == "apply": luahook_disable_all(self.config, self.device) luahook_crack_all(self.config, self.device) else: logger.error(f"Unknown Hook Operation: {op}") def get_next_task(self): while 1: task = self.config.get_next() self.config.task = task self.config.bind(task) from module.base.resource import release_resources if self.config.task.command != 'Alas': release_resources(next_task=task.command) if task.next_run > datetime.now(): self.config.scheduler_watcher.no_task() logger.info(f'Wait until {task.next_run} for task `{task.command}`') self.is_first_task = False method = self.config.Optimization_WhenTaskQueueEmpty if method == 'close_game': logger.info('Close game during wait') self.device.app_stop() release_resources() self.device.release_during_wait() if not self.wait_until(task.next_run): del_cached_property(self, 'config') continue if task.command != 'Restart': self.config.task_call('Restart') del_cached_property(self, 'config') continue elif method == 'goto_main': logger.info('Goto main page during wait') self.run('goto_main') release_resources() self.device.release_during_wait() if not self.wait_until(task.next_run): del_cached_property(self, 'config') continue elif method == 'stay_there': logger.info('Stay there during wait') release_resources() self.device.release_during_wait() if not self.wait_until(task.next_run): del_cached_property(self, 'config') continue else: logger.warning(f'Invalid Optimization_WhenTaskQueueEmpty: {method}, fallback to stay_there') release_resources() self.device.release_during_wait() if not self.wait_until(task.next_run): del_cached_property(self, 'config') continue break AzurLaneConfig.is_hoarding_task = False return task.command def get_task_name(self, task): try: return read_file(filepath_i18n('zh-CN'))["Task"][task]["name"] except KeyError: return task def is_force_enabled_task(self, task): return task in ["Research", "Reward", "Commission"] def loop(self): if self.is_azur: CrackResource(self.config, self.device).ensure() if self.is_azur and self.config.full_config.Hook_HookGeneral_Enable and self.config.full_config.Hook_HookGeneral_RestartEveryTime: logger.info("Hook enabled, do restart") self.restart() logger.set_file_logger(self.config_name) logger.info(f'Start scheduler loop: {self.config_name}') while 1: # Check update event from GUI if self.stop_event is not None: if self.stop_event.is_set(): logger.info("Update event detected") logger.info(f"Alas [{self.config_name}] exited.") break # Check game server maintenance self.checker.wait_until_available() if self.checker.is_recovered(): # There is an accidental bug hard to reproduce # Sometimes, config won't be updated due to blocking # even though it has been changed # So update it once recovered del_cached_property(self, 'config') logger.info('Server or network is recovered. Restart game client') self.config.task_call('Restart') # Get task task = self.get_next_task() # Init device and change server _ = self.device self.device.config = self.config # Skip first restart if self.is_first_task and task == 'Restart': logger.info('Skip task `Restart` at scheduler start') self.config.task_delay(server_update=True) del_cached_property(self, 'config') continue # Run logger.info(f'Scheduler: Start task `{task}`') if self.is_azur: self.config.scheduler_watcher.switch_task(task) self.device.stuck_record_clear() self.device.click_record_clear() logger.hr(task, level=0) success = self.run(inflection.underscore(task)) logger.info(f'Scheduler: End task `{task}`') self.is_first_task = False # Check failures if not success: if not self.failed_task_counter.count_task(task): logger.critical(f"Task `{task}` failed {self.max_retry_times_for_same_task_failed} or more times.") logger.critical("Possible reason #1: You haven't used it correctly. " "Please read the help text of the options.") logger.critical("Possible reason #2: There is a problem with this task. " "Please contact developers or try to fix it yourself.") logger.critical('Request human takeover') handle_notify( self.config.Error_OnePushConfig, title=f"Alas <{self.config_name}> crashed", content=f"<{self.config_name}> RequestHumanTakeover\nTask `{task}` failed {self.max_retry_times_for_same_task_failed} or more times.", ) if self.is_enabled_disable_task: if not self.repeated_failed_task_counter.count_once(): if not self.is_force_enabled_task(task): msg = "This task has caused many errors already, disable it" logger.warning(msg) self.config.disable_task(task) self.failed_task_counter.reset_task(task) self.repeated_failed_task_counter.reset() handle_notify( self.config.Error_OnePushConfig, title=f"Alas <{self.config_name}> disabled `{self.get_task_name(task)}`", content=msg, ) else: msg = "This task has caused many errors already, but it is force enabled, delay it" logger.warning(msg) self.config.task_delay(target=datetime.now() + timedelta(hours=6), task=task) self.failed_task_counter.reset_task(task) self.repeated_failed_task_counter.reset() handle_notify( self.config.Error_OnePushConfig, title=f"Alas <{self.config_name}> delay `{self.get_task_name(task)}`", content=msg, ) else: exit(1) if success: self.failed_task_counter.reset_task(task) del_cached_property(self, 'config') continue elif self.config.Error_HandleError: # self.config.task_delay(success=False) del_cached_property(self, 'config') self.checker.check_now() continue else: break if __name__ == '__main__': alas = AzurLaneAutoScript() alas.loop()