python-yzy/main.py

# This sample code uses the Appium python client v2
# pip install Appium-Python-Client
# Then you can paste this into a file and simply run with Python
import random
import types
from typing import Union

from appium import webdriver
from selenium.common import TimeoutException
from selenium.webdriver.support.ui import WebDriverWait
from appium.webdriver.common.appiumby import AppiumBy
from datetime import datetime
import time
import os
import sys
import getopt
# import winsound

# For W3C actions
from selenium.webdriver.common.action_chains import ActionChains
from selenium.webdriver.common.actions import interaction
from selenium.webdriver.common.actions.action_builder import ActionBuilder
from selenium.webdriver.common.actions.pointer_input import PointerInput
from selenium.webdriver.support import expected_conditions

isProfession = False
driver = ''
# 起始id
startId = 10006
# 结束id
endId = 12900
# ignores
ignores = []
# indicate, higher priority than start-end for no-seq targets
targets = []

card_id = "15700779"
card_pwd = "244796"

phone = '13974787400'
phone_pwd = 'Xj654123'

hide_guide = False
stop = False
retry_times_max = 1  # 搜索loading时，间隔几次重新点击搜索
retry_current = 0  # 暂时没有用途了
batch_count = 3  # 有几个批次
subject_default_top = True  # 默认理科、物理
plan_change_years = [2,3]  # 计划取几年？从 1 开始
profession_change_years = [1]  # 专业取几年？从 1 开始
# 登录方式 1 卡号 2 手机密码
loginType = 1
# 数据类型 1 招生计划 2 历年录取 3 两者都有
dataType = 2


def randomSleep():
    rd = random.choice([0, 0.5, 0.8, 0.75, 1, 2, 1.5, 1.2, 1.8])
    if rd: time.sleep(rd / 4.0)


def init():
    global driver
    global isProfession
    print('程序启动')
    caps = {}
    caps["platformName"] = "Android"
    caps["appium:platformVersion"] = "6.0"
    caps["appium:deviceName"] = "emulator-5554"
    caps["appium:appPackage"] = "com.eagersoft.youzy.youzy"
    # .mvvm.ui.login.LoginAndRegisterSelectActivity
    # .mvvm.ui.college.FindCollegeActivity
    caps["appium:appActivity"] = ".mvvm.ui.login.LoginAndRegisterSelectActivity"
    caps["appium:resetKeyboard"] = True
    caps["appium:ensureWebviewsHavePages"] = True
    caps["appium:nativeWebScreenshot"] = True
    caps["appium:newCommandTimeout"] = 3600
    caps["appium:connectHardwareKeyboard"] = True

    driver = webdriver.Remote("http://127.0.0.1:4723/wd/hub", caps)
    # 不要混合使用隐式和显式等待。这样做会导致不可预测的等待时间。例如，将隐式等待设置为10秒，将显式等待设置为15秒， 可能会导致在20秒后发生超时。
    # driver.implicitly_wait(2)


def start():
    init()
    login()


def alarm():
    if sys.platform.startswith('win'):
        pass  # winsound.Beep(1000, 3000)
    elif sys.platform.startswith('darwin'):
        os.system("say --voice=\"Mei-Jia\" 出错了，请尽快处理")


def getForkRange():
    if targets:
        return targets
    else:
        return range(startId, endId + 1)


def login():
    if loginType == 1:
        login_by_card()
    elif loginType == 2:
        login_by_phone()
    else:
        login_by_card()


