169 lines
5.6 KiB
Python
169 lines
5.6 KiB
Python
"""
|
||
建立 Appium 连接脚本
|
||
运行此脚本会创建 Appium session 并保存 session 信息到文件
|
||
"""
|
||
import json
|
||
import os
|
||
import logging
|
||
import globals.driver_utils as driver_utils
|
||
import globals.global_variable as global_variable
|
||
|
||
# Session 信息保存路径
|
||
SESSION_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "appium_session.json")
|
||
|
||
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s: %(message)s")
|
||
|
||
|
||
def create_connection():
|
||
"""创建 Appium 连接并保存 session 信息"""
|
||
try:
|
||
# 获取设备ID
|
||
device_id = driver_utils.get_device_id()
|
||
if not device_id:
|
||
logging.error("未找到可用设备")
|
||
return None, None, None
|
||
|
||
logging.info(f"使用设备: {device_id}")
|
||
global_variable.GLOBAL_DEVICE_ID = device_id
|
||
|
||
# 确保 Appium 服务器运行
|
||
if not driver_utils.check_server_status(4723):
|
||
logging.info("Appium 服务器未运行，正在启动...")
|
||
driver_utils.start_appium_server()
|
||
|
||
# 授予权限
|
||
if driver_utils.grant_appium_permissions(device_id):
|
||
logging.info(f"设备 {device_id} 授予Appium权限成功")
|
||
else:
|
||
logging.warning(f"设备 {device_id} 授予Appium权限失败")
|
||
|
||
# 初始化驱动
|
||
driver, wait = driver_utils.init_appium_driver(device_id)
|
||
|
||
if not driver:
|
||
logging.error("驱动初始化失败")
|
||
return None, None, None
|
||
|
||
# 保存 session 信息到文件
|
||
session_info = {
|
||
"session_id": driver.session_id,
|
||
"server_url": driver.command_executor._url,
|
||
"device_id": device_id,
|
||
"capabilities": dict(driver.capabilities) if driver.capabilities else {}
|
||
}
|
||
|
||
with open(SESSION_FILE, "w", encoding="utf-8") as f:
|
||
json.dump(session_info, f, ensure_ascii=False, indent=2)
|
||
|
||
logging.info(f"Session 信息已保存到: {SESSION_FILE}")
|
||
logging.info(f"Session ID: {driver.session_id}")
|
||
|
||
return driver, wait, device_id
|
||
|
||
except Exception as e:
|
||
logging.error(f"创建连接失败: {e}")
|
||
return None, None, None
|
||
|
||
|
||
def get_connection():
|
||
"""获取现有连接（如果存在且有效）"""
|
||
if not os.path.exists(SESSION_FILE):
|
||
logging.info("Session 文件不存在，需要创建新连接")
|
||
return None, None, None
|
||
|
||
try:
|
||
with open(SESSION_FILE, "r", encoding="utf-8") as f:
|
||
session_info = json.load(f)
|
||
|
||
session_id = session_info.get("session_id")
|
||
device_id = session_info.get("device_id")
|
||
|
||
if not session_id or not device_id:
|
||
logging.warning("Session 文件信息不完整")
|
||
return None, None, None
|
||
|
||
# 尝试 attach 到现有 session
|
||
driver, wait = attach_to_session(session_id, device_id)
|
||
|
||
if driver:
|
||
global_variable.GLOBAL_DEVICE_ID = device_id
|
||
logging.info(f"成功连接到现有 Session: {session_id}")
|
||
return driver, wait, device_id
|
||
else:
|
||
logging.warning("无法连接到现有 Session")
|
||
return None, None, None
|
||
|
||
except Exception as e:
|
||
logging.error(f"获取连接失败: {e}")
|
||
return None, None, None
|
||
|
||
|
||
def attach_to_session(session_id, device_id):
|
||
"""Attach 到现有的 Appium session"""
|
||
from appium import webdriver
|
||
from appium.options.android import UiAutomator2Options
|
||
from selenium.webdriver.support.ui import WebDriverWait
|
||
|
||
try:
|
||
server_url = "http://127.0.0.1:4723/wd/hub"
|
||
|
||
# 创建 options，但不启动新应用
|
||
options = UiAutomator2Options()
|
||
options.platform_name = "Android"
|
||
options.device_name = device_id
|
||
options.automation_name = "UiAutomator2"
|
||
options.udid = device_id
|
||
|
||
# 连接到现有 session
|
||
driver = webdriver.Remote(
|
||
server_url,
|
||
options=options
|
||
)
|
||
|
||
# 手动设置 session_id
|
||
driver.session_id = session_id
|
||
driver.command_executor._commands["getSession"] = ("GET", "/session/$sessionId")
|
||
|
||
# 验证 session 是否有效
|
||
try:
|
||
current_package = driver.current_package
|
||
logging.info(f"Session 有效，当前应用: {current_package}")
|
||
except Exception as e:
|
||
logging.error(f"Session 无效: {e}")
|
||
return None, None
|
||
|
||
wait = WebDriverWait(driver, 20)
|
||
return driver, wait
|
||
|
||
except Exception as e:
|
||
logging.error(f"Attach 到 session 失败: {e}")
|
||
return None, None
|
||
|
||
|
||
def close_connection():
|
||
"""关闭连接并清理 session 文件"""
|
||
global_variable.GLOBAL_DEVICE_ID = None
|
||
|
||
if os.path.exists(SESSION_FILE):
|
||
os.remove(SESSION_FILE)
|
||
logging.info("Session 文件已删除")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 作为独立脚本运行时，创建新连接
|
||
driver, wait, device_id = create_connection()
|
||
|
||
if driver:
|
||
logging.info("连接建立成功！")
|
||
logging.info("可以运行 add_transition_point.py 来添加转点")
|
||
logging.info("按 Ctrl+C 退出并保持连接...")
|
||
|
||
try:
|
||
# 保持连接，等待用户操作
|
||
import time
|
||
while True:
|
||
time.sleep(1)
|
||
except KeyboardInterrupt:
|
||
logging.info("用户中断，保持连接状态")
|
||
else:
|
||
logging.error("连接建立失败") |