study.py

# -*- coding:utf-8 -*- 

"""
作者：luoyulong
日期：2021年01月29日
"""
import contextlib
import ctypes
import logging
import shlex
import subprocess
import sys
import time
from ctypes import *
from enum import Enum

import uiautomator2 as u2
import urllib3
import os
from answer_questions.auto_study import AutoStudy
from answer_questions.weekly_module import WeeklyQuestions
from unit.check_date_time import check_date, check_current_date
from unit.get_android_battery import get_battery_info
from unit.unit import *
from rich.progress import track

from push.send_mail import SendMail
from answer_questions.challenge_module import ChallengeQuestions
from answer_questions.daily_module import DailyQuestions
from answer_questions.double_battle_module import DoubleBattle
from answer_questions.four_man_battle_module import FourManAimHigh
from answer_questions.competition import Competition
from listen_reading.listen_module import WatchingVideos, NewListen
from unit.machine_code import Register
from listen_reading.reading_module import Reading
# from unit.screen_off import screen_off
from answer_questions.special_module import SpecialQuestions
from subscribe.subscribe_module import Subscribe
from unit.unit import *

from terminal_layout.extensions.choice import *
from terminal_layout import *

# from rich import logging

class_dict = {0: DailyQuestions,
              1: WeeklyQuestions,
              2: SpecialQuestions,
              3: ChallengeQuestions,
              4: ChallengeQuestions,
              5: FourManAimHigh,
              6: DoubleBattle,
              7: Reading,
              8: NewListen,
              9: Subscribe,
              10: WatchingVideos,
              11: Competition,
              12: 1}


def menu():
    c = Choice('Which function you hope? (press <esc> to exit) ',
               [value.value for value in Arguments],
               icon_style=StringStyle(fore=Fore.blue),
               selected_style=StringStyle(fore=Fore.blue))
    return c.get_choice()


def check_arguments(func):
    def wrapper(*args, **kwargs):
        sys_arg = sys.argv
        argument = sys_arg[1] if len(sys_arg) > 1 else "?"
        if len(sys_arg) < 2:
            input_ = input("[ Input the function you wish: ]")
            if input_ == "Main":
                func(*args, **kwargs)
            elif input_ == "Menu":
                index, value = menu()
                print(f"Prepare to run {index}:-> {value} ")
                if index == 4:
                    class_dict.get(index)(cycle=True).entry()
                elif index == 12:
                    func(*args, **kwargs)
                else:
                    class_dict.get(index)().entry()
        elif argument == "Main":
            func(*args, **kwargs)
        elif argument == "3":
            class_dict.get(int(argument))(cycle=True).entry()
        else:
            class_dict.get(int(argument))().entry()

    return wrapper


