YiLin/cjgc_data
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1. actions.py 2.add_trasition.py 3.check_upload.py

Browse Source
This commit is contained in:
YiLin
2026-06-09 14:52:29 +08:00
parent 1e17816a89
commit de5173ded6
14 changed files with 1208 additions and 359 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
actions.py
View File

@@ -1,7 +1,9 @@
import logging
import os
import sys
import time
import subprocess
import argparse
from tkinter import E
from appium import webdriver
from appium.options.android import UiAutomator2Options
@@ -143,8 +145,14 @@ class DeviceAutomation:
logging.error(f"设备 {self.device_id} 初始化驱动失败: {str(e)}")
raise
def run_automation(self):
"""根据当前应用状态处理相应的操作"""
def run_automation(self, task_data=None):
"""根据当前应用状态处理相应的操作
Args:
task_data: 可选的任务数据字典，包含 user_name, line_name, line_num, id 等字段。
如果传入，则使用传入值设置全局变量；
如果不传，则使用硬编码的默认值。
"""
try:
max_retry = 3 # 限制最大重试次数
retry_count = 0
@@ -193,19 +201,21 @@ class DeviceAutomation:
task_count = 0
# 获取测量任务
logging.info(f"设备 {self.device_id} 获取测量任务 (第{task_count + 1}次)")
# task_data = apis.get_measurement_task()
# logging.info(f"设备 {self.device_id} 获取到的测量任务: {task_data}")
task_data = {
"id": 21,
"user_name": "czsczq115ykl",
"name": "czsczq115ykl",
"line_num": "L220076",
"line_name": "CZSCZQ-11-五工区-列衣隧道进口-D3K638+507-D3K638+607-山区",
"remaining": "0",
"status": 0
}
# 获取测量任务：优先使用传入的 task_data，否则使用硬编码默认值
if task_data is None:
logging.info(f"设备 {self.device_id} 获取测量任务 (第{task_count + 1}次)")
# task_data = apis.get_measurement_task()
# logging.info(f"设备 {self.device_id} 获取到的测量任务: {task_data}")
task_data = {
"id": 21,
"user_name": "czsczq115ykl",
"name": "czsczq115ykl",
"line_num": "L220179",
"line_name": "CZSCZQ-11-五工区-德达隧道6#斜井左线-DK632+707-DK632+705-山区",
"remaining": "0",
"status": 0
}
if not task_data:
logging.info(f"设备 {self.device_id} 未获取到状态为1的测量任务，等待后重试")
time.sleep(1) # 等待1秒后重试
@@ -264,6 +274,14 @@ class DeviceAutomation:
# 主执行逻辑
if __name__ == "__main__":
parser = argparse.ArgumentParser(description='DeviceAutomation 脚本 - 执行设备自动化流程')
parser.add_argument('--user_name', type=str, default='', help='用户名')
parser.add_argument('--line_name', type=str, default='', help='项目/线路名称')
parser.add_argument('--line_num', type=str, default='', help='线路编码')
parser.add_argument('--account_id', type=int, default=0, help='账号ID')
args = parser.parse_args()
device_id = create_link.setup_adb_wireless()
automation = None
try:
@@ -271,12 +289,27 @@ if __name__ == "__main__":
logging.error("未能获取设备 ID，无法继续执行自动化流程")
else:
automation = DeviceAutomation(device_id=device_id)
success = automation.run_automation()
# 如果传入了命令行参数，则构建 task_data 并传入
if args.user_name or args.line_name or args.line_num or args.account_id:
task_data = {
"id": args.account_id,
"user_name": args.user_name,
"line_num": args.line_num,
"line_name": args.line_name,
}
logging.info(f"使用命令行参数: {task_data}")
success = automation.run_automation(task_data=task_data)
else:
logging.info("未传入命令行参数，使用默认 task_data")
success = automation.run_automation()
if success:
logging.info(f"设备 {automation.device_id} 自动化流程执行成功")
print("自动化流程执行成功")
else:
logging.error(f"设备 {automation.device_id} 自动化流程执行失败")
print("自动化流程执行失败")
# 保持脚本运行，不关闭连接
logging.info("自动化流程执行完成，保持连接状态...")

146
add_transition_point.py
View File

