<feat> 繁星2.0配置上传到3.0的接口

<feat> 如果场景改变了也需要重新部署任务 <feat> 收到分析结果的日志修改

<feat> 繁星2.0配置上传到3.0的接口
<feat> 如果场景改变了也需要重新部署任务 <feat> 收到分析结果的日志修改
xmh
Commit e3f76479 authored Jan 12, 2022 by xmh
Showing 5 changed files with 76 additions and 17 deletions
fanxing-forward/src/main/java/com/viontech/fanxing/forward/batch/processor/BehaviorProcessor.java
fanxing-forward/src/main/java/com/viontech/fanxing/forward/batch/processor/TrafficFlowProcessor.java
fanxing-forward/src/main/java/com/viontech/fanxing/forward/batch/processor/TrafficProcessor.java
fanxing-task/src/main/java/com/viontech/fanxing/task/controller/web/TaskController.java
fanxing-task/src/main/java/com/viontech/fanxing/task/utils/TaskUtils.java
--- a/fanxing-forward/src/main/java/com/viontech/fanxing/forward/batch/processor/BehaviorProcessor.java
+++ b/fanxing-forward/src/main/java/com/viontech/fanxing/forward/batch/processor/BehaviorProcessor.java
@@ -2,7 +2,6 @@ package com.viontech.fanxing.forward.batch.processor;
 import com.alibaba.fastjson.JSONObject;
 import com.viontech.fanxing.commons.model.Behavior;
-import com.viontech.fanxing.forward.util.CacheUtils;
 import com.viontech.keliu.util.DateUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
@@ -12,7 +11,6 @@ import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.ItemStreamException;
 import org.springframework.stereotype.Component;