class Main(object):
    device = None

    def __init__(self):
        self.emulator_name = ""
        self.emulator_path = ""
        self.__auto_start = 0
        self.__wait_time = 0
        self.__run_delay = 150
        self.__is_screen_off = False
        self.__shutdown = False
        self.time_start = time.time()
        self.time_used = None
        self.time_end = None
        self.app = None
        self.connected_devices = None
        self.read_db()
        self.log = Log().get_logger()

    # @check_current_date
    @check_arguments
    def run_order(self):
        u2.setup_logger(level=40, disableStderrLogger=True)
        urllib3.add_stderr_logger(logging.ERROR)
        self.add_environ()
        self.emulator_run()
        # self.__cancel_edit_mod()

        self.connected_devices = self.get_devices()
        self.__screen_off()
        self.connect_device()
        # self.__cancel_edit_mod()
        # self.hide_app()
        self.start_project()
        self.study_entry()
        self.end_project()
        self.is_shutdown()

    @staticmethod
    def add_environ():
        with contextlib.suppress(Exception):
            os.environ["PATH"] += os.pathsep + os.getcwd()

    @staticmethod
    def __cancel_edit_mod():
        kernel32 = ctypes.windll.kernel32
        # try:
        #     kernel32.SetConsoleMode(kernel32.GetStdHandle(-10), (0x4 | 0x80 | 0x20 | 0x2 | 0x10 | 0x1 | 0x00 | 0x100))
        # except Exception:
        #     pass

    def read_db(self):
        conn = sqlite3.connect('config/data.sqlite')
        cursor = conn.cursor()
        self.emulator_name, self.emulator_path, self.__auto_start, self.__run_delay = cursor.execute('select name,path,auto_start,run_delay from emulator').fetchone()
        cursor.execute('select * from power')
        __power = cursor.fetchone()
        a = __power[0]
        self.__shutdown = bool(__power[0])
        self.__wait_time = int(__power[1])
        self.__is_screen_off = bool(__power[2])
        conn.close()

    def emulator_run(self):
        if not self.__auto_start:
            self.log.info("不启用模拟器,进入手机模式. . .")
            return
        while True:
            if not self.__is_emulator_running(self.emulator_name):
                shell_process = subprocess.Popen([self.emulator_path], shell=True)
                self.log.info(f"初始化模拟器. . .进程ID：{shell_process.pid}")
                for _ in track(range(self.__run_delay), description="[0] process:"):
                    time.sleep(0.1)
            else:
                self.log.info(f"模拟器正在运行--监听进程:{self.emulator_name}")
                break

    def connect_device(self):
        connected_devices = self.get_devices()
        if len(connected_devices) > 1:
            self.log.info("发现超过1个的安卓设备连接到PC上")
        elif len(connected_devices) <= 0:
            self.log.info("未检测到设备！请检查模拟器或手机是否已正确连接并开启了调试模式，程序在2秒后重试...")
            time.sleep(2)
        # 尝试连接设备
        for device_name in connected_devices:
            try:
                self.device = u2.connect(device_name) if device_name else u2.connect()
                self.device.debug = False  # 取消掉debug
                self.log.info(f"连接设备成功！设备名:{device_name} ")
                if not self.__auto_start: #　手机模式,点亮屏幕
                    self.log.info("~！点亮屏幕")
                    self.device.screen_on()
                return
            except Exception as error:
                self.log.info(f"连接设备失败！设备名:{device_name}; error code:{error}")
        self._restart_emulator()
        self.connect_device()

    def _restart_emulator(self):
        self.log.warning("重启模拟器...")
        command = f"taskkill /f /t /im {self.emulator_name}"
        try:
            os.popen(command)
            self.emulator_run()
        except Exception as error:
            self.log.error(f"重启模拟器...失败！ {error}")

    def hide_app(self):
        self.hide_command("adb shell pm hide de.robv.android.xposed.installer")
        self.hide_command("adb shell pm hide club.fishine.webviewdebughook")
        self.hide_command("adb shell pm hide io.appium.android.apis")

    @staticmethod
    def hide_command( command):
        with contextlib.suppress(Exception):
            subprocess.check_call(shlex.split(command), stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL)
            # self.log.error(e)

    def start_project(self):
        try:
            self.app = CreateAPP(driver=self.device, logger=self.log)
            self.app.start_first()
            self.log.debug("启动app")
        except Exception as error_project:
            self.log.debug("error:从不是创建控件的线程访问它" % error_project)

    def study_entry(self):
        while True:
            try:
                complete = AutoStudy(device=self.device, logger=self.log, app=self.app).complete
                time_sleep(2)
                if complete:
                    break
            except Exception as study_entry_error:
                self.log.debug(f'study_entry_error:{study_entry_error}')
                self.app.restart()
            time_sleep(2)

    def end_project(self):
        self.time_end = time.time()
        self.time_used = self.time_end - self.time_start
        self.log.info("本次学习共用时约:%.0f分" % (self.time_used / 60))

    def is_shutdown(self):
        battery_level = get_battery_info()
        if battery_level < 50:
            self.log.info("电量低于50%，充电1小时")
            time.sleep(60*60)
        if self.__shutdown:
            os.system(f"shutdown -s -f -t  {self.__wait_time} ")
            for _ in track(range(self.__wait_time), description="[!] 自动关机:"):
                time.sleep(1)
        else:
            self.log.info("不自动关机，即将退出程序. . .")
            time_sleep(3)

    def __is_emulator_running(self, process_name):
        # os.system("mode con cols=60 lines=20")
        try:
            self.log.debug(f'tasklist | findstr {process_name}')
            process = len(os.popen(f'tasklist | findstr {process_name}').readlines())
            if process >= 1:
                self.log.debug(f'模拟器-->{process_name}<-- 正在运行--{process}个实例')
                return True
            else:
                self.log.debug(f'模拟器-->{process_name}<-- is not running')
                return False
        except Exception as e:
            self.log.error(f"程序错误| {e}")
            return False

    def __screen_off(self):
        if self.__is_screen_off:
            HWND_BROADCAST = 0xffff
            WM_SYSCOMMAND = 0x0112
            SC_MONITORPOWER = 0xF170
            MonitorPowerOff = 2
            SW_SHOW = 5
            windll.user32.PostMessageW(HWND_BROADCAST, WM_SYSCOMMAND,
                                       SC_MONITORPOWER, MonitorPowerOff)
            shell32 = windll.LoadLibrary("shell32.dll")
            shell32.ShellExecuteW(None, 'open', 'rundll32.exe',
                                  'USER32', '', SW_SHOW)
        else:
            self.log.debug("跳过屏幕检测")

    @staticmethod
    def get_devices():
        cmd = r'adb devices'  # % apk_file
        pr = subprocess.Popen(cmd, stdout=subprocess.PIPE, shell=True)
        pr.wait()  # 不会马上返回输出的命令，需要等待
        out = pr.stdout.readlines()  # out = pr.stdout.read().decode("UTF-8")
        result = {}
        for i in out[1:-1]:
            d = str(i).split("'")[1].split("\\t")
            device = d[0]
            statue = d[1].replace('\\r', '').replace('\\n', '')
            result[device] = statue
        return result  # 手机设备列表


class Arguments(Enum):
    Daily = "Daily"
    Weekly = "Weekly"
    Special = "Special"
    Challenge = "Challenge"
    ChallengeCycle = "ChallengeCycle"
    FourMan = "FourMan"
    DoubleMan = "DoubleMan"
    Read = "Read"
    Listen = "Listen"
    Subscribe = "Subscribe"
    BaiLing = "BaiLing"
    Competition = "Competition"
    Main = "Main"


if __name__ == '__main__':
    Main().run_order()
    # DailyQuestions().entry()