importconfig.py
5.73 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
# -*- coding: utf-8 -*-
# @Time : 2021/1/11 22:14
# @Author : Young Lee
# @Email : young_lee2017@163.com
import time
import os
from selenium.webdriver import ActionChains
from jumppage import jump_page
def check_send_status(driver, review_list, target_task_ele, import_result):
""" 检查配置下发状态 """
if target_task_ele and review_list:
for name in review_list:
target_task_ele.find_element_by_css_selector(f"ol>li[title='{name}']").click()
mask_ele = driver.find_element_by_class_name("el-loading-mask")
flag = False
for i in range(30):
if mask_ele.get_attribute('style') == 'display: none;':
break
time.sleep(0.3)
if i == 29:
import_result[name] = 'review: loding failed'
flag = True
# 任务信息加载失败则跳过本次检测
if flag:
continue
# 移动鼠标到参数、区域、标定设置之上，获取配置下发状态
set_send_status = {}
for set_ele in driver.find_elements_by_css_selector(".table-set>span"):
set_type = set_ele.get_attribute('textContent')
ActionChains(driver).move_to_element(set_ele).perform()
set_status = driver.find_element_by_css_selector(
"div[role='tooltip'][aria-hidden='false']").get_attribute('textContent')
set_send_status[set_type] = set_status
# 记录配置下发状态
import_result[name] = set_send_status
def import_config(driver, task_name, subtasks_cfg_info, retry_check_times=5):
"""导入任务配置
:param task_name: 大任务名称
:param subtasks_cfg_info: 配置信息 {<任务名称>: <本地配置文件路径>, ...}
:param retry_check_times: 导入配置响应状态监测的最大次数，默认5次
:return: {任务名称: 导入结果}，导入成功时'导入结果'为配置下发状态（dict），导入失败时'导入结果'为失败原因（str）
"""
import_result = {}
review_list = []
target_task_ele = None
jump_page(driver, '任务管理', '场景设置')
tasks_ele = driver.find_elements_by_css_selector(".tasklist>li")
for task_ele in tasks_ele:
task_title = task_ele.find_element_by_css_selector(".task-title")
ele_name = task_title.get_attribute('textContent')
if ele_name != task_name:
continue
target_task_ele = task_ele
tag_name = driver.find_element_by_css_selector(".tasklist>li>*:nth-last-child(1)").get_attribute('tagName')
if tag_name != 'OL':
task_title.click()
subtasks_ele = task_ele.find_elements_by_css_selector('ol>li')
subtasks_title = [title.get_attribute('title') for title in subtasks_ele]
# 挨个任务导入
for name, cfg in subtasks_cfg_info.items():
if name not in subtasks_title or not os.path.exists(cfg):
import_result[name] = 'preparation: subtask/config not found'
continue
subtasks_ele[subtasks_title.index(name)].click()
mask_ele = driver.find_element_by_class_name("el-loading-mask")
for i in range(30):
if mask_ele.get_attribute('style') == 'display: none;':
break
time.sleep(0.3)
# 清除已存在的alert
while True:
last_ele = driver.find_element_by_css_selector("#theme-name>*:nth-last-child(1)")
if last_ele.get_attribute('tagName') == 'DIV' and last_ele.get_attribute('role') == 'alert':
try:
driver.execute_script("document.querySelector(\"div[role='alert']\").remove()")
except Exception:
pass
else:
break
# 导入配置
driver.find_element_by_xpath("//div[@class='setting-box']/div[contains(text(), '配置导入')]/span").click()
driver.find_element_by_css_selector(
"div[aria-label='配置导入'] .el-dialog__body .upload-demo input").send_keys(cfg)
driver.find_element_by_xpath(
"//div[@aria-label='配置导入']/div[@class='el-dialog__footer']/span/button/span[text()='确 定']").click()
response_class = driver.find_element_by_css_selector(
"div[role='alert']:nth-last-child(1)>i").get_attribute('class')
response_message = driver.find_element_by_css_selector("div[role='alert']>p").get_attribute('textContent')
print(f"导入操作响应信息: {response_message}")
if 'el-icon-success' in response_class:
review_list.append(name)
import_result[name] = None
else:
import_result[name] = 'import: import failed'
break
# 复检配置导入状态
time.sleep(1)
while review_list:
check_send_status(driver, review_list, target_task_ele, import_result)
for name in review_list:
status = import_result[name]
if isinstance(status, dict) and '下发中' not in status.values():
review_list.remove(name)
retry_check_times -= 1
if retry_check_times < 1:
break
return import_result
if __name__ == '__main__':
from unitinit import *
from login import login_fx
login_fx(driver, login_url, user, passwd)
# cfg_path = r"\\192.168.9.1\t测试\任务信息库\config\100辆车_6f52d8381807461b96f102a2d73c5c62.tar.gz"
for i in range(30):
import_result = import_config(driver, '测试任务njF9Y', task_cfg_info)
print(import_result)
driver.quit()