mirror of
https://github.com/0O0o0oOoO00/Alas.git
synced 2026-05-14 16:19:25 +08:00
103 lines
3.6 KiB
Python
103 lines
3.6 KiB
Python
from module.logger import logger
|
|
import time
|
|
import json
|
|
from module.base.base import ModuleBase
|
|
from module.base.timer import Timer
|
|
from module.cloud_phone.chinac_api import *
|
|
from module.config.config import deep_get
|
|
from module.exception import ChinacError
|
|
|
|
|
|
class CloudPhoneRestart(ModuleBase):
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
|
|
config = self.config.data
|
|
self.access_key = deep_get(config, "CloudPhoneRestart.CloudPhoneSetting.AccessKey", "")
|
|
self.access_key_secret = deep_get(config, "CloudPhoneRestart.CloudPhoneSetting.AccessKeySecret", "")
|
|
self.region = deep_get(config, "CloudPhoneRestart.CloudPhoneSetting.Region", "")
|
|
self.phone_id = deep_get(config, "CloudPhoneRestart.CloudPhoneSetting.PhoneId", "")
|
|
|
|
self.api = None
|
|
|
|
def check_config(self):
|
|
def value_check(v: str):
|
|
if not v:
|
|
raise ValueError("Invalid cloud phone setting")
|
|
|
|
value_check(self.access_key)
|
|
value_check(self.access_key_secret)
|
|
value_check(self.region)
|
|
value_check(self.phone_id)
|
|
|
|
def load_chinac_api(self):
|
|
self.api = ChinacOpenApi(self.access_key, self.access_key_secret)
|
|
|
|
def phone_restart(self):
|
|
self.api.set_action(CloudPhoneAction.RebootCloudPhone)
|
|
self.api.set_http_method("POST")
|
|
self.api.set_request_params({
|
|
"CloudPhones": [
|
|
{
|
|
"Region": self.region,
|
|
"Id": self.phone_id
|
|
}
|
|
],
|
|
"Operate": "reboot"
|
|
})
|
|
res = self.api.do()
|
|
info = json.loads(res["Info"])
|
|
status_code = res["Status"]
|
|
if status_code != 200 and info["code"] == 10000:
|
|
raise ChinacError(f"Chinac api error, code: {status_code}")
|
|
logger.info("Cloud phone restart request success")
|
|
|
|
def is_ready(self):
|
|
self.api.set_action(CloudPhoneAction.DescribeCloudPhone)
|
|
self.api.set_http_method("POST")
|
|
self.api.set_request_params({
|
|
'Region': self.region,
|
|
"Id": self.phone_id
|
|
})
|
|
res = self.api.do()
|
|
info = json.loads(res["Info"])
|
|
status_code = res["Status"]
|
|
if status_code != 200 and info["code"] == 10000:
|
|
raise ChinacError(f"Chinac api error, code: {status_code}")
|
|
phone_status = info["data"]["BasicInfo"]["Status"]
|
|
logger.info(f"Cloud phone current status: {phone_status}")
|
|
if phone_status == "START":
|
|
logger.info("Cloud phone ready")
|
|
return True
|
|
return False
|
|
|
|
def wait_for_adb(self):
|
|
logger.info("Wait for adb of cloud phone")
|
|
time.sleep(deep_get(self.config.data, "CloudPhoneRestart.CloudPhoneSetting.AdbWaitTime", 60))
|
|
|
|
def wait_for_ready(self):
|
|
wait_timer = Timer(deep_get(self.config.data, "CloudPhoneRestart.CloudPhoneSetting.PhoneWaitTime", 60))
|
|
wait_timer.start()
|
|
while 1:
|
|
if wait_timer.reached():
|
|
raise ChinacError("Waiting for cloud phone restart too long")
|
|
if self.is_ready():
|
|
break
|
|
time.sleep(1)
|
|
|
|
def run(self):
|
|
try:
|
|
self.check_config()
|
|
self.load_chinac_api()
|
|
self.phone_restart()
|
|
self.wait_for_ready()
|
|
self.wait_for_adb()
|
|
self.config.task_delay(server_update=True)
|
|
self.config.task_call('Restart')
|
|
except Exception as e:
|
|
logger.exception(e)
|
|
self.config.task_delay(server_update=True)
|
|
self.config.task_call('Restart')
|
|
return False
|
|
return True
|