305 lines
12 KiB
Python
305 lines
12 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.screenshot_page import ScreenshotPage
|
||
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, project_name=None):
|
||
self.device_id = device_id
|
||
self.project_name = project_name
|
||
|
||
# 设置项目名称到全局变量
|
||
if project_name:
|
||
global_variable.set_current_project_name(project_name)
|
||
logging.info(f"设备 {self.device_id} 已设置项目名称: {project_name}")
|
||
|
||
# 初始化权限
|
||
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.screenshot_page = ScreenshotPage(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)
|
||
|
||
|
||
# 获取状态为3的线路。
|
||
apis.get_line_info_and_save_global(user_name=global_variable.get_username());
|
||
|
||
# 点击测量导航栏按钮
|
||
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)
|
||
return True
|
||
|
||
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, project_name=None):
|
||
"""
|
||
供其他页面或模块调用的静态方法
|
||
执行完整的自动化流程
|
||
|
||
参数:
|
||
device_id: 可选的设备ID，如果为None则自动获取
|
||
upload_time: 上传时间
|
||
project_name: 项目名称
|
||
|
||
返回:
|
||
bool: 自动化流程执行结果（True/False）
|
||
"""
|
||
automation = None
|
||
try:
|
||
# 创建自动化实例
|
||
automation = DeviceAutomation(device_id=device_id, project_name=project_name)
|
||
|
||
# 执行自动化流程
|
||
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)}")
|
||
|
||
|
||
|
||
|