def login_success():
    # 弹出了验证窗口，请在20s内通过验证
    entrance_xpath = "/hierarchy/android.widget.FrameLayout/android.widget.FrameLayout/" \
                     "android.widget.LinearLayout/android.widget.FrameLayout/android.widget.LinearLayout/" \
                     "android.widget.FrameLayout/android.view.ViewGroup/androidx.viewpager.widget.ViewPager/" \
                     "android.view.ViewGroup/android.widget.LinearLayout/android.widget.FrameLayout/" \
                     "android.view.ViewGroup[1]/android.view.ViewGroup/android.view.ViewGroup[2]/" \
                     "androidx.viewpager.widget.ViewPager/androidx.recyclerview.widget.RecyclerView/" \
                     "android.view.ViewGroup[1]"
    modal_id = "com.eagersoft.youzy.youzy:id/click_close_dialog_activity_pop"
    try:
        def load(x):
            modal = find_ele_by_id(modal_id)
            if modal is not None:
                modal.click()
                return True
            else:
                if find_ele_by_xpath(entrance_xpath) is not None:
                    return True
                else:
                    return False
            return False
    except:
        print('超时了')
        sys.exit()

    alarm()
    WebDriverWait(driver, 120).until(load)
    # 登录成功并且关闭了弹窗
    try:
        # 进入学校列表
        entrance = find_ele_by_xpath(entrance_xpath)
        entrance.click()
        # 开始工作
        global_begin_time = time.time()
        college_range = getForkRange()
        for i in college_range:
            if i in ignores: continue
            current_no = college_range.index(i) + 1
            time_diff = time.time() - global_begin_time
            print('进度统计:', current_no, 'OF', len(college_range), 'IN', time_diff / 3600, 'Hours')
            if not stop:
                auto_click(i)
            else:
                print('结束')
                return
            time_diff = time.time() - global_begin_time
            current_speed1 = time_diff / current_no
            current_speed2 = current_no * 3600 / time_diff
            print('速率统计:', current_speed1, 'Seconds of Each.', current_speed2, 'Completed Per Hour.')
        # auto_click(startId)
    except Exception as e:
        print('错误：', repr(e))


def login_by_card():
    el1 = find_ele_by_id("com.eagersoft.youzy.youzy:id/click_login_register_card")
    el1.click()
    el3 = find_ele_by_id("com.eagersoft.youzy.youzy:id/cardNumber")
    el3.send_keys(card_id)
    el4 = find_ele_by_id("com.eagersoft.youzy.youzy:id/password")
    el4.send_keys(card_pwd)
    el5 = find_ele_by_id("com.eagersoft.youzy.youzy:id/checkbox")
    el5.click()
    el6 = find_ele_by_id("com.eagersoft.youzy.youzy:id/click_next")
    el6.click()

    login_success()


def login_by_phone():
    el2 = driver.find_element(by=AppiumBy.ID, value="com.eagersoft.youzy.youzy:id/click_login_register_mobile")
    el2.click()
    time.sleep(1)
    el3 = driver.find_element(by=AppiumBy.ID, value="com.eagersoft.youzy.youzy:id/click_login_by_password")
    el3.click()
    time.sleep(0.5)
    el4 = driver.find_element(by=AppiumBy.ID, value="com.eagersoft.youzy.youzy:id/account")
    el4.send_keys(phone)
    time.sleep(1)
    el5 = driver.find_element(by=AppiumBy.ID, value="com.eagersoft.youzy.youzy:id/password")
    el5.send_keys(phone_pwd)
    time.sleep(1)
    el6 = driver.find_element(by=AppiumBy.ID, value="com.eagersoft.youzy.youzy:id/checkbox")
    el6.click()
    time.sleep(1)
    el7 = driver.find_element(by=AppiumBy.ID, value="com.eagersoft.youzy.youzy:id/click_login_register_mobile")
    el7.click()
    time.sleep(1)
    login_success()


def change_tab(tabIndex=1) -> bool:
    print('切换tab')
    # 540 740 1213
    # 招生计划
    el_plan = find_ele_by_xpath(
        "/hierarchy/android.widget.FrameLayout/android.widget.FrameLayout/android.widget.LinearLayout/"
        "android.widget.FrameLayout/android.widget.LinearLayout/android.widget.FrameLayout/"
        "android.widget.RelativeLayout/android.view.ViewGroup/android.view.ViewGroup[1]/"
        "android.widget.LinearLayout/android.widget.HorizontalScrollView/"
        "android.widget.LinearLayout/androidx.appcompat.app.ActionBar.Tab/"
        "android.view.ViewGroup/android.widget.TextView[contains(@text,'招生计划')]", 5)
    # print(el_plan)
    print(el_plan)
    change_success = False
    if tabIndex == 1:
        if el_plan is not None:
            el_plan.click()
            change_success = True
    elif tabIndex == 2:
        try:
            # 历年录取
            el_enter = find_ele_by_xpath(
                "/hierarchy/android.widget.FrameLayout/android.widget.FrameLayout/android.widget.LinearLayout/"
                "android.widget.FrameLayout/android.widget.LinearLayout/android.widget.FrameLayout/"
                "android.widget.RelativeLayout/android.view.ViewGroup/android.view.ViewGroup[1]/"
                "android.widget.LinearLayout/android.widget.HorizontalScrollView/"
                "android.widget.LinearLayout/androidx.appcompat.app.ActionBar.Tab/android.view.ViewGroup/"
                "android.widget.TextView[contains(@text,'历年录取')]", 5)
            # print(el_enter)
            if el_enter is not None:
                el_enter.click()
                change_success = True
        except:
            if el_plan is not None:
                el_plan.click()
            # 历年录取
            # time.sleep(1)  # tab会横向滑动，稍等会
            el_enter = find_ele_by_xpath(
                "/hierarchy/android.widget.FrameLayout/android.widget.FrameLayout/android.widget.LinearLayout/"
                "android.widget.FrameLayout/android.widget.LinearLayout/android.widget.FrameLayout/"
                "android.widget.RelativeLayout/android.view.ViewGroup/android.view.ViewGroup[1]/"
                "android.widget.LinearLayout/android.widget.HorizontalScrollView/"
                "android.widget.LinearLayout/androidx.appcompat.app.ActionBar.Tab/android.view.ViewGroup/"
                "android.widget.TextView[contains(@text,'历年录取')]")
            # print(el_enter)
            if el_enter is not None:
                el_enter.click()
                change_success = True
    else:
        if el_plan is not None:
            el_plan.click()
            change_success = True
    return change_success


