435 lines
20 KiB
Python
435 lines
20 KiB
Python
# actions.py 主自动化脚本
|
||
import os
|
||
import logging
|
||
import time
|
||
import subprocess
|
||
from appium import webdriver
|
||
from appium.options.android import UiAutomator2Options
|
||
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 globals.ids as ids
|
||
import globals.global_variable as global_variable # 导入全局变量模块
|
||
import permissions # 导入权限处理模块
|
||
import globals.apis as apis
|
||
from globals.driver_utils import init_appium_driver, ensure_appium_server_running, safe_quit_driver, is_app_launched, launch_app_manually
|
||
from page_objects.login_page import LoginPage
|
||
from page_objects.download_tabbar_page import DownloadTabbarPage
|
||
from page_objects.screenshot_page import ScreenshotPage
|
||
from page_objects.upload_config_page import UploadConfigPage
|
||
from page_objects.more_download_page import MoreDownloadPage
|
||
|
||
|
||
# 配置日志
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s - %(levelname)s: %(message)s",
|
||
handlers=[
|
||
logging.FileHandler("appium_automation.log"),
|
||
logging.StreamHandler()
|
||
]
|
||
)
|
||
|
||
class DeviceAutomation(object):
|
||
|
||
|
||
def __init__(self, device_id=None):
|
||
self.device_id = device_id
|
||
|
||
# 初始化权限
|
||
if permissions.grant_appium_permissions(self.device_id):
|
||
logging.info(f"设备 {self.device_id} 权限授予成功")
|
||
else:
|
||
logging.warning(f"设备 {self.device_id} 权限授予失败")
|
||
|
||
# 确保Appium服务器正在运行
|
||
ensure_appium_server_running(4723)
|
||
|
||
# 初始化驱动
|
||
self.init_driver()
|
||
# 先拼接，后创建测试结果目录
|
||
self.results_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'test_results')
|
||
os.makedirs(self.results_dir, exist_ok=True)
|
||
|
||
|
||
def init_driver(self):
|
||
"""初始化Appium驱动"""
|
||
try:
|
||
# 使用全局函数初始化驱动
|
||
self.driver, self.wait = init_appium_driver(self.device_id)
|
||
# 初始化页面对象
|
||
logging.info(f"设备 {self.device_id} 开始初始化页面对象")
|
||
self.login_page = LoginPage(self.driver, self.wait)
|
||
self.download_tabbar_page = DownloadTabbarPage(self.driver, self.wait, self.device_id)
|
||
self.screenshot_page = ScreenshotPage(self.driver, self.wait, self.device_id)
|
||
self.upload_config_page = UploadConfigPage(self.driver, self.wait, self.device_id)
|
||
self.more_download_page = MoreDownloadPage(self.driver, self.wait,self.device_id)
|
||
logging.info(f"设备 {self.device_id} 所有页面对象初始化完成")
|
||
|
||
# 检查应用是否成功启动
|
||
if is_app_launched(self.driver):
|
||
logging.info(f"设备 {self.device_id} 沉降观测App已成功启动")
|
||
else:
|
||
logging.warning(f"设备 {self.device_id} 应用可能未正确启动")
|
||
# 手动启动应用
|
||
launch_app_manually(self.driver, self.device_id)
|
||
|
||
except Exception as e:
|
||
logging.error(f"设备 {self.device_id} 初始化驱动失败: {str(e)}")
|
||
raise
|
||
|
||
def is_app_launched(self):
|
||
"""检查应用是否已启动"""
|
||
try:
|
||
return is_app_launched(self.driver)
|
||
except Exception as e:
|
||
logging.error(f"设备 {self.device_id} 检查应用启动状态时出错: {str(e)}")
|
||
return False
|
||
|
||
def is_element_present(self, by, value):
|
||
"""检查元素是否存在"""
|
||
try:
|
||
self.driver.find_element(by, value)
|
||
return True
|
||
except NoSuchElementException:
|
||
return False
|
||
|
||
def handle_app_state(self):
|
||
"""根据当前应用状态处理相应的操作"""
|
||
try:
|
||
login_btn_exists = self.login_page.is_login_page()
|
||
if not login_btn_exists:
|
||
logging.error(f"设备 {self.device_id} 未知应用状态，无法确定当前页面，跳转到登录页面")
|
||
if self.navigate_to_login_page(self.driver, self.device_id):
|
||
logging.info(f"设备 {self.device_id} 成功跳转到登录页面")
|
||
return self.handle_app_state() # 递归调用处理登录后的状态
|
||
else:
|
||
logging.error(f"设备 {self.device_id} 跳转到登录页面失败")
|
||
return False
|
||
|
||
# 处理登录页面状态
|
||
logging.info(f"设备 {self.device_id} 检测到登录页面，执行登录操作")
|
||
max_retries = 1
|
||
login_success = False
|
||
|
||
for attempt in range(max_retries + 1):
|
||
if self.login_page.login():
|
||
login_success = True
|
||
break
|
||
else:
|
||
if attempt < max_retries:
|
||
logging.warning(f"设备 {self.device_id} 登录失败，准备重试 ({attempt + 1}/{max_retries})")
|
||
time.sleep(2) # 等待2秒后重试
|
||
else:
|
||
logging.error(f"设备 {self.device_id} 登录失败，已达到最大重试次数")
|
||
|
||
if not login_success:
|
||
return False
|
||
|
||
logging.info(f"设备 {self.device_id} 登录成功，继续执行更新操作")
|
||
time.sleep(1)
|
||
|
||
# # 执行更新操作
|
||
# if not self.download_tabbar_page.download_tabbar_page_manager():
|
||
# logging.error(f"设备 {self.device_id} 更新操作执行失败")
|
||
# return False
|
||
|
||
# 获取状态为3的线路。
|
||
apis.get_line_info_and_save_global(user_name=global_variable.get_username());
|
||
# # 虚拟数据替代
|
||
# global_variable.GLOBAL_UPLOAD_BREAKPOINT_DICT = {'CDWZQ-2标-龙骨湾右线大桥-0-7号墩-平原': 'L156372', 'CDWZQ-2标-蓝家湾特大 桥-31-31-平原': 'L159206'}
|
||
# global_variable.GLOBAL_UPLOAD_BREAKPOINT_LIST = list(global_variable.GLOBAL_UPLOAD_BREAKPOINT_DICT.keys())
|
||
|
||
# 点击测量导航栏按钮
|
||
measure_page_btn = WebDriverWait(self.driver, 5).until(
|
||
EC.element_to_be_clickable((AppiumBy.ID, "com.bjjw.cjgc:id/img_3_layout"))
|
||
)
|
||
measure_page_btn.click()
|
||
|
||
|
||
# 处理平差
|
||
self.screenshot_page.screenshot_page_manager(self.device_id)
|
||
|
||
|
||
# 检查是否有需要上传的断点
|
||
if not global_variable.get_upload_breakpoint_list():
|
||
logging.info(f"设备 {self.device_id} 断点列表为空，无需执行上传操作")
|
||
return False
|
||
|
||
# 点击上传导航栏按钮
|
||
upload_page_btn = WebDriverWait(self.driver, 5).until(
|
||
EC.element_to_be_clickable((AppiumBy.ID, "com.bjjw.cjgc:id/img_2_layout"))
|
||
)
|
||
upload_page_btn.click()
|
||
upload_page_btn.click()
|
||
time.sleep(0.5)
|
||
|
||
upload_success_count = 0 # 上传成功的断点数量
|
||
# 实现连续获取断点列表直到两次相同
|
||
previous_breakpoint_list = None
|
||
current_breakpoint_list = None
|
||
max_attempts = 10 # 最大尝试次数
|
||
attempt_count = 0
|
||
while True:
|
||
# 获取当前页面的断点列表
|
||
|
||
# 从当前页面获取断点列表名称
|
||
current_breakpoint_list = self.upload_config_page.get_breakpoint_list_from_page()
|
||
attempt_count += 1
|
||
if current_breakpoint_list is None:
|
||
logging.warning(f"设备 {self.device_id} 没有获取到断点列表，结束上传流程")
|
||
break
|
||
|
||
logging.info(f"第 {attempt_count} 次获取断点列表: {current_breakpoint_list}")
|
||
|
||
# 检查是否与上一次获取的列表相同
|
||
if previous_breakpoint_list is not None and current_breakpoint_list == previous_breakpoint_list:
|
||
logging.info("连续两次获取到相同的断点列表，结束上传")
|
||
# 下滑一次再次检查是否一致
|
||
# 获取列表容器
|
||
list_container = self.driver.find_element(AppiumBy.ID, "com.bjjw.cjgc:id/upload_result_list")
|
||
|
||
# 计算滑动坐标
|
||
start_x = list_container.location['x'] + list_container.size['width'] // 2
|
||
# 向下滑动
|
||
start_y = list_container.location['y'] + list_container.size['height'] * 0.95
|
||
end_y = list_container.location['y'] + list_container.size['height'] * 0.05
|
||
logging.info("向下滑动列表")
|
||
self.driver.swipe(start_x, start_y, start_x, end_y, 500)
|
||
current_breakpoint_list = self.upload_config_page.get_breakpoint_list_from_page()
|
||
if previous_breakpoint_list is not None and current_breakpoint_list == previous_breakpoint_list:
|
||
break
|
||
|
||
previous_breakpoint_list = current_breakpoint_list.copy() if current_breakpoint_list else None
|
||
# 遍历当前页面列表，逐个执行上传操作
|
||
|
||
for breakpoint_name in current_breakpoint_list:
|
||
|
||
# # 遍历断点列表，逐个执行上传操作
|
||
# upload_success_count = 0
|
||
# for breakpoint_name in global_variable.get_upload_breakpoint_list():
|
||
try:
|
||
# 检查断点名称是否在当前页面列表中
|
||
if breakpoint_name not in current_breakpoint_list:
|
||
logging.warning(f"设备 {self.device_id} 断点 '{breakpoint_name}' 不在当前页面列表中，跳过上传")
|
||
continue
|
||
|
||
logging.info(f"设备 {self.device_id} 开始处理断点 '{breakpoint_name}' 的上传")
|
||
|
||
# 安全地获取断点信息
|
||
# breakpoint_names = global_variable.get_upload_breakpoint_dict()
|
||
# logging.info(f"设备 {self.device_id} 断点字典 '{breakpoint_names}' 中包含断点 '{breakpoint_name}'")
|
||
ui_breakpoint_dict = {}
|
||
for key, value in global_variable.get_upload_breakpoint_dict().items():
|
||
# 去掉末尾的"-平原"
|
||
if key.endswith('-平原'):
|
||
ui_key = key[:-3] # 去掉最后3个字符"-平原"
|
||
else:
|
||
ui_key = key
|
||
ui_breakpoint_dict[ui_key] = value
|
||
|
||
line_num = ui_breakpoint_dict.get(breakpoint_name)
|
||
logging.info(f"设备 {self.device_id} 断点 '{breakpoint_name}' 对应的line_num为 {line_num}")
|
||
|
||
if line_num is None:
|
||
logging.warning(f"设备 {self.device_id} 断点 '{breakpoint_name}' 在字典中未找到，跳过上传")
|
||
continue
|
||
if not line_num:
|
||
logging.warning(f"设备 {self.device_id} 断点 '{breakpoint_name}' 未获取到line_num，跳过上传")
|
||
continue
|
||
|
||
# 执行上传配置管理，传入当前断点名称
|
||
if self.upload_config_page.upload_config_page_manager(self.results_dir, breakpoint_name, line_num):
|
||
logging.info(f"设备 {self.device_id} 断点 '{breakpoint_name}' 上传成功")
|
||
upload_success_count += 1
|
||
else:
|
||
logging.error(f"设备 {self.device_id} 断点 '{breakpoint_name}' 上传失败")
|
||
for i in range(3):
|
||
if self.upload_config_page.upload_config_page_manager(self.results_dir, breakpoint_name, line_num):
|
||
logging.info(f"设备 {self.device_id} 断点 '{breakpoint_name}' 重试上传成功")
|
||
upload_success_count += 1
|
||
break
|
||
else:
|
||
logging.error(f"设备 {self.device_id} 断点 '{breakpoint_name}' 上传失败，第 {i+1} 次重试")
|
||
|
||
except Exception as e:
|
||
logging.error(f"设备 {self.device_id} 处理断点 '{breakpoint_name}' 时发生异常: {str(e)}")
|
||
|
||
logging.info(f"设备 {self.device_id} 上传配置管理执行完成，成功上传 {upload_success_count}/{len(global_variable.get_upload_breakpoint_list())} 个断点")
|
||
|
||
# 如果所有断点都上传成功，返回True；否则返回False
|
||
all_upload_success = upload_success_count == len(global_variable.get_upload_breakpoint_list())
|
||
if all_upload_success:
|
||
logging.info(f"设备 {self.device_id} 所有断点上传成功")
|
||
# 把上传成功的断点写入日志文件"上传成功的断点.txt"
|
||
with open(os.path.join(self.results_dir, "上传成功的断点.txt"), "w", encoding='utf-8') as f:
|
||
for bp in global_variable.get_upload_success_breakpoint_list():
|
||
f.write(f"{bp}\n")
|
||
else:
|
||
logging.warning(f"设备 {self.device_id} 部分断点上传失败")
|
||
# 把上传成功的断点写入日志文件"上传成功的断点.txt"
|
||
with open(os.path.join(self.results_dir, "上传成功的断点.txt"), "w", encoding='utf-8') as f:
|
||
for bp in global_variable.get_upload_success_breakpoint_list():
|
||
f.write(f"{bp}\n")
|
||
# 把上传失败的断点写入日志文件"上传失败的断点.txt"
|
||
with open(os.path.join(self.results_dir, "上传失败的断点.txt"), "w", encoding='utf-8') as f:
|
||
for bp in set(global_variable.get_upload_breakpoint_list())-set(global_variable.get_upload_success_breakpoint_list()):
|
||
f.write(f"{bp}\n")
|
||
|
||
return all_upload_success
|
||
|
||
except Exception as e:
|
||
logging.error(f"设备 {self.device_id} 处理应用状态时出错: {str(e)}")
|
||
return False
|
||
|
||
|
||
|
||
def check_and_click_confirm_popup_appium(self):
|
||
"""
|
||
适用于Appium的弹窗检测函数
|
||
|
||
Returns:
|
||
bool: 是否成功处理弹窗
|
||
"""
|
||
try:
|
||
from appium.webdriver.common.appiumby import AppiumBy
|
||
|
||
# 使用self.driver而不是参数
|
||
if not hasattr(self, 'driver') or self.driver is None:
|
||
logging.warning("driver未初始化，无法检测弹窗")
|
||
return False
|
||
|
||
# 检查弹窗消息
|
||
message_elements = self.driver.find_elements(AppiumBy.XPATH, '//android.widget.TextView[@text="是否退出测量界面？"]')
|
||
|
||
if message_elements:
|
||
logging.info("检测到退出测量界面弹窗")
|
||
|
||
# 点击"是"按钮
|
||
confirm_buttons = self.driver.find_elements(AppiumBy.XPATH, '//android.widget.Button[@text="是" and @resource-id="android:id/button1"]')
|
||
if confirm_buttons:
|
||
confirm_buttons[0].click()
|
||
logging.info("已点击'是'按钮")
|
||
time.sleep(1)
|
||
return True
|
||
else:
|
||
logging.warning("未找到'是'按钮")
|
||
return False
|
||
else:
|
||
return False
|
||
|
||
except Exception as e:
|
||
logging.error(f"Appium检测弹窗时发生错误: {str(e)}")
|
||
return False
|
||
|
||
|
||
|
||
def navigate_to_login_page(self, driver, device_id):
|
||
"""
|
||
补充的跳转页面函数：当设备处于未知状态时，尝试跳转到登录页面
|
||
|
||
参数:
|
||
driver: 已初始化的Appium WebDriver对象
|
||
device_id: 设备ID，用于日志记录
|
||
"""
|
||
try:
|
||
target_package = 'com.bjjw.cjgc'
|
||
target_activity = '.activity.LoginActivity'
|
||
# 使用ADB命令启动Activity
|
||
try:
|
||
logging.info(f"尝试使用ADB命令启动LoginActivity: {target_package}/{target_activity}")
|
||
adb_command = f"adb -s {device_id} shell am start -n {target_package}/{target_activity}"
|
||
result = subprocess.run(adb_command, shell=True, capture_output=True, text=True)
|
||
if result.returncode == 0:
|
||
logging.info(f"使用ADB命令启动LoginActivity成功")
|
||
time.sleep(2) # 等待Activity启动
|
||
return True
|
||
else:
|
||
logging.warning(f"ADB命令执行失败: {result.stderr}")
|
||
except Exception as adb_error:
|
||
logging.warning(f"执行ADB命令时出错: {adb_error}")
|
||
except Exception as e:
|
||
logging.error(f"跳转到登录页面过程中发生未预期错误: {e}")
|
||
|
||
# 所有尝试都失败
|
||
return False
|
||
|
||
|
||
|
||
|
||
def run_automation(self):
|
||
"""运行自动化流程"""
|
||
try:
|
||
success = self.handle_app_state()
|
||
# success = self.test_handle_app_state()
|
||
if success:
|
||
logging.info(f"设备 {self.device_id} 自动化流程执行成功")
|
||
else:
|
||
logging.error(f"设备 {self.device_id} 自动化流程执行失败")
|
||
return success
|
||
except Exception as e:
|
||
logging.error(f"设备 {self.device_id} 自动化执行过程中发生错误: {str(e)}")
|
||
return False
|
||
finally:
|
||
self.quit()
|
||
|
||
def quit(self):
|
||
"""关闭驱动"""
|
||
safe_quit_driver(getattr(self, 'driver', None), self.device_id)
|
||
|
||
@staticmethod
|
||
def start_upload(device_id=None, upload_time=None):
|
||
"""
|
||
供其他页面或模块调用的静态方法
|
||
执行完整的自动化流程
|
||
|
||
参数:
|
||
device_id: 可选的设备ID，如果为None则自动获取
|
||
|
||
返回:
|
||
bool: 自动化流程执行结果（True/False）
|
||
"""
|
||
automation = None
|
||
try:
|
||
# 创建自动化实例
|
||
automation = DeviceAutomation(device_id=device_id)
|
||
|
||
# 执行自动化流程
|
||
success = automation.run_automation()
|
||
|
||
if success:
|
||
logging.info("自动化流程执行成功")
|
||
else:
|
||
logging.error("自动化流程执行失败")
|
||
|
||
return success
|
||
|
||
except Exception as e:
|
||
logging.error(f"自动化流程执行出错: {str(e)}")
|
||
return False
|
||
finally:
|
||
# 确保资源被清理
|
||
if automation:
|
||
automation.quit()
|
||
|
||
# 主执行逻辑
|
||
if __name__ == "__main__":
|
||
|
||
# 单个设备配置 - 现在DeviceAutomation会自动获取设备ID
|
||
|
||
try:
|
||
automation = DeviceAutomation(device_id="192.168.1.100:5556")
|
||
success = automation.run_automation()
|
||
|
||
if success:
|
||
logging.info(f"设备自动化流程执行成功")
|
||
else:
|
||
logging.error(f"设备自动化流程执行失败")
|
||
except Exception as e:
|
||
logging.error(f"设备执行出错: {str(e)}")
|
||
|
||
|
||
|
||
|