createtask.py
6.18 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
137
138
139
# -*- coding: utf-8 -*-
# @Time : 2020/12/13 14:47
# @Author : Young Lee
# @Email : young_lee2017@163.com
from selenium.webdriver.common.keys import Keys
from jumppage import jump_page
import time
import requests
def check_exist(tasks_url, task_name):
""" 调用任务api查询是否有同名任务 """
try:
list_data = requests.get(tasks_url, timeout=7).json()["list_data"]
if list_data:
exist_tasks = [task['task_name'] for task in list_data]
return True if task_name in exist_tasks else False
else:
return False
except Exception:
return False
def create_task(tasks_url, driver, task_name, task_algo_type="交通", resource_use=1, running_time=None, store_conf=None, device_type=None):
""""创建任务
:param task_name: 大任务名称
:param task_algo_type: 算法类型，默认"交通"
:param resource_use: 场景占用，默认1
:param running_time: 默认None(全天), 需要指定时间段时需要传入列表[开始时间, 结束时间]
:param device_type: 默认None(随机下发), 需要指定下发到某设备需要传入设备类型
:return: {first_task_name: first_task_status}
"""
whether_exising = check_exist(tasks_url, task_name)
jump_page(driver, "任务管理", "任务设置")
driver.find_element_by_xpath("//div[@class='content']/div[1]//span[text()='新建任务']").click()
# 填入任务名称
driver.find_element_by_css_selector("label[for='task_name'] + div input").send_keys(task_name)
# 选择算法类型
driver.find_element_by_css_selector("label[for='task_algo_type'] + div input").click()
time.sleep(0.5)
driver.find_element_by_xpath(
f"//label[@for='task_algo_type']/following-sibling::div//li/span[text()='{task_algo_type}']"
).click()
# 选择场景占用
driver.find_element_by_css_selector("label[for='resource_use'] + div input").click()
time.sleep(0.5)
driver.find_element_by_xpath(f"//label[@for='resource_use']/following-sibling::div//li[{resource_use}]").click()
# 选择任务运行时间段
if running_time:
start_time = running_time[0]
end_time = running_time[1]
driver.find_element_by_xpath("//label[text()='时间计划']/following-sibling::*//input").click() # 点开时间计划进行选择
time.sleep(0.5)
driver.find_element_by_xpath("//span[text()='时间']").click()
start_time_ele = driver.find_element_by_css_selector("input[placeholder='开始时间']")
start_time_ele.send_keys(Keys.CONTROL,'a') # 全选开始时间文本
start_time_ele.send_keys(start_time) # 输入开始时间
end_time_ele = driver.find_element_by_css_selector("input[placeholder='结束时间']")
end_time_ele.send_keys(Keys.CONTROL,'a') # 全选结束时间文本
end_time_ele.send_keys(end_time) # 输入结束时间
driver.find_element_by_css_selector("button[class='el-time-panel__btn confirm']").click() # 点击确定
# 选择存储配置
driver.find_element_by_css_selector("label[for='store_conf_unid'] + div input").click() # 点开存储配置列表
time.sleep(0.5)
if store_conf:
stor_num = driver.execute_script("return document.evaluate(\"//label[@for='store_conf_unid']/following-sibling"
"::div//ul\", document).iterateNext().childNodes.length")
if stor_num == 0:
return "未发现存储配置"
ul_xpath = "//label[@for='store_conf_unid']/following-sibling::div//ul/li"
li_eles = driver.find_elements_by_xpath(ul_xpath)
stor_names = [ele.get_attribute('textContent') for ele in li_eles]
if store_conf in stor_names:
driver.find_element_by_xpath(f"{ul_xpath}/span[text()='{store_conf}']").click() # 选择指定存储配置
else:
print(f"未找到存储配置{store_conf}, 选择列表内第一个替代")
driver.find_element_by_css_selector("label[for='store_conf_unid'] + div li:nth-child(1)").click() # 选择第一个存储配置
else:
print("未指定存储配置，自动选择列表内第一个")
driver.find_element_by_css_selector("label[for='store_conf_unid'] + div li:nth-child(1)").click() # 选择第一个存储配置
# 选择设备类型
if device_type:
driver.find_element_by_xpath("//label[text()='设备类型']/following-sibling::div").click() # 展开设备类型列表
time.sleep(0.5)
driver.find_element_by_xpath(f"//li[text()=' {device_type} ']").click() # 选择指定设备类型
# 单击确定进行新建
driver.find_element_by_css_selector(
"div[aria-label='新建任务']>div[class='el-dialog__footer'] button:nth-child(2)"
).click()
if whether_exising:
print("存在同名任务")
try:
click_result = driver.find_element_by_css_selector("div[role='alert']>p").get_attribute('textContent')
driver.find_element_by_xpath("//div[@aria-label='新建任务']/div[@class='el-dialog__footer']/span/button/"
"span[text()='取 消']").click()
return click_result
except Exception:
return
time.sleep(2)
# 获得当前第一个任务的名称和部署状态
first_task_name = driver.find_element_by_css_selector(
"table[class='el-table__body'] tr:nth-child(1) td:nth-child(3)>div"
).text
first_task_status = driver.find_element_by_css_selector(
"table[class='el-table__body'] tr:nth-child(1) td:nth-child(4)>div"
).text
return {first_task_name: first_task_status}
if __name__ == "__main__":
from unitinit import *
from login import login_fx
login_fx(driver, login_url, user, passwd)
task_info = create_task(
tasks_url, driver, "测试任务njF9Y",
task_algo_type="交通",
# resource_use=1,
# running_time=["00:00:00", "12:00:00"],
store_conf="基础",
# device_type="9.133_gpu0"
)
print(task_info)
# from createsubtask import create_subtask_batch
# create_subtask_batch(driver, '100辆车', {'短视频': ['100辆车']}, monitor_status=True, monitor_times=20)
driver.quit()