init py of yzy

main.py

@@ -0,0 +1,567 @@
+# 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 = 10001
+# 结束id
+endId = 11500
+# ignores
+ignores = []
+# indicate, higher priority than start-end for no-seq targets
+targets = []
+
+card_id = "16416117"
+card_pwd = "396254"
+
+phone = '13974787400'
+phone_pwd = 'Xj654123'
+
+hide_guide = False
+stop = False
+retry_times_max = 1  # 搜索loading时，间隔几次重新点击搜索
+retry_current = 0  # 暂时没有用途了
+batch_count = 2  # 有几个批次
+subject_default_top = True  # 默认理科、物理
+plan_change_years = [2,3]  # 计划取几年？从 2 开始
+profession_change_years = [1,2,3,4]  # 专业取几年？从 2 开始
+# 登录方式 1 卡号 2 手机密码
+loginType = 1
+# 数据类型 1 招生计划 2 历年录取 3 两者都有
+dataType = 1
+
+
+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:
+            # 从第二个年份开始切换
+            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()