-import javax.annotation.Resource;
 import java.util.Date;
 /**
@@ -24,8 +22,6 @@ import java.util.Date;
 @Component
 @Slf4j
 public class BehaviorProcessor implements ItemStream, ItemProcessor<JSONObject, Behavior> {
-    @Resource
-    private CacheUtils cacheUtils;
    @Override
    public Behavior process(JSONObject item) throws Exception {
@@ -35,7 +31,8 @@ public class BehaviorProcessor implements ItemStream, ItemProcessor<JSONObject, 
        if (StringUtils.isEmpty(eventRefid)) {
            eventRefid = eventData.getString("ID");
        }
-        log.info("收到 behavior 消息 , eventId:{}", eventRefid);
+        String taskName = item.getString("task_name");
+        log.info("收到 behavior 消息 , 任务名称:[{}] , 消息ID:[{}]", taskName, eventRefid);
        Behavior behavior = new Behavior();
        Long taskId = item.getLong("taskId");
@@ -43,7 +40,6 @@ public class BehaviorProcessor implements ItemStream, ItemProcessor<JSONObject, 
        String eventDt = item.getString("event_dt");
        String eventCate = item.getString("event_cate");
        String channelUnid = item.getString("vchan_refid");
-        String taskName = item.getString("task_name");
        Date eventTime = DateUtil.parse(DateUtil.FORMAT_FULL, eventDt);
        String picArray = item.getString("pic_path_array");

--- a/fanxing-forward/src/main/java/com/viontech/fanxing/forward/batch/processor/TrafficFlowProcessor.java
+++ b/fanxing-forward/src/main/java/com/viontech/fanxing/forward/batch/processor/TrafficFlowProcessor.java
@@ -30,9 +30,6 @@ import java.util.List;
 @Slf4j
 public class TrafficFlowProcessor implements ItemProcessor<JSONObject, TrafficFlowContent>, ItemStream {
-    @Resource
-    private CacheUtils cacheUtils;
    @Override
    public TrafficFlowContent process(JSONObject item) throws Exception {
        JSONObject eventData = item.getJSONObject("event_data");
@@ -40,7 +37,8 @@ public class TrafficFlowProcessor implements ItemProcessor<JSONObject, TrafficFl
        if (StringUtils.isEmpty(eventRefid)) {
            eventRefid = eventData.getString("ID");
        }
-        log.info("收到 flow 消息 , eventId:{}", eventRefid);
+        String taskName = item.getString("task_name");
+        log.info("收到 flow 消息 , 任务名称:[{}] , 消息ID:[{}]", taskName, eventRefid);
        TrafficFlowContent content = new TrafficFlowContent();
        FlowEvent flowEvent = new FlowEvent();

--- a/fanxing-forward/src/main/java/com/viontech/fanxing/forward/batch/processor/TrafficProcessor.java
+++ b/fanxing-forward/src/main/java/com/viontech/fanxing/forward/batch/processor/TrafficProcessor.java
@@ -5,7 +5,6 @@ import com.alibaba.fastjson.JSONObject;
 import com.viontech.fanxing.commons.model.Traffic;
 import com.viontech.fanxing.commons.model.TrafficFace;
 import com.viontech.fanxing.forward.model.TrafficContent;
-import com.viontech.fanxing.forward.util.CacheUtils;
 import com.viontech.keliu.util.DateUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
@@ -15,7 +14,6 @@ import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.ItemStreamException;
 import org.springframework.stereotype.Component;
-import javax.annotation.Resource;
 import java.util.ArrayList;
 import java.util.Date;
 import java.util.Optional;
@@ -31,9 +29,6 @@ import java.util.Optional;
 @Slf4j
 public class TrafficProcessor implements ItemProcessor<JSONObject, TrafficContent>, ItemStream {
-    @Resource
-    private CacheUtils cacheUtils;
    @Override
    public TrafficContent process(JSONObject item) throws Exception {
@@ -42,7 +37,8 @@ public class TrafficProcessor implements ItemProcessor<JSONObject, TrafficConten
        if (StringUtils.isEmpty(eventRefid)) {
            eventRefid = eventData.getString("ID");
        }
-        log.info("收到 traffic 消息 , eventId:{}", eventRefid);
+        String taskName = item.getString("task_name");
+        log.info("收到 traffic 消息 , 任务名称:[{}] , 消息ID:[{}]", taskName, eventRefid);
        TrafficContent content = new TrafficContent();
        Traffic traffic = new Traffic();

--- a/fanxing-task/src/main/java/com/viontech/fanxing/task/controller/web/TaskController.java
+++ b/fanxing-task/src/main/java/com/viontech/fanxing/task/controller/web/TaskController.java
 package com.viontech.fanxing.task.controller.web;
+import cn.hutool.core.io.IoUtil;
+import cn.hutool.core.util.ZipUtil;
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
@@ -20,6 +22,8 @@ import com.viontech.fanxing.task.service.OpsClientService;
 import com.viontech.fanxing.task.utils.SceneUtils;
 import com.viontech.keliu.util.JsonMessageUtil;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.compress.archivers.ArchiveEntry;
+import org.apache.commons.compress.archivers.tar.TarArchiveInputStream;
 import org.apache.commons.io.IOUtils;
 import org.springframework.dao.DuplicateKeyException;
 import org.springframework.http.MediaType;
@@ -29,11 +33,13 @@ import org.springframework.web.multipart.MultipartFile;
 import javax.annotation.Resource;
 import javax.servlet.http.HttpServletResponse;
+import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.net.URLEncoder;
 import java.nio.charset.StandardCharsets;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.stream.Collectors;
 import static com.viontech.keliu.util.JsonMessageUtil.getSuccessJsonMsg;
@@ -198,6 +204,62 @@ public class TaskController extends TaskBaseController {
    }
    /**
+     * 上传场景配置
+     *
+     * @return
+     */
+    @PostMapping("uploadRoiAndConfig2")
+    public Object uploadRoiAndConfig2(@RequestParam Long taskId, @RequestParam MultipartFile file) throws IOException {
+        Task task = taskService.selectByPrimaryKey(taskId);
+        Assert.notNull(task, "找不到对应任务");
+        Channel channel = opsClientService.getChannelByChannelUnid(task.getChannelUnid());
+        String scene = task.getScene();
+        JSONArray sceneArr = JSON.parseArray(scene);
+        JSONObject source;
+        TarArchiveInputStream tais = new TarArchiveInputStream(new ByteArrayInputStream(ZipUtil.unGzip(file.getInputStream())));
+        ArchiveEntry nextEntry = tais.getNextEntry();
+        if (nextEntry != null) {
+            byte[] bytes = IoUtil.read(tais).toByteArray();
+            tais.close();
+            try {
+                source = JSON.parseObject(new String(bytes, StandardCharsets.UTF_8));
+            } catch (Exception e) {
+                log.error("场景配置文件解析失败", e);
+                return JsonMessageUtil.getErrorJsonMsg("场景配置文件解析失败");
+            }
+        } else {
+            return JsonMessageUtil.getErrorJsonMsg("解析失败");
+        }
+        JSONArray scenes = Optional.of(source).map(x -> x.getJSONArray("mtasks"))
+                .map(x -> x.getJSONObject(0))
+                .map(x -> x.getJSONArray("scenes")).orElse(new JSONArray());
+        for (int i = 0; i < Math.min(scenes.size(), sceneArr.size()); i++) {
+            JSONObject sourceScene = scenes.getJSONObject(i);
+            JSONObject targetScene = sceneArr.getJSONObject(i);
+            // 合并配置
+            SceneUtils.INSTANCE.mergeScene(sourceScene, targetScene);
+            // 合并之后,因为需要设备信息的配置,所以需要重新构建
+            JSONObject config = targetScene.getJSONObject("config");
+            if (config != null) {
+                String xml = config.getString("xml");
+                ConfigBuilder configBuilder = new ConfigBuilder();
+                String c = configBuilder.buildDefaultConfig(xml)
+                        .buildVchanInfo(channel.getName(), channel.getChannelUnid())
+                        .build();
+                config.put("xml", c);
+            }
+        }
+        TaskVo temp = new TaskVo();
+        temp.setId(taskId);
+        temp.setScene(sceneArr.toJSONString());
+        return update(taskId, temp);
+    }
+    /**
     * 下载场景配置和roi
     *
     * @param taskId    任务ID

--- a/fanxing-task/src/main/java/com/viontech/fanxing/task/utils/TaskUtils.java
+++ b/fanxing-task/src/main/java/com/viontech/fanxing/task/utils/TaskUtils.java
@@ -16,6 +16,7 @@ import com.viontech.fanxing.task.service.VAServerService;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.redisson.api.RMap;
@@ -31,6 +32,7 @@ import java.util.concurrent.TimeUnit;
 * @date 2021/12/21
 */
 @SuppressWarnings("ALL")
+@Slf4j
 public enum TaskUtils {
    INSTANCE;
@@ -49,7 +51,11 @@ public enum TaskUtils {
     * @param present  现在的任务信息
     */
    public boolean needRebuild(Task original, Task present) {
-        return (!original.getRuntimeConf().equals(present.getRuntimeConf())) || (!original.getStoreConfigId().equals(present.getStoreConfigId()));
+        return (!original.getRuntimeConf().equals(present.getRuntimeConf()))
+                || (!original.getStoreConfigId().equals(present.getStoreConfigId()))
+                || (!original.getResourceNeed().equals(present.getResourceNeed()))
+                || (!original.getVaType().equals(present.getVaType()))
+                || (!original.getScene().equals(present.getScene()));
    }
    public void checkRuntimeConf(TaskData taskData, VAServerService vaServerService, TaskDataService taskDataService) {
@@ -126,6 +132,7 @@ public enum TaskUtils {
                    }
                    break;
            }
+            log.error("任务无法启动,");
            throw new FanXingException("任务资源信息异常");
        }