feat: 数据扩增逻辑重做

taais-modules/taais-biz/src/main/java/com/taais/biz/constant/BizConstant.java

@@ -126,6 +126,8 @@ public class BizConstant {
     public static final String DOCKER_OBJ_MATCH_PATH = "/objectMatch";
     public static final String TYPE_OBJ_TRACE = "OBJ_TRACE";
     public static final String DOCKER_MAT_TASK = "/obj_track";
+
+    public static final String TYPE_DATA_EXPAND = "DATA_EXPAND";
 //    public static final String MULTI_OBJ_TRACE_URL = "127.0.0.1:10027/objTrace";
 //    public static final String MULTI_OBJ_MATCH_URL = "127.0.0.1:10028/imgMatch";
 }

taais-modules/taais-biz/src/main/java/com/taais/biz/controller/PublicController.java

@@ -8,6 +8,7 @@ import com.taais.biz.domain.bo.*;
 import com.taais.biz.domain.dto.TaskResultDTO;
 import com.taais.biz.service.*;
 import com.taais.biz.service.impl.AlgorithmModelServiceImpl;
+import com.taais.biz.service.impl.DataAmplificationTaskServiceImpl;
 import com.taais.biz.service.impl.ObjectTraceMergeServiceImpl;
 import com.taais.biz.service.impl.TargetIdentificationSubtaskDetailsServiceImpl;
 import com.taais.biz.service.service.impl.ObjectMatchServiceImpl;
