WindyMadman/AzurLaneAutoScript
1
0
Fork 0
mirror of https://github.com/LmeSzinc/AzurLaneAutoScript.git synced 2025-01-08 12:07:36 +08:00
Code Issues Packages Projects Releases Wiki Activity

Add: Expose scheduler and server

Browse Source
This commit is contained in:
LmeSzinc 2021-09-14 17:32:23 +08:00
parent 11290af0fa
commit a504e97f4c
4 changed files with 49 additions and 302 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@ -3,6 +3,8 @@ log
config/backup/*
config/*.ini
!config/template.ini
config/*.yaml
!config/template.yaml
*.pyw
!alas_cn.pyw
!alas_en.pyw

246
alas.py
View File

@ -1,32 +1,41 @@
import os
import re
import time
from datetime import datetime
from module.config.config import AzurLaneConfig
import inflection
from cached_property import cached_property
from module.commission.commission import RewardCommission
from module.config.config import AzurLaneConfig, TaskEnd
from module.config.db import Database
from module.device.device import Device
from module.exception import *
from module.handler.login import LoginHandler
from module.handler.sensitive_info import handle_sensitive_image, handle_sensitive_logs
from module.logger import logger, pyw_name, log_file
from module.logger import logger
from module.research.research import RewardResearch
from module.tactical.tactical_class import RewardTacticalClass
class AzurLaneAutoScript:
def __init__(self, ini_name=''):
if not ini_name:
ini_name = pyw_name
ini_name = ini_name.lower()
self.config = AzurLaneConfig(ini_name)
self.device = None
def __init__(self, config_name='alas'):
self.config_name = config_name
Database().update_config(config_name)
@cached_property
def config(self):
config = AzurLaneConfig(config_name=self.config_name)
return config
@cached_property
def device(self):
device = Device(config=self.config)
return device
def run(self, command):
logger.attr('Command', command)
self.config.start_time = datetime.now()
self.device = Device(config=self.config)
while 1:
try:
self.__getattribute__(command.lower())()
break
self.__getattribute__(command)()
return True
except TaskEnd:
return True
except GameNotRunningError as e:
logger.warning(e)
az = LoginHandler(self.config, device=self.device)
@ -56,201 +65,36 @@ class AzurLaneAutoScript:
except Exception as e:
logger.exception(e)
self.save_error_log()
break
return False
def save_error_log(self):
"""
Save last 60 screenshots in ./log/error/<timestamp>
Save logs to ./log/error/<timestamp>/log.txt
"""
if self.config.ENABLE_ERROR_LOG_AND_SCREENSHOT_SAVE:
folder = f'./log/error/{int(time.time() * 1000)}'
logger.info(f'Saving error: {folder}')
os.mkdir(folder)
for data in logger.screenshot_deque:
image_time = datetime.strftime(data['time'], '%Y-%m-%d_%H-%M-%S-%f')
image = handle_sensitive_image(data['image'])
image.save(f'{folder}/{image_time}.png')
with open(log_file, 'r', encoding='utf-8') as f:
start = 0
for index, line in enumerate(f.readlines()):
if re.search('\+-{15,}\+', line):
start = index
with open(log_file, 'r', encoding='utf-8') as f:
text = f.readlines()[start - 2:]
text = handle_sensitive_logs(text)
with open(f'{folder}/log.txt', 'w', encoding='utf-8') as f:
f.writelines(text)
pass
def reward_when_finished(self):
from module.reward.reward import Reward
az = Reward(self.config, device=self.device)
az.reward_loop()
self.update_check()
def research(self):
RewardResearch(config=self.config, device=self.device).run()
def setting(self):
for key, value in self.config.config['Setting'].items():
if key == 'azurstat_id':
print(f'{key} = <sensitive_infomation>')
else:
print(f'{key} = {value}')
def commission(self):
RewardCommission(config=self.config, device=self.device).run()
logger.hr('Settings saved')
self.update_check()
self.config.config_check()
def tactical(self):
RewardTacticalClass(config=self.config, device=self.device).run()
def update_check(self):
from module.update import Update
ad = Update(self.config)
if self.config.UPDATE_CHECK:
ad.get_local_commit()
def loop(self):
while 1:
logger.info(f'Scheduler: Start task `{self.config.task}`')
success = self.run(inflection.underscore(self.config.task))
def reward(self):
for key, value in self.config.config['Reward'].items():
print(f'{key} = {value}')
logger.info(f'Scheduler: End task `{self.config.task}`')
del self.__dict__['config']
logger.hr('Reward Settings saved')
self.update_check()
self.reward_when_finished()
if not success:
break
def emulator(self):
for key, value in self.config.config['Emulator'].items():
if key == 'github_token':
print(f'{key} = <sensitive_infomation>')
else:
print(f'{key} = {value}')
logger.hr('Emulator saved')
self.update_check()
from module.handler.login import LoginHandler
az = LoginHandler(self.config, device=self.device)
if az.app_ensure_start():
from module.reward.reward import Reward
az = Reward(self.config, device=self.device)
az.reward()
else:
az.device.screenshot()
def main(self):
"""
Method to run main chapter.
"""
from module.campaign.run import CampaignRun
az = CampaignRun(self.config, device=self.device)
az.run(self.config.CAMPAIGN_NAME)
self.reward_when_finished()
def daily(self):
"""
Method to run daily missions.
"""
from module.reward.reward import Reward
az = Reward(self.config, device=self.device)
az.daily_wrapper_run()
self.reward_when_finished()
def event(self):
"""
Method to run event.
"""
from module.campaign.run import CampaignRun
az = CampaignRun(self.config, device=self.device)
az.run(self.config.EVENT_STAGE, folder=self.config.EVENT_NAME)
self.reward_when_finished()
def sos(self):
"""
Method to SOS maps.
"""
from module.sos.sos import CampaignSos
az = CampaignSos(self.config, device=self.device)
az.run()
self.reward_when_finished()
def war_archives(self):
"""
Method to War Archives maps.
"""
from module.war_archives.war_archives import CampaignWarArchives
az = CampaignWarArchives(self.config, device=self.device)
az.run(self.config.WAR_ARCHIVES_STAGE, folder=self.config.WAR_ARCHIVES_NAME)
self.reward_when_finished()
def raid(self):
from module.raid.run import RaidRun
az = RaidRun(self.config, device=self.device)
az.run()
self.reward_when_finished()
def event_daily_ab(self):
from module.event.campaign_ab import CampaignAB
az = CampaignAB(self.config, device=self.device)
az.run_event_daily()
self.reward_when_finished()
def semi_auto(self):
from module.daemon.daemon import AzurLaneDaemon
az = AzurLaneDaemon(self.config, device=self.device)
az.daemon()
def c11_affinity_farming(self):
from module.campaign.run import CampaignRun
az = CampaignRun(self.config, device=self.device)
az.run('campaign_1_1_affinity_farming')
self.reward_when_finished()
def c72_mystery_farming(self):
from module.campaign.run import CampaignRun
az = CampaignRun(self.config, device=self.device)
az.run('campaign_7_2_mystery_farming')
self.reward_when_finished()
def c124_leveling(self):
from module.campaign.run import CampaignRun
az = CampaignRun(self.config, device=self.device)
az.run('campaign_12_4_leveling')
self.reward_when_finished()
def c122_leveling(self):
from module.campaign.run import CampaignRun
az = CampaignRun(self.config, device=self.device)
az.run('campaign_12_2_leveling')
self.reward_when_finished()
def gems_farming(self):
from module.campaign.gems_farming import GemsFarming
az = GemsFarming(self.config, device=self.device)
az.run(self.config.GEMS_STAGE)
self.reward_when_finished()
def retire(self):
from module.retire.retirement import Retirement
az = Retirement(self.config, device=self.device)
az.device.screenshot()
az.retire_ships(amount=2000)
def os_semi_auto(self):
from module.daemon.os_daemon import AzurLaneDaemon
az = AzurLaneDaemon(self.config, device=self.device)
az.daemon()
def os_clear_map(self):
from module.campaign.os_run import OSCampaignRun
az = OSCampaignRun(self.config, device=self.device)
az.run()
def os_world_clear(self):
from module.campaign.os_run import OSCampaignRun
az = OSCampaignRun(self.config, device=self.device)
az.run_clear_os_world()
self.reward_when_finished()
def os_fully_auto(self):
from module.campaign.os_run import OSCampaignRun
az = OSCampaignRun(self.config, device=self.device)
az.run_operation_siren()
self.reward_when_finished()
# alas = AzurLaneAutoScript()
# alas.reward()
if __name__ == '__main__':
alas = AzurLaneAutoScript()
alas.loop()

99
module/config/scheduler.py
View File

@ -1,99 +0,0 @@
import time
import inflection
from cached_property import cached_property
from module.commission.commission import RewardCommission
from module.config.config import AzurLaneConfig, TaskEnd
from module.config.db import Database
from module.device.device import Device
from module.exception import *
from module.handler.login import LoginHandler
from module.logger import logger
from module.research.research import RewardResearch
from module.tactical.tactical_class import RewardTacticalClass
class AzurLaneAutoScript:
def __init__(self, config_name='alas'):
self.config_name = config_name
Database().update_config(config_name)
@cached_property
def config(self):
config = AzurLaneConfig(config_name=self.config_name)
return config
@cached_property
def device(self):
device = Device(config=self.config)
return device
def run(self, command):
while 1:
try:
self.__getattribute__(command)()
return True
except TaskEnd as e:
return True
except GameNotRunningError as e:
logger.warning(e)
az = LoginHandler(self.config, device=self.device)
az.app_restart()
az.ensure_no_unfinished_campaign()
continue
except GameTooManyClickError as e:
logger.warning(e)
self.save_error_log()
az = LoginHandler(self.config, device=self.device)
az.handle_game_stuck()
continue
except GameStuckError as e:
logger.warning(e)
self.save_error_log()
az = LoginHandler(self.config, device=self.device)
az.handle_game_stuck()
continue
except LogisticsRefreshBugHandler as e:
logger.warning(e)
self.save_error_log()
az = LoginHandler(self.config, device=self.device)
az.device.app_stop()
time.sleep(600)
az.app_ensure_start()
continue
except Exception as e:
logger.exception(e)
self.save_error_log()
return False
def save_error_log(self):
"""
Save last 60 screenshots in ./log/error/<timestamp>
Save logs to ./log/error/<timestamp>/log.txt
"""
pass
def research(self):
RewardResearch(config=self.config, device=self.device).run()
def commission(self):
RewardCommission(config=self.config, device=self.device).run()
def tactical(self):
RewardTacticalClass(config=self.config, device=self.device).run()
def loop(self):
while 1:
logger.info(f'Scheduler: Start task: {self.config.task}')
success = self.run(inflection.underscore(self.config.task))
logger.info(f'Scheduler: End task: {self.config.task}')
del self.__dict__['config']
if not success:
break
scheduler = AzurLaneAutoScript()
scheduler.loop()

4
module/config/websocket.py → module/config/std_server.py
View File

@ -8,7 +8,7 @@ from module.config.db import Database
sys.stdin.reconfigure(encoding='utf-8')
class WebsocketServer(Database):
class StdServer(Database):
def response_error(self, msg):
self.response({'query': 'error', 'data': str(msg)})
@ -72,6 +72,6 @@ class WebsocketServer(Database):
if __name__ == '__main__':
server = WebsocketServer()
server = StdServer()
res = server.parse('{"query":"upsert_db","data":{"task":"Main","group":"_info","arg":"_info","lang":"zh-CN","name":"中文","help":"Main._info._info.help","type":"","value":"","row":1,"option":{},"config":"alas"}}')
print(res)