截图平差未测试
This commit is contained in:
618
check_station.py
618
check_station.py
@@ -257,6 +257,21 @@ class CheckStation:
|
||||
# 错误处理,可以继续循环或退出
|
||||
print(f"已处理{over_station_num}个站点")
|
||||
|
||||
# 滑动列表到底部
|
||||
if not self.scroll_list_to_bottom(device_id):
|
||||
logging.error(f"设备 {device_id} 下滑列表到底部失败")
|
||||
return False
|
||||
|
||||
# 2. 点击最后一个spinner
|
||||
if not self.click_last_spinner_with_retry(device_id):
|
||||
logging.error(f"设备 {device_id} 点击最后一个spinner失败")
|
||||
return False
|
||||
|
||||
# 3. 再下滑一次
|
||||
if not self.scroll_down_once(device_id):
|
||||
logging.warning(f"设备 {device_id} 再次下滑失败,但继续执行")
|
||||
|
||||
|
||||
# # 截图
|
||||
# self.driver.save_screenshot("check_station.png")
|
||||
if not self.take_screenshot():
|
||||
@@ -264,8 +279,26 @@ class CheckStation:
|
||||
|
||||
# 打完数据,截图完毕,点击平差处理按钮
|
||||
if not self.click_adjustment_button(device_id):
|
||||
self.logger.error(f"设备 {device_id} 点击平差处理按钮失败")
|
||||
logging.error(f"设备 {device_id} 点击平差处理按钮失败")
|
||||
return False
|
||||
|
||||
# 检查是否在测量页面,在就重新执行选择断点,滑动列表到底部,点击最后一个spinner, 再下滑一次,点击平差处理按钮平差
|
||||
if not self.handle_back_navigation(item, device_id):
|
||||
logging.error(f"{item}平差失败")
|
||||
|
||||
|
||||
# 检测并处理"是 保留成果"弹窗
|
||||
if not self.handle_adjustment_result_dialog():
|
||||
logging.error("处理平差结果弹窗失败")
|
||||
|
||||
|
||||
# 检查在不在测量列表页面,不在就点击返回按钮并处理弹窗
|
||||
if self.check_measurement_list(device_id):
|
||||
logging.error(f"设备 {device_id} 未在测量列表页面")
|
||||
# 点击返回按钮并处理弹窗
|
||||
if not self.execute_back_navigation_steps(device_id):
|
||||
logging.error(f"设备 {device_id} 处理返回按钮确认失败")
|
||||
|
||||
return True
|
||||
|
||||
def click_adjustment_button(self, device_id):
|
||||
@@ -279,7 +312,7 @@ class CheckStation:
|
||||
bool: 是否成功点击
|
||||
"""
|
||||
try:
|
||||
self.logger.info(f"设备 {device_id} 查找平差处理按钮")
|
||||
logging.info(f"设备 {device_id} 查找平差处理按钮")
|
||||
|
||||
# 查找平差处理按钮
|
||||
adjustment_button = self.driver.find_element(AppiumBy.ID, "com.bjjw.cjgc:id/point_measure_btn")
|
||||
@@ -287,24 +320,595 @@ class CheckStation:
|
||||
# 验证按钮文本
|
||||
button_text = adjustment_button.text
|
||||
if "平差处理" not in button_text:
|
||||
self.logger.warning(f"设备 {device_id} 按钮文本不匹配,期望'平差处理',实际: {button_text}")
|
||||
logging.warning(f"设备 {device_id} 按钮文本不匹配,期望'平差处理',实际: {button_text}")
|
||||
|
||||
if adjustment_button.is_displayed() and adjustment_button.is_enabled():
|
||||
self.logger.info(f"设备 {device_id} 点击平差处理按钮")
|
||||
logging.info(f"设备 {device_id} 点击平差处理按钮")
|
||||
adjustment_button.click()
|
||||
time.sleep(3) # 等待平差处理完成
|
||||
return True
|
||||
else:
|
||||
self.logger.error(f"设备 {device_id} 平差处理按钮不可点击")
|
||||
logging.error(f"设备 {device_id} 平差处理按钮不可点击")
|
||||
return False
|
||||
|
||||
except NoSuchElementException:
|
||||
self.logger.error(f"设备 {device_id} 未找到平差处理按钮")
|
||||
logging.error(f"设备 {device_id} 未找到平差处理按钮")
|
||||
return False
|
||||
except Exception as e:
|
||||
self.logger.error(f"设备 {device_id} 点击平差处理按钮时发生错误: {str(e)}")
|
||||
logging.error(f"设备 {device_id} 点击平差处理按钮时发生错误: {str(e)}")
|
||||
return False
|
||||
|
||||
def handle_back_navigation(self, breakpoint_name, device_id):
|
||||
"""
|
||||
完整的返回导航处理流程
|
||||
|
||||
Args:
|
||||
breakpoint_name: 断点名称
|
||||
device_id: 设备ID
|
||||
|
||||
Returns:
|
||||
bool: 整个返回导航流程是否成功
|
||||
"""
|
||||
try:
|
||||
# time.sleep(2)
|
||||
logging.info(f"已点击平差处理按钮,检查是否在测量页面")
|
||||
|
||||
# 检测是否存在测量列表
|
||||
has_measurement_list = self.check_measurement_list(device_id)
|
||||
if not has_measurement_list:
|
||||
logging.info(f"设备 {device_id} 存在测量列表,重新执行平差流程")
|
||||
|
||||
# 把断点名称给find_keyword
|
||||
if not self.find_keyword(breakpoint_name):
|
||||
logging.error(f"设备 {device_id} 未找到包含 {breakpoint_name} 的文件名")
|
||||
return False
|
||||
|
||||
if not self.handle_measurement_dialog():
|
||||
logging.error(f"设备 {device_id} 处理测量弹窗失败")
|
||||
return False
|
||||
|
||||
if not self.check_apply_btn():
|
||||
logging.error(f"设备 {device_id} 检查平差处理按钮失败")
|
||||
return False
|
||||
|
||||
|
||||
# 4. 点击平差处理按钮
|
||||
if not self.click_adjustment_button(device_id):
|
||||
logging.error(f"设备 {device_id} 点击平差处理按钮失败")
|
||||
return False
|
||||
|
||||
logging.info(f"重新选择断点并点击平差处理按钮成功")
|
||||
return True
|
||||
|
||||
else:
|
||||
logging.info(f"不在测量页面,继续执行后续返回操作")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"设备 {device_id} 处理返回导航时发生错误: {str(e)}")
|
||||
return False
|
||||
|
||||
def handle_adjustment_result_dialog(self):
|
||||
"""处理平差结果确认弹窗"""
|
||||
try:
|
||||
logging.info("开始检测平差结果弹窗")
|
||||
|
||||
# 等待弹窗出现(最多等待5秒)
|
||||
warning_dialog = WebDriverWait(self.driver, 5).until(
|
||||
EC.presence_of_element_located((AppiumBy.ID, "android:id/parentPanel"))
|
||||
)
|
||||
|
||||
# 验证弹窗内容
|
||||
alert_title = warning_dialog.find_element(AppiumBy.ID, "android:id/alertTitle")
|
||||
alert_message = warning_dialog.find_element(AppiumBy.ID, "android:id/message")
|
||||
|
||||
logging.info(f"检测到弹窗 - 标题: {alert_title.text}, 消息: {alert_message.text}")
|
||||
|
||||
# 确认是目标弹窗
|
||||
if "警告" in alert_title.text and "是否保留测量成果" in alert_message.text:
|
||||
logging.info("确认是平差结果确认弹窗")
|
||||
|
||||
# 点击"是 保留成果"按钮
|
||||
yes_button = warning_dialog.find_element(AppiumBy.ID, "android:id/button1")
|
||||
if yes_button.text == "是 保留成果":
|
||||
yes_button.click()
|
||||
logging.info("已点击'是 保留成果'按钮")
|
||||
|
||||
# 等待弹窗消失
|
||||
WebDriverWait(self.driver, 5).until(
|
||||
EC.invisibility_of_element_located((AppiumBy.ID, "android:id/parentPanel"))
|
||||
)
|
||||
logging.info("弹窗已关闭")
|
||||
return True
|
||||
else:
|
||||
logging.error(f"按钮文本不匹配,期望'是 保留成果',实际: {yes_button.text}")
|
||||
return False
|
||||
else:
|
||||
logging.warning("弹窗内容不匹配,不是目标弹窗")
|
||||
return False
|
||||
|
||||
except TimeoutException:
|
||||
logging.info("未检测到平差结果弹窗,继续流程")
|
||||
return True # 没有弹窗也是正常情况
|
||||
except Exception as e:
|
||||
logging.error(f"处理平差结果弹窗时出错: {str(e)}")
|
||||
return False
|
||||
|
||||
def check_measurement_list(self, device_id):
|
||||
"""
|
||||
检查是否存在测量列表
|
||||
|
||||
Args:
|
||||
device_id: 设备ID
|
||||
|
||||
Returns:
|
||||
bool: 如果不存在测量列表返回True,存在返回False
|
||||
"""
|
||||
try:
|
||||
# 等待线路列表容器出现
|
||||
self.wait.until(
|
||||
EC.presence_of_element_located((AppiumBy.ID, ids.MEASURE_LIST_ID))
|
||||
)
|
||||
logging.info("线路列表容器已找到")
|
||||
|
||||
# 如果存在MEASURE_LIST_ID,说明有测量列表,不需要执行后续步骤
|
||||
logging.info(f"设备 {device_id} 存在测量列表,无需执行后续返回操作")
|
||||
return False
|
||||
|
||||
except TimeoutException:
|
||||
# 等待超时,说明没有测量列表
|
||||
logging.info(f"设备 {device_id} 未找到测量列表,可以继续执行后续步骤")
|
||||
return True
|
||||
except Exception as e:
|
||||
logging.error(f"设备 {device_id} 检查测量列表时发生错误: {str(e)}")
|
||||
return True
|
||||
|
||||
def find_keyword(self, fixed_filename):
|
||||
"""查找指定关键词并点击,支持向下和向上滑动查找"""
|
||||
try:
|
||||
# if not check_session_valid(self.driver, self.device_id):
|
||||
# self.logger.warning(f"设备 {self.device_id} 会话无效,尝试重新连接驱动...")
|
||||
# if not reconnect_driver(self.device_id, self.driver):
|
||||
# self.logger.error(f"设备 {self.device_id} 驱动重连失败")
|
||||
|
||||
# 等待线路列表容器出现
|
||||
self.wait.until(
|
||||
EC.presence_of_element_located((AppiumBy.ID, ids.MEASURE_LIST_ID))
|
||||
)
|
||||
self.logger.info("线路列表容器已找到")
|
||||
|
||||
max_scroll_attempts = 100 # 最大滚动尝试次数
|
||||
scroll_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()
|
||||
|
||||
# 向下滑动列表以加载更多项目
|
||||
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 get_current_items(self):
|
||||
"""获取当前页面中的所有项目文本"""
|
||||
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
|
||||
|
||||
return item_texts
|
||||
except Exception as e:
|
||||
self.logger.error(f"获取当前项目失败: {str(e)}")
|
||||
return []
|
||||
|
||||
def click_item_by_text(self, text):
|
||||
"""点击指定文本的项目"""
|
||||
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
|
||||
|
||||
self.logger.warning(f"未找到可点击的项目: {text}")
|
||||
return False
|
||||
except Exception as e:
|
||||
self.logger.error(f"点击项目失败: {str(e)}")
|
||||
return False
|
||||
|
||||
def handle_measurement_dialog(self):
|
||||
"""处理测量弹窗 - 选择继续测量"""
|
||||
try:
|
||||
self.logger.info("检查测量弹窗...")
|
||||
|
||||
# 直接尝试点击"继续测量"按钮
|
||||
continue_btn = WebDriverWait(self.driver, 2).until(
|
||||
EC.element_to_be_clickable((AppiumBy.ID, "com.bjjw.cjgc:id/measure_continue_btn"))
|
||||
)
|
||||
continue_btn.click()
|
||||
self.logger.info("已点击'继续测量'按钮")
|
||||
return True
|
||||
|
||||
except TimeoutException:
|
||||
self.logger.info("未找到继续测量按钮,可能没有弹窗")
|
||||
return True # 没有弹窗也认为是成功的
|
||||
except Exception as e:
|
||||
self.logger.error(f"点击继续测量按钮时出错: {str(e)}")
|
||||
return False
|
||||
|
||||
# 检查有没有平差处理按钮
|
||||
def check_apply_btn(self):
|
||||
"""检查是否有平差处理按钮"""
|
||||
try:
|
||||
apply_btn = WebDriverWait(self.driver, 1).until(
|
||||
EC.element_to_be_clickable((AppiumBy.ID, "com.bjjw.cjgc:id/point_measure_btn"))
|
||||
)
|
||||
if apply_btn.is_displayed():
|
||||
logging.info("进入平差页面")
|
||||
else:
|
||||
self.logger.info("没有找到'平差处理'按钮")
|
||||
return True
|
||||
except TimeoutException:
|
||||
self.logger.info("未找到平差处理按钮")
|
||||
return False # 没有弹窗也认为是成功的
|
||||
except Exception as e:
|
||||
self.logger.error(f"点击平差处理按钮时出错: {str(e)}")
|
||||
return False
|
||||
|
||||
|
||||
|
||||
def execute_back_navigation_steps(self, device_id):
|
||||
"""
|
||||
执行实际的返回导航步骤
|
||||
|
||||
Args:
|
||||
device_id: 设备ID
|
||||
|
||||
Returns:
|
||||
bool: 导航是否成功
|
||||
"""
|
||||
try:
|
||||
# 1. 首先点击返回按钮
|
||||
if not self.click_back_button(device_id):
|
||||
logging.error(f"设备 {device_id} 点击返回按钮失败")
|
||||
return False
|
||||
|
||||
# 2. 处理返回确认弹窗
|
||||
logging.info(f"已点击返回按钮,等待处理返回确认弹窗")
|
||||
if not self.handle_confirmation_dialog(device_id):
|
||||
logging.error(f"设备 {device_id} 处理返回确认弹窗失败")
|
||||
# return False
|
||||
|
||||
|
||||
# 3. 验证是否成功返回到上一页面
|
||||
time.sleep(0.5) # 等待页面跳转完成
|
||||
|
||||
# 可以添加页面验证逻辑,比如检查是否返回到预期的页面
|
||||
# 这里可以根据实际应用添加特定的页面元素验证
|
||||
|
||||
logging.info(f"设备 {device_id} 返回导航流程完成")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"设备 {device_id} 执行返回导航步骤时发生错误: {str(e)}")
|
||||
return False
|
||||
|
||||
|
||||
def scroll_list_to_bottom(self, device_id, max_swipes=60):
|
||||
"""
|
||||
下滑列表到最底端
|
||||
|
||||
Args:
|
||||
device_id: 设备ID
|
||||
max_swipes: 最大下滑次数
|
||||
|
||||
Returns:
|
||||
bool: 是否滑动到底部
|
||||
"""
|
||||
try:
|
||||
logging.info(f"设备 {device_id} 开始下滑列表到底部")
|
||||
|
||||
# 获取列表元素
|
||||
list_view = self.driver.find_element(AppiumBy.ID, "com.bjjw.cjgc:id/auto_data_list")
|
||||
logging.info(f"时间戳1: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())}")
|
||||
|
||||
same_content_count = 0
|
||||
|
||||
# 初始化第一次的子元素文本
|
||||
initial_child_elements = list_view.find_elements(AppiumBy.CLASS_NAME, "android.widget.TextView")
|
||||
logging.info(f"时间戳2: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())}")
|
||||
# current_child_texts = "|".join([
|
||||
# elem.text.strip() for elem in initial_child_elements
|
||||
# if elem.text and elem.text.strip()
|
||||
# ])
|
||||
current_child_texts = "|".join(
|
||||
elem_text.strip() for elem in initial_child_elements
|
||||
if (elem_text := elem.text) and elem_text.strip()
|
||||
)
|
||||
logging.info(f"时间戳3: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())}")
|
||||
for i in range(max_swipes):
|
||||
# 执行下滑操作
|
||||
self.driver.execute_script("mobile: scrollGesture", {
|
||||
'elementId': list_view.id,
|
||||
'direction': 'down',
|
||||
'percent': 0.8,
|
||||
'duration': 500
|
||||
})
|
||||
|
||||
# 获取滑动后的子元素文本
|
||||
new_child_elements = list_view.find_elements(AppiumBy.CLASS_NAME, "android.widget.TextView")
|
||||
new_child_texts = "|".join(
|
||||
elem_text.strip() for elem in new_child_elements
|
||||
if (elem_text := elem.text) and elem_text.strip()
|
||||
)
|
||||
|
||||
# 判断内容是否变化:若连续3次相同,认为到达底部
|
||||
if new_child_texts == current_child_texts:
|
||||
same_content_count += 1
|
||||
if same_content_count >= 2:
|
||||
logging.info(f"设备 {device_id} 列表已滑动到底部,共滑动 {i+1} 次")
|
||||
return True
|
||||
else:
|
||||
same_content_count = 0 # 内容变化,重置计数
|
||||
current_child_texts = new_child_texts # 更新上一次内容
|
||||
|
||||
logging.debug(f"设备 {device_id} 第 {i+1} 次下滑完成,当前子元素文本: {new_child_texts[:50]}...") # 打印部分文本
|
||||
|
||||
logging.warning(f"设备 {device_id} 达到最大下滑次数 {max_swipes},可能未完全到底部")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"设备 {device_id} 下滑列表时发生错误: {str(e)}")
|
||||
return False
|
||||
|
||||
def click_last_spinner_with_retry(self, device_id, max_retries=2):
|
||||
"""带重试机制的点击方法"""
|
||||
for attempt in range(max_retries):
|
||||
try:
|
||||
if self.click_last_spinner(device_id):
|
||||
return True
|
||||
logging.warning(f"设备 {device_id} 第{attempt + 1}次点击失败,准备重试")
|
||||
time.sleep(0.5) # 重试前等待
|
||||
except Exception as e:
|
||||
logging.error(f"设备 {device_id} 第{attempt + 1}次尝试失败: {str(e)}")
|
||||
|
||||
logging.error(f"设备 {device_id} 所有重试次数已用尽")
|
||||
return False
|
||||
|
||||
def click_last_spinner(self, device_id):
|
||||
"""
|
||||
点击最后一个spinner
|
||||
|
||||
Args:
|
||||
device_id: 设备ID
|
||||
|
||||
Returns:
|
||||
bool: 是否成功点击
|
||||
"""
|
||||
try:
|
||||
logging.info(f"设备 {device_id} 查找最后一个spinner")
|
||||
|
||||
# 查找所有的spinner元素
|
||||
spinners = self.driver.find_elements(AppiumBy.ID, "com.bjjw.cjgc:id/spinner")
|
||||
|
||||
if not spinners:
|
||||
logging.error(f"设备 {device_id} 未找到任何spinner元素")
|
||||
return False
|
||||
|
||||
# 获取最后一个spinner
|
||||
last_spinner = spinners[-1]
|
||||
|
||||
if not (last_spinner.is_displayed() and last_spinner.is_enabled()):
|
||||
logging.error(f"设备 {device_id} 最后一个spinner不可点击")
|
||||
return False
|
||||
|
||||
# 点击操作
|
||||
logging.info(f"设备 {device_id} 点击最后一个spinner")
|
||||
last_spinner.click()
|
||||
|
||||
# 执行额外一次下滑操作
|
||||
self.scroll_down_once(device_id)
|
||||
|
||||
max_retries = 3 # 最大重试次数
|
||||
retry_count = 0
|
||||
wait_timeout = 5 # 增加等待时间到5秒
|
||||
|
||||
while retry_count < max_retries:
|
||||
try:
|
||||
# 确保device_id正确设置,使用全局变量作为备用
|
||||
if not hasattr(self, 'device_id') or not self.device_id:
|
||||
# 优先使用传入的device_id,其次使用全局变量
|
||||
self.device_id = device_id if device_id else global_variable.get_device_id()
|
||||
|
||||
# 使用self.device_id,确保有默认值
|
||||
actual_device_id = self.device_id if self.device_id else global_variable.get_device_id()
|
||||
|
||||
if not check_session_valid(self.driver, actual_device_id):
|
||||
logging.warning(f"设备 {actual_device_id} 会话无效,尝试重新连接驱动...")
|
||||
try:
|
||||
# 使用正确的设备ID进行重连
|
||||
new_driver, new_wait = reconnect_driver(actual_device_id, self.driver)
|
||||
if new_driver:
|
||||
self.driver = new_driver
|
||||
self.wait = new_wait
|
||||
logging.info(f"设备 {actual_device_id} 驱动重连成功")
|
||||
else:
|
||||
logging.error(f"设备 {actual_device_id} 驱动重连失败")
|
||||
retry_count += 1
|
||||
continue
|
||||
except Exception as e:
|
||||
logging.error(f"设备 {actual_device_id} 驱动重连异常: {str(e)}")
|
||||
retry_count += 1
|
||||
continue
|
||||
|
||||
# 点击spinner(如果是重试,需要重新获取元素)
|
||||
if retry_count > 0:
|
||||
spinners = self.driver.find_elements(AppiumBy.CLASS_NAME, "android.widget.Spinner")
|
||||
if not spinners:
|
||||
logging.error(f"设备 {device_id} 未找到spinner元素")
|
||||
retry_count += 1
|
||||
continue
|
||||
last_spinner = spinners[-1]
|
||||
if not (last_spinner.is_displayed() and last_spinner.is_enabled()):
|
||||
logging.error(f"设备 {device_id} spinner不可点击")
|
||||
retry_count += 1
|
||||
continue
|
||||
logging.info(f"设备 {device_id} 重新点击spinner")
|
||||
last_spinner.click()
|
||||
# 重试时也执行下滑操作
|
||||
self.scroll_down_once(device_id)
|
||||
|
||||
# 等待下拉菜单出现,增加等待时间到5秒
|
||||
wait = WebDriverWait(self.driver, wait_timeout)
|
||||
detail_show = wait.until(
|
||||
EC.presence_of_element_located((AppiumBy.ID, "com.bjjw.cjgc:id/detailshow"))
|
||||
)
|
||||
|
||||
if detail_show.is_displayed():
|
||||
logging.info(f"设备 {device_id} spinner点击成功,下拉菜单已展开")
|
||||
return True
|
||||
else:
|
||||
logging.error(f"设备 {device_id} 下拉菜单未显示")
|
||||
retry_count += 1
|
||||
continue
|
||||
|
||||
except Exception as wait_error:
|
||||
error_msg = str(wait_error)
|
||||
logging.error(f"设备 {device_id} 等待下拉菜单超时 (第{retry_count+1}次尝试): {error_msg}")
|
||||
|
||||
# 检查是否是连接断开相关的错误
|
||||
if not check_session_valid(self.driver, self.device_id):
|
||||
logging.warning(f"设备 {self.device_id} 会话无效,尝试重新连接驱动...")
|
||||
# if any(keyword in error_msg for keyword in ['socket hang up', 'Could not proxy command']):
|
||||
# logging.warning(f"设备 {device_id} 检测到连接相关错误,尝试重连...")
|
||||
if not reconnect_driver(self.device_id, self.driver):
|
||||
logging.error(f"设备 {device_id} 驱动重连失败")
|
||||
|
||||
retry_count += 1
|
||||
if retry_count < max_retries:
|
||||
logging.info(f"设备 {device_id} 将在1秒后进行第{retry_count+1}次重试")
|
||||
time.sleep(1) # 等待1秒后重试
|
||||
|
||||
logging.error(f"设备 {device_id} 经过{max_retries}次重试后仍无法展开下拉菜单")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"设备 {device_id} 点击最后一个spinner时发生错误: {str(e)}")
|
||||
return False
|
||||
|
||||
def scroll_down_once(self, device_id):
|
||||
"""
|
||||
再次下滑一次
|
||||
|
||||
Args:
|
||||
device_id: 设备ID
|
||||
|
||||
Returns:
|
||||
bool: 是否成功下滑
|
||||
"""
|
||||
try:
|
||||
logging.info(f"设备 {device_id} 执行额外一次下滑")
|
||||
|
||||
# 获取列表元素
|
||||
list_view = self.driver.find_element(AppiumBy.ID, "com.bjjw.cjgc:id/auto_data_list")
|
||||
|
||||
# 执行下滑操作
|
||||
self.driver.execute_script("mobile: scrollGesture", {
|
||||
'elementId': list_view.id,
|
||||
'direction': 'down',
|
||||
'percent': 0.5
|
||||
})
|
||||
|
||||
time.sleep(0.2)
|
||||
logging.info(f"设备 {device_id} 额外下滑完成")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"设备 {device_id} 额外下滑时发生错误: {str(e)}")
|
||||
return False
|
||||
|
||||
def take_screenshot(self):
|
||||
"""
|
||||
|
||||
Reference in New Issue
Block a user