@@ -1,146 +0,0 @@
"""
添加转点脚本
使用已建立的 Appium 连接来添加转点
"""
import json
import os
import logging
import sys
from selenium.common.exceptions import TimeoutException, NoSuchElementException
from appium.webdriver.common.appiumby import AppiumBy
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.ui import WebDriverWait
# 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 load_session_info():
"""从文件加载 session 信息"""
if not os.path.exists(SESSION_FILE):
logging.error(f"Session 文件不存在: {SESSION_FILE}")
logging.error("请先运行 connect_appium.py 建立连接")
return None
try:
with open(SESSION_FILE, "r", encoding="utf-8") as f:
return json.load(f)
except Exception as e:
logging.error(f"读取 session 文件失败: {e}")
return None
def attach_to_session(session_id, device_id):
"""Attach 到现有的 Appium session"""
from appium import webdriver
from appium.options.android import UiAutomator2Options
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
# 验证 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 get_driver_from_connection():
"""从已建立的连接获取 driver"""
session_info = load_session_info()
if not session_info:
return None, None, None
session_id = session_info.get("session_id")
device_id = session_info.get("device_id")
if not session_id or not device_id:
logging.error("Session 信息不完整")
return None, None, None
driver, wait = attach_to_session(session_id, device_id)
return driver, wait, device_id
def add_transition_point(driver, wait, device_id=None):
"""
添加转点
参数:
driver: WebDriver 实例
wait: WebDriverWait 实例
device_id: 设备ID（可选）
返回:
bool: 是否成功
"""
device_str = f"设备 {device_id} " if device_id else ""
try:
# 查找并点击添加转点按钮
add_transition_btn = wait.until(
EC.element_to_be_clickable((AppiumBy.ID, "com.bjjw.cjgc:id/btn_add_ZPoint"))
)
add_transition_btn.click()
logging.info(f"{device_str}已点击添加转点按钮")
return True
except TimeoutException:
logging.error(f"{device_str}等待添加转点按钮超时")
return False
except Exception as e:
logging.error(f"{device_str}添加转点时出错: {str(e)}")
return False
def main():
"""主函数"""
# 获取连接
driver, wait, device_id = get_driver_from_connection()
if not driver:
logging.error("无法获取 Appium 连接，请先运行 connect_appium.py")
return False
logging.info(f"成功连接到设备: {device_id}")
# 添加转点
result = add_transition_point(driver, wait, device_id)
if result:
logging.info("添加转点成功！")
else:
logging.error("添加转点失败")
return result
if __name__ == "__main__":
success = main()
sys.exit(0 if success else 1)

10
add_trasition.py
View File

@@ -70,11 +70,13 @@ class CheckStation:
return False
try:
# 查找并点击添加转点按钮
# 查找并点击添加转点按钮（双击确保点到）
add_transition_btn = self.wait.until(
EC.element_to_be_clickable((AppiumBy.ID, "com.bjjw.cjgc:id/btn_add_ZPoint"))
)
add_transition_btn.click()
time.sleep(0.3)
add_transition_btn.click()
logging.info("已点击添加转点按钮")
return True
except TimeoutException:
@@ -1061,4 +1063,8 @@ def get_excel_from_url(url):
if __name__ == "__main__":
check_station = CheckStation()
# check_station.run()
check_station.add_run()
success = check_station.add_run()
if success:
print("添加转点成功")
else:
print("添加转点失败")

BIN
check_station.png
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 143 KiB

4
check_station.py
View File

@@ -110,11 +110,13 @@ class CheckStation:
return False
try:
# 查找并点击添加转点按钮
# 查找并点击添加转点按钮（双击确保点到）
add_transition_btn = self.wait.until(
EC.element_to_be_clickable((AppiumBy.ID, "com.bjjw.cjgc:id/btn_add_ZPoint"))
)
add_transition_btn.click()
time.sleep(0.3)
add_transition_btn.click()
logging.info("已点击添加转点按钮")
return True
except TimeoutException:

1042
check_upload.py Normal file
View File

File diff suppressed because it is too large Load Diff

169
connect_appium.py
View File

