# 测量标签栏页面操作 # page_objects/measure_tabbar_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, StaleElementReferenceException import logging import time import globals.ids as ids import globals.global_variable as global_variable # 导入全局变量模块 from globals.driver_utils import check_session_valid, reconnect_driver, go_main_click_tabber_button # 导入会话检查和重连函数 # import globals.driver_utils as driver_utils # 导入全局变量模块 class MeasureTabbarPage: def __init__(self, driver, wait,device_id): self.driver = driver self.wait = wait self.logger = logging.getLogger(__name__) self.seen_items = set() # 记录已经看到的项目，用于检测是否滚动到底部 self.all_items = set() # 记录所有看到的项目，用于检测是否已经查看过所有项目 # 获取设备ID用于重连操作 self.device_id = device_id def is_measure_tabbar_visible(self): """文件列表是否可见""" try: return self.driver.find_element(AppiumBy.ID, ids.MEASURE_LIST_ID).is_displayed() except NoSuchElementException: self.logger.warning("文件列表未找到") return False except Exception as e: self.logger.error(f"文件列表可见性时发生意外错误: {str(e)}") return False def click_measure_tabbar(self): """点击测量标签栏""" try: measure_tab = self.wait.until( EC.element_to_be_clickable((AppiumBy.ID, ids.MEASURE_TABBAR_ID)) ) measure_tab.click() self.logger.info("已点击测量标签栏") time.sleep(1) # 等待测量页面加载完成 self.wait.until( lambda driver: self.is_measure_tabbar_visible() ) return True except TimeoutException: self.logger.error("等待测量标签栏可点击超时") return False except Exception as e: self.logger.error(f"点击测量标签栏时出错: {str(e)}") return False # def scroll_list(self): # """滑动列表以加载更多项目""" # try: # # 获取列表容器 # list_container = self.driver.find_element(AppiumBy.ID, ids.MEASURE_LIST_ID) # # 计算滑动坐标 # start_x = list_container.location['x'] + list_container.size['width'] // 2 # start_y = list_container.location['y'] + list_container.size['height'] * 0.8 # end_y = list_container.location['y'] + list_container.size['height'] * 0.2 # # 执行滑动 # self.driver.swipe(start_x, start_y, start_x, end_y, 1000) # self.logger.info("已滑动列表") # # 等待新内容加载 # time.sleep(2) # return True # except Exception as e: # self.logger.error(f"滑动列表失败: {str(e)}") # return False def scroll_list(self, direction="down"): """滑动列表以加载更多项目 Args: direction: 滑动方向，"down"表示向下滑动，"up"表示向上滑动 Returns: bool: 滑动是否成功执行，对于向上滑动，如果滑动到顶则返回False """ max_retry = 2 retry_count = 0 while retry_count <= max_retry: # 检查会话是否有效 if not check_session_valid(self.driver, self.device_id): self.logger.warning(f"会话无效，尝试重新连接... (尝试 {retry_count + 1}/{max_retry})") try: # 尝试重新连接驱动 self.driver, self.wait = reconnect_driver(self.device_id, self.driver) self.logger.info("驱动重新连接成功") except Exception as reconnect_error: self.logger.error(f"驱动重新连接失败: {str(reconnect_error)}") retry_count += 1 if retry_count > max_retry: self.logger.error("达到最大重试次数，滑动列表失败") return False time.sleep(1) continue try: # 获取列表容器 list_container = self.driver.find_element(AppiumBy.ID, ids.MEASURE_LIST_ID) # 计算滑动坐标 start_x = list_container.location['x'] + list_container.size['width'] // 2 if direction == "down": # 向下滑动 start_y = list_container.location['y'] + list_container.size['height'] * 0.95 end_y = list_container.location['y'] + list_container.size['height'] * 0.05 else: # 向上滑动 # 记录滑动前的项目，用于判断是否滑动到顶 before_scroll_items = self.get_current_items() start_y = list_container.location['y'] + list_container.size['height'] * 0.05 end_y = list_container.location['y'] + list_container.size['height'] * 0.95 # 执行滑动 self.driver.swipe(start_x, start_y, start_x, end_y, 1000) # 等待新内容加载 time.sleep(1) # 向上滑动时，检查是否滑动到顶 if direction == "up": after_scroll_items = self.get_current_items() # 如果滑动后的项目与滑动前的项目相同，说明已经滑动到顶 if after_scroll_items == before_scroll_items: self.logger.info("已滑动到列表顶部，列表内容不变") return False return True except Exception as e: error_msg = str(e) self.logger.error(f"滑动列表失败: {error_msg}") # 如果是连接相关的错误，尝试重连 if any(keyword in error_msg.lower() for keyword in ['socket hang up', 'could not proxy command', 'session']): retry_count += 1 if retry_count > max_retry: self.logger.error("达到最大重试次数，滑动列表失败") return False self.logger.warning(f"尝试重新连接后重试... (尝试 {retry_count}/{max_retry})") time.sleep(1) else: # 非连接错误，直接返回False return False def get_current_items(self): """获取当前页面中的所有项目文本""" max_retry = 2 retry_count = 0 while retry_count <= max_retry: # 检查会话是否有效 if not check_session_valid(self.driver, self.device_id): self.logger.warning(f"会话无效，尝试重新连接... (尝试 {retry_count + 1}/{max_retry})") try: # 尝试重新连接驱动 self.driver, self.wait = reconnect_driver(self.device_id, self.driver) self.logger.info("驱动重新连接成功") except Exception as reconnect_error: self.logger.error(f"驱动重新连接失败: {str(reconnect_error)}") retry_count += 1 if retry_count > max_retry: self.logger.error("达到最大重试次数，获取当前项目失败") return [] time.sleep(0.1) # 等待1秒后重试 continue try: items = self.driver.find_elements(AppiumBy.ID, ids.MEASURE_LISTVIEW_ID) item_texts = [] for item in items: try: title_element = item.find_element(AppiumBy.ID, ids.MEASURE_NAME_TEXT_ID) if title_element and title_element.text: item_texts.append(title_element.text) except NoSuchElementException: continue except Exception as item_error: self.logger.warning(f"处理项目时出错: {str(item_error)}") continue return item_texts except Exception as e: error_msg = str(e) self.logger.error(f"获取当前项目失败: {error_msg}") # 如果是连接相关的错误，尝试重连 if any(keyword in error_msg.lower() for keyword in ['socket hang up', 'could not proxy command', 'session']): retry_count += 1 if retry_count > max_retry: self.logger.error("达到最大重试次数，获取当前项目失败") return [] self.logger.warning(f"尝试重新连接后重试... (尝试 {retry_count}/{max_retry})") time.sleep(1) else: # 非连接错误，直接返回空列表 return [] def click_item_by_text(self, text): """点击指定文本的项目""" max_retry = 2 retry_count = 0 while retry_count <= max_retry: # 检查会话是否有效 if not check_session_valid(self.driver, self.device_id): self.logger.warning(f"会话无效，尝试重新连接... (尝试 {retry_count + 1}/{max_retry})") try: # 尝试重新连接驱动 self.driver, self.wait = reconnect_driver(self.device_id, self.driver) self.logger.info("驱动重新连接成功") except Exception as reconnect_error: self.logger.error(f"驱动重新连接失败: {str(reconnect_error)}") retry_count += 1 if retry_count > max_retry: self.logger.error("达到最大重试次数，点击项目失败") return False time.sleep(1) continue try: # 查找包含指定文本的项目 items = self.driver.find_elements(AppiumBy.ID, ids.MEASURE_LISTVIEW_ID) for item in items: try: title_element = item.find_element(AppiumBy.ID, ids.MEASURE_NAME_TEXT_ID) if title_element and title_element.text == text: title_element.click() self.logger.info(f"已点击项目: {text}") return True except NoSuchElementException: continue except Exception as item_error: self.logger.warning(f"处理项目时出错: {str(item_error)}") continue self.logger.warning(f"未找到可点击的项目: {text}") return False except Exception as e: error_msg = str(e) self.logger.error(f"点击项目失败: {error_msg}") # 如果是连接相关的错误，尝试重连 if any(keyword in error_msg.lower() for keyword in ['socket hang up', 'could not proxy command', 'session']): retry_count += 1 if retry_count > max_retry: self.logger.error("达到最大重试次数，点击项目失败") return False self.logger.warning(f"尝试重新连接后重试... (尝试 {retry_count}/{max_retry})") time.sleep(1) else: # 非连接错误，直接返回False return False def find_keyword(self, fixed_filename): """查找指定关键词并点击，支持向下和向上滑动查找""" try: # 等待线路列表容器出现 self.wait.until( EC.presence_of_element_located((AppiumBy.ID, ids.MEASURE_LIST_ID)) ) # self.logger.info("线路列表容器已找到") max_scroll_attempts = 50 # 最大滚动尝试次数 scroll_count = 0 found_items_count = 0 # 记录已找到的项目数量 last_items_count = 0 # 记录上一次找到的项目数量 previous_items = set() # 记录前一次获取的项目集合，用于检测是否到达边界 # 首先尝试向下滑动查找 while scroll_count < max_scroll_attempts: # 获取当前页面中的所有项目 current_items = self.get_current_items() # self.logger.info(f"当前页面找到 {len(current_items)} 个项目: {current_items}") # 检查目标文件是否在当前页面中 if fixed_filename in current_items: # self.logger.info(f"找到目标文件: {fixed_filename}") # 点击目标文件 if self.click_item_by_text(fixed_filename): return True else: self.logger.error(f"点击目标文件失败: {fixed_filename}") return False # 检查是否到达底部：连续两次获取的项目相同 if current_items == previous_items and len(current_items) > 0: self.logger.info("连续两次获取的项目相同，已到达列表底部") break # 更新前一次项目集合 previous_items = current_items.copy() # # 记录所有看到的项目 # self.all_items.update(current_items) # # 检查是否已经查看过所有项目 # if len(current_items) > 0 and found_items_count == len(self.all_items): # self.logger.info("已向下查看所有项目，未找到目标文件") # break # # return False # found_items_count = len(self.all_items) # 向下滑动列表以加载更多项目 if not self.scroll_list(direction="down"): self.logger.error("向下滑动列表失败") return False scroll_count += 1 self.logger.info(f"第 {scroll_count} 次向下滑动，继续查找...") # 如果向下滑动未找到，尝试向上滑动查找 self.logger.info("向下滑动未找到目标，开始向上滑动查找") # 重置滚动计数 scroll_count = 0 while scroll_count < max_scroll_attempts: # 向上滑动列表 # 如果返回False，说明已经滑动到顶 if not self.scroll_list(direction="up"): # 检查是否是因为滑动到顶而返回False if "已滑动到列表顶部" in self.logger.handlers[0].buffer[-1].message: self.logger.info("已滑动到列表顶部，停止向上滑动") break else: self.logger.error("向上滑动列表失败") return False # 获取当前页面中的所有项目 current_items = self.get_current_items() # self.logger.info(f"向上滑动后找到 {len(current_items)} 个项目: {current_items}") # 检查目标文件是否在当前页面中 if fixed_filename in current_items: self.logger.info(f"找到目标文件: {fixed_filename}") # 点击目标文件 if self.click_item_by_text(fixed_filename): return True else: self.logger.error(f"点击目标文件失败: {fixed_filename}") return False scroll_count += 1 self.logger.info(f"第 {scroll_count} 次向上滑动，继续查找...") self.logger.warning(f"经过 {max_scroll_attempts * 2} 次滑动仍未找到目标文件") return False except TimeoutException: self.logger.error("等待线路列表元素超时") return False except Exception as e: self.logger.error(f"查找关键词时出错: {str(e)}") return False def measure_tabbar_page_manager(self): """执行测量操作""" try: # 跳转到测量页面 if not go_main_click_tabber_button(self.driver, self.device_id, "com.bjjw.cjgc:id/img_3_layout"): logging.error(f"设备 {self.device_id} 跳转到测量页面失败") return False # # 点击测量标签栏 # if not self.click_measure_tabbar(): # self.logger.error("点击测量标签栏失败") # return False # 固定文件名 fixed_filename = global_variable.GLOBAL_CURRENT_PROJECT_NAME self.logger.info(f"开始查找测量数据: {fixed_filename}") # 重置已看到的项目集合 self.seen_items = set() self.all_items = set() # 查找并点击测量数据 if self.find_keyword(fixed_filename): self.logger.info("成功找到并点击测量数据") return True else: self.logger.warning("未找到测量数据") return False except Exception as e: self.logger.error(f"执行测量操作时出错: {str(e)}") return False