From 1e17816a895b9405e56564539e3eb16339905ccb Mon Sep 17 00:00:00 2001 From: YiLin <482244139@qq.com> Date: Sat, 6 Jun 2026 18:08:22 +0800 Subject: [PATCH] =?UTF-8?q?action=E5=BB=BA=E7=AB=8B=E8=BF=9E=E6=8E=A5?= =?UTF-8?q?=EF=BC=88=E4=B8=BB=E9=A3=9F=E4=BA=86=E5=BC=80=E5=A7=8B=E6=B5=8B?= =?UTF-8?q?=E9=87=8F=E4=B9=8B=E5=90=8E=E7=9A=84=E9=80=BB=E8=BE=91=EF=BC=89?= =?UTF-8?q?=EF=BC=8Cadd=5Ftrasition=EF=BC=88=E8=B0=83=E7=94=A8run=5Fadd()?= =?UTF-8?q?=E5=87=BD=E6=95=B0)=E5=8A=A0=E8=BD=AC=E7=82=B9=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- __pycache__/check_station.cpython-312.pyc | Bin 45100 -> 46806 bytes actions.py | 86 +- add_transition_point.py | 146 +++ add_trasition.py | 1064 +++++++++++++++++ connect_appium.py | 169 +++ globals/__pycache__/apis.cpython-312.pyc | Bin 20873 -> 20873 bytes .../__pycache__/create_link.cpython-312.pyc | Bin 9819 -> 10484 bytes globals/create_link.py | 19 +- ...ection_mileage_config_page.cpython-312.pyc | Bin 46109 -> 46106 bytes page_objects/section_mileage_config_page.py | 2 +- 10 files changed, 1427 insertions(+), 59 deletions(-) create mode 100644 add_transition_point.py create mode 100644 add_trasition.py create mode 100644 connect_appium.py diff --git a/__pycache__/check_station.cpython-312.pyc b/__pycache__/check_station.cpython-312.pyc index 845dd87694ea4aa9020b4b09dc3836ec1d97aa76..cc79aecf3e6c1908025c7b07ca11c32e4507519b 100644 GIT binary patch delta 3352 zcmZ`*dr(tX8owtmk^nJC0)#*!5?daIM*=9;uA;Klu2x&L>eensl1g|4ytye4Zwk`Z zM@1A4DvFNpii?W&b{DHw-PdRz&V-s>nJn97XGn-MyHncH-JS02%%1OtKuTxtAHRF< z_x;ZAdz^F6y_ZJ?fBJ_Y_KVn9IYlnR{q(B!*JBN-=ox296(jjSB@}g(vLZLaFX|HU zx_vT*Tfhi`3l}TEYYWxIf_@Zwb_=Znw`e=%7PBH&SS_F^`mT&(qC#GZvI?h|5>`?r z!*R}Jku!!7;-O!tirbA$bePFXrkGJ|6f1t4q6sbRwTi=DM76JE@va_qGAm%C`bk8% z+)`F*l~iYt_#nP3LzU!MMod!PWQ-7!;eLJQsz_;hg!EBXWQ}f)C4wS{GGbQDG-sN@ zimT%AnceEl$z~vGstKt5Zp3UQE9#}JaVLd%>B=;Gy;O~t3Z(+40k`bd39YkPFeMO1 z{K=_yA*xBmQdQ}=M5o4FY5f10q5jc~d66?@&U8l4Hwd@9lo$@BBB2r@Cz3#!urdN? zdSpt#OPs;Wj+CEKLJ{F7Me>_e%zQpNfvTMFX!`D!v+}BZY{`w+Xu=Jq#jM;)S(B3~ z;-#w0BwoBg&bJ6Fqh%D=yN^443koPT^N_&o2hAcCy zOpVN%NEJ`f`GxvHR~#g?E|TA*Pp6pK;i?Q2JquA0iyu3k$Y$!0pnV;qis0tBWf5Ej zE9<2q%VH97+`^O=>#Rr-E9I8GBO_9KLZb%MKs5?CQw=C{&MU4XM7@$>>Ya;nbFCJ} zvd!YOo2#PQ#^e?in-?ssDBM=?R)NX7&}=O( z%qu7?%CqMB_ZQ z@(+GEe)F#X{KdeT3&uQSe!lR@3_UQ?_nwaS&6rs3c$S|ZF(sQYaVZ3j@|K`Eq+5YkFzQEqWu^U$cN3OUO z#N$8MIo^FL6m-#?C$Qu$8Q8-a^G~FjP&!=q+(NV^m41Uz=$`_tBj6L;9rZ587iD)^ z?c42)PiS{Av>n_9;FehG>unCJFYafyrfqc=+PccY*l5~S&(Nvh7cur)n@{YlvDxbJ z=jm^v8k}MHLrXM07bMC6-UOb~$ygYBouhn*%i>_{Op{Mp_qL5L51SpXT3_4`EH!o~ z<@|GwSCu=e z)j!r23~39z+M5ez@{g*9>wV$rz<=OSzcxKh*EV_-YM@f9aVHG{xcyajPIR! zWGOy=Zgn>G8*MURQHg@wekXpO;FsB|H4B9g1yVqH`WmzFp;=7E6`^k|7kKW0NKQAv z;x-b%x3V`NmW!J|TY~Db!PJl1u*b9g!9bJfK0#5kjA+JhE`(3WV zc3u$wW5)w7r!q^7F5}I$|3dfitB%paskZsVo7R=AUpW~Vg!$U|WDdb&K0qh|{8n89 z`i%R!ZmkS;;GCwvp+2l^)}u=}ulZW)V;~L_n9Qm*l+_5aG#BE$UKQTil!i?^mqq^t zm`eaIvX>-B4sADEkO_Q>uqx-li3fk@wl^1qx3*zes#pW0?m2>hcX`rLV!; zLcar(fbSDJ>Kf=V7zyf}K2g1esiep8$#efjHC*rcK>^ukWfxy=p&6LLZ&m&iR11&$ z09b%lfL#D>0Nnsr0R{jL1Hi8d{Vu=*fX@J$0Ga`w0zk{@8h{LdYykc}i{VoGUXY*$ zyy>a|MPv8XOytHFuRd7D9}uu}LLbi*@-yy>t*EqBRENJ`p(KB(_q}v+ZJpIsWBUnh J#McLQ{tsQ>&&dD) delta 1622 zcmZ{kYfMx}6vyw`d-uu$TNcYm8emVD? z|IGa7%xI;e)$xT14#l2CzsDfJ;efPvsYT9D%RPmFy0)T`#-L>nd42n;P)SB|~6lOGJSmUiO7zHx#XEcZFdN)l^i} zu@sSKDqdV^$Ii@E5aJ`52`U6}?8#HmjC)sRLr4^hp^+!8x*VnIByY!SS;wG>H)p?P zh8}e1Y=dUrp0joi9OI*#8cmKKQbiFSCG-)FV?p5`(9PEuy=L}O#z@Tbc?*jRd}X;g zg+-fld@q*etYtb??Eh5qy?MT}BHv3zRBES^>aVM-2nH~z&${;0+}|%bp=o zy?@Qql+y`k2<^CU>*F?>hJ8S2C7i{JTj#?dzq++vh6vBwo&s=?SC@kf`>XWMO`#xKfy^@KzKtSBn!CRZsws(3tw0>XiEHdVT4}^E;g$XA!>LQbRppLX+=$Y7kl0w>zDV!kHLYC;4vI)3-+6y7~5TE>V9X>7;TC9%}hmiucc zWYk|HIy=`}SXf5KVkp(MRY7(MOFJ*a7`nTX)jA<*NyT_Qm2dCbqKG|xbYL}H>g`Td zXY8kUFI7f`3SpH4>Upr+5esi(*OB!xZ6eF)_OiqLyCb(?#eL^Fbx-f>2@`T-&Q5>G zuW5|gX|UGYx70O35xrS#8R20<4`If|^L^=X1GT`}~r)Y$o{}%B?E!V`i5w|XPW0p6! zthmruKn-LKG*+RK9IS0*21TaYkfGH1wVmvHT=&jx*u{OPJ~TlDZ@;&p&5)~t+Vq{P zqOmoEFd;-}Bs3A43EhMN!XV)w;SiyPFhaOMpi^V@gzE(I&gh9@=>&Q=rms;fAASER z6}r$g>;b?Dku5 E0Rl?N6aWAK diff --git a/actions.py b/actions.py index 852e0a3..ec9a58e 100644 --- a/actions.py +++ b/actions.py @@ -24,27 +24,6 @@ import globals.create_link as create_link class DeviceAutomation: - # def __init__(self, device_id=None): - # # 如果没有提供设备ID，则自动获取 - # if device_id is None: - # self.device_id = driver_utils.get_device_id() - # else: - # self.device_id = device_id - - # # 初始化权限 - # if driver_utils.grant_appium_permissions(self.device_id): - # logging.info(f"设备 {self.device_id} 授予Appium权限成功") - # else: - # logging.warning(f"设备 {self.device_id} 授予Appium权限失败") - - # # 确保Appium服务器正在运行,不在运行则启动 - # if not driver_utils.check_server_status(4723): - # driver_utils.start_appium_server() - - # # 初始化Appium驱动和页面对象 - # self.init_driver() - # # 创建测试结果目录 - # self.results_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'test_results') def __init__(self, driver=None, wait=None, device_id=None): self.driver = driver self.wait = wait @@ -213,9 +192,7 @@ class DeviceAutomation: return False task_count = 0 - max_tasks = 1 # 最大任务数量，防止无限循环 - - # while task_count < max_tasks: + # 获取测量任务 logging.info(f"设备 {self.device_id} 获取测量任务 (第{task_count + 1}次)") # task_data = apis.get_measurement_task() @@ -243,15 +220,9 @@ class DeviceAutomation: logging.info(f"设备 {self.device_id} 当前要处理的项目名称：{global_variable.GLOBAL_CURRENT_PROJECT_NAME}") # 执行测量操作 - # logging.info(f"设备 {self.device_id} 开始执行测量操作") if not self.measure_tabbar_page.measure_tabbar_page_manager(): logging.error(f"设备 {self.device_id} 测量操作执行失败") - # # 返回到测量页面 - # self.driver.back() - # self.check_and_click_confirm_popup_appium() - - # continue # 继续下一个任务 logging.info(f"设备 {self.device_id} 测量页面操作执行成功") @@ -259,41 +230,33 @@ class DeviceAutomation: logging.info(f"设备 {self.device_id} 开始执行断面里程配置") if not self.section_mileage_config_page.section_mileage_config_page_manager(): logging.error(f"设备 {self.device_id} 断面里程配置执行失败") - # continue # 继续下一个任务 # 任务完成后短暂等待 logging.info(f"设备 {self.device_id} 第{task_count}个任务完成") task_count += 1 - # # +++++++++++++white+++++++++++++++++ - - # logging.info(f"设备 {self.device_id} 已完成{task_count}个任务，结束打数据流程") - # if task_count == 0: - # logging.error(f"没有完成打数据的线路，结束任务") + # # 执行打数据加转点 + # if not self.check_station.run(): + # logging.error(f"设备 {self.device_id} 打数据加转点执行失败") # return False - # 执行打数据加转点 - if not self.check_station.run(): - logging.error(f"设备 {self.device_id} 打数据加转点执行失败") - return False + # # 执行上传配置管理，传入当前断点名称 + # breakpoint_name = global_variable.GLOBAL_CURRENT_PROJECT_NAME + # line_num = global_variable.GLOBAL_LINE_NUM + # 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} 次重试") - # 执行上传配置管理，传入当前断点名称 - breakpoint_name = global_variable.GLOBAL_CURRENT_PROJECT_NAME - line_num = global_variable.GLOBAL_LINE_NUM - 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 task_count > 0 + # return task_count > 0 except Exception as e: logging.error(f"设备 {self.device_id} 处理应用状态时出错: {str(e)}") @@ -314,6 +277,15 @@ if __name__ == "__main__": logging.info(f"设备 {automation.device_id} 自动化流程执行成功") else: logging.error(f"设备 {automation.device_id} 自动化流程执行失败") + + # 保持脚本运行，不关闭连接 + logging.info("自动化流程执行完成，保持连接状态...") + logging.info("按 Ctrl+C 退出并关闭连接") + try: + while True: + time.sleep(1) + except KeyboardInterrupt: + logging.info("用户中断，准备关闭连接...") except Exception as e: logging.error(f"设备执行出错: {str(e)}") finally: diff --git a/add_transition_point.py b/add_transition_point.py new file mode 100644 index 0000000..de1d23a --- /dev/null +++ b/add_transition_point.py @@ -0,0 +1,146 @@ +""" +添加转点脚本 +使用已建立的 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 new file mode 100644 index 0000000..de829d1 --- /dev/null +++ b/add_trasition.py @@ -0,0 +1,1064 @@ +import logging +import time +import requests +import pandas as pd +from io import BytesIO +from datetime import datetime +import os +import subprocess +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 + + + + + +class CheckStation: + def __init__(self, driver=None, wait=None,device_id=None): + """初始化CheckStation对象""" + 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) + + + 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() + 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_station_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 add_run(self): + if not self.add_transition_point(): + print("添加转点失败") + return False + return True + + def main_run(self): + if not self.add_transition_point(): + logging.error("添加转点失败") + return False + if not self.take_screenshot(): + logging.error("截图失败") + return False + logging.info("检查站点成功") + return True + + def run(self): + # last_station_num = 0 + + url = f"https://database.yuxindazhineng.com/team-bucket/69378c5b4f42d83d9504560d/前测点表/20260309/CDWZQ-2标-龙家沟左线大桥-0-11号墩-平原.xlsx" + station_data = self.get_excel_from_url(url) + print(station_data) + station_quantity = len(station_data) #总站点数量 + over_station_num = 0 #已完成的站点数量 + over_station_list = [] #已完成的站点列表 + while over_station_num < station_quantity: + try: + # 键盘输出线路编号 + station_num_input = input("请输入线路编号：") + if not station_num_input.isdigit(): # 检查输入是否为数字 + print("输入错误：请输入一个整数") + continue + station_num = int(station_num_input) # 转为整数 + + if station_num in over_station_list: + print("已处理该站点，跳过") + continue + + # if last_station_num == station_num: + # print("输入与上次相同，跳过处理") + # continue + # last_station_num = station_num + + result = self.check_station_exists(station_data, station_num) + if result == "error": + print("处理错误：站点不存在") + # 错误处理逻辑，比如记录日志、发送警报等 + elif result == "add": + print("执行添加操作") + # 添加转点 + if not self.add_transition_point(): + print("添加转点失败") + # 可以决定是否继续循环 + continue + over_station_num += 1 + else: # result == "pass" + print("跳过处理") + over_station_num += 1 + + over_station_list.append(station_num) + + # 可以添加适当的延时，避免CPU占用过高 + # time.sleep(1) + + except KeyboardInterrupt: + print("程序被用户中断") + break + except Exception as e: + print(f"发生错误: {e}") + time.sleep(20) + # 错误处理，可以继续循环或退出 + print(f"已处理{over_station_num}个站点") + + # 滑动列表到底部 + 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_station.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} 处理返回按钮确认失败") + + 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__": + check_station = CheckStation() + # check_station.run() + check_station.add_run() \ No newline at end of file diff --git a/connect_appium.py b/connect_appium.py new file mode 100644 index 0000000..a18fefe --- /dev/null +++ b/connect_appium.py @@ -0,0 +1,169 @@ +""" +建立 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 dd90e9f92b5da1f9d5e945c4c2a19fbc7a7f9bd2..09edfdb407f219a894d504245df7449d1953f518 100644 GIT binary patch delta 28 icmeBN%-Fe@k?S-sFBbz4d_KC7>o_l?=;lkj+noV_atS>E delta 28 icmeBN%-Fe@k?S-sFBbz4cpu!zb)1*cbn_+N?alyqdI+%q diff --git a/globals/__pycache__/create_link.cpython-312.pyc b/globals/__pycache__/create_link.cpython-312.pyc index 8f844f349e718ad14d1736571e69757b538f38c3..39c63f423bd794de8a96d9e27c7ecfffa79659e7 100644 GIT binary patch delta 1794 zcmZuxeN0#{!iT}Yl4Yo?Vj|fV`PW3VJOmsf|7@WY63vL2|9fxY(YLvqyx)EAo^yWZ z
WJpQ=V_KVeO0%X1TU~kk{{*}!N|DCj38CXBV+VAVR!$c&$Kp>=FF~8B**Bw9C z)puTuiH#V#kHNN?m)xxkGo{00o=O&hX-K&MkRXEMCERCpC!lgR3IyG_M)`sO!K0wF z8-NZC(Pt!$q(xc;dddLkhe>N}hY0{74*_DUN6aV~*XHbF`fXoC!=Ri>I>aE|1t3-l zMx=`x0l)`2Hh~yLK`?oUl>h`Y2Z)+S<;J4uKoB!4)YCgJBwXfA27;47Wns-(Bzv^r znE~aYyePC{mg6|VBmfRb`nQ2(Kzfq=Yb0?aGDhu09Dqh#$@vqem`?GHObWkX>tRS) z49K(-zp@|N3$J;*vu(xOYUAzL;a z;y=5D0#&q)$QE_)2vnP}TOY};%EXQAPX$Q}lwHViU%+$CLGY zM?H+C3Iik!lI)F{amw$9`|v%#r*?m~WM#P)AXDSmo*FV8xq%MlI0iuIxn{8e2a66Q zj*vSNs|myl2=HsOH#1jK>ti3Ue=(7{J-0FU<@%NDagK-{j(x3+CnKHMU| zdTJs3y4-$xzAAi1?&y*`yX5fM`O34aE|2W1T67&;bMc=wFMErxHLua5mzP|}x7;*< z$3AS=^D~{reGJz?@(3}cSLp5$S+a@k6b|D|$?xWD3A1&L;GauNposr2Z3$e2kljKW aptW3n_cx8_fwo{tTezhIT28Daw*LXDLz_YX delta 1278 zcmZux&1(}u6rb6TY?94pqe)G(N%PSJnt)&tp-QDzX)S_ZrHCSyQk+)0tu{`!mBOqk z2%d5%&PlWf>89z6M`?PfQ%3;X8rK7Q}_-kaUulizQ~KN&_8 z!E^0z(q13<5I6A3mT5?Mq@6IoY1R<9=tU4UZC{_Q*B4fv&eWHwL&rEyPvEKcjr61> zy;S)7%wU2jJ6K>L-&a`)EQ}QDn+2clH!bglv+4H(#$Mn2*0$uU+ z&Pn1dyaR2B$BnQ^7ah}s0kY#PVki08WUlE(18fp3dW=oaAs9?5%FxGC9dMoLbSsc~ zT4cg}5af3@gi#%l^h^)L>a-1xz#qa=+t}P;c7J?KrP&CgBzMaoD3tlXStJqp{`aHayBc)*kvWi{N zuL7RZ!7BUxi;029V+o@vqy$%9_7T=lts>v1qkN)puIDP0qm~d^A#y_G`Gw*ae5oBR zVXXcXzYDJS@8wK-UUX$21csC)|# z^X+mmwQsD>(S`Z789LWkS#lcUW8$C7HJyO63vc7C#I;JEC$vKQPo*wdqhPE_z+0Rd nnt+2o5Mqe04}8@Rii=v5@Y!kwKj*KjcS7?RCw@KT->aE_dXvz? diff --git a/globals/create_link.py b/globals/create_link.py index 62eb24e..b3ea13a 100644 --- a/globals/create_link.py +++ b/globals/create_link.py @@ -191,7 +191,24 @@ def setup_adb_wireless(): target_port = 4723 print(f"🚀 开始无线 ADB 建链（端口 {target_port}）") - # 获取USB连接的设备 + # ===== 优先检查是否已有无线连接 ===== + wireless_connections = check_wireless_connections(target_port) + if wireless_connections: + print(f"📡 发现 {len(wireless_connections)} 个现有无线连接") + for conn_id, ip, port in wireless_connections: + if port == target_port: + print(f"✅ 已存在目标端口 {target_port} 的无线连接: {conn_id}") + # 验证连接是否有效 + verify_result = run_command("adb devices") + if conn_id in verify_result and "device" in verify_result.split(conn_id)[1][:10]: + print(f"✅ 连接验证通过") + # 启动 Appium（如果未运行） + start_appium() + return conn_id + else: + print(f"⚠️ 连接验证失败，准备重新建立") + + # ===== 获取USB连接的设备 ===== devices_output = run_command("adb devices") lines = devices_output.splitlines()[1:] 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 5f768ad4b39d2dc24575e9f9bbe27e86425b91d2..5c3e8732ba06ba8420ea94c65e94579c1abeb73c 100644 GIT binary patch delta 65 zcmV-H0KWg7=mMJP0u9Rz4GI7N0008xBWgCW4bQ9rDw7bc6%6IBjOK`t@4SrVzJ;>^ Xt$qUsNz@EV*b^6%?;IGjp0IxdH?ta; delta 52 zcmbRBf@$sxCf?J$yj%=G(ATP*Wx0{}{2E5x$#>Vtu{~S6`01P#oByq;V`lNW!sENS IVZ$tD01~|w!T