538 lines
24 KiB
Python
538 lines
24 KiB
Python
# 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 |