def change_time(count: int = 2):
    print('切换年份')
    tv_year = find_ele_by_id("com.eagersoft.youzy.youzy:id/tv_year")
    tv_year.click()
    el_year = find_ele_by_xpath(f"/hierarchy/android.widget.FrameLayout/android.widget"
                                f".LinearLayout/android.widget.FrameLayout/android.widget"
                                f".FrameLayout/android.widget.FrameLayout/android.widget"
                                f".FrameLayout/android.view.ViewGroup/android.widget"
                                f".FrameLayout/android.view.ViewGroup/android.view."
                                f"ViewGroup/android.widget.ScrollView/android.widget"
                                f".LinearLayout/android.view.ViewGroup/android.widget"
                                f".LinearLayout/androidx.recyclerview.widget"
                                f".RecyclerView/android.view.ViewGroup[{count}]/android.widget.TextView")
    el_year.click()


def open_profession():
    print('切换专业分数线')
    el_switch = find_ele_by_xpath(
        "/hierarchy/android.widget.FrameLayout/android.widget.FrameLayout/android.widget.LinearLayout/"
        "android.widget.FrameLayout/android.widget.LinearLayout/android.widget.FrameLayout/"
        "android.widget.RelativeLayout/android.view.ViewGroup/android.view.ViewGroup[1]/"
        "androidx.viewpager.widget.ViewPager/android.view.ViewGroup/android.widget.ScrollView/"
        "android.view.ViewGroup/android.widget.LinearLayout/android.view.View")
    el_switch_mid_x = el_switch.location['x'] + el_switch.size['width'] / 2
    el_switch_mid_y = el_switch.location['y'] + el_switch.size['height'] / 2
    print(el_switch, el_switch_mid_x, el_switch_mid_y)
    actions = ActionChains(driver)
    actions.w3c_actions = ActionBuilder(driver, mouse=PointerInput(interaction.POINTER_TOUCH, "touch"))
    actions.w3c_actions.pointer_action.move_to_location(el_switch_mid_x, el_switch_mid_y)
    actions.w3c_actions.pointer_action.pointer_down()
    actions.w3c_actions.pointer_action.pause(0.1)
    actions.w3c_actions.pointer_action.release()
    actions.perform()


def change_batch(count):
    # 3个科目都点击一次
    print('打开批次面板', count)
    for index in range(count):
        # el_dropdown = find_ele_by_id("com.eagersoft.youzy.youzy:id/ll_select_college_batch")
        el_dropdown = WebDriverWait(driver, 5, 0.5).until(expected_conditions.presence_of_element_located(
            (AppiumBy.ID, "com.eagersoft.youzy.youzy:id/ll_select_college_batch")))
        if el_dropdown is not None:
            el_dropdown.click()
        else:
            # 表示元素有，但不可见，需要滚动至可见
            driver.execute_script("arguments[0].scrollIntoView();", el_dropdown)
            el_dropdown.click()
        batch_item_path = f"/hierarchy/android.widget.FrameLayout/android.widget.LinearLayout/android.widget" \
                          f".FrameLayout/android.widget.FrameLayout/android.widget.FrameLayout/android.widget" \
                          f".FrameLayout/android.view.ViewGroup/android.widget.FrameLayout/android.view" \
                          f".ViewGroup/android.view.ViewGroup/android.widget.ScrollView/android.widget" \
                          f".LinearLayout/android.view.ViewGroup/android.widget.LinearLayout/androidx" \
                          f".recyclerview.widget.RecyclerView/android.view.ViewGroup[{index + 1}]"
        try:
            el_batch_item = find_ele_by_xpath(batch_item_path)
            el_batch_item.click()
        except Exception as ex:
            print('切换批次异常，索引', index, '乎略异常', ex)
            el_ddl_close = find_ele_by_id('com.eagersoft.youzy.youzy:id/iv_close')
            el_ddl_close.click()
        randomSleep()


