createsubtask.py
5.91 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
140
141
142
143
144
145
146
147
148
149
# -*- coding: utf-8 -*-
# @Time : 2020/12/14 9:05
# @Author : Young Lee
# @Email : young_lee2017@163.com
import time
from findtask import find_task
from monitor_subtask_status import monitor_subtask_status
def select_videos(videos_list, add_videos):
""" 勾选视频 """
for add_video in add_videos:
try:
check_ele = videos_list.find_element_by_xpath(f".//span[text()='{add_video}']/../preceding-sibling::label")
if "is-checked" not in check_ele.get_attribute("class"):
check_ele.click()
except Exception:
print(f"未找到视频: {add_video}")
def create_subtask(driver, group_path, add_videos):
"""创建子任务
:param group_path: 找寻视频的路径
:param add_videos: list, 要添加的视频列表
"""
print(f"开始勾选>> '{group_path}'下的视频列表: {add_videos}")
group_structure = group_path.split("/")
if group_structure[0] == "外部设备":
group_structure[0] = 1
elif group_structure[0] == "内部设备":
group_structure[0] = 2
if len(group_structure) < 2:
print("填写的视频寻找路径不全[内部设备]！")
return
elif group_structure[0] == "短视频":
group_structure[0] = 3
else:
print("所给video_path错误！")
return
videos_list = driver.find_element_by_css_selector(f"div[role='tree']>div:nth-child({group_structure[0]})")
whether_expanded = videos_list.find_element_by_css_selector(
"div[class='el-tree-node__content']>span:nth-child(1)"
).get_attribute("class")
if "expanded" not in whether_expanded:
videos_list.find_element_by_css_selector(
"div[class='el-tree-node__content']>span[class='custom-tree-node']"
).click()
time.sleep(1)
ele = "."
detail_ele = ele
if group_structure[0] == 1 or group_structure[0] == 3:
spec_videos_list = videos_list.find_element_by_xpath("./div[@role='group']")
select_videos(spec_videos_list, add_videos)
else:
for i in range(1, len(group_structure)):
ele = ele + "/div[@role='group']/div[@role='treeitem']"
detail_ele = f"{ele}/div[@class='el-tree-node__content']/span[@class='custom-tree-node']/span[text()='{group_structure[i]}']"
try:
whether_expanded = videos_list.find_element_by_xpath(
f"{detail_ele}/../preceding-sibling::span"
).get_attribute("class")
except Exception as e:
print(f"未找到{group_structure[i]}", e)
return
if "expanded" not in whether_expanded:
videos_list.find_element_by_xpath(detail_ele).click()
spec_videos_list = videos_list.find_element_by_xpath(
f"{detail_ele}/../../following-sibling::div[@role='group']")
select_videos(spec_videos_list, add_videos)
# videos_list.find_element_by_xpath(f"{ele}//span[text()='{group_structure[-1]}']").click()
def create_subtask_batch(driver, task_name, videos_batch_info, monitor_status=False, monitor_times=30):
"""批量添加子任务
:param task_name: 要添加子任务的大任务名称
:param videos_batch_info: {group_path(视频寻找路径,/分割): add_videos(视频名称，list)}
:param monitor_status: 是否开启子任务下发状态监测，默认False
:param monitor_times: 重复监测次数(下发状态为"工作中"时停止)，默认为30次，两次监测间隔2S
:return: monitor_status为True时返回subtasks_status: [subtask_name, subtask_status, subtask_id]
monitor_status为False时返回已下发任务列表
"""
subtasks_status = []
# 寻找任务
subtasks_id_before_create = find_task(driver, task_name)
if not subtasks_id_before_create:
return
# 打开‘修改视频源配置’
driver.find_element_by_xpath(f"//div[text()='{task_name}']/../following-sibling::td[last()]//span[contains(@class,"
f"'el-icon-video-camera-solid')]").click()
# 批量勾选视频
for group_path, add_videos in videos_batch_info.items():
create_subtask(driver, group_path, add_videos)
# 点击确定下发任务
driver.find_element_by_xpath(
"//div[@aria-label='视频源配置修改']/div[@class='el-dialog__footer']//span[text()='确 定']"
).click()
time.sleep(1)
if monitor_status:
for i in range(monitor_times):
print(f"第{i+1}次:")
# subtasks_status = monitor_subtask_status(driver, task_name, subtasks_id_before_create)
subtasks_status = monitor_subtask_status(driver, task_name)
if "部署中" not in [value[1] for value in subtasks_status]:
break
time.sleep(2)
return subtasks_status
else:
subtasks_status = monitor_subtask_status(driver, task_name, subtasks_id_before_create)
return [value[0] for value in subtasks_status]
if __name__ == "__main__":
from unitinit import *
from login import login_fx
task_name = "顺德逆行"
group_path1 = "内部设备/123123/9.179"
add_videos1 = ["泥头车", "运输车"]
group_path2 = "短视频"
add_videos2 = ["大货车合并文件"]
group_path3 = "外部设备"
add_videos3 = ["34020000001339990002", "34020000001339990003"]
# batch_info = {group_path1: add_videos1, group_path2: add_videos2, group_path3: add_videos3}
batch_info = {group_path1: add_videos1}
login_fx(driver, login_url, user, passwd)
# print(find_task(driver, task_name))
# print(monitor_subtask_status(driver, task_name, []))
retrun_value = create_subtask_batch(driver, task_name, batch_info)
print(retrun_value)
# create_subtask(driver, task_name, {group_path: add_videos})
# start = time.time()
# find_task(driver, task_name)
# end = time.time()
# print(end-start)
# time.sleep(10)
driver.quit()