# test_more_download_page.py from appium.webdriver.common.appiumby import AppiumBy from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from selenium.common.exceptions import TimeoutException, NoSuchElementException import logging import time from datetime import datetime from globals.driver_utils import launch_app_manually from page_objects.login_page import LoginPage class MoreDownloadPage: def __init__(self, driver, wait,device_id): self.driver = driver self.wait = wait self.device_id = device_id self.logger = logging.getLogger(__name__) def is_on_more_download_page(self): """通过下载历史数据按钮来判断是否在更多下载页面""" try: # 使用下载历史数据按钮的resource-id来检查 download_history_locator = (AppiumBy.ID, "com.bjjw.cjgc:id/download_history") self.wait.until(EC.presence_of_element_located(download_history_locator)) self.logger.info("已确认在更多下载页面") return True except TimeoutException: self.logger.warning("未找到下载历史数据按钮，不在更多下载页面") return False except Exception as e: self.logger.error(f"检查更多下载页面时发生意外错误: {str(e)}") return False def click_download_button(self): """点击下载按钮""" try: # 点击下载历史数据按钮 download_button = self.wait.until( EC.element_to_be_clickable((AppiumBy.ID, "com.bjjw.cjgc:id/download_history")) ) download_button.click() self.logger.info("已点击下载历史数据按钮") # 等待下载操作开始 # time.sleep(3) return True except TimeoutException: self.logger.error("等待下载按钮可点击超时") return False except Exception as e: self.logger.error(f"点击下载按钮时出错: {str(e)}") return False def click_download_original_data(self): """点击下载原始数据按钮并处理日期选择""" try: # 点击下载原始数据按钮 download_original_btn = self.wait.until( EC.element_to_be_clickable((AppiumBy.ID, "com.bjjw.cjgc:id/download_org")) ) download_original_btn.click() self.logger.info("已点击下载原始数据按钮") # 等待日期选择弹窗出现 # time.sleep(2) # 点击选择开始日期 start_date_btn = self.wait.until( EC.element_to_be_clickable((AppiumBy.ID, "com.bjjw.cjgc:id/date")) ) start_date_btn.click() self.logger.info("已点击选择开始日期") # 等待日期选择器出现 # time.sleep(2) # 滑动年份选择器 - 向上滑动1/5的距离 if not self._swipe_year_wheel(): self.logger.error("滑动年份选择器失败") return False # 点击日期选择器的确定按钮 confirm_btn = self.wait.until( EC.element_to_be_clickable((AppiumBy.ID, "com.bjjw.cjgc:id/okBtn")) ) confirm_btn.click() self.logger.info("已确认日期选择") # 等待日期选择器关闭 # time.sleep(2) # 假设弹窗有确定按钮，点击它开始下载 try: # 尝试查找并点击下载弹窗的确定按钮 download_confirm_btn = WebDriverWait(self.driver, 5).until( EC.element_to_be_clickable((AppiumBy.XPATH, "//android.widget.Button[contains(@text, '确定') or contains(@text, '下载')]")) ) download_confirm_btn.click() self.logger.info("已点击下载确认按钮") except TimeoutException: self.logger.warning("未找到下载确认按钮，可能不需要确认") # 等待下载开始 # time.sleep(3) return True except TimeoutException: self.logger.error("等待下载原始数据按钮可点击超时") return False except Exception as e: self.logger.error(f"点击下载原始数据时出错: {str(e)}") return False def click_download_result_data(self): """点击下载成果数据按钮并处理日期选择""" try: # 点击下载成果数据按钮 download_result_btn = self.wait.until( EC.element_to_be_clickable((AppiumBy.ID, "com.bjjw.cjgc:id/download_result")) ) download_result_btn.click() self.logger.info("已点击下载成果数据按钮") # 等待日期选择弹窗出现 # time.sleep(2) # 点击选择开始日期 start_date_btn = self.wait.until( EC.element_to_be_clickable((AppiumBy.ID, "com.bjjw.cjgc:id/date")) ) start_date_btn.click() self.logger.info("已点击选择开始日期") # 等待日期选择器出现 # time.sleep(2) # 滑动年份选择器 - 向上滑动1/5的距离 if not self._swipe_year_wheel(): self.logger.error("滑动年份选择器失败") return False # 点击日期选择器的确定按钮 confirm_btn = self.wait.until( EC.element_to_be_clickable((AppiumBy.ID, "com.bjjw.cjgc:id/okBtn")) ) confirm_btn.click() self.logger.info("已确认日期选择") # 等待日期选择器关闭 # time.sleep(2) # 假设弹窗有确定按钮，点击它开始下载 try: # 尝试查找并点击下载弹窗的确定按钮 download_confirm_btn = WebDriverWait(self.driver, 5).until( EC.element_to_be_clickable((AppiumBy.XPATH, "//android.widget.Button[contains(@text, '确定') or contains(@text, '下载')]")) ) download_confirm_btn.click() self.logger.info("已点击下载确认按钮") except TimeoutException: self.logger.warning("未找到下载确认按钮，可能不需要确认") # 等待下载开始 # time.sleep(3) return True except TimeoutException: self.logger.error("等待下载成果数据按钮可点击超时") return False except Exception as e: self.logger.error(f"点击下载成果数据时出错: {str(e)}") return False def _swipe_year_wheel(self): """滑动年份选择器的滚轮""" try: # # 获取年份选择器滚轮元素 # year_wheel = self.driver.find_element(AppiumBy.ID, "com.bjjw.cjgc:id/wheelView1") # 获取当前日期 current_date = datetime.now() current_month = current_date.month # 检查月份是否小于等于3 if current_month <= 3: self.logger.info(f"当前月份为{current_month}月，需要滑动年份选择器") # 获取年份选择器滚轮元素 year_wheel = self.driver.find_element(AppiumBy.ID, "com.bjjw.cjgc:id/wheelView1") # 获取滚轮的位置和尺寸 location = year_wheel.location size = year_wheel.size # 计算滚轮中心点坐标 center_x = location['x'] + size['width'] // 2 center_y = location['y'] + size['height'] // 2 # 计算滑动距离 - 滚轮高度的1/5 swipe_distance = size['height'] // 5 # 滑动一次年份选择器 # 执行滑动操作 - 从中心向上滑动1/5高度 self.driver.swipe(center_x, center_y - swipe_distance, center_x, center_y, 500) self.logger.info("已滑动一次年份选择器") else: self.logger.info(f"当前月份为{current_month}月，不需要滑动年份选择器") # 获取月份选择器滚轮元素 month_wheel = self.driver.find_element(AppiumBy.ID, "com.bjjw.cjgc:id/wheelView2") # 获取滚轮的位置和尺寸 location = month_wheel.location size = month_wheel.size # 计算滚轮中心点坐标 center_x = location['x'] + size['width'] // 2 center_y = location['y'] + size['height'] // 2 # 计算滑动距离 - 滚轮高度的1/5 swipe_distance = size['height'] // 5 for i in range(3): # 执行滑动操作 - 从中心向上滑动1/5高度 self.driver.swipe(center_x, center_y - swipe_distance, center_x, center_y, 500) self.logger.info(f"已{i}次滑动月份选择器") return True except Exception as e: self.logger.error(f"滑动年份选择器时出错: {str(e)}") return False # def wait_for_loading_dialog(self, timeout=900, download_type="unknown"): # """ # 检查特定结构的加载弹窗的出现和消失 # 参数: # timeout: 最大等待时间，默认10分钟(600秒) # 返回: # bool: 如果加载弹窗出现并消失返回True，否则返回False # """ # try: # self.logger.info(f"开始检查{download_type}加载弹窗...") # # 首先检查加载弹窗是否出现 # start_time = time.time() # loading_appeared = False # # 等待加载弹窗出现（最多等待30秒） # while time.time() - start_time < 30: # try: # # 根据提供的结构查找加载弹窗 # # 查找包含ProgressBar和"loading..."文本的弹窗 # loading_indicators = [ # (AppiumBy.XPATH, "//android.widget.FrameLayout[@resource-id='android:id/content']/android.widget.LinearLayout[@resource-id='android:id/parentPanel']//android.widget.ProgressBar"), # (AppiumBy.XPATH, "//android.widget.TextView[@resource-id='android:id/message' and @text='loading...']"), # (AppiumBy.XPATH, "//android.widget.FrameLayout[@resource-id='android:id/content']//android.widget.ProgressBar"), # (AppiumBy.XPATH, "//*[contains(@text, 'loading...')]") # ] # for by, value in loading_indicators: # try: # element = self.driver.find_element(by, value) # if element.is_displayed(): # loading_appeared = True # self.logger.info("数据下载已开始") # self.logger.info("检测到加载弹窗出现") # break # except: # continue # if loading_appeared: # break # except Exception as e: # pass # time.sleep(1) # # 如果加载弹窗没有出现，直接返回True # if not loading_appeared: # self.logger.info("未检测到加载弹窗，继续执行") # return True # # 等待加载弹窗消失 # self.logger.info("等待加载弹窗消失...") # disappearance_start_time = time.time() # while time.time() - disappearance_start_time < timeout: # try: # # 检查加载弹窗是否还存在 # loading_still_exists = False # for by, value in loading_indicators: # try: # element = self.driver.find_element(by, value) # if element.is_displayed(): # loading_still_exists = True # break # except: # continue # if not loading_still_exists: # self.logger.info("加载弹窗已消失") # return True # # 每1分钟记录一次状态 # if int(time.time() - disappearance_start_time) % 60 == 0: # elapsed_time = int(time.time() - disappearance_start_time) # self.logger.info(f"加载弹窗仍在显示，已等待{elapsed_time//60}分钟") # except Exception as e: # # 如果出现异常，可能弹窗已经消失 # self.logger.info("加载弹窗可能已消失") # return True # time.sleep(1) # # 如果超时，记录错误并返回False # self.logger.error(f"加载弹窗在{timeout}秒后仍未消失") # # return False # # 检查是否有loading加载窗口 # try: # self.logger.info(f"检查{download_type}下载是否有loading加载窗口...") # loading_indicators = [ # (AppiumBy.XPATH, "//android.widget.FrameLayout[@resource-id='android:id/content']/android.widget.LinearLayout[@resource-id='android:id/parentPanel']//android.widget.ProgressBar"), # (AppiumBy.XPATH, "//android.widget.TextView[@resource-id='android:id/message' and @text='loading...']"), # (AppiumBy.XPATH, "//android.widget.FrameLayout[@resource-id='android:id/content']//android.widget.ProgressBar"), # (AppiumBy.XPATH, "//*[contains(@text, 'loading...')]") # ] # loading_exists = False # for by, value in loading_indicators: # try: # element = self.driver.find_element(by, value) # if element.is_displayed(): # loading_exists = True # break # except: # continue # if loading_exists: # self.logger.info(f"检测到{download_type}下载的loading加载窗口，执行重新打开应用操作") # # 手动启动应用 # launch_app_manually(self.driver, self.device_id, "com.bjjw.cjgc", ".activity.LoginActivity") # self.logger.info("已重新启动沉降观测应用") # # 点击登录 # login_page = LoginPage(self.driver, self.wait) # if login_page.is_login_page(): # if login_page.login(): # self.logger.info("登录成功") # else: # self.logger.error("登录失败") # return False # else: # self.logger.info("应用已登录，无需重复登录") # self.logger.info(f"{download_type}下载超时处理完成，应用已重启") # # 点击img_5_layout（更多下载按钮） # more_download_btn = self.wait.until( # EC.element_to_be_clickable((AppiumBy.ID, "com.bjjw.cjgc:id/img_5_layout")) # ) # more_download_btn.click() # self.logger.info("已点击更多下载按钮") # self.more_download_page_manager_2(download_type) # # 等待页面加载 # time.sleep(1) # else: # self.logger.info(f"未检测到{download_type}下载的loading加载窗口") # except Exception as e: # self.logger.error(f"检查{download_type}下载的loading加载窗口时出错: {str(e)}") # # 出错时继续执行，不影响主流程 # return False # except Exception as e: # self.logger.error(f"检查加载弹窗时出错: {str(e)}") # return False def _is_loading_present(self): """私有辅助方法：检测当前页面是否存在加载弹窗""" loading_indicators = [ (AppiumBy.XPATH, "//android.widget.FrameLayout[@resource-id='android:id/content']//android.widget.ProgressBar"), (AppiumBy.XPATH, "//android.widget.TextView[@resource-id='android:id/message' and @text='loading...']"), (AppiumBy.XPATH, "//*[contains(@text, 'loading...')]") ] for by, value in loading_indicators: try: element = self.driver.find_element(by, value) if element.is_displayed(): return True except: continue return False def wait_for_loading_dialog(self, timeout=1200, download_type="unknown", retry_count=0): """ 检查加载弹窗的出现和消失，支持最多1次重试（总共执行2次） """ try: self.logger.info(f"开始检查 {download_type} 加载弹窗 (尝试次数: {retry_count + 1})...") # 1. 等待加载弹窗出现（最多30秒） start_time = time.time() loading_appeared = False while time.time() - start_time < 30: if self._is_loading_present(): loading_appeared = True self.logger.info(f"检测到 {download_type} 加载弹窗出现") break time.sleep(1) if not loading_appeared: self.logger.info(f"未检测到 {download_type} 加载弹窗，视为直接通过") return True # 2. 等待加载弹窗消失 disappearance_start_time = time.time() while time.time() - disappearance_start_time < timeout: if not self._is_loading_present(): self.logger.info(f"{download_type} 加载弹窗已消失") return True # 每60秒打印一次日志 if int(time.time() - disappearance_start_time) % 60 == 0: self.logger.info(f"等待中...已耗时 {int(time.time() - disappearance_start_time)//60} 分钟") time.sleep(2) # 3. 超时处理逻辑 self.logger.error(f"{download_type} 加载弹窗在 {timeout} 秒后仍未消失") # 检查是否还可以重试（retry_count=0 时执行重试，即第二次执行） if retry_count < 1: self.logger.warning(f"检测到超时，准备进行第 {retry_count + 2} 次尝试（重启应用）...") # 执行重启逻辑 from globals.driver_utils import launch_app_manually from page_objects.login_page import LoginPage launch_app_manually(self.driver, self.device_id, "com.bjjw.cjgc", ".activity.LoginActivity") login_page = LoginPage(self.driver, self.wait) if login_page.is_login_page(): login_page.login() # 重新导航到下载页面 more_download_btn = self.wait.until( EC.element_to_be_clickable((AppiumBy.ID, "com.bjjw.cjgc:id/img_5_layout")) ) more_download_btn.click() # 递归调用：增加 retry_count return self.more_download_page_manager_2(download_type, retry_count = retry_count + 1) else: self.logger.error(f"{download_type} 已达到最大重试次数，操作失败") return False except Exception as e: self.logger.error(f"检查加载弹窗时发生异常: {str(e)}") return False def more_download_page_manager_2(self, download_type, retry_count=0): """ 修改后的 manager_2，透传 retry_count 参数 """ try: if not self.is_on_more_download_page(): return False self.click_download_button() if download_type in ["原始数据", "历史数据"]: self.click_download_original_data() return self.wait_for_loading_dialog(download_type="原始数据", retry_count=retry_count) elif download_type == "成果数据": self.click_download_result_data() return self.wait_for_loading_dialog(download_type="成果数据", retry_count=retry_count) return False except Exception as e: self.logger.error(f"manager_2 执行出错: {str(e)}") return False def more_download_page_manager(self): """执行更多下载页面管理操作""" try: self.logger.info("开始执行更多下载页面操作") # 检查是否在更多下载页面 if not self.is_on_more_download_page(): self.logger.error("不在更多下载页面") return False # 点击下载历史数据按钮 if not self.click_download_button(): self.logger.error("点击下载历史数据按钮失败") return False # 等待下载历史数据页面加载完成 # time.sleep(3) # 点击下载原始数据按钮 if not self.click_download_original_data(): self.logger.error("点击下载原始数据按钮失败") return False # 等待下载操作完成 time.sleep(1) # 使用wait_for_loading_dialog函数等待下载过程中的加载弹窗消失 if not self.wait_for_loading_dialog(download_type="原始数据"): self.logger.warning("下载过程中的加载弹窗未在预期时间内消失，但操作已完成") # 等待一段时间，确保原始数据下载完成 time.sleep(1) # 点击下载成果数据按钮 if not self.click_download_result_data(): self.logger.error("点击下载成果数据按钮失败") return False # 使用wait_for_loading_dialog函数等待下载过程中的加载弹窗消失 if not self.wait_for_loading_dialog(download_type="成果数据"): self.logger.warning("成果数据下载过程中的加载弹窗未在预期时间内消失，但操作已完成") self.logger.info("更多下载页面操作执行完成") return True except Exception as e: self.logger.error(f"执行更多下载页面操作时出错: {str(e)}") return False