def cicle():
    change_batch(batch_count)
    # time.sleep(0.5)
    # 切换科目
    subtype_tab = find_ele_by_id("com.eagersoft.youzy.youzy:id/tv_subject_type")
    if subtype_tab is not None:
        subtype_tab.click()
    # time.sleep(0.5)

    subject_index = 1
    if subject_default_top: subject_index = 2
    subject_path = f"/hierarchy/android.widget.FrameLayout/android.widget.LinearLayout/android.widget" \
                   f".FrameLayout/android.widget.FrameLayout/android.widget.FrameLayout/android.widget" \
                   f".FrameLayout/android.view.ViewGroup/android.widget.FrameLayout/android.view.ViewGroup/android" \
                   f".view.ViewGroup/android.widget.ScrollView/android.widget.LinearLayout/android.view" \
                   f".ViewGroup/android.widget.LinearLayout/androidx.recyclerview.widget.RecyclerView/android" \
                   f".view.ViewGroup[{subject_index}]"
    el29 = find_ele_by_xpath(subject_path)
    if el29 is not None:
        el29.click()
    # time.sleep(0.5)
    change_batch(batch_count - 1)
    # time.sleep(0.5)


def get_data(tab_index, change_years):
    if change_tab(tab_index):
        # time.sleep(1)
        if tab_index == 2:
            open_profession()
            # time.sleep(1)
        if change_years[0] == 1: # 如果第 1 个是默认年份，直接开始批次切换
            cicle()
        for year in change_years:
            if year == 1:
                cicle()
            else:
                # 从第二个年份开始切换
                change_time(year)
                cicle()
    else:
        raise Exception("切换tab失败")


def is_error() -> bool:
    try:
        return (WebDriverWait(driver, 0.5, 0.25).until(expected_conditions.visibility_of_element_located(
            (AppiumBy.ID, "com.eagersoft.youzy.youzy:id/errorViewRelativeLayout"))) or
                WebDriverWait(driver, 0.5, 0.25).until(
                    expected_conditions.visibility_of_element_located(
                        (AppiumBy.ID, "com.eagersoft.youzy.youzy:id/load_more_load_fail_view"))))
    except:
        return False


def retry():
    try:
        retry_button = WebDriverWait(driver, 2, 0.5).until(
            expected_conditions.visibility_of_element_located(
                (AppiumBy.ID, "com.eagersoft.youzy.youzy:id/errorStateButton")))
        if retry_button is not None:
            retry_button.click()
        else:
            retry_button = WebDriverWait(driver, 2, 0.5).until(
                expected_conditions.visibility_of_element_located(
                    (AppiumBy.ID, "com.eagersoft.youzy.youzy:id/tv_prompt")))
            if retry_button is not None:
                retry_button.click()
    except Exception as ex:
        print('retry ignored', ex)


def is_loading() -> bool:
    try:
        return (WebDriverWait(driver, 0.5, 0.25).until(expected_conditions.visibility_of_element_located(
            (AppiumBy.ID, "com.eagersoft.youzy.youzy:id/image_loading"))) or
                WebDriverWait(driver, 0.5, 0.25).until(expected_conditions.visibility_of_element_located(
                    (AppiumBy.ID, "com.eagersoft.youzy.youzy:id/load_more_load_end_view"))))
    except:
        return False


