405 lines
18 KiB
Python
405 lines
18 KiB
Python
# 更新基站页面操作
|
||
# page_objects/download_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
|
||
from datetime import datetime
|
||
|
||
import globals.ids as ids # 导入元素ID
|
||
import globals.global_variable as global_variable # 导入全局变量
|
||
from globals.driver_utils import check_session_valid, reconnect_driver
|
||
|
||
class DownloadTabbarPage:
|
||
def __init__(self, driver, wait, device_id):
|
||
self.driver = driver
|
||
self.wait = wait
|
||
self.device_id = device_id
|
||
self.logger = logging.getLogger(__name__)
|
||
# 添加默认的目标日期值
|
||
self.target_year = 2022
|
||
self.target_month = 9
|
||
self.target_day = 22
|
||
|
||
def is_download_tabbar_visible(self):
|
||
"""检查下载标签栏是否可见"""
|
||
try:
|
||
return self.driver.find_element(AppiumBy.ID, ids.DOWNLOAD_TABBAR_ID).is_displayed()
|
||
except NoSuchElementException:
|
||
self.logger.warning("下载标签栏元素未找到")
|
||
return False
|
||
except Exception as e:
|
||
self.logger.error(f"检查下载标签栏可见性时发生意外错误: {str(e)}")
|
||
return False
|
||
|
||
def click_download_tabbar(self):
|
||
"""点击下载标签栏"""
|
||
try:
|
||
download_tab = self.wait.until(
|
||
EC.element_to_be_clickable((AppiumBy.ID, ids.DOWNLOAD_TABBAR_ID))
|
||
)
|
||
download_tab.click()
|
||
self.logger.info("已点击下载标签栏")
|
||
|
||
# 使用显式等待替代固定等待
|
||
self.wait.until(
|
||
lambda driver: self.is_download_tabbar_visible()
|
||
)
|
||
return True
|
||
except TimeoutException:
|
||
self.logger.error("等待下载标签栏可点击超时")
|
||
return False
|
||
except Exception as e:
|
||
self.logger.error(f"点击下载标签栏时出错: {str(e)}")
|
||
return False
|
||
|
||
def update_work_base(self):
|
||
"""更新工作基点"""
|
||
try:
|
||
# 点击更新工作基点
|
||
update_work_base = self.wait.until(
|
||
EC.element_to_be_clickable((AppiumBy.ID, ids.UPDATE_WORK_BASE))
|
||
)
|
||
update_work_base.click()
|
||
self.logger.info("已点击更新工作基点")
|
||
|
||
# 等待更新完成 - 可以添加更具体的等待条件
|
||
# 例如等待某个进度条消失或成功提示出现
|
||
time.sleep(2) # 暂时保留，但建议替换为显式等待
|
||
return True
|
||
except TimeoutException:
|
||
self.logger.error("等待更新工作基点按钮可点击超时")
|
||
return False
|
||
except Exception as e:
|
||
self.logger.error(f"更新工作基点时出错: {str(e)}")
|
||
return False
|
||
|
||
def _get_current_date(self):
|
||
"""获取当前开始日期控件的日期值，支持多种格式解析"""
|
||
try:
|
||
date_element = self.wait.until(
|
||
EC.visibility_of_element_located((AppiumBy.ID, ids.DATE_START))
|
||
)
|
||
date_text = date_element.text.strip()
|
||
self.logger.info(f"获取到当前开始日期: {date_text}")
|
||
|
||
# 尝试多种日期格式解析
|
||
date_formats = [
|
||
"%Y-%m-%d", # 匹配 '2025-08-12' 格式
|
||
"%Y年%m月%d日", # 匹配 '2025年08月12日' 格式
|
||
"%Y/%m/%d" # 可选：添加其他可能的格式
|
||
]
|
||
|
||
for fmt in date_formats:
|
||
try:
|
||
return datetime.strptime(date_text, fmt)
|
||
except ValueError:
|
||
continue # 尝试下一种格式
|
||
|
||
# 如果所有格式都匹配失败
|
||
self.logger.error(f"日期格式解析错误: 无法识别的格式，日期文本: {date_text}")
|
||
return None
|
||
|
||
except TimeoutException:
|
||
self.logger.error("获取当前日期超时")
|
||
return None
|
||
except Exception as e:
|
||
self.logger.error(f"获取当前日期失败: {str(e)}")
|
||
return None
|
||
|
||
def update_level_line(self):
|
||
"""更新水准线路，修改为设置2022年9月22日"""
|
||
try:
|
||
# 点击更新水准线路
|
||
update_level_line = self.wait.until(
|
||
EC.element_to_be_clickable((AppiumBy.ID, ids.UPDATE_LEVEL_LINE))
|
||
)
|
||
update_level_line.click()
|
||
self.logger.info("已点击更新水准线路")
|
||
|
||
# 获取原始开始日期
|
||
original_date = self._get_current_date()
|
||
if not original_date:
|
||
self.logger.error("无法获取原始开始日期，更新水准线路失败")
|
||
return False
|
||
|
||
# 点击开始日期
|
||
date_start = self.wait.until(
|
||
EC.element_to_be_clickable((AppiumBy.ID, ids.DATE_START))
|
||
)
|
||
date_start.click()
|
||
self.logger.info("已点击开始日期控件")
|
||
|
||
# # 处理时间选择器，设置为2022年9月22日
|
||
# if not self.handle_time_selector(2022, 9, 22, original_date):
|
||
# self.logger.error("处理时间选择失败")
|
||
# return False
|
||
# 处理时间选择器，滚动选择年份
|
||
if not self.handle_year_selector():
|
||
self.logger.error("时间选择器滑动年份，处理时间失败")
|
||
return False
|
||
|
||
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")
|
||
|
||
# 获取滚轮的位置和尺寸
|
||
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("已滑动年份选择器")
|
||
return True
|
||
|
||
except Exception as e:
|
||
self.logger.error(f"滑动年份选择器时出错: {str(e)}")
|
||
return False
|
||
def _scroll_to_value(self, picker_id, target_value, original_value, max_attempts=20):
|
||
"""滚动选择器到目标值，基于原始值计算滚动次数"""
|
||
try:
|
||
# 计算需要滚动的次数（绝对值）
|
||
scroll_count = abs(int(target_value) - int(original_value))
|
||
self.logger.info(f"需要滚动{scroll_count}次将{picker_id}从{original_value}调整到{target_value}")
|
||
|
||
# 确定滚动方向
|
||
direction = "down" if int(target_value) > int(original_value) else "up"
|
||
|
||
# 获取选择器元素
|
||
picker = self.wait.until(
|
||
EC.visibility_of_element_located((AppiumBy.ID, picker_id))
|
||
)
|
||
|
||
# 计算滚动坐标
|
||
x = picker.location['x'] + picker.size['width'] // 2
|
||
self.logger.info(f"水平位置x为{x}")
|
||
y_center = picker.location['y'] + picker.size['height'] // 2
|
||
self.logger.info(f"垂直位置中点y_center为{y_center}")
|
||
# start_y = y_center if direction == "down" else picker.location['y']
|
||
# end_y = picker.location['y'] if direction == "down" else y_center
|
||
# 关键修改：计算选择器高度的五分之一（滑动距离）
|
||
height_fifth = picker.size['height'] // 5 # 1/5高度
|
||
|
||
# 根据方向计算起点和终点，确保滑动距离为 height_fifth
|
||
if direction == "down":
|
||
# 向下滚动：从中心点向上滑动1/5高度
|
||
start_y = y_center
|
||
end_y = y_center - height_fifth # 终点 = 中心点 - 1/5高度
|
||
self.logger.info(f"down垂直开始位置start_y为{y_center}，垂直结束位置end_y为{end_y}")
|
||
|
||
else:
|
||
# 向上滚动：从中心点向下滑动1/5高度
|
||
start_y = y_center
|
||
end_y = y_center + height_fifth # 终点 = 中心点 + 1/5高度
|
||
self.logger.info(f"up垂直开始位置start_y为{y_center}，垂直结束位置end_y为{end_y}")
|
||
# 执行滚动操作
|
||
for _ in range(scroll_count):
|
||
self.driver.swipe(x, start_y, x, end_y, 500)
|
||
time.sleep(0.5) # 等待滚动稳定
|
||
return True # 循环scroll_count次后直接返回
|
||
# # 验证当前值
|
||
# current_value = picker.text
|
||
# if current_value == str(target_value):
|
||
# self.logger.info(f"{picker_id}已达到目标值: {target_value}")
|
||
# return True
|
||
|
||
# 最终验证
|
||
# final_value = picker.text
|
||
# if final_value == str(target_value):
|
||
# self.logger.info(f"{picker_id}已达到目标值: {target_value}")
|
||
# return True
|
||
# else:
|
||
# self.logger.error(f"{picker_id}滚动{scroll_count}次后未达到目标值，当前值: {final_value}")
|
||
# return False
|
||
|
||
except StaleElementReferenceException:
|
||
self.logger.warning("元素状态已过期，重新获取")
|
||
return False
|
||
except Exception as e:
|
||
self.logger.error(f"滚动选择器出错: {str(e)}")
|
||
return False
|
||
|
||
def handle_year_selector(self):
|
||
"""处理时间选择器，滚动选择年份"""
|
||
try:
|
||
# 等待时间选择器出现
|
||
self.wait.until(
|
||
EC.visibility_of_element_located((AppiumBy.ID, ids.ALERT_DIALOG))
|
||
)
|
||
self.logger.info("时间选择对话框已出现")
|
||
|
||
# 滚动选择年份
|
||
if not self._swipe_year_wheel():
|
||
self.logger.error("滚动选择年份失败")
|
||
return False
|
||
|
||
# 点击确认按钮
|
||
confirm_btn = self.wait.until(
|
||
EC.element_to_be_clickable((AppiumBy.ID, ids.SCRCOLL_CONFIRM)) # 日期选择器确认按钮
|
||
)
|
||
confirm_btn.click()
|
||
self.logger.info("已确认时间选择")
|
||
|
||
# 点击对话框确认按钮（UPDATE_LEVEL_LINE_CONFIRM）
|
||
confirm_btn = self.wait.until(
|
||
EC.element_to_be_clickable((AppiumBy.ID, ids.UPDATE_LEVEL_LINE_CONFIRM)) # 选择日期对话框确认按钮
|
||
)
|
||
confirm_btn.click()
|
||
self.logger.info("已点击对话框确认按钮commit")
|
||
|
||
# 等待加载对话框出现
|
||
custom_dialog = self.wait.until(
|
||
EC.visibility_of_element_located((AppiumBy.ID, ids.LOADING_DIALOG))
|
||
)
|
||
self.logger.info("检测到loading对话框出现")
|
||
|
||
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} 驱动重连失败")
|
||
|
||
# 新增：等待加载对话框消失（表示更新完成）
|
||
WebDriverWait(self.driver, 300).until(
|
||
EC.invisibility_of_element_located((AppiumBy.ID, ids.LOADING_DIALOG))
|
||
)
|
||
self.logger.info("loading对话框已消失，更新完成")
|
||
|
||
return True
|
||
|
||
except TimeoutException as e:
|
||
# 明确超时发生在哪个环节
|
||
self.logger.error("处理时间选择时超时：可能是等待对话框出现、日期选择器元素或确认按钮超时", exc_info=True)
|
||
return False
|
||
except Exception as e:
|
||
# 细分不同环节的异常
|
||
self.logger.error(f"处理时间选择对话框时出错：可能是日期滚动、选择器确认或对话框确认环节失败 - {str(e)}", exc_info=True)
|
||
return False
|
||
|
||
def handle_time_selector(self, target_year, target_month, target_day, original_date=None):
|
||
"""处理时间选择器，选择起始时间并确认"""
|
||
self.logger.info(f"传入handle_time_selector的初始日期: {original_date}")
|
||
try:
|
||
# 等待时间选择器出现
|
||
self.wait.until(
|
||
EC.visibility_of_element_located((AppiumBy.ID, ids.ALERT_DIALOG))
|
||
)
|
||
self.logger.info("时间选择对话框已出现")
|
||
|
||
# 如果没有提供原始日期，使用当前日期控件的值
|
||
if not original_date:
|
||
original_date = self._get_current_date()
|
||
if not original_date:
|
||
self.logger.error("无法获取原始日期，处理时间选择失败")
|
||
return False
|
||
|
||
# 滚动选择年份
|
||
if not self._scroll_to_value(ids.SCRCOLL_YEAR, target_year, original_date.year):
|
||
self.logger.error("滚动选择年份失败")
|
||
return False
|
||
|
||
# 滚动选择月份
|
||
if not self._scroll_to_value(ids.SCRCOLL_MONTH, target_month, original_date.month):
|
||
self.logger.error("滚动选择月份失败")
|
||
return False
|
||
|
||
# 滚动选择日期
|
||
if not self._scroll_to_value(ids.SCRCOLL_DAY, target_day, original_date.day):
|
||
self.logger.error("滚动选择日期失败")
|
||
return False
|
||
|
||
# 点击确认按钮
|
||
confirm_btn = self.wait.until(
|
||
EC.element_to_be_clickable((AppiumBy.ID, ids.SCRCOLL_CONFIRM)) # 日期选择器确认按钮
|
||
)
|
||
confirm_btn.click()
|
||
self.logger.info("已确认时间选择")
|
||
|
||
# 点击对话框确认按钮（UPDATE_LEVEL_LINE_CONFIRM）
|
||
confirm_btn = self.wait.until(
|
||
EC.element_to_be_clickable((AppiumBy.ID, ids.UPDATE_LEVEL_LINE_CONFIRM)) # 选择日期对话框确认按钮
|
||
)
|
||
confirm_btn.click()
|
||
self.logger.info("已点击对话框确认按钮commit")
|
||
|
||
# 新增：等待加载对话框出现
|
||
custom_dialog = self.wait.until(
|
||
EC.visibility_of_element_located((AppiumBy.ID, ids.LOADING_DIALOG))
|
||
)
|
||
self.logger.info("检测到loading对话框出现")
|
||
|
||
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} 驱动重连失败")
|
||
|
||
# 新增：等待加载对话框消失（表示更新完成）
|
||
WebDriverWait(self.driver, 300).until(
|
||
EC.invisibility_of_element_located((AppiumBy.ID, ids.LOADING_DIALOG))
|
||
)
|
||
self.logger.info("loading对话框已消失，更新完成")
|
||
|
||
'''点击commit确认按钮后，loading弹窗会出现，等待其加载完成后关闭
|
||
检测导航栏中的测量tabbar是否出现来确定是否返回True
|
||
'''
|
||
# measure_tabbar_btn = self.wait.until(
|
||
# EC.visibility_of_element_located((AppiumBy.ID, ids.MEASURE_TABBAR_ID))
|
||
# )
|
||
# self.logger.info("检测测量tabbar按钮出现")
|
||
|
||
return True
|
||
|
||
except TimeoutException as e:
|
||
# 明确超时发生在哪个环节
|
||
self.logger.error("处理时间选择时超时：可能是等待对话框出现、日期选择器元素或确认按钮超时", exc_info=True)
|
||
return False
|
||
except Exception as e:
|
||
# 细分不同环节的异常
|
||
self.logger.error(f"处理时间选择对话框时出错：可能是日期滚动、选择器确认或对话框确认环节失败 - {str(e)}", exc_info=True)
|
||
return False
|
||
|
||
def download_tabbar_page_manager(self):
|
||
"""执行基础更新操作"""
|
||
try:
|
||
# 执行基础更新流程
|
||
self.logger.info(f"设备 {global_variable.GLOBAL_DEVICE_ID} 开始执行更新流程")
|
||
|
||
# 点击下载标签栏
|
||
if not self.click_download_tabbar():
|
||
self.logger.error(f"设备 {global_variable.GLOBAL_DEVICE_ID} 点击下载标签栏失败")
|
||
return False
|
||
|
||
# 更新工作基点
|
||
if not self.update_work_base():
|
||
self.logger.error(f"设备 {global_variable.GLOBAL_DEVICE_ID} 更新工作基点失败")
|
||
return False
|
||
|
||
# 更新水准线路
|
||
if not self.update_level_line():
|
||
self.logger.error(f"设备 {global_variable.GLOBAL_DEVICE_ID} 更新水准线路失败")
|
||
return False
|
||
|
||
self.logger.info(f"设备 {global_variable.GLOBAL_DEVICE_ID} 更新操作执行成功")
|
||
return True
|
||
except Exception as e:
|
||
self.logger.error(f"设备 {global_variable.GLOBAL_DEVICE_ID} 执行更新操作时出错: {str(e)}")
|
||
return False |