diff --git a/actions.py b/actions.py index ec9a58e..fbbb1af 100644 --- a/actions.py +++ b/actions.py @@ -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("自动化流程执行完成，保持连接状态...") diff --git a/add_transition_point.py b/add_transition_point.py deleted file mode 100644 index de1d23a..0000000 --- a/add_transition_point.py +++ /dev/null @@ -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) \ No newline at end of file diff --git a/add_trasition.py b/add_trasition.py index de829d1..45e1a4c 100644 --- a/add_trasition.py +++ b/add_trasition.py @@ -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() \ No newline at end of file + success = check_station.add_run() + if success: + print("添加转点成功") + else: + print("添加转点失败") \ No newline at end of file diff --git a/check_station.png b/check_station.png deleted file mode 100644 index ff218a3..0000000 Binary files a/check_station.png and /dev/null differ diff --git a/check_station.py b/check_station.py index 77cb982..bf56e64 100644 --- a/check_station.py +++ b/check_station.py @@ -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: diff --git a/check_upload.py b/check_upload.py new file mode 100644 index 0000000..e3fe95c --- /dev/null +++ b/check_upload.py @@ -0,0 +1,1042 @@ +import logging +import time +import requests +import pandas as pd +from io import BytesIO +from datetime import datetime +import os +import sys +import subprocess +import argparse +import globals.global_variable as global_variable +import globals.driver_utils as driver_utils # 导入驱动工具模块 +from selenium.common.exceptions import TimeoutException, NoSuchElementException, StaleElementReferenceException +from appium.webdriver.common.appiumby import AppiumBy +from selenium.webdriver.support import expected_conditions as EC +from selenium.webdriver.support.ui import WebDriverWait +import globals.ids as ids +from page_objects.upload_config_page import UploadConfigPage + + + + + +class checkUpload: + def __init__(self, driver=None, wait=None,device_id=None): + """初始化checkUpload对象""" + if device_id is None: + self.device_id = driver_utils.get_device_id() + else: + self.device_id = device_id + if driver is None or wait is None: + self.driver, self.wait = driver_utils.init_appium_driver(self.device_id) + else: + self.driver = driver + self.wait = wait + try: + if not driver_utils.check_session_valid(self.driver, self.device_id): + logging.warning(f"设备 {self.device_id} 会话无效，尝试重新连接驱动...") + self.driver, self.wait = driver_utils.reconnect_driver(self.device_id, self.driver) + if not self.driver: + logging.error(f"设备 {self.device_id} 驱动重连失败") + return False + except Exception as inner_e: + logging.warning(f"设备 {self.device_id} 检查会话状态时出错: {str(inner_e)}") + return False + + + # 检查应用是否成功启动 + if driver_utils.is_app_launched(self.driver): + logging.info(f"设备 {self.device_id} 沉降观测App已成功启动") + else: + logging.warning(f"设备 {self.device_id} 应用可能未正确启动") + driver_utils.check_app_status(self.driver) + + + self.upload_config_page = UploadConfigPage(self.driver, self.wait, self.device_id) + + + def get_measure_data(self): + + # 模拟获取测量数据 + pass + + + def add_transition_point(self): + """添加转点""" + try: + if not driver_utils.check_session_valid(self.driver, self.device_id): + logging.warning(f"设备 {self.device_id} 会话无效，尝试重新连接驱动...") + self.driver, self.wait = driver_utils.reconnect_driver(self.device_id, self.driver) + if not self.driver: + logging.error(f"设备 {self.device_id} 驱动重连失败") + return False + except Exception as inner_e: + logging.warning(f"设备 {self.device_id} 检查会话状态时出错: {str(inner_e)}") + 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: + logging.error("等待添加转点按钮超时") + return False + except Exception as e: + logging.error(f"添加转点时出错: {str(e)}") + return False + + def get_excel_from_url(self, url): + """ + 从URL获取Excel文件并解析为字典 + Excel只有一列数据（A列），每行是站点值 + + Args: + url: Excel文件的URL地址 + + Returns: + dict: 解析后的站点数据字典 {行号: 值}，失败返回None + """ + try: + print(f"正在从URL获取数据: {url}") + response = requests.get(url, timeout=30) + response.raise_for_status() # 检查请求是否成功 + + # 使用pandas读取Excel数据，指定没有表头，只读第一个sheet + excel_data = pd.read_excel( + BytesIO(response.content), + header=None, # 没有表头 + sheet_name=0, # 只读取第一个sheet + dtype=str # 全部作为字符串读取 + ) + + station_dict = {} + + # 解析Excel数据：使用行号+1作为站点编号，A列的值作为站点值 + print("解析Excel数据（使用行号作为站点编号）...") + for index, row in excel_data.iterrows(): + station_num = index + 1 # 行号从1开始作为站点编号 + station_value = str(row[0]).strip() if pd.notna(row[0]) else "" + + if station_value: # 只保存非空值 + station_dict[station_num] = station_value + + print(f"成功解析Excel，共{len(station_dict)}条数据") + return station_dict + + except requests.exceptions.RequestException as e: + print(f"请求URL失败: {e}") + return None + except Exception as e: + print(f"解析Excel失败: {e}") + return None + + def check_upload_exists(self, station_data: dict, station_num: int) -> str: + """ + 根据站点编号检查该站点的值是否以Z开头 + + Args: + station_data: 站点数据字典 {编号: 值} + station_num: 要检查的站点编号 + + Returns: + str: 如果站点存在且以Z开头返回"add"，否则返回"pass" + """ + if station_num not in station_data: + print(f"站点{station_num}不存在") + return "error" + + value = station_data[station_num] + str_value = str(value).strip() + is_z = str_value.upper().startswith('Z') + + result = "add" if is_z else "pass" + print(f"站点{station_num}: {value} -> {result}") + return result + + def run(self): + + # 滑动列表到底部 + if not self.scroll_list_to_bottom(self.device_id): + logging.error(f"设备 {self.device_id} 下滑列表到底部失败") + return False + + # 2. 点击最后一个spinner + if not self.click_last_spinner_with_retry(self.device_id): + logging.error(f"设备 {self.device_id} 点击最后一个spinner失败") + return False + + # 3. 再下滑一次 + if not self.scroll_down_once(self.device_id): + logging.warning(f"设备 {self.device_id} 再次下滑失败，但继续执行") + + + # # 截图 + # self.driver.save_screenshot("check_upload.png") + if not self.take_screenshot(): + logging.error(f"设备 {self.device_id} 截图失败") + + # 打完数据，截图完毕，点击平差处理按钮 + if not self.click_adjustment_button(self.device_id): + logging.error(f"设备 {self.device_id} 点击平差处理按钮失败") + return False + + item = global_variable.GLOBAL_CURRENT_PROJECT_NAME + if item.endswith('-平原'): + item = item[:-3] # 去掉最后3个字符"-平原" + # 检查是否在测量页面，在就重新执行选择断点，滑动列表到底部，点击最后一个spinner， 再下滑一次，点击平差处理按钮平差 + if not self.handle_back_navigation(item, self.device_id): + logging.error(f"{item}平差失败") + + + # 检测并处理"是 保留成果"弹窗 + if not self.handle_adjustment_result_dialog(): + logging.error("处理平差结果弹窗失败") + + + # 检查在不在测量列表页面，不在就点击返回按钮并处理弹窗 + if self.check_measurement_list(self.device_id): + logging.error(f"设备 {self.device_id} 未在测量列表页面") + # 点击返回按钮并处理弹窗 + if not self.execute_back_navigation_steps(self.device_id): + logging.error(f"设备 {self.device_id} 处理返回按钮确认失败") + # # 点击返回按钮并处理弹窗 + # if not self.execute_back_navigation_steps(self.device_id): + # logging.error(f"设备 {self.device_id} 处理返回按钮确认失败") + + + # 执行上传配置管理，传入当前断点名称 + breakpoint_name = global_variable.GLOBAL_CURRENT_PROJECT_NAME + line_num = global_variable.GLOBAL_LINE_NUM + self.results_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'test_results') + 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} 次重试") + + + return True + + def click_adjustment_button(self, device_id): + """ + 点击平差处理按钮 + + Args: + device_id: 设备ID + + Returns: + bool: 是否成功点击 + """ + try: + logging.info(f"设备 {device_id} 查找平差处理按钮") + + # 查找平差处理按钮 + adjustment_button = self.driver.find_element(AppiumBy.ID, "com.bjjw.cjgc:id/point_measure_btn") + + # 验证按钮文本 + button_text = adjustment_button.text + if "平差处理" not in button_text: + logging.warning(f"设备 {device_id} 按钮文本不匹配，期望'平差处理'，实际: {button_text}") + + if adjustment_button.is_displayed() and adjustment_button.is_enabled(): + logging.info(f"设备 {device_id} 点击平差处理按钮") + adjustment_button.click() + time.sleep(3) # 等待平差处理完成 + return True + else: + logging.error(f"设备 {device_id} 平差处理按钮不可点击") + return False + + except NoSuchElementException: + logging.error(f"设备 {device_id} 未找到平差处理按钮") + return False + except Exception as e: + logging.error(f"设备 {device_id} 点击平差处理按钮时发生错误: {str(e)}") + return False + + def handle_back_navigation(self, breakpoint_name, device_id): + """ + 完整的返回导航处理流程 + + Args: + breakpoint_name: 断点名称 + device_id: 设备ID + + Returns: + bool: 整个返回导航流程是否成功 + """ + try: + # time.sleep(2) + logging.info(f"已点击平差处理按钮，检查是否在测量页面") + + # 检测是否存在测量列表 + has_measurement_list = self.check_measurement_list(device_id) + if not has_measurement_list: + logging.info(f"设备 {device_id} 存在测量列表，重新执行平差流程") + + # 把断点名称给find_keyword + if not self.find_keyword(breakpoint_name): + logging.error(f"设备 {device_id} 未找到包含 {breakpoint_name} 的文件名") + return False + + if not self.handle_measurement_dialog(): + logging.error(f"设备 {device_id} 处理测量弹窗失败") + return False + + if not self.check_apply_btn(): + logging.error(f"设备 {device_id} 检查平差处理按钮失败") + return False + + + # 4. 点击平差处理按钮 + if not self.click_adjustment_button(device_id): + logging.error(f"设备 {device_id} 点击平差处理按钮失败") + return False + + logging.info(f"重新选择断点并点击平差处理按钮成功") + return True + + else: + logging.info(f"不在测量页面，继续执行后续返回操作") + return True + + except Exception as e: + logging.error(f"设备 {device_id} 处理返回导航时发生错误: {str(e)}") + return False + + def handle_adjustment_result_dialog(self): + """处理平差结果确认弹窗""" + try: + logging.info("开始检测平差结果弹窗") + + # 等待弹窗出现（最多等待5秒） + warning_dialog = WebDriverWait(self.driver, 5).until( + EC.presence_of_element_located((AppiumBy.ID, "android:id/parentPanel")) + ) + + # 验证弹窗内容 + alert_title = warning_dialog.find_element(AppiumBy.ID, "android:id/alertTitle") + alert_message = warning_dialog.find_element(AppiumBy.ID, "android:id/message") + + logging.info(f"检测到弹窗 - 标题: {alert_title.text}, 消息: {alert_message.text}") + + # 确认是目标弹窗 + if "警告" in alert_title.text and "是否保留测量成果" in alert_message.text: + logging.info("确认是平差结果确认弹窗") + + # 点击"是 保留成果"按钮 + yes_button = warning_dialog.find_element(AppiumBy.ID, "android:id/button1") + if yes_button.text == "是 保留成果": + yes_button.click() + logging.info("已点击'是 保留成果'按钮") + + # 等待弹窗消失 + WebDriverWait(self.driver, 5).until( + EC.invisibility_of_element_located((AppiumBy.ID, "android:id/parentPanel")) + ) + logging.info("弹窗已关闭") + return True + else: + logging.error(f"按钮文本不匹配，期望'是 保留成果'，实际: {yes_button.text}") + return False + else: + logging.warning("弹窗内容不匹配，不是目标弹窗") + return False + + except TimeoutException: + logging.info("未检测到平差结果弹窗，继续流程") + return True # 没有弹窗也是正常情况 + except Exception as e: + logging.error(f"处理平差结果弹窗时出错: {str(e)}") + return False + + def check_measurement_list(self, device_id): + """ + 检查是否存在测量列表 + + Args: + device_id: 设备ID + + Returns: + bool: 如果不存在测量列表返回True，存在返回False + """ + try: + # 等待线路列表容器出现 + self.wait.until( + EC.presence_of_element_located((AppiumBy.ID, ids.MEASURE_LIST_ID)) + ) + logging.info("线路列表容器已找到") + + # 如果存在MEASURE_LIST_ID，说明有测量列表，不需要执行后续步骤 + logging.info(f"设备 {device_id} 存在测量列表，无需执行后续返回操作") + return False + + except TimeoutException: + # 等待超时，说明没有测量列表 + logging.info(f"设备 {device_id} 未找到测量列表，可以继续执行后续步骤") + return True + except Exception as e: + logging.error(f"设备 {device_id} 检查测量列表时发生错误: {str(e)}") + return True + + def find_keyword(self, fixed_filename): + """查找指定关键词并点击，支持向下和向上滑动查找""" + try: + # if not check_session_valid(self.driver, self.device_id): + # logging.warning(f"设备 {self.device_id} 会话无效，尝试重新连接驱动...") + # if not reconnect_driver(self.device_id, self.driver): + # logging.error(f"设备 {self.device_id} 驱动重连失败") + + # 等待线路列表容器出现 + self.wait.until( + EC.presence_of_element_located((AppiumBy.ID, ids.MEASURE_LIST_ID)) + ) + logging.info("线路列表容器已找到") + + max_scroll_attempts = 100 # 最大滚动尝试次数 + scroll_count = 0 + previous_items = set() # 记录前一次获取的项目集合，用于检测是否到达边界 + + # 首先尝试向下滑动查找 + while scroll_count < max_scroll_attempts: + # 获取当前页面中的所有项目 + current_items = self.get_current_items() + logging.info(f"当前页面找到 {len(current_items)} 个项目: {current_items}") + + # 检查目标文件是否在当前页面中 + if fixed_filename in current_items: + logging.info(f"找到目标文件: {fixed_filename}") + # 点击目标文件 + if self.click_item_by_text(fixed_filename): + return True + else: + logging.error(f"点击目标文件失败: {fixed_filename}") + return False + + # 检查是否到达底部：连续两次获取的项目相同 + if current_items == previous_items and len(current_items) > 0: + logging.info("连续两次获取的项目相同，已到达列表底部") + break + + # 更新前一次项目集合 + previous_items = current_items.copy() + + # 向下滑动列表以加载更多项目 + if not self.scroll_list(direction="down"): + logging.error("向下滑动列表失败") + return False + + scroll_count += 1 + logging.info(f"第 {scroll_count} 次向下滑动，继续查找...") + + # 如果向下滑动未找到，尝试向上滑动查找 + logging.info("向下滑动未找到目标，开始向上滑动查找") + + # 重置滚动计数 + scroll_count = 0 + + while scroll_count < max_scroll_attempts: + # 向上滑动列表 + # 如果返回False，说明已经滑动到顶 + if not self.scroll_list(direction="up"): + # 检查是否是因为滑动到顶而返回False + if "已滑动到列表顶部" in logging.handlers[0].buffer[-1].message: + logging.info("已滑动到列表顶部，停止向上滑动") + break + else: + logging.error("向上滑动列表失败") + return False + + # 获取当前页面中的所有项目 + current_items = self.get_current_items() + # logging.info(f"向上滑动后找到 {len(current_items)} 个项目: {current_items}") + + # 检查目标文件是否在当前页面中 + if fixed_filename in current_items: + logging.info(f"找到目标文件: {fixed_filename}") + # 点击目标文件 + if self.click_item_by_text(fixed_filename): + return True + else: + logging.error(f"点击目标文件失败: {fixed_filename}") + return False + + scroll_count += 1 + logging.info(f"第 {scroll_count} 次向上滑动，继续查找...") + + logging.warning(f"经过 {max_scroll_attempts * 2} 次滑动仍未找到目标文件") + return False + + except TimeoutException: + logging.error("等待线路列表元素超时") + return False + except Exception as e: + logging.error(f"查找关键词时出错: {str(e)}") + return False + + def get_current_items(self): + """获取当前页面中的所有项目文本""" + try: + items = self.driver.find_elements(AppiumBy.ID, ids.MEASURE_LISTVIEW_ID) + item_texts = [] + + for item in items: + try: + title_element = item.find_element(AppiumBy.ID, ids.MEASURE_NAME_TEXT_ID) + if title_element and title_element.text: + item_texts.append(title_element.text) + except NoSuchElementException: + continue + + return item_texts + except Exception as e: + logging.error(f"获取当前项目失败: {str(e)}") + return [] + + def click_item_by_text(self, text): + """点击指定文本的项目""" + try: + # 查找包含指定文本的项目 + items = self.driver.find_elements(AppiumBy.ID, ids.MEASURE_LISTVIEW_ID) + + for item in items: + try: + title_element = item.find_element(AppiumBy.ID, ids.MEASURE_NAME_TEXT_ID) + if title_element and title_element.text == text: + title_element.click() + logging.info(f"已点击项目: {text}") + return True + except NoSuchElementException: + continue + + logging.warning(f"未找到可点击的项目: {text}") + return False + except Exception as e: + logging.error(f"点击项目失败: {str(e)}") + return False + + def handle_measurement_dialog(self): + """处理测量弹窗 - 选择继续测量""" + try: + logging.info("检查测量弹窗...") + + # 直接尝试点击"继续测量"按钮 + continue_btn = WebDriverWait(self.driver, 2).until( + EC.element_to_be_clickable((AppiumBy.ID, "com.bjjw.cjgc:id/measure_continue_btn")) + ) + continue_btn.click() + logging.info("已点击'继续测量'按钮") + return True + + except TimeoutException: + logging.info("未找到继续测量按钮，可能没有弹窗") + return True # 没有弹窗也认为是成功的 + except Exception as e: + logging.error(f"点击继续测量按钮时出错: {str(e)}") + return False + + # 检查有没有平差处理按钮 + def check_apply_btn(self): + """检查是否有平差处理按钮""" + try: + apply_btn = WebDriverWait(self.driver, 1).until( + EC.element_to_be_clickable((AppiumBy.ID, "com.bjjw.cjgc:id/point_measure_btn")) + ) + if apply_btn.is_displayed(): + logging.info("进入平差页面") + else: + logging.info("没有找到'平差处理'按钮") + return True + except TimeoutException: + logging.info("未找到平差处理按钮") + return False # 没有弹窗也认为是成功的 + except Exception as e: + logging.error(f"点击平差处理按钮时出错: {str(e)}") + return False + + + + def execute_back_navigation_steps(self, device_id): + """ + 执行实际的返回导航步骤 + + Args: + device_id: 设备ID + + Returns: + bool: 导航是否成功 + """ + try: + # 1. 首先点击返回按钮 + if not self.click_back_button(device_id): + logging.error(f"设备 {device_id} 点击返回按钮失败") + return False + + # 2. 处理返回确认弹窗 + logging.info(f"已点击返回按钮，等待处理返回确认弹窗") + if not self.handle_confirmation_dialog(device_id): + logging.error(f"设备 {device_id} 处理返回确认弹窗失败") + # return False + + + # 3. 验证是否成功返回到上一页面 + time.sleep(0.5) # 等待页面跳转完成 + + # 可以添加页面验证逻辑，比如检查是否返回到预期的页面 + # 这里可以根据实际应用添加特定的页面元素验证 + + logging.info(f"设备 {device_id} 返回导航流程完成") + return True + + except Exception as e: + logging.error(f"设备 {device_id} 执行返回导航步骤时发生错误: {str(e)}") + return False + + def click_back_button(self, device_id): + """点击手机系统返回按钮""" + try: + self.driver.back() + logging.info("已点击手机系统返回按钮") + return True + except Exception as e: + logging.error(f"点击手机系统返回按钮失败: {str(e)}") + return False + + def handle_confirmation_dialog(self, device_id, timeout=2): + """ + 处理确认弹窗，点击"是"按钮 + + Args: + device_id: 设备ID + timeout: 等待弹窗的超时时间 + + Returns: + bool: 是否成功处理弹窗 + """ + # 等待弹窗出现（最多等待2秒） + try: + max_attempts = 2 + for attempt in range(max_attempts): + try: + dialog_message = WebDriverWait(self.driver, timeout).until( + EC.presence_of_element_located((AppiumBy.XPATH, "//android.widget.TextView[@text='是否退出测量界面？']")) + ) + + logging.info(f"设备 {device_id} 检测到确认弹窗 (第 {attempt + 1} 次)") + + # 查找并点击"是"按钮 + confirm_button = self.driver.find_element( + AppiumBy.XPATH, + "//android.widget.Button[@text='是' and @resource-id='android:id/button1']" + ) + + if confirm_button.is_displayed() and confirm_button.is_enabled(): + logging.info(f"设备 {device_id} 点击确认弹窗的'是'按钮 (第 {attempt + 1} 次)") + confirm_button.click() + time.sleep(0.5) + + # 如果是第一次尝试，继续检查是否还有弹窗 + if attempt < max_attempts - 1: + logging.info(f"设备 {device_id} 等待 1 秒后检查是否还有弹窗") + time.sleep(0.5) + continue + return True + else: + logging.error(f"设备 {device_id} '是'按钮不可点击") + return False + except TimeoutException: + # 超时未找到弹窗，认为没有弹窗，返回成功 + logging.info(f"设备 {device_id} 等待 {timeout} 秒未发现确认弹窗，可能没有弹窗，返回成功") + return True + + except Exception as e: + logging.error(f"设备 {device_id} 处理确认弹窗时出错: {str(e)}") + return False + + + + def scroll_list_to_bottom(self, device_id, max_swipes=60): + """ + 下滑列表到最底端 + + Args: + device_id: 设备ID + max_swipes: 最大下滑次数 + + Returns: + bool: 是否滑动到底部 + """ + try: + logging.info(f"设备 {device_id} 开始下滑列表到底部") + + # 获取列表元素 + list_view = self.driver.find_element(AppiumBy.ID, "com.bjjw.cjgc:id/auto_data_list") + logging.info(f"时间戳1: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())}") + + same_content_count = 0 + + # 初始化第一次的子元素文本 + initial_child_elements = list_view.find_elements(AppiumBy.CLASS_NAME, "android.widget.TextView") + logging.info(f"时间戳2: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())}") + # current_child_texts = "|".join([ + # elem.text.strip() for elem in initial_child_elements + # if elem.text and elem.text.strip() + # ]) + current_child_texts = "|".join( + elem_text.strip() for elem in initial_child_elements + if (elem_text := elem.text) and elem_text.strip() + ) + logging.info(f"时间戳3: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())}") + for i in range(max_swipes): + # 执行下滑操作 + self.driver.execute_script("mobile: scrollGesture", { + 'elementId': list_view.id, + 'direction': 'down', + 'percent': 0.8, + 'duration': 500 + }) + + # 获取滑动后的子元素文本 + new_child_elements = list_view.find_elements(AppiumBy.CLASS_NAME, "android.widget.TextView") + new_child_texts = "|".join( + elem_text.strip() for elem in new_child_elements + if (elem_text := elem.text) and elem_text.strip() + ) + + # 判断内容是否变化：若连续3次相同，认为到达底部 + if new_child_texts == current_child_texts: + same_content_count += 1 + if same_content_count >= 2: + logging.info(f"设备 {device_id} 列表已滑动到底部，共滑动 {i+1} 次") + return True + else: + same_content_count = 0 # 内容变化，重置计数 + current_child_texts = new_child_texts # 更新上一次内容 + + logging.debug(f"设备 {device_id} 第 {i+1} 次下滑完成，当前子元素文本: {new_child_texts[:50]}...") # 打印部分文本 + + logging.warning(f"设备 {device_id} 达到最大下滑次数 {max_swipes}，可能未完全到底部") + return True + + except Exception as e: + logging.error(f"设备 {device_id} 下滑列表时发生错误: {str(e)}") + return False + + def click_last_spinner_with_retry(self, device_id, max_retries=2): + """带重试机制的点击方法""" + for attempt in range(max_retries): + try: + if self.click_last_spinner(device_id): + return True + logging.warning(f"设备 {device_id} 第{attempt + 1}次点击失败，准备重试") + time.sleep(0.5) # 重试前等待 + except Exception as e: + logging.error(f"设备 {device_id} 第{attempt + 1}次尝试失败: {str(e)}") + + logging.error(f"设备 {device_id} 所有重试次数已用尽") + return False + + def click_last_spinner(self, device_id): + """ + 点击最后一个spinner + + Args: + device_id: 设备ID + + Returns: + bool: 是否成功点击 + """ + try: + logging.info(f"设备 {device_id} 查找最后一个spinner") + + # 查找所有的spinner元素 + spinners = self.driver.find_elements(AppiumBy.ID, "com.bjjw.cjgc:id/spinner") + + if not spinners: + logging.error(f"设备 {device_id} 未找到任何spinner元素") + return False + + # 获取最后一个spinner + last_spinner = spinners[-1] + + if not (last_spinner.is_displayed() and last_spinner.is_enabled()): + logging.error(f"设备 {device_id} 最后一个spinner不可点击") + return False + + # 点击操作 + logging.info(f"设备 {device_id} 点击最后一个spinner") + last_spinner.click() + + # 执行额外一次下滑操作 + self.scroll_down_once(device_id) + + max_retries = 3 # 最大重试次数 + retry_count = 0 + wait_timeout = 5 # 增加等待时间到5秒 + + while retry_count < max_retries: + try: + # 确保device_id正确设置，使用全局变量作为备用 + if not hasattr(self, 'device_id') or not self.device_id: + # 优先使用传入的device_id，其次使用全局变量 + self.device_id = device_id if device_id else global_variable.get_device_id() + + # 使用self.device_id，确保有默认值 + actual_device_id = self.device_id if self.device_id else global_variable.get_device_id() + + if not driver_utils.check_session_valid(self.driver, actual_device_id): + logging.warning(f"设备 {actual_device_id} 会话无效，尝试重新连接驱动...") + try: + # 使用正确的设备ID进行重连 + new_driver, new_wait = reconnect_driver(actual_device_id, self.driver) + if new_driver: + self.driver = new_driver + self.wait = new_wait + logging.info(f"设备 {actual_device_id} 驱动重连成功") + else: + logging.error(f"设备 {actual_device_id} 驱动重连失败") + retry_count += 1 + continue + except Exception as e: + logging.error(f"设备 {actual_device_id} 驱动重连异常: {str(e)}") + retry_count += 1 + continue + + # 点击spinner（如果是重试，需要重新获取元素） + if retry_count > 0: + spinners = self.driver.find_elements(AppiumBy.CLASS_NAME, "android.widget.Spinner") + if not spinners: + logging.error(f"设备 {device_id} 未找到spinner元素") + retry_count += 1 + continue + last_spinner = spinners[-1] + if not (last_spinner.is_displayed() and last_spinner.is_enabled()): + logging.error(f"设备 {device_id} spinner不可点击") + retry_count += 1 + continue + logging.info(f"设备 {device_id} 重新点击spinner") + last_spinner.click() + # 重试时也执行下滑操作 + self.scroll_down_once(device_id) + + # 等待下拉菜单出现，增加等待时间到5秒 + wait = WebDriverWait(self.driver, wait_timeout) + detail_show = wait.until( + EC.presence_of_element_located((AppiumBy.ID, "com.bjjw.cjgc:id/detailshow")) + ) + + if detail_show.is_displayed(): + logging.info(f"设备 {device_id} spinner点击成功，下拉菜单已展开") + return True + else: + logging.error(f"设备 {device_id} 下拉菜单未显示") + retry_count += 1 + continue + + except Exception as wait_error: + error_msg = str(wait_error) + logging.error(f"设备 {device_id} 等待下拉菜单超时 (第{retry_count+1}次尝试): {error_msg}") + + # 检查是否是连接断开相关的错误 + if not driver_utils.check_session_valid(self.driver, self.device_id): + logging.warning(f"设备 {self.device_id} 会话无效，尝试重新连接驱动...") + # if any(keyword in error_msg for keyword in ['socket hang up', 'Could not proxy command']): + # logging.warning(f"设备 {device_id} 检测到连接相关错误，尝试重连...") + if not reconnect_driver(self.device_id, self.driver): + logging.error(f"设备 {device_id} 驱动重连失败") + + retry_count += 1 + if retry_count < max_retries: + logging.info(f"设备 {device_id} 将在1秒后进行第{retry_count+1}次重试") + time.sleep(1) # 等待1秒后重试 + + logging.error(f"设备 {device_id} 经过{max_retries}次重试后仍无法展开下拉菜单") + return False + + except Exception as e: + logging.error(f"设备 {device_id} 点击最后一个spinner时发生错误: {str(e)}") + return False + + def scroll_down_once(self, device_id): + """ + 再次下滑一次 + + Args: + device_id: 设备ID + + Returns: + bool: 是否成功下滑 + """ + try: + logging.info(f"设备 {device_id} 执行额外一次下滑") + + # 获取列表元素 + list_view = self.driver.find_element(AppiumBy.ID, "com.bjjw.cjgc:id/auto_data_list") + + # 执行下滑操作 + self.driver.execute_script("mobile: scrollGesture", { + 'elementId': list_view.id, + 'direction': 'down', + 'percent': 0.5 + }) + + time.sleep(0.2) + logging.info(f"设备 {device_id} 额外下滑完成") + return True + + except Exception as e: + logging.error(f"设备 {device_id} 额外下滑时发生错误: {str(e)}") + return False + + def take_screenshot(self): + """ + 通过Appium驱动截取设备屏幕 + + 参数: + filename_prefix: 断点名称 + + 返回: + bool: 操作是否成功 + """ + try: + # 获取项目名称 + project_name = global_variable.GLOBAL_USERNAME or "用户名" + filename_prefix = global_variable.GLOBAL_CURRENT_PROJECT_NAME or "平差页面截图" + + # 获取当前日期 + date_str = datetime.now().strftime("%Y%m%d") + # if not date_str: + # date_str = datetime.now().strftime("%Y%m%d") + + # 获取当前时间（如果没有提供），并确保格式合法（不含冒号） + time_str = datetime.now().strftime("%H%M%S") + + # 创建D盘下的截图目录结构：D:\uploadInfo\picture\项目名\年月日 + screenshots_dir = os.path.join("D:\\", "uploadInfo", "picture", project_name, date_str) + + # 确保目录存在 + try: + os.makedirs(screenshots_dir, exist_ok=True) + logging.info(f"截图目录: {screenshots_dir}") + except Exception as dir_error: + logging.error(f"创建截图目录失败: {str(dir_error)}") + return False + line_code = global_variable.GLOBAL_LINE_NUM + if not line_code: + logging.error(f"未找到与断点名称 {filename_prefix} 对应的线路编码") + line_code = "unknown" + + # 截图保存 + screenshot_file = os.path.join( + screenshots_dir, + f"{line_code}_{filename_prefix}_{time_str}.png" + ) + + # 尝试保存截图 + try: + success = self.driver.save_screenshot(screenshot_file) + if success: + logging.info(f"截图已保存: {screenshot_file}") + # 验证文件是否真的存在 + if os.path.exists(screenshot_file): + logging.info(f"截图文件验证存在: {screenshot_file}") + else: + logging.warning(f"截图文件保存成功但验证不存在: {screenshot_file}") + return True + else: + logging.error(f"Appium截图保存失败: {screenshot_file}") + return False + except Exception as save_error: + logging.error(f"保存截图时发生错误: {str(save_error)}") + return False + + except Exception as e: + logging.error(f"截图时发生错误: {str(e)}") + return False + + +def get_excel_from_url(url): + """ + 从URL获取Excel文件并解析为字典 + Excel只有一列数据（A列），每行是站点值 + + Args: + url: Excel文件的URL地址 + + Returns: + dict: 解析后的站点数据字典 {行号: 值}，失败返回None + """ + try: + print(f"正在从URL获取数据: {url}") + response = requests.get(url, timeout=30) + response.raise_for_status() # 检查请求是否成功 + + # 使用pandas读取Excel数据，指定没有表头，只读第一个sheet + excel_data = pd.read_excel( + BytesIO(response.content), + header=None, # 没有表头 + sheet_name=0, # 只读取第一个sheet + dtype=str # 全部作为字符串读取 + ) + + station_dict = {} + + # 解析Excel数据：使用行号+1作为站点编号，A列的值作为站点值 + print("解析Excel数据（使用行号作为站点编号）...") + for index, row in excel_data.iterrows(): + station_num = index + 1 # 行号从1开始作为站点编号 + station_value = str(row[0]).strip() if pd.notna(row[0]) else "" + + if station_value: # 只保存非空值 + station_dict[station_num] = station_value + + print(f"成功解析Excel，共{len(station_dict)}条数据") + return station_dict + + except requests.exceptions.RequestException as e: + print(f"请求URL失败: {e}") + return None + except Exception as e: + print(f"解析Excel失败: {e}") + return None + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description='check_upload 脚本 - 执行上传检查流程') + 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() + + # 设置全局变量 + global_variable.GLOBAL_USERNAME = args.user_name + global_variable.GLOBAL_CURRENT_PROJECT_NAME = args.line_name + global_variable.GLOBAL_LINE_NUM = args.line_num + global_variable.GLOBAL_ACCOUNT_ID = args.account_id + + logging.info(f"全局变量已设置: user_name={global_variable.GLOBAL_USERNAME}, " + f"line_name={global_variable.GLOBAL_CURRENT_PROJECT_NAME}, " + f"line_num={global_variable.GLOBAL_LINE_NUM}, " + f"account_id={global_variable.GLOBAL_ACCOUNT_ID}") + + check_upload = checkUpload() + success = check_upload.run() + if success: + print("上传检查执行成功") + else: + print("上传检查执行失败") \ No newline at end of file diff --git a/connect_appium.py b/connect_appium.py deleted file mode 100644 index a18fefe..0000000 --- a/connect_appium.py +++ /dev/null @@ -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("连接建立失败") \ No newline at end of file diff --git a/globals/__pycache__/apis.cpython-312.pyc b/globals/__pycache__/apis.cpython-312.pyc index 09edfdb..2b1e579 100644 Binary files a/globals/__pycache__/apis.cpython-312.pyc and b/globals/__pycache__/apis.cpython-312.pyc differ diff --git a/globals/__pycache__/global_variable.cpython-312.pyc b/globals/__pycache__/global_variable.cpython-312.pyc index ce0d191..2207843 100644 Binary files a/globals/__pycache__/global_variable.cpython-312.pyc and b/globals/__pycache__/global_variable.cpython-312.pyc differ diff --git a/globals/apis.py b/globals/apis.py index 1ec7884..b3d364d 100644 --- a/globals/apis.py +++ b/globals/apis.py @@ -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: diff --git a/globals/global_variable.py b/globals/global_variable.py index 91262ec..b24c5b2 100644 --- a/globals/global_variable.py +++ b/globals/global_variable.py @@ -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 = {} diff --git a/page_objects/__pycache__/section_mileage_config_page.cpython-312.pyc b/page_objects/__pycache__/section_mileage_config_page.cpython-312.pyc index 5c3e873..13d736d 100644 Binary files a/page_objects/__pycache__/section_mileage_config_page.cpython-312.pyc and b/page_objects/__pycache__/section_mileage_config_page.cpython-312.pyc differ diff --git a/page_objects/login_page.py b/page_objects/login_page.py index fd66518..2b5956e 100644 --- a/page_objects/login_page.py +++ b/page_objects/login_page.py @@ -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): """检查登录是否成功""" diff --git a/page_objects/section_mileage_config_page.py b/page_objects/section_mileage_config_page.py index 8a0b71b..26747c7 100644 --- a/page_objects/section_mileage_config_page.py +++ b/page_objects/section_mileage_config_page.py @@ -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