def find_ele_by_id(ele_id: str, timeout: int = 2, poll_frequency: int = 0.5, deep: int = 0):
    loading = False
    error = False
    try:
        def find(x):
            nonlocal loading
            nonlocal error
            if is_error():
                print("error by id", x)
                error = True
                loading = False
                return False
            if is_loading():
                print("loading")
                loading = True
                error = False
                return False
            return x.find_element(by=AppiumBy.ID, value=ele_id) is not None

        WebDriverWait(driver, timeout, poll_frequency).until(find)
        return driver.find_element(by=AppiumBy.ID, value=ele_id)
    except:
        if loading:
            print("新一轮loading", deep)
            if retry_times_max > 0 and deep > 0 and deep % retry_times_max == 0:
                try:
                    print('重新触发查询', deep, '%', retry_times_max)
                    btn_search = driver.find_element(by=AppiumBy.ID, value="com.eagersoft.youzy.youzy:id/tv_search")
                    btn_search.click()
                except Exception as e:
                    print('重新触发查询 except', e)
            return find_ele_by_id(ele_id, timeout, poll_frequency, deep + 1)
        elif error:
            print("error,重试")
            retry()
            return find_ele_by_id(ele_id, timeout * 2, poll_frequency, deep + 1)
        else:
            return None


def find_ele_by_xpath(ele_xpath: str, timeout: int = 2, poll_frequency: int = 0.5):
    loading = False
    error = False
    try:
        def find(x):
            nonlocal loading
            nonlocal error
            if is_error():
                print("error by path", x)
                error = True
                loading = False
                return False
            if is_loading():
                print("loading")
                loading = True
                error = False
                return False
            return x.find_element(by=AppiumBy.XPATH, value=ele_xpath)

        return WebDriverWait(driver, timeout, poll_frequency).until(find)
    except:
        if loading:
            print("新一轮，loading")
            return find_ele_by_xpath(ele_xpath, timeout, poll_frequency)
        elif error:
            print("新一轮，重试")
            retry()
            return find_ele_by_xpath(ele_xpath, timeout, poll_frequency)
        else:
            return None


def find_college(college_id: Union[str, int]) -> Union[bool, 'WebElement']:
    el23 = find_ele_by_id("com.eagersoft.youzy.youzy:id/et_input")
    el23.send_keys(college_id)
    el24 = find_ele_by_id("com.eagersoft.youzy.youzy:id/tv_search")
    el24.click()
    empty_container = find_ele_by_id("com.eagersoft.youzy.youzy:id/emptyViewRelativeLayout", timeout=2)
    if empty_container is not None:
        return False
    else:
        college = find_ele_by_id("com.eagersoft.youzy.youzy:id/cl_parent")
        if college is not None:
            return college
    return False


def back():
    el_back = find_ele_by_id("com.eagersoft.youzy.youzy:id/iv_back")
    if el_back is not None:
        print("返回页面")
        el_back.click()


def auto_click(id):
    global hide_guide
    global stop
    global retry_current
    global dataType
    global loginType
    print('当前id：', id, datetime.now())
    # 开始搜索
    el_search = find_ele_by_id("com.eagersoft.youzy.youzy:id/click_search")
    if el_search is not None:
        el_search.click()
    college = find_college(id)
    if isinstance(college, bool):
        return
    else:
        college.click()
        retry_current = 0
        try:
            if not hide_guide:
                print('等待关闭引导')
                time.sleep(6)
                # actions = ActionChains(driver)
                # actions.w3c_actions = ActionBuilder(driver, mouse=PointerInput(interaction.POINTER_TOUCH, "touch"))
                # actions.w3c_actions.pointer_action.move_to_location(507, 1537)
                # actions.w3c_actions.pointer_action.pointer_down()
                # actions.w3c_actions.pointer_action.pause(0.1)
                # actions.w3c_actions.pointer_action.release()
                # actions.perform()
                hide_guide = True

            # time.sleep(1)
            if dataType == 1:
                get_data(1, plan_change_years)
            elif dataType == 2:
                get_data(2, profession_change_years)
            elif dataType == 3:
                get_data(1, plan_change_years)
                get_data(2, profession_change_years)
            # 回退两次，清除学校列表缓存
            back()
            back()
        except Exception as ex:
            print('get_data exception', ex)
            alarm()
            back()
            back()
            auto_click(id)


if __name__ == '__main__':
    try:
        args = sys.argv[1:]
        longopts = ['loginType=', 'dataType=']
        shortopts = []
        options, args = getopt.getopt(args, shortopts, longopts)
        for name, value in options:
            print(name, value)
            if name in ('--loginType'):
                loginType = int(value)
            if name in '--dataType':
                dataType = int(value)
    except:
        print('运行参数解析错误')
    if dataType == 1:
        print('招生计划')
    elif dataType == 2:
        print('历年录取')
    else:
        print('招生计划和历年录取')
    start()