@@ -1,169 +0,0 @@
"""
建立 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("连接建立失败")

BIN
globals/__pycache__/apis.cpython-312.pyc
View File

Binary file not shown.

BIN
globals/__pycache__/global_variable.cpython-312.pyc
View File

Binary file not shown.

4
globals/apis.py
View File

@@ -91,8 +91,8 @@ def get_breakpoint_list():
"id": 21,
"user_name": "czsczq115ykl",
"name": "czsczq115ykl",
"line_num": "L220076",
"line_name": "CZSCZQ-11-五工区-列衣隧道进口-D3K638+507-D3K638+607-山区",
"line_num": "L220179",
"line_name": "CZSCZQ-11-五工区-德达隧道6#斜井左线-DK632+707-DK632+705-山区",
"status": 3
}]
else:

4
globals/global_variable.py
View File

@@ -2,8 +2,8 @@
GLOBAL_DEVICE_ID = "" # 设备ID
GLOBAL_USERNAME = "czyuzongwen" # 用户名
GLOBAL_ACCOUNT_ID = "" # 账号ID
GLOBAL_CURRENT_PROJECT_NAME = "" # 当前测试项目名称
GLOBAL_LINE_NUM = "" # 线路编码
GLOBAL_CURRENT_PROJECT_NAME = "CZSCZQ-11-五工区-德达隧道6#斜井左线-DK632+707-DK632+705-山区" # 当前测试项目名称
GLOBAL_LINE_NUM = "L220179" # 线路编码
GLOBAL_BREAKPOINT_STATUS_CODES = [0,3] # 要获取的断点状态码列表
GLOBAL_UPLOAD_BREAKPOINT_LIST = []
GLOBAL_UPLOAD_BREAKPOINT_DICT = {}

BIN
page_objects/__pycache__/section_mileage_config_page.cpython-312.pyc
View File

Binary file not shown.

98
page_objects/login_page.py
View File

@@ -185,6 +185,104 @@ class LoginPage:
logging.error(f"登录过程中出错: {str(e)}")
return False
def login_get_username(self, username=None):
"""执行登录操作"""
try:
logging.info("正在执行登录操作...")
# 获取文本框中已有的用户名
username_field = self.wait.until(
EC.element_to_be_clickable((AppiumBy.ID, ids.LOGIN_USERNAME))
)
existing_username = username_field.text
# 日志记录获取到的已有用户名（若为空，也需明确记录，避免后续误解）
if existing_username.strip(): # 去除空格后判断是否有有效内容
logging.info(f"已获取文本框中的已有用户名: {existing_username}")
else:
logging.info("文本框中未检测到已有用户名（内容为空）")
# 将用户名写入全局变量中
global_variable.GLOBAL_USERNAME = existing_username # 关键：给全局变量赋值
# 1. 定位密码输入框
password_field = self.wait.until(
EC.element_to_be_clickable((AppiumBy.ID, ids.LOGIN_PASSWORD))
)
# 2. 清空密码框（如果需要）
try:
password_field.clear()
# time.sleep(0.5) # 等待清除完成
except:
# 如果clear方法不可用，尝试其他方式
pass
accounts = apis.get_accounts_from_server("68ef0e02b0138d25e2ac9918")
# 检查accounts是否为None，如果是则设为空列表
if accounts is None:
logging.warning("获取账户列表返回None，设为空列表")
accounts = []
matches = [acc for acc in accounts if acc.get("username") == existing_username]
password = None
account_id = False
if matches:
password = matches[0].get("password")
# ✅ 关键：把 account_id 存入全局变量
account_id = matches[0].get("account_id", False)
# 只有 account_id 存在时才存全局
if account_id is not False:
global_variable.GLOBAL_ACCOUNT_ID = account_id
logging.info(f"匹配到账号信息：username={existing_username}, account_id={account_id}")
else:
logging.warning(f"账号 {existing_username} 未返回 account_id，已设为 False")
password_field.send_keys(password)
# 4. 可选：隐藏键盘
try:
self.driver.hide_keyboard()
except:
pass
# 点击登录按钮
max_retries = 3
retry_count = 0
while retry_count < max_retries:
login_btn = self.wait.until(
EC.element_to_be_clickable((AppiumBy.ID, ids.LOGIN_BTN))
)
login_btn.click()
logging.info(f"已点击登录按钮 (尝试 {retry_count + 1}/{max_retries})")
# 等待登录完成
time.sleep(3)
# 检查是否登录成功
if self.is_login_successful():
logging.info("登录成功")
return True
else:
logging.warning("登录后未检测到主页面元素，准备重试")
retry_count += 1
if retry_count < max_retries:
logging.info(f"等待2秒后重新尝试登录...")
time.sleep(2)
logging.error(f"登录失败，已尝试 {max_retries} 次")
return False
except Exception as e:
logging.error(f"登录过程中出错: {str(e)}")
return False
def is_login_successful(self):
"""检查登录是否成功"""

25
page_objects/section_mileage_config_page.py
View File

@@ -741,11 +741,13 @@ class SectionMileageConfigPage:
def add_transition_point(self):
"""添加转点"""
try:
# 查找并点击添加转点按钮
# 查找并点击添加转点按钮（双击确保点到）
add_transition_btn = self.wait.until(
EC.element_to_be_clickable((AppiumBy.ID, "com.bjjw.cjgc:id/btn_add_ZPoint"))
)
add_transition_btn.click()
time.sleep(0.3)
add_transition_btn.click()
self.logger.info("已点击添加转点按钮")
return True
except TimeoutException:
@@ -981,7 +983,7 @@ class SectionMileageConfigPage:
try:
self.logger.info("检查线路弹出测量弹窗...")
# 直接尝试点击"继续测量"按钮
# 直接尝试点击"重新测量"按钮
remeasure_btn = WebDriverWait(self.driver, 2).until(
EC.element_to_be_clickable((AppiumBy.ID, "com.bjjw.cjgc:id/measure_remeasure_all_btn"))
)
@@ -1086,25 +1088,6 @@ class SectionMileageConfigPage:
if not self.click_start_measure_btn():
return False
# if not self.check_station_page.run():
# self.logger.error("检查站页面运行失败")
# return False
# # 添加断点到列表
# if not self.add_breakpoint_to_tested_list():
# return False
# # 点击返回按钮
# if not self.click_back_button():
# return False
# # 测量结束。点击手机物理返回按钮,返回测量页面
# # 点击了手机独步导航栏返回键
# if not self.click_system_back_button():
# return False
self.logger.info("断面里程配置完成，待执行测量")
return True