278 lines
11 KiB
Python
278 lines
11 KiB
Python
import logging
|
||
import time
|
||
import requests
|
||
import pandas as pd
|
||
from io import BytesIO
|
||
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
|
||
|
||
|
||
|
||
|
||
|
||
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
|
||
# 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()
|
||
|
||
# 检查应用是否成功启动
|
||
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)
|
||
|
||
# @staticmethod
|
||
# def get_device_id() -> str:
|
||
# """
|
||
# 获取设备ID，优先使用已连接设备，否则使用全局配置
|
||
# """
|
||
# try:
|
||
# # 检查已连接设备
|
||
# result = subprocess.run(
|
||
# ["adb", "devices"],
|
||
# capture_output=True,
|
||
# text=True,
|
||
# timeout=10
|
||
# )
|
||
|
||
# # 解析设备列表
|
||
# for line in result.stdout.strip().split('\n')[1:]:
|
||
# if line.strip() and "device" in line and "offline" not in line:
|
||
# device_id = line.split('\t')[0]
|
||
# logging.info(f"使用已连接设备: {device_id}")
|
||
# global_variable.GLOBAL_DEVICE_ID = device_id
|
||
# return device_id
|
||
|
||
# except Exception as e:
|
||
# logging.warning(f"设备检测失败: {e}")
|
||
|
||
# # 使用全局配置
|
||
# device_id = global_variable.GLOBAL_DEVICE_ID
|
||
# logging.info(f"使用全局配置设备: {device_id}")
|
||
# return device_id
|
||
|
||
def get_measure_data(self):
|
||
# 模拟获取测量数据
|
||
pass
|
||
|
||
# def add_transition_point(self):
|
||
# # 添加转点逻辑
|
||
# print("添加转点")
|
||
# return True
|
||
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()
|
||
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 main_run(self):
|
||
return self.add_transition_point()
|
||
|
||
def run(self, station_num: int):
|
||
# 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}个站点")
|
||
|
||
# 截图
|
||
self.driver.save_screenshot("check_station.png")
|
||
return True
|
||
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() |