@@ -79,6 +80,9 @@ public class PublicController extends BaseController {
     @Resource
     ObjectMatchServiceImpl objectMatchService;
 
+    @Resource
+    DataAmplificationTaskServiceImpl dataAmplificationTaskService;
+
     @PostMapping("/taskResult")
     public CommonResult<Void> taskResult(@RequestBody TaskResultDTO resultDTO) {
         log.info("taskResult start,params:{}", resultDTO);
@@ -139,6 +143,22 @@ public class PublicController extends BaseController {
             objectMatchService.update(bo);
         } else if (BizConstant.TYPE_DATA_PROCESS.equals(bizType)) {
             errorMsg = dataProcessService.taskResult(resultDTO);
+        } else if (BizConstant.TYPE_DATA_EXPAND.equals(bizType)) {
+            DataAmplificationTaskBo bo = dataAmplificationTaskService.getById(resultDTO.getBizId());
+            if (bo != null) {
+                try {
+                    bo.setInputImagePath(String.valueOf(Integer.parseInt(bo.getOutputImagePath()) - 1));
+                    if (bo.getInputImagePath().equals("0")) {
+                        bo.setEndTime(new Date());
+                        bo.setCostSecond((int) ((bo.getEndTime().getTime() - bo.getStartTime().getTime()) / 1000));
+                        bo.setStatus(resultDTO.getStatus() == 200 ? BizConstant.TASK_STATUS_SUCCEED : BizConstant.TASK_STATUS_FAILED);
+                    }
+                } catch (Exception e) {
+                    log.error("error: {}", e.getMessage());
+                } finally {
+                    dataAmplificationTaskService.update(bo);
+                }
+            }
         } else {
             log.error("这种情况是不可能发生的,参数：{}",resultDTO);
             return CommonResult.fail("这种情况是不可能发生的");

taais-modules/taais-biz/src/main/java/com/taais/biz/service/impl/DataAmplificationTaskServiceImpl.java

@@ -51,6 +51,10 @@ public class DataAmplificationTaskServiceImpl extends BaseServiceImpl<DataAmplif
         return this.getOneAs(query().where(DATA_AMPLIFICATION_TASK.ID.eq(id)), DataAmplificationTaskVo.class);
     }
 
+    public DataAmplificationTaskBo getById(Long id) {
+        return this.getOneAs(query().where(DATA_AMPLIFICATION_TASK.ID.eq(id)), DataAmplificationTaskBo.class);
+    }
+
     @Override
     public List<DataAmplificationTaskVo> selectList(DataAmplificationTaskBo taskBo) {
         QueryWrapper queryWrapper = buildQueryWrapper(taskBo);

taais-modules/taais-biz/src/main/java/com/taais/biz/service/impl/DataServiceImpl.java

@@ -2,6 +2,10 @@ package com.taais.biz.service.impl;
 
 import cn.hutool.core.util.ObjectUtil;
 import cn.hutool.http.HttpRequest;
+import cn.hutool.http.HttpUtil;
+import cn.hutool.json.JSONUtil;
+import com.alibaba.fastjson2.JSON;
+import com.esotericsoftware.minlog.Log;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.mybatisflex.core.paginate.Page;
@@ -27,6 +31,7 @@ import com.taais.common.core.utils.StringUtils;
 import com.taais.common.core.utils.file.FileUploadUtils;
 import com.taais.common.core.utils.file.FileUtils;
 import com.taais.common.core.utils.file.UnPackedUtil;
+import com.taais.common.core.utils.uuid.UUID;
 import com.taais.common.json.utils.JsonUtils;
 import com.taais.common.orm.core.page.PageQuery;
 import com.taais.common.orm.core.service.impl.BaseServiceImpl;
@@ -55,6 +60,8 @@ import java.time.format.DateTimeFormatter;
 import java.util.*;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import static com.taais.biz.constant.BizConstant.*;
+import static com.taais.biz.constant.BizConstant.TASK_STATUS_FAILED;
 import static com.taais.biz.domain.table.DataTableDef.DATA;
 
 /**
@@ -79,7 +86,7 @@ public class DataServiceImpl extends BaseServiceImpl<DataMapper, Data> implement
     private DataMapper dataMapper;
 
     @Resource
-    private IDataAmplificationTaskService dataAmplificationTaskService;
+    private DataAmplificationTaskServiceImpl dataAmplificationTaskService;
 
     @Resource
     private ISysDictDataService dictDataService;
@@ -439,215 +446,117 @@ public class DataServiceImpl extends BaseServiceImpl<DataMapper, Data> implement
     @Override
     @Transactional
     public CommonResult<Boolean> amplifyForData(String id) {
-        DataAmplificationTaskVo taskVo = dataAmplificationTaskService.selectById(Long.valueOf(id));
-        DataAmplifyDto dataAmplifyDto = new DataAmplifyDto();
-        dataAmplifyDto.setId(taskVo.getId().toString());
-        dataAmplifyDto.setAugmentationType(taskVo.getAugmentationType());
-        List<Map<String, String>> maps = JsonUtils.parseArray(taskVo.getParameters(), Map.class);
-        dataAmplifyDto.setOtherParams(maps);
-        dataAmplifyDto.setBatchNum(taskVo.getDataBatchNums());
-        return this.doAmplify(dataAmplifyDto);
-    }
-
-    private CommonResult<Boolean> doAmplify(DataAmplifyDto dataAmplifyDto) {
-        this.updateEmp(Long.valueOf(dataAmplifyDto.getId()));
-        String[] split = dataAmplifyDto.getBatchNum().split(",");
-        SysDictDataVo sysDictDataVo = dictDataService.selectDictDataByTypeAndLabel("python_api_address", "python_expand_data");
-
-        for (String batchNum : split) {
-            QueryWrapper query = query();
-            query.eq(Data::getBatchNum, batchNum);
-            List<Data> dataList = dataMapper.selectListByQuery(query);
-            if (dataList.isEmpty()) {
-                this.updateFail(Long.valueOf(dataAmplifyDto.getId()));
-                continue;
-            }
+        DataAmplificationTaskBo taskVo = dataAmplificationTaskService.getById(Long.valueOf(id));
+        taskVo.setStartTime(new Date());
+        List<Map<String, String>> otherParams = JsonUtils.parseArray(taskVo.getParameters(), Map.class);
+        Map<String, String> realOtherParams = new HashMap<>();
+        otherParams.forEach(param -> {
+            realOtherParams.put(param.get("agName"), param.containsKey("value") ? param.get("value") : param.get("defaultValue"));
+        });
+        // move data
+        String[] batches = taskVo.getDataBatchNums().split(",");
+        String filepath = "/" + UUID.randomUUID().toString().replace("-", "_");
+        for (String batch : batches) {
+            copyFilesToPath(batch, filepath, "e_");
+        }
 
-            List<Data> dataListInfo = dataList.stream().filter(data -> !StringUtils.isEmpty(data.getUrl())).toList();
-            if (dataListInfo.isEmpty()) {
-                this.updateFail(Long.valueOf(dataAmplifyDto.getId()));
-                continue;
-            }
-            String filePath = TaaisConfig.getUploadPath();
-            LocalDate currentDate = LocalDate.now();
-            // 定义日期格式器
-            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy/MM/dd");
-            String formattedDate = currentDate.format(formatter);
-            filePath = filePath + File.separator + formattedDate;
-            String finalFilePath = filePath;
-            Map<String, String> otherParams = new HashMap<>();
-            for (Map<String, String> param : dataAmplifyDto.getOtherParams()) {
-                otherParams.put(param.get("agName"), param.get("defaultValue"));
+        // post request
+        Map<String, String> params = new HashMap<>();
+        params.put("bizType", TYPE_DATA_EXPAND);
+        params.put("bizId", String.valueOf(taskVo.getId()));
+        params.put("logPath", DOCKER_BASE_PATH + filepath + "/log.log");
+        params.put("sourcePath", PATH_PREFIX + filepath);
+        params.put("otherParams", JSONUtil.toJsonStr(realOtherParams));
+
+        String[] outputs = taskVo.getOutputImagePath().split(";");
+        log.info("check outputs: {} ; {}", outputs, outputs[0]);
+        taskVo.setInputImagePath(String.valueOf(outputs.length));
+        for (String path : outputs) {
+            final String resultPath = TargetIdentificationTaskServiceImpl.PATH_PREFIX + path;
+            params.put("resultPath", resultPath);
+
+            log.info("check data expand: {}", params);
+            try {
+                String url = "http://127.0.0.1:11001/augment";
+                String res = HttpUtil.post(url, JSONUtil.toJsonStr(params));
+                taskVo.setStatus(TASK_STATUS_PROCESSING);
+            } catch (Exception e) {
+                log.error("HTTP请求失败",e);
+                taskVo.setStatus(TASK_STATUS_FAILED);
+                return CommonResult.fail("HTTP请求失败" + e.getMessage());
+            } finally {
+                dataAmplificationTaskService.update(taskVo);
             }
-            dataListInfo.forEach(dataInfo -> {
-                try {
-                    boolean success = true;
-                    String message = "";
-                    Date startTime = new Date();
-
-                    //循环调用Python扩增接口
-                    Map<String, Object> bodyJson = new HashMap<>();
-                    bodyJson.put("bizType", "数据扩增");
-                    bodyJson.put("bizId", dataAmplifyDto.getId());
-                    bodyJson.put("augmentationType", dataAmplifyDto.getAugmentationType());
-                    bodyJson.put("inputImagePath", dataInfo.getUrl());
-                    String outputImagePath = finalFilePath + AMPLIFY + System.currentTimeMillis();
-                    File desc = new File(outputImagePath);
-                    if (!desc.exists()) {
-                        log.info("创建文件目录: {}", desc.mkdirs());
-                    }
-                    bodyJson.put("outputImagePath", outputImagePath);
-                    bodyJson.put("otherParams", otherParams);
-                    String logPath = TaaisConfig.getProfile() + "/task/log/" + dataAmplifyDto.getId() + "_log.log";
-                    System.out.println("logPath===>" + logPath);
-                    bodyJson.put("logPath", logPath);
-                    //实际请求接口，接口未提供，暂且注释
-                    String response = HttpRequest.post(sysDictDataVo.getDictValue())
-                        .body(JsonUtils.toJsonString(bodyJson))
-                        .execute().body();
-//                String response = "{\"status\":200,\"msg\":\"扩增成功\"}";
-                    ObjectMapper objectMapper = new ObjectMapper();
-                    JsonNode rootNode = objectMapper.readTree(response);
-                    String resultCode = rootNode.path(RESULT_CODE).asText();
-                    //判断接口是否响应成功
-                    if (!RESULT_STATUS.equals(resultCode)) {
-                        success = false;
-                    }
-                    message = rootNode.path("msg").asText();
-                    //处理当前目录文件，并进行入库
-                    saveDataInfo(outputImagePath, dataInfo);
-
-                    DataAmplificationTaskVo taskVo = dataAmplificationTaskService.selectById(Long.valueOf(dataAmplifyDto.getId()));
-                    Date endTime = new Date();
-                    DataAmplificationTaskBo update = new DataAmplificationTaskBo();
-                    if (taskVo.getInputImagePath() != null) {
-                        update.setInputImagePath(taskVo.getInputImagePath() + "|" + dataInfo.getUrl());
-                    } else {
-                        update.setInputImagePath(dataInfo.getUrl());
-                    }
-                    if (taskVo.getOutputImagePath() != null) {
-                        update.setOutputImagePath(taskVo.getOutputImagePath() + "|" + outputImagePath);
-                    } else {
-                        update.setOutputImagePath(outputImagePath);
-                    }
-                    update.setId(Long.valueOf(dataAmplifyDto.getId()));
-                    update.setStartTime(startTime);
-                    update.setEndTime(endTime);
-                    update.setCostSecond((int) (endTime.getTime() - startTime.getTime()));
-                    update.setStatus(success ? BizConstant.TASK_STATUS_SUCCEED : BizConstant.TASK_STATUS_FAILED);
-                    update.setLog(taskVo.getLog() + "|" + message);
-                    update.setVersion(taskVo.getVersion());
-                    update.setRemarks(taskVo.getRemarks() + "|" + (int) (endTime.getTime() - startTime.getTime()));
-                    dataAmplificationTaskService.update(update);
-                } catch (Exception e) {
-                    throw new RuntimeException(e);
-                }
-            });
         }
-        //根据批次号获取该批次的所有文件数据
-        return CommonResult.success();
-    }
 
-    private void updateFail(Long id) {
-        DataAmplificationTaskVo taskVo = dataAmplificationTaskService.selectById(id);
-        DataAmplificationTaskBo update = new DataAmplificationTaskBo();
-        update.setId(taskVo.getId());
-        update.setStatus(BizConstant.TASK_STATUS_FAILED);
-        update.setLog(taskVo.getLog() + "|" + "该批次下没有文件数据，请重新选择批次！");
-        update.setVersion(taskVo.getVersion());
-        dataAmplificationTaskService.update(update);
+        return CommonResult.success();
     }
 
-    private void updateEmp(Long id) {
-        DataAmplificationTaskVo taskVo = dataAmplificationTaskService.selectById(id);
-        DataAmplificationTaskBo update = new DataAmplificationTaskBo();
-        update.setOutputImagePath("");
-        update.setInputImagePath("");
-        update.setId(id);
-        update.setLog("");
-        update.setVersion(taskVo.getVersion());
-        update.setRemarks("");
-        dataAmplificationTaskService.update(update);
-    }
 
+    public static final String PATH_PREFIX = TaaisConfig.getProfile() + "/amplify";
+    public static final String WORK_DIR = TaaisConfig.getProfile();
     /**
-     * 解析目标目录所有文件，进行文件更名并入库
-     *
-     * @param directoryPath：目录地址
-     * @param dataInfo：深拷贝对象
+     * 移动文件到对应文件夹
+     * @param batch
+     * @param path
+     * @param namePrefix 文件名前缀
      */
-    public void saveDataInfo(String directoryPath, Data dataInfo) {
-        try {
-            // 获取指定目录下所有文件
-            File directory = new File(directoryPath);
-            List<File> extractedImagesFileList = new ArrayList<>();
-            if (directory.exists() && directory.isDirectory()) {
-                File[] files = directory.listFiles();
-                if (files != null) {
-                    for (File file : files) {
-                        initFileInfo(directoryPath, extractedImagesFileList, file.isDirectory(), file.getName());
-                    }
-                }
-            }
+    private void copyFilesToPath(String batch, String path, String namePrefix) {
+        String[] batches = batch.split(",");
 
-            // 获取ID集合
-            List<Long> ids = dataMapper.getIds(extractedImagesFileList.size());
-            if (ids.isEmpty()) {
-                return;
-            }
-
-            List<Data> dataList = new ArrayList<>();
-            AtomicInteger countSize = new AtomicInteger();
-            extractedImagesFileList.forEach(fileInfo -> {
-                Long id = ids.get(countSize.get());
-                Data data = new Data();
-                BeanUtils.copyProperties(dataInfo, data);
+        File dir = new File(PATH_PREFIX + path);
+        if (!dir.exists()) {
+            dir.mkdirs();
+        }
+        dir = new File(PATH_PREFIX + path + "/images");
+        if (!dir.exists()) {
+            dir.mkdirs();
+        }
+        dir = new File(PATH_PREFIX + path + "/labels");
+        if (!dir.exists()) {
+            dir.mkdirs();
+        }
+        dir = new File(PATH_PREFIX + path + "/result");
+        if (!dir.exists()) {
+            dir.mkdirs();
+        }
 
-                if (checkLabeled(fileInfo.getPath())) {
-                    data.setLabeled(Boolean.TRUE);
-                } else {
-                    data.setLabeled(Boolean.FALSE);
+        System.out.println(dir.getAbsolutePath());
+
+        for (String batchNum : batches) {
+            List<DataVo> dataVoList = getDataByBatchNum(batchNum);
+            for (DataVo dataVo : dataVoList) {
+                String[] strings = dataVo.getUrl().split("/profile");
+                String relativePath = WORK_DIR + strings[strings.length - 1];
+                relativePath = relativePath.replace("\\", "/").replace("//", "/");
+                File file = new File(relativePath);
+                System.out.println(file.getAbsolutePath());
+                if (file.exists()) {
+                    try {
+                        File dist = new File(PATH_PREFIX + path + "/images/" + namePrefix + file.getName());
+                        org.apache.commons.io.FileUtils.copyFile(file, dist);
+                        System.out.println("file dist: " + dist.getAbsolutePath());
+                    } catch (IOException e) {
+                        Log.debug("bug found");
+                        continue;
+                    }
                 }
-
-                try {
-                    Path path = Paths.get(fileInfo.getAbsolutePath());
-                    BasicFileAttributes attrs = Files.readAttributes(path, BasicFileAttributes.class);
-                    Instant creationTime = attrs.lastModifiedTime().toInstant();
-                    Date date = Date.from(creationTime);
-
-                    data.setId(id);
-                    data.setGatherTime(date);
-                    data.setName(fileInfo.getName());
-
-                    // 更改图片文件名称
-                    String fileHeaderSuffix = StringUtils.substring(fileInfo.getName(), fileInfo.getName().lastIndexOf("."), fileInfo.getName().length());
-                    String destInfo = fileInfo.getPath().replaceAll(fileInfo.getName(), "");
-                    File newFile = new File(destInfo, id + fileHeaderSuffix);
-                    File oldFile = new File(destInfo, fileInfo.getName());
-                    log.info("saveDataInfo更改用户上传图片文件名称:{}", oldFile.renameTo(newFile));
-
-                    String imagePath = FileUploadUtils.getPathFileName(destInfo, id + fileHeaderSuffix);
-                    data.setUrl(imagePath);
-
-                    if (data.getLabeled()) {
-                        String labeledPath = fileInfo.getPath().replaceFirst("[.][^.]+$", "") + ".txt";
-                        File labeledNewFile = new File(destInfo, id + ".txt");
-                        File labeledOldFile = new File(labeledPath);
-                        log.info("saveDataInfo更改用户上传标注文件名称:{}", labeledOldFile.renameTo(labeledNewFile));
-                        String labelUrl = FileUploadUtils.getPathFileName(destInfo, id + ".txt");
-                        data.setLabelurl(labelUrl);
+                if (dataVo.getLabelurl() != null) {
+                    strings = dataVo.getLabelurl().split("/profile");
+                    relativePath = WORK_DIR + strings[strings.length - 1];
+                    relativePath = relativePath.replace("\\", "/").replace("//", "/");
+                    file = new File(relativePath);
+                    if (file.exists()) {
+                        try {
+                            File dist = new File(PATH_PREFIX + path + "/labels/" + namePrefix + file.getName());
+                            org.apache.commons.io.FileUtils.copyFile(file, dist);
+                            System.out.println("file dist: " + dist.getAbsolutePath());
+                        } catch (IOException e) {
+                            Log.debug("bug found");
+                        }
                     }
-                    dataList.add(data);
-                } catch (IOException e) {
-                    throw new RuntimeException(e);
                 }
-                countSize.getAndIncrement();
-            });
-
-            dataMapper.insertBatch(dataList);
-        } catch (Exception e) {
-            log.error("[saveDataInfo]数据集处理出现未知异常.e:", e);
+            }
         }
     }
-
 }

taais-modules/taais-biz/src/main/java/com/taais/biz/service/impl/TargetIdentificationTaskServiceImpl.java

@@ -222,7 +222,7 @@ public class TargetIdentificationTaskServiceImpl extends BaseServiceImpl<TargetI
         // 创建扩增子任务
         if (taskDto.getHasTrainAugmentation()) {
             if (taskDto.getTrainAugmentationParams() != null && StringUtils.isNotEmpty(taskDto.getTrainAugmentationParams().getOtherParams())) {
-                createDataAmplifyTask(taskBo.getId(), taskDto);
+                createDataAmplifyTask(taskBo.getId(), taskDto, records);
             }
         }
         return null;
@@ -535,27 +535,27 @@ public class TargetIdentificationTaskServiceImpl extends BaseServiceImpl<TargetI
         }
     }
 
-    private void createDataAmplifyTask(Long taskId, CreateTargetIdentificationTaskDto taskDto) {
+    private void createDataAmplifyTask(Long taskId, CreateTargetIdentificationTaskDto taskDto, Map<String, String> records) {
         List<String> trainBatchNumList = taskDto.getTrainBatchNumList();
-        for (String batchNumber : trainBatchNumList) {
-            TargetIdentificationSubtaskBo subtask = new TargetIdentificationSubtaskBo();
-            subtask.setName("数据扩增");
-            subtask.setStatus(BizConstant.TASK_STATUS_PENDING);
-            subtask.setTaskId(taskId);
-            subtask.setRemarks("DEFAULT_REMARK");
-            TargetIdentificationSubtask insertSubtask = subtaskService.insertSubtask(subtask);
-
-            DataAmplifyDto amplifyDto = taskDto.getTrainAugmentationParams();
-            DataAmplificationTaskBo dataAmplificationTaskBo = new DataAmplificationTaskBo();
-            dataAmplificationTaskBo.setSubTaskId(insertSubtask.getId());
-            dataAmplificationTaskBo.setName(amplifyDto.getTaskName());
-            dataAmplificationTaskBo.setStatus(BizConstant.TASK_STATUS_PENDING);
-            dataAmplificationTaskBo.setDataBatchNums(batchNumber);
-            dataAmplificationTaskBo.setAugmentationType(amplifyDto.getAugmentationType());
-            dataAmplificationTaskBo.setParameters(JsonUtils.toJsonString(amplifyDto.getOtherParams()));
-            dataAmplificationTaskBo.setDelFlag(0);
-            dataAmplificationTaskService.insert(dataAmplificationTaskBo);
-        }
+
+        TargetIdentificationSubtaskBo subtask = new TargetIdentificationSubtaskBo();
+        subtask.setName("数据扩增");
+        subtask.setStatus(BizConstant.TASK_STATUS_PENDING);
+        subtask.setTaskId(taskId);
+        subtask.setRemarks("DEFAULT_REMARK");
+        TargetIdentificationSubtask insertSubtask = subtaskService.insertSubtask(subtask);
+
+        DataAmplifyDto amplifyDto = taskDto.getTrainAugmentationParams();
+        DataAmplificationTaskBo dataAmplificationTaskBo = new DataAmplificationTaskBo();
+        dataAmplificationTaskBo.setSubTaskId(insertSubtask.getId());
+        dataAmplificationTaskBo.setName(amplifyDto.getTaskName());
+        dataAmplificationTaskBo.setStatus(BizConstant.TASK_STATUS_PENDING);
+        dataAmplificationTaskBo.setDataBatchNums(amplifyDto.getBatchNum());
+        dataAmplificationTaskBo.setAugmentationType(amplifyDto.getAugmentationType());
+        dataAmplificationTaskBo.setParameters(JsonUtils.toJsonString(amplifyDto.getOtherParams()));
+        dataAmplificationTaskBo.setOutputImagePath(String.join(";", records.values()));
+        dataAmplificationTaskBo.setDelFlag(0);
+        dataAmplificationTaskService.insert(dataAmplificationTaskBo);
     }