|
@@ -2,19 +2,21 @@ package com.taais.biz.service.impl;
|
|
|
|
|
|
import cn.hutool.core.util.ObjectUtil;
|
|
import cn.hutool.core.util.ObjectUtil;
|
|
import cn.hutool.http.HttpRequest;
|
|
import cn.hutool.http.HttpRequest;
|
|
-import cn.hutool.http.HttpUtil;
|
|
|
|
-import com.fasterxml.jackson.core.JsonProcessingException;
|
|
|
|
import com.fasterxml.jackson.databind.JsonNode;
|
|
import com.fasterxml.jackson.databind.JsonNode;
|
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
|
import com.mybatisflex.core.paginate.Page;
|
|
import com.mybatisflex.core.paginate.Page;
|
|
import com.mybatisflex.core.query.QueryWrapper;
|
|
import com.mybatisflex.core.query.QueryWrapper;
|
|
|
|
+import com.taais.biz.constant.BizConstant;
|
|
import com.taais.biz.domain.Data;
|
|
import com.taais.biz.domain.Data;
|
|
|
|
+import com.taais.biz.domain.bo.DataAmplificationTaskBo;
|
|
import com.taais.biz.domain.bo.DataBo;
|
|
import com.taais.biz.domain.bo.DataBo;
|
|
import com.taais.biz.domain.vo.BatchDataResult;
|
|
import com.taais.biz.domain.vo.BatchDataResult;
|
|
import com.taais.biz.domain.dto.DataAmplifyDto;
|
|
import com.taais.biz.domain.dto.DataAmplifyDto;
|
|
|
|
+import com.taais.biz.domain.vo.DataAmplificationTaskVo;
|
|
import com.taais.biz.domain.vo.DataSelectVo;
|
|
import com.taais.biz.domain.vo.DataSelectVo;
|
|
import com.taais.biz.domain.vo.DataVo;
|
|
import com.taais.biz.domain.vo.DataVo;
|
|
import com.taais.biz.mapper.DataMapper;
|
|
import com.taais.biz.mapper.DataMapper;
|
|
|
|
+import com.taais.biz.service.IDataAmplificationTaskService;
|
|
import com.taais.biz.service.IDataService;
|
|
import com.taais.biz.service.IDataService;
|
|
import com.taais.common.core.config.TaaisConfig;
|
|
import com.taais.common.core.config.TaaisConfig;
|
|
import com.taais.common.core.core.domain.CommonResult;
|
|
import com.taais.common.core.core.domain.CommonResult;
|
|
@@ -34,6 +36,7 @@ import org.redisson.api.RedissonClient;
|
|
import org.slf4j.Logger;
|
|
import org.slf4j.Logger;
|
|
import org.slf4j.LoggerFactory;
|
|
import org.slf4j.LoggerFactory;
|
|
import org.springframework.beans.BeanUtils;
|
|
import org.springframework.beans.BeanUtils;
|
|
|
|
+import org.springframework.beans.factory.annotation.Value;
|
|
import org.springframework.stereotype.Service;
|
|
import org.springframework.stereotype.Service;
|
|
import org.springframework.transaction.annotation.Transactional;
|
|
import org.springframework.transaction.annotation.Transactional;
|
|
import org.springframework.web.multipart.MultipartFile;
|
|
import org.springframework.web.multipart.MultipartFile;
|
|
@@ -48,7 +51,9 @@ import java.time.Instant;
|
|
import java.time.LocalDate;
|
|
import java.time.LocalDate;
|
|
import java.time.format.DateTimeFormatter;
|
|
import java.time.format.DateTimeFormatter;
|
|
import java.util.*;
|
|
import java.util.*;
|
|
|
|
+import java.util.concurrent.atomic.AtomicBoolean;
|
|
import java.util.concurrent.atomic.AtomicInteger;
|
|
import java.util.concurrent.atomic.AtomicInteger;
|
|
|
|
+import java.util.concurrent.atomic.AtomicReference;
|
|
import java.util.stream.Collectors;
|
|
import java.util.stream.Collectors;
|
|
|
|
|
|
import static com.taais.biz.domain.table.DataTableDef.DATA;
|
|
import static com.taais.biz.domain.table.DataTableDef.DATA;
|
|
@@ -68,13 +73,16 @@ public class DataServiceImpl extends BaseServiceImpl<DataMapper, Data> implement
|
|
private static final String RAR = ".rar";
|
|
private static final String RAR = ".rar";
|
|
private static final String TXT = ".txt";
|
|
private static final String TXT = ".txt";
|
|
private static final String[] VALID_EXTENSIONS = {".jpg", ".jpeg", ".png"};
|
|
private static final String[] VALID_EXTENSIONS = {".jpg", ".jpeg", ".png"};
|
|
- private static final String PYTHON_DATA_AMPLIFY_API = "http://127.0.0.1:9096/api/data/amplify";
|
|
|
|
|
|
+ private static final String PYTHON_DATA_AMPLIFY_API = "http://127.0.0.1:11001/augment";
|
|
private static final String RESULT_CODE = "status";
|
|
private static final String RESULT_CODE = "status";
|
|
private static final String RESULT_STATUS = "200";
|
|
private static final String RESULT_STATUS = "200";
|
|
private static final String AMPLIFY = "/AMPLIFY/";
|
|
private static final String AMPLIFY = "/AMPLIFY/";
|
|
@Resource
|
|
@Resource
|
|
private DataMapper dataMapper;
|
|
private DataMapper dataMapper;
|
|
|
|
|
|
|
|
+ @Resource
|
|
|
|
+ private IDataAmplificationTaskService dataAmplificationTaskService;
|
|
|
|
+
|
|
@Override
|
|
@Override
|
|
public QueryWrapper query() {
|
|
public QueryWrapper query() {
|
|
return super.query().from(DATA);
|
|
return super.query().from(DATA);
|
|
@@ -278,7 +286,6 @@ public class DataServiceImpl extends BaseServiceImpl<DataMapper, Data> implement
|
|
if (dataList.isEmpty()) {
|
|
if (dataList.isEmpty()) {
|
|
return CommonResult.fail("该批次下没有文件数据,请重新选择批次!");
|
|
return CommonResult.fail("该批次下没有文件数据,请重新选择批次!");
|
|
}
|
|
}
|
|
- //TODO: 此处需要定义任务开始,把相关任务信息添加上(任务名称、任务开始时间、任务类型),然后再处理文件。
|
|
|
|
|
|
|
|
List<Data> dataListInfo = dataList.stream().filter(data -> !StringUtils.isEmpty(data.getUrl())).toList();
|
|
List<Data> dataListInfo = dataList.stream().filter(data -> !StringUtils.isEmpty(data.getUrl())).toList();
|
|
if (dataListInfo.isEmpty()) {
|
|
if (dataListInfo.isEmpty()) {
|
|
@@ -291,6 +298,9 @@ public class DataServiceImpl extends BaseServiceImpl<DataMapper, Data> implement
|
|
String formattedDate = currentDate.format(formatter);
|
|
String formattedDate = currentDate.format(formatter);
|
|
filePath = filePath + File.separator + formattedDate;
|
|
filePath = filePath + File.separator + formattedDate;
|
|
String finalFilePath = filePath;
|
|
String finalFilePath = filePath;
|
|
|
|
+ Date startTime = new Date();
|
|
|
|
+ AtomicBoolean success = new AtomicBoolean(true);
|
|
|
|
+ AtomicReference<String> message = new AtomicReference<>("");
|
|
dataListInfo.forEach(dataInfo -> {
|
|
dataListInfo.forEach(dataInfo -> {
|
|
try {
|
|
try {
|
|
//循环调用Python扩增接口
|
|
//循环调用Python扩增接口
|
|
@@ -305,23 +315,33 @@ public class DataServiceImpl extends BaseServiceImpl<DataMapper, Data> implement
|
|
bodyJson.put("outputImagePath", outputImagePath);
|
|
bodyJson.put("outputImagePath", outputImagePath);
|
|
bodyJson.put("otherParams", dataAmplifyDto.getOtherParams());
|
|
bodyJson.put("otherParams", dataAmplifyDto.getOtherParams());
|
|
//实际请求接口,接口未提供,暂且注释
|
|
//实际请求接口,接口未提供,暂且注释
|
|
-// String response = HttpRequest.post(PYTHON_DATA_AMPLIFY_API)
|
|
|
|
-// .body(JsonUtils.toJsonString(bodyJson))
|
|
|
|
-// .execute().body();
|
|
|
|
- String response = "{\"status\":200,\"msg\":\"扩增成功\"}";
|
|
|
|
|
|
+ String response = HttpRequest.post(PYTHON_DATA_AMPLIFY_API)
|
|
|
|
+ .body(JsonUtils.toJsonString(bodyJson))
|
|
|
|
+ .execute().body();
|
|
|
|
+// String response = "{\"status\":200,\"msg\":\"扩增成功\"}";
|
|
ObjectMapper objectMapper = new ObjectMapper();
|
|
ObjectMapper objectMapper = new ObjectMapper();
|
|
JsonNode rootNode = objectMapper.readTree(response);
|
|
JsonNode rootNode = objectMapper.readTree(response);
|
|
String resultCode = rootNode.path(RESULT_CODE).asText();
|
|
String resultCode = rootNode.path(RESULT_CODE).asText();
|
|
//判断接口是否响应成功
|
|
//判断接口是否响应成功
|
|
if (!RESULT_STATUS.equals(resultCode)) {
|
|
if (!RESULT_STATUS.equals(resultCode)) {
|
|
- throw new RuntimeException("调用Python接口返回扩增失败");
|
|
|
|
|
|
+ success.set(true);
|
|
}
|
|
}
|
|
|
|
+ message.set(rootNode.path("msg").asText());
|
|
//处理当前目录文件,并进行入库
|
|
//处理当前目录文件,并进行入库
|
|
saveDataInfo(outputImagePath, dataInfo);
|
|
saveDataInfo(outputImagePath, dataInfo);
|
|
} catch (Exception e) {
|
|
} catch (Exception e) {
|
|
throw new RuntimeException(e);
|
|
throw new RuntimeException(e);
|
|
}
|
|
}
|
|
});
|
|
});
|
|
|
|
+ Date endTime = new Date();
|
|
|
|
+ DataAmplificationTaskBo update = new DataAmplificationTaskBo();
|
|
|
|
+ update.setId(Long.valueOf(dataAmplifyDto.getId()));
|
|
|
|
+ update.setStartTime(startTime);
|
|
|
|
+ update.setEndTime(endTime);
|
|
|
|
+ update.setCostSecond((int) (endTime.getTime() - startTime.getTime()));
|
|
|
|
+ update.setStatus(success.get() ? BizConstant.TASK_STATUS_SUCCEED : BizConstant.TASK_STATUS_FAILED);
|
|
|
|
+ update.setLog(message.get());
|
|
|
|
+ dataAmplificationTaskService.update(update);
|
|
return CommonResult.fail("该批次下没有文件数据,请重新选择批次!");
|
|
return CommonResult.fail("该批次下没有文件数据,请重新选择批次!");
|
|
}
|
|
}
|
|
|
|
|
|
@@ -436,6 +456,143 @@ public class DataServiceImpl extends BaseServiceImpl<DataMapper, Data> implement
|
|
return cacheList;
|
|
return cacheList;
|
|
}
|
|
}
|
|
|
|
|
|
|
|
+ @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);
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+ @Value("${taais.profile}")
|
|
|
|
+ private String profile;
|
|
|
|
+
|
|
|
|
+ private CommonResult<Boolean> doAmplify(DataAmplifyDto dataAmplifyDto) {
|
|
|
|
+ this.updateEmp(Long.valueOf(dataAmplifyDto.getId()));
|
|
|
|
+ String[] split = dataAmplifyDto.getBatchNum().split(",");
|
|
|
|
+ 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;
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+ 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"));
|
|
|
|
+ }
|
|
|
|
+ Map<String, String> param = dataAmplifyDto.getOtherParams()
|
|
|
|
+ .stream().collect(Collectors.toMap(item -> item.get("agName"), item -> item.get("defaultValue")));
|
|
|
|
+ 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 = profile + "/task/log/" + dataAmplifyDto.getId() + "_log.log";
|
|
|
|
+ System.out.println("logPath===>" + logPath);
|
|
|
|
+ bodyJson.put("logPath", logPath);
|
|
|
|
+ //实际请求接口,接口未提供,暂且注释
|
|
|
|
+ String response = HttpRequest.post(PYTHON_DATA_AMPLIFY_API)
|
|
|
|
+ .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 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);
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+ 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);
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
|
|
/**
|
|
/**
|
|
* 解析目标目录所有文件,进行文件更名并入库
|
|
* 解析目标目录所有文件,进行文件更名并入库
|