fix:1.修改天气预报和源项重建模块测试过程中的问题
This commit is contained in:
parent
c1773b6a90
commit
a55343a10e
|
|
@ -3,7 +3,7 @@ package org.jeecg.common.constant.enums;
|
|||
/**
|
||||
* 输运模拟任务置顶说明枚举
|
||||
*/
|
||||
public enum TransportTaskTopEnum {
|
||||
public enum TaskTopEnum {
|
||||
|
||||
/**
|
||||
* 未置顶
|
||||
|
|
@ -17,7 +17,7 @@ public enum TransportTaskTopEnum {
|
|||
|
||||
private Integer value;
|
||||
|
||||
TransportTaskTopEnum(Integer value) {
|
||||
TaskTopEnum(Integer value) {
|
||||
this.value = value;
|
||||
}
|
||||
|
||||
|
|
@ -86,6 +86,7 @@ public class SourceRebuildTask implements Serializable {
|
|||
*/
|
||||
@NotNull(message = "srs时间周期-开始日期不能为空",groups = {InsertGroup.class, UpdateGroup.class})
|
||||
@JsonFormat(timezone = "GMT+8", pattern = "yyyy-MM-dd")
|
||||
@DateTimeFormat(pattern = "yyyy-MM-dd")
|
||||
@TableField(value = "srs_start_time")
|
||||
private LocalDate srsStartTime;
|
||||
|
||||
|
|
@ -94,6 +95,7 @@ public class SourceRebuildTask implements Serializable {
|
|||
*/
|
||||
@NotNull(message = "srs时间周期-结束日期不能为空",groups = {InsertGroup.class, UpdateGroup.class})
|
||||
@JsonFormat(timezone = "GMT+8", pattern = "yyyy-MM-dd")
|
||||
@DateTimeFormat(pattern = "yyyy-MM-dd")
|
||||
@TableField(value = "srs_end_time")
|
||||
private LocalDate srsEndTime;
|
||||
|
||||
|
|
@ -127,14 +129,16 @@ public class SourceRebuildTask implements Serializable {
|
|||
* 释放源开始释放时间
|
||||
*/
|
||||
@TableField(value = "release_start_time")
|
||||
@JsonFormat(timezone = "GMT+8", pattern = "yyyy-MM-dd")
|
||||
@JsonFormat(timezone = "GMT+8", pattern = "yyyy-MM-dd HH:mm:ss")
|
||||
@DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss")
|
||||
private LocalDateTime releaseStartTime;
|
||||
|
||||
/**
|
||||
* 释放源结束释放时间
|
||||
*/
|
||||
@TableField(value = "release_end_time")
|
||||
@JsonFormat(timezone = "GMT+8", pattern = "yyyy-MM-dd")
|
||||
@JsonFormat(timezone = "GMT+8", pattern = "yyyy-MM-dd HH:mm:ss")
|
||||
@DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss")
|
||||
private LocalDateTime releaseEndTime;
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -39,7 +39,7 @@ public class RebuildTaskConsumerHandler {
|
|||
MessageConsumerThread messageConsumerThread = new MessageConsumerThread();
|
||||
messageConsumerThread.setName("rebuild-task-thread");
|
||||
messageConsumerThread.start();
|
||||
log.info("启动源项重建任务消费线程----------------------");
|
||||
log.info("启动源项重建任务消费线程");
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -28,4 +28,10 @@ public interface SourceRebuildTaskService extends IService<SourceRebuildTask> {
|
|||
* @param task
|
||||
*/
|
||||
void setInpectionFailed(SourceRebuildTask task);
|
||||
|
||||
/**
|
||||
* 取消置顶
|
||||
* @param taskId
|
||||
*/
|
||||
void setCancelTaskTop(Integer taskId);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
|
|||
import lombok.RequiredArgsConstructor;
|
||||
import org.jeecg.common.constant.CommonConstant;
|
||||
import org.jeecg.common.constant.enums.SourceRebuildTaskStatusEnum;
|
||||
import org.jeecg.common.constant.enums.TaskTopEnum;
|
||||
import org.jeecg.common.properties.ServerProperties;
|
||||
import org.jeecg.common.util.RedisUtil;
|
||||
import org.jeecg.modules.base.entity.SourceRebuildTask;
|
||||
|
|
@ -11,6 +12,7 @@ import org.jeecg.modules.base.mapper.SourceRebuildTaskMapper;
|
|||
import org.jeecg.rebuild.service.SourceRebuildTaskService;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
import java.util.Objects;
|
||||
|
||||
/**
|
||||
* 源项重建任务
|
||||
|
|
@ -62,4 +64,21 @@ public class SourceRebuildTaskServiceImpl extends ServiceImpl<SourceRebuildTaskM
|
|||
task.setTaskStatus(SourceRebuildTaskStatusEnum.INSPECTION_FAILED.getValue());
|
||||
this.baseMapper.updateById(task);
|
||||
}
|
||||
|
||||
/**
|
||||
* 取消置顶
|
||||
* @param taskId
|
||||
*/
|
||||
@Transactional(rollbackFor = RuntimeException.class)
|
||||
@Override
|
||||
public void setCancelTaskTop(Integer taskId) {
|
||||
SourceRebuildTask task = this.baseMapper.selectById(taskId);
|
||||
if (Objects.isNull(task)) {
|
||||
throw new RuntimeException("此任务不存在");
|
||||
}
|
||||
if (TaskTopEnum.TOP.getValue().equals(task.getTopTask())) {
|
||||
task.setTopTask(TaskTopEnum.NOT_TOP.getValue());
|
||||
this.baseMapper.updateById(task);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2,14 +2,11 @@ package org.jeecg.rebuild.task;
|
|||
|
||||
import cn.hutool.core.collection.CollUtil;
|
||||
import cn.hutool.core.date.DateUtil;
|
||||
import cn.hutool.core.io.FileUtil;
|
||||
import cn.hutool.core.util.ArrayUtil;
|
||||
import com.jcraft.jsch.JSchException;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.commons.lang3.time.StopWatch;
|
||||
import org.jeecg.common.constant.enums.SourceRebuildReleaseSourceEnum;
|
||||
import org.jeecg.common.constant.enums.SourceRebuildTaskStatusEnum;
|
||||
import org.jeecg.common.constant.enums.TransportTaskStatusEnum;
|
||||
import org.jeecg.common.properties.ServerProperties;
|
||||
import org.jeecg.common.properties.SourceRebuildProperties;
|
||||
import org.jeecg.jsch.JSchRemoteRunner;
|
||||
|
|
@ -24,24 +21,17 @@ import org.jeecg.transport.flexparttask.ProgressQueue;
|
|||
import org.rosuda.REngine.REXP;
|
||||
import org.rosuda.REngine.REXPDouble;
|
||||
import org.rosuda.REngine.REXPInteger;
|
||||
import org.rosuda.REngine.REXPMismatchException;
|
||||
import org.rosuda.REngine.Rserve.RConnection;
|
||||
import org.rosuda.REngine.Rserve.RserveException;
|
||||
import org.springframework.util.CollectionUtils;
|
||||
import java.io.File;
|
||||
import java.math.BigDecimal;
|
||||
import java.math.RoundingMode;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.nio.file.Paths;
|
||||
import java.text.DateFormat;
|
||||
import java.text.DecimalFormat;
|
||||
import java.time.Duration;
|
||||
import java.time.LocalDateTime;
|
||||
import java.time.format.DateTimeFormatter;
|
||||
import java.util.*;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.stream.Stream;
|
||||
|
||||
@Slf4j
|
||||
public class SourceRebuildTaskExec extends Thread{
|
||||
|
|
@ -104,11 +94,9 @@ public class SourceRebuildTaskExec extends Thread{
|
|||
for (SourceRebuildMonitoringData monitoringData : taskMonitoringDatas) {
|
||||
StringBuilder srsPath = new StringBuilder();
|
||||
srsPath.append(sourceRebuildProperties.getSrsFilePath());
|
||||
// srsPath.append(File.separator);
|
||||
srsPath.append("/");
|
||||
srsPath.append(File.separator);
|
||||
srsPath.append(DateUtil.format(monitoringData.getCollectStop(),"yyyyMMdd"));
|
||||
// srsPath.append(File.separator);
|
||||
srsPath.append("/");
|
||||
srsPath.append(File.separator);
|
||||
srsPath.append(sourceRebuildProperties.getXeGZFileDirSign());
|
||||
srsDirsPath.add(srsPath.toString());
|
||||
}
|
||||
|
|
@ -175,6 +163,8 @@ public class SourceRebuildTaskExec extends Thread{
|
|||
try{
|
||||
//修改任务状态为执行中
|
||||
this.sourceRebuildTaskService.updateTaskStatus(this.sourceRebuildTask.getId(), SourceRebuildTaskStatusEnum.IN_OPERATION.getValue());
|
||||
//任务开始运行后,如果之前是置顶状态,则设置任务取消置顶
|
||||
this.sourceRebuildTaskService.setCancelTaskTop(this.sourceRebuildTask.getId());
|
||||
//如果此任务已存在历史日志,先清除
|
||||
sourceRebuildTaskLogService.delete(this.sourceRebuildTask.getId());
|
||||
String startRunLog = "----------------------------------------开始执行任务----------------------------------------";
|
||||
|
|
@ -216,8 +206,7 @@ public class SourceRebuildTaskExec extends Thread{
|
|||
this.generateLog(monitorDataLog);
|
||||
this.generateLog(title);
|
||||
//根据监测数据生成文件(input_subexp1.dat)
|
||||
// String inputPath = sourceRebuildProperties.getRInput()+File.separator+ "input_subexp1.dat";
|
||||
String inputPath = sourceRebuildProperties.getRInput()+"/"+ "input_subexp1.dat";
|
||||
String inputPath = sourceRebuildProperties.getRInput()+File.separator+ "input_subexp1.dat";
|
||||
//格式化监测数据
|
||||
List<String> lines = new ArrayList<>();
|
||||
lines.add(title);
|
||||
|
|
@ -244,8 +233,7 @@ public class SourceRebuildTaskExec extends Thread{
|
|||
String srsDataLog = "----------------------------------------生成SRS数据输入文件----------------------------------------";
|
||||
this.generateLog(srsDataLog);
|
||||
this.srsFilesPath.forEach(this::generateLog);
|
||||
// String inputPath = sourceRebuildProperties.getRInput()+File.separator+ "srsfilelist_subexp1.dat";
|
||||
String inputPath = sourceRebuildProperties.getRInput()+"/"+ "srsfilelist_subexp1.dat";
|
||||
String inputPath = sourceRebuildProperties.getRInput()+File.separator+ "srsfilelist_subexp1.dat";
|
||||
jschRemoteRunner.writeFile(inputPath,String.join("",this.srsFilesPath));
|
||||
}
|
||||
|
||||
|
|
@ -330,7 +318,6 @@ public class SourceRebuildTaskExec extends Thread{
|
|||
" )" +
|
||||
" })" +
|
||||
"})";
|
||||
// REXP outputREXP = conn.eval("capture.output({source(\"R_scripts/main.R\")})");
|
||||
REXP outputREXP = conn.eval(rCmd);
|
||||
// 将输出作为字符串数组获取
|
||||
String[] outputLines = outputREXP.asStrings();
|
||||
|
|
@ -370,8 +357,7 @@ public class SourceRebuildTaskExec extends Thread{
|
|||
private String getOutputPath(){
|
||||
StringBuilder outputPath = new StringBuilder();
|
||||
outputPath.append(sourceRebuildProperties.getROutput());
|
||||
outputPath.append("/");
|
||||
// outputPath.append(File.separator);
|
||||
outputPath.append(File.separator);
|
||||
outputPath.append(sourceRebuildTask.getTaskName());
|
||||
//创建任务输出目录
|
||||
this.jschRemoteRunner.mkdir(outputPath.toString());
|
||||
|
|
|
|||
|
|
@ -1,23 +1,15 @@
|
|||
package org.jeecg.transport.consumer;
|
||||
|
||||
import cn.hutool.core.collection.CollUtil;
|
||||
import com.alibaba.fastjson2.JSON;
|
||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
|
||||
import org.apache.rocketmq.client.exception.MQClientException;
|
||||
import org.apache.rocketmq.common.message.MessageExt;
|
||||
import org.jeecg.common.constant.CommonConstant;
|
||||
import org.jeecg.common.constant.RocketMQTopConstant;
|
||||
import org.jeecg.common.constant.enums.TransportTaskStatusEnum;
|
||||
import org.jeecg.common.constant.enums.TransportTaskTopEnum;
|
||||
import org.jeecg.common.properties.DataFusionProperties;
|
||||
import org.jeecg.common.properties.ServerProperties;
|
||||
import org.jeecg.common.properties.SystemStorageProperties;
|
||||
import org.jeecg.common.properties.TransportSimulationProperties;
|
||||
import org.jeecg.common.util.RedisUtil;
|
||||
import org.jeecg.modules.base.dto.TransportTaskDTO;
|
||||
import org.jeecg.modules.base.entity.TransportTask;
|
||||
import org.jeecg.modules.base.mapper.*;
|
||||
import org.jeecg.transport.consumer.china.AbstractTaskMsgHandler;
|
||||
|
|
@ -25,10 +17,8 @@ import org.jeecg.transport.consumer.china.Server11TaskHandler;
|
|||
import org.jeecg.transport.service.StationDataService;
|
||||
import org.jeecg.transport.service.StationsModValService;
|
||||
import org.jeecg.transport.service.TransportTaskService;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Objects;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
|
|
@ -59,7 +49,7 @@ public class TranTaskConsumerHandler{
|
|||
MessageConsumerThread messageConsumerThread = new MessageConsumerThread();
|
||||
messageConsumerThread.setName("transport-task-thread");
|
||||
messageConsumerThread.start();
|
||||
log.info("启动输运模拟任务消费线程----------------------");
|
||||
log.info("启动输运模拟任务消费线程");
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -89,6 +89,8 @@ public class BackwardTaskExec extends AbstractTaskExec {
|
|||
super.setTaskRunFlag();
|
||||
//修改任务状态为执行中
|
||||
super.transportTaskService.updateTaskStatus(super.transportTask.getId(), TransportTaskStatusEnum.IN_OPERATION.getValue());
|
||||
//任务开始运行后,如果之前是置顶状态,则设置任务取消置顶
|
||||
this.transportTaskService.setCancelTaskTop(this.transportTask.getId());
|
||||
//如果此任务已存在历史日志,先清除
|
||||
super.transportTaskService.deleteTaskLog(super.transportTask.getId());
|
||||
//执行模拟
|
||||
|
|
|
|||
|
|
@ -105,6 +105,8 @@ public class ForwardTaskExec extends AbstractTaskExec {
|
|||
super.setTaskRunFlag();
|
||||
//修改任务状态为执行中
|
||||
super.transportTaskService.updateTaskStatus(super.transportTask.getId(), TransportTaskStatusEnum.IN_OPERATION.getValue());
|
||||
//任务开始运行后,如果之前是置顶状态,则设置任务取消置顶
|
||||
this.transportTaskService.setCancelTaskTop(this.transportTask.getId());
|
||||
//如果此任务已存在历史日志,先清除
|
||||
super.transportTaskService.deleteTaskLog(super.transportTask.getId());
|
||||
//执行模拟
|
||||
|
|
|
|||
|
|
@ -49,4 +49,10 @@ public interface TransportTaskService extends IService<TransportTask> {
|
|||
* @param transportTask
|
||||
*/
|
||||
void setInpectionFailed(TransportTask transportTask);
|
||||
|
||||
/**
|
||||
* 取消置顶
|
||||
* @param taskId
|
||||
*/
|
||||
void setCancelTaskTop(Integer taskId);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -13,8 +13,6 @@ import org.jeecg.modules.base.mapper.*;
|
|||
import org.jeecg.transport.service.TransportTaskService;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.*;
|
||||
|
||||
/**
|
||||
|
|
@ -53,7 +51,7 @@ public class TransportTaskServiceImpl extends ServiceImpl<TransportTaskMapper,Tr
|
|||
transportTask.setTaskStatus(status);
|
||||
transportTask.setTimeConsuming(minute);
|
||||
//任务执行完成,如果之前任务有设置置顶,则修改为默认值
|
||||
transportTask.setTopTask(TransportTaskTopEnum.NOT_TOP.getValue());
|
||||
transportTask.setTopTask(TaskTopEnum.NOT_TOP.getValue());
|
||||
this.baseMapper.updateById(transportTask);
|
||||
redisUtil.hdel(CommonConstant.HOST_TASK_STATE,CommonConstant.TRAN_TASK_STATE_PRE+serverProperties.getHost());
|
||||
}
|
||||
|
|
@ -128,4 +126,22 @@ public class TransportTaskServiceImpl extends ServiceImpl<TransportTaskMapper,Tr
|
|||
transportTask.setTaskStatus(TransportTaskStatusEnum.INSPECTION_FAILED.getValue());
|
||||
this.baseMapper.updateById(transportTask);
|
||||
}
|
||||
|
||||
/**
|
||||
* 取消置顶
|
||||
*
|
||||
* @param taskId
|
||||
*/
|
||||
@Transactional(rollbackFor = RuntimeException.class)
|
||||
@Override
|
||||
public void setCancelTaskTop(Integer taskId) {
|
||||
TransportTask transportTask = this.baseMapper.selectById(taskId);
|
||||
if (Objects.isNull(transportTask)) {
|
||||
throw new RuntimeException("此任务不存在");
|
||||
}
|
||||
if (TaskTopEnum.TOP.getValue().equals(transportTask.getTopTask())) {
|
||||
transportTask.setTopTask(TaskTopEnum.NOT_TOP.getValue());
|
||||
this.baseMapper.updateById(transportTask);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -37,7 +37,7 @@ public class WeatherTaskConsumerHandler {
|
|||
MessageConsumerThread messageConsumerThread = new MessageConsumerThread();
|
||||
messageConsumerThread.setName("weather-task-thread");
|
||||
messageConsumerThread.start();
|
||||
log.info("启动天气预报任务消费线程----------------------");
|
||||
log.info("启动天气预报任务消费线程");
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -42,4 +42,10 @@ public interface WeatherTaskService extends IService<WeatherTask> {
|
|||
*/
|
||||
void setInpectionFailed(WeatherTask weatherTask);
|
||||
|
||||
/**
|
||||
* 取消置顶
|
||||
* @param taskId
|
||||
*/
|
||||
void setCancelTaskTop(Integer taskId);
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
|||
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.jeecg.common.constant.CommonConstant;
|
||||
import org.jeecg.common.constant.enums.TaskTopEnum;
|
||||
import org.jeecg.common.constant.enums.WeatherTaskStatusEnum;
|
||||
import org.jeecg.common.properties.ServerProperties;
|
||||
import org.jeecg.common.util.RedisUtil;
|
||||
|
|
@ -14,6 +15,7 @@ import org.jeecg.modules.base.mapper.WeatherTaskMapper;
|
|||
import org.jeecg.weather.service.WeatherTaskService;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
import java.util.Objects;
|
||||
|
||||
/**
|
||||
* 天气预报预测任务管理
|
||||
|
|
@ -90,4 +92,22 @@ public class WeatherTaskServiceImpl extends ServiceImpl<WeatherTaskMapper, Weath
|
|||
weatherTask.setTaskStatus(WeatherTaskStatusEnum.INSPECTION_FAILED.getValue());
|
||||
this.updateById(weatherTask);
|
||||
}
|
||||
|
||||
/**
|
||||
* 取消置顶
|
||||
*
|
||||
* @param taskId
|
||||
*/
|
||||
@Transactional(rollbackFor = RuntimeException.class)
|
||||
@Override
|
||||
public void setCancelTaskTop(Integer taskId) {
|
||||
WeatherTask weatherTask = this.baseMapper.selectById(taskId);
|
||||
if (Objects.isNull(weatherTask)) {
|
||||
throw new RuntimeException("此任务不存在");
|
||||
}
|
||||
if (TaskTopEnum.TOP.getValue().equals(weatherTask.getTopTask())) {
|
||||
weatherTask.setTopTask(TaskTopEnum.NOT_TOP.getValue());
|
||||
this.baseMapper.updateById(weatherTask);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -3,9 +3,12 @@ package org.jeecg.weather.task;
|
|||
import cn.hutool.core.collection.CollUtil;
|
||||
import cn.hutool.core.io.FileUtil;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.commons.codec.digest.DigestUtils;
|
||||
import org.apache.commons.io.FileUtils;
|
||||
import org.apache.commons.lang3.ArrayUtils;
|
||||
import org.apache.commons.lang3.time.StopWatch;
|
||||
import org.jeecg.common.constant.enums.WeatherDataSourceEnum;
|
||||
import org.jeecg.common.constant.enums.WeatherFileSuffixEnum;
|
||||
import org.jeecg.common.constant.enums.WeatherForecastDatasourceEnum;
|
||||
import org.jeecg.common.constant.enums.WeatherTaskStatusEnum;
|
||||
import org.jeecg.common.util.Grib2TimeReader;
|
||||
|
|
@ -38,6 +41,18 @@ public class WeatherForecastTaskExec extends AbstractWeatherTask {
|
|||
* 适配flexpart格式化存储路径(临时)
|
||||
*/
|
||||
private final static String FLEXPART_FORMAT_DIR = "flexpart_format";
|
||||
/**
|
||||
* 如果日志包含RuntimeError说明 ai-models运行出现错误了,把任务状态修改为-1
|
||||
*/
|
||||
private final static String EXCEPTION_LOG1 = "RuntimeError";
|
||||
/**
|
||||
* 如果日志包含RuntimeError说明 ai-models运行出现错误了,把任务状态修改为-1
|
||||
*/
|
||||
private final static String EXCEPTION_LOG2 = "Traceback";
|
||||
/**
|
||||
* 如果在预测阶段出现错误,则此字段值为true,后续数据格式化以及入库阶段不在往下走
|
||||
*/
|
||||
private boolean failureFlag = false;
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
|
|
@ -50,21 +65,29 @@ public class WeatherForecastTaskExec extends AbstractWeatherTask {
|
|||
public void execute() {
|
||||
StopWatch stopWatch = new StopWatch();
|
||||
stopWatch.start();
|
||||
String taskErrorLog = "";
|
||||
try{
|
||||
super.setRedisFlag();
|
||||
//修改任务状态为执行中
|
||||
this.weatherTaskService.updateTaskStatus(this.weatherTask.getId(), WeatherTaskStatusEnum.IN_OPERATION.getValue());
|
||||
//任务开始运行后,如果之前是置顶状态,则设置任务取消置顶
|
||||
this.weatherTaskService.setCancelTaskTop(this.weatherTask.getId());
|
||||
//如果此任务已存在历史日志,先清除
|
||||
this.weatherTaskService.deleteTaskLog(this.weatherTask.getId());
|
||||
//执行模拟
|
||||
this.execSimulation();
|
||||
}catch (Exception e){
|
||||
this.failureFlag = true;
|
||||
taskErrorLog = "任务执行失败,原因:"+e.getMessage();
|
||||
throw e;
|
||||
}finally {
|
||||
if(!this.failureFlag){
|
||||
//执行完成
|
||||
this.execComplete(stopWatch,WeatherTaskStatusEnum.COMPLETED.getValue(), "");
|
||||
}catch (Exception e){
|
||||
String taskErrorLog = "任务执行失败,原因:"+e.getMessage();
|
||||
}else {
|
||||
//执行失败
|
||||
this.execComplete(stopWatch,WeatherTaskStatusEnum.FAILURE.getValue(), taskErrorLog);
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -101,10 +124,13 @@ public class WeatherForecastTaskExec extends AbstractWeatherTask {
|
|||
super.serverProperties.getUsername(),super.serverProperties.getPassword());
|
||||
//预测气象数据
|
||||
this.forecast();
|
||||
//如果forecastFailureFlag值为true,说明预测阶段出现错误,不在走适配flexpart和数据入库流程
|
||||
if(!failureFlag){
|
||||
//适配flexpart
|
||||
this.convertGrib(jschRemoteRunner);
|
||||
//检查预测结果数据并存储到数据库
|
||||
this.checkDataAndSaveToDB();
|
||||
}
|
||||
//删除以任务id命名的目录
|
||||
String delPath = "";
|
||||
if(WeatherDataSourceEnum.PANGU.getKey().equals(weatherTask.getPredictionModel())){
|
||||
|
|
@ -149,6 +175,10 @@ public class WeatherForecastTaskExec extends AbstractWeatherTask {
|
|||
.bodyToFlux(String.class)
|
||||
.doOnNext(log->{
|
||||
ProgressQueue.getInstance().offer(new ProgressEvent(this.weatherTask.getId(),log));
|
||||
if((log.startsWith(EXCEPTION_LOG1) || log.startsWith(EXCEPTION_LOG2)) && !this.failureFlag){
|
||||
this.failureFlag = true;
|
||||
this.weatherTaskService.updateTaskStatus(this.weatherTask.getId(),WeatherTaskStatusEnum.FAILURE.getValue());
|
||||
}
|
||||
})
|
||||
.doOnError(e->{
|
||||
throw new RuntimeException(e);
|
||||
|
|
@ -236,11 +266,16 @@ public class WeatherForecastTaskExec extends AbstractWeatherTask {
|
|||
ProgressQueue.getInstance().offer(new ProgressEvent(this.weatherTask.getId(),log));
|
||||
return;
|
||||
}
|
||||
//把预测好的及格式化后的气象文件移动到最终目录
|
||||
|
||||
String formatFilesFinalPath = this.getFormatFilesFinalStoragePath();
|
||||
String sourceFilesFinalPath = this.getSourceFilesFinalStoragePath();
|
||||
FileUtil.move(new File(gribCopyPath),new File(sourceFilesFinalPath),true);
|
||||
FileUtil.move(new File(flexpartFormatPath),new File(formatFilesFinalPath),true);
|
||||
try{
|
||||
//把预测好的及格式化后的气象文件移动到最终目录
|
||||
this.moveAll(gribCopyPath,sourceFilesFinalPath);
|
||||
this.moveAll(flexpartFormatPath,formatFilesFinalPath);
|
||||
}catch (Exception e){
|
||||
throw new RuntimeException("文件移动出现错误",e);
|
||||
}
|
||||
//处理文件入库
|
||||
List<WeatherData> dataList = new ArrayList<>();
|
||||
for(File sourceFile : sourceFiles){
|
||||
|
|
@ -252,8 +287,9 @@ public class WeatherForecastTaskExec extends AbstractWeatherTask {
|
|||
weatherData.setDataStartTime(Grib2TimeReader.readValidTime(sourceFilesFinalPath+File.separator+sourceFile.getName()));
|
||||
weatherData.setDataSource(weatherTask.getPredictionModel());
|
||||
weatherData.setFilePath(sourceFilesFinalPath+File.separator+sourceFile.getName());
|
||||
weatherData.setFormatFilePath(formatFilesFinalPath+File.separator+sourceFile.getName());
|
||||
weatherData.setFormatFilePath(formatFilesFinalPath+File.separator+sourceFile.getName().substring(0,sourceFile.getName().lastIndexOf(".")+1)+ WeatherFileSuffixEnum.GRIB2.getValue());
|
||||
weatherData.setTaskId(this.weatherTask.getId());
|
||||
weatherData.setMd5Value(this.getGribFileMD5(sourceFilesFinalPath+File.separator+sourceFile.getName()));
|
||||
dataList.add(weatherData);
|
||||
}catch (Exception e){
|
||||
String logContent = "读取"+sourceFilesFinalPath+File.separator+sourceFile.getName()+"文件时间参数出现错误";
|
||||
|
|
@ -266,6 +302,36 @@ public class WeatherForecastTaskExec extends AbstractWeatherTask {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 移动所有文件到指定目录
|
||||
* @param srcDirPath
|
||||
* @param destDirPath
|
||||
*/
|
||||
private void moveAll(String srcDirPath,String destDirPath) throws IOException {
|
||||
File srcDir = new File(srcDirPath);
|
||||
File destDir = new File(destDirPath);
|
||||
if(!destDir.exists()){
|
||||
FileUtils.forceMkdir(destDir);
|
||||
}
|
||||
File[] files = srcDir.listFiles();
|
||||
if(ArrayUtils.isEmpty(files)){
|
||||
return;
|
||||
}
|
||||
for(File file : files){
|
||||
FileUtils.moveToDirectory(file,destDir,false);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取GRIB文件的MD5唯一值
|
||||
*/
|
||||
private String getGribFileMD5(String filePath) throws IOException {
|
||||
try (FileInputStream fis = new FileInputStream(filePath)) {
|
||||
// 底层自动采用流式读取,内存占用极低
|
||||
return DigestUtils.md5Hex(fis);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取盘古模型请求命令
|
||||
* @return
|
||||
|
|
@ -307,7 +373,7 @@ public class WeatherForecastTaskExec extends AbstractWeatherTask {
|
|||
map.put("lead_time",this.weatherTask.getLeadTime());
|
||||
map.put("class","od");
|
||||
map.put("assets","assets-graphcast");
|
||||
map.put("workdir",systemStorageProperties.getPanguModelExecPath());
|
||||
map.put("workdir",systemStorageProperties.getGraphcastModelExecPath());
|
||||
map.put("split_dir",getGribCopyPath());
|
||||
map.put("path",buildOutputFilePath());
|
||||
|
||||
|
|
@ -347,7 +413,7 @@ public class WeatherForecastTaskExec extends AbstractWeatherTask {
|
|||
if(WeatherDataSourceEnum.PANGU.getKey().equals(weatherTask.getPredictionModel())){
|
||||
path.append(systemStorageProperties.getPanguModelExecPath());
|
||||
}else if(WeatherDataSourceEnum.GRAPHCAST.getKey().equals(weatherTask.getPredictionModel())){
|
||||
path.append(systemStorageProperties.getPanguModelExecPath());
|
||||
path.append(systemStorageProperties.getGraphcastModelExecPath());
|
||||
}
|
||||
path.append(File.separator);
|
||||
path.append(this.weatherTask.getId());
|
||||
|
|
@ -380,9 +446,9 @@ public class WeatherForecastTaskExec extends AbstractWeatherTask {
|
|||
*/
|
||||
private String getSourceFilesFinalStoragePath(){
|
||||
if(WeatherDataSourceEnum.PANGU.getKey().equals(weatherTask.getPredictionModel())){
|
||||
return systemStorageProperties.getPanguModelExecPath()+"/source";
|
||||
return systemStorageProperties.getPanguDataPath()+"/source";
|
||||
}else if(WeatherDataSourceEnum.GRAPHCAST.getKey().equals(weatherTask.getPredictionModel())){
|
||||
return systemStorageProperties.getGraphcastModelExecPath()+"/source";
|
||||
return systemStorageProperties.getGraphcastDataPath()+"/source";
|
||||
}
|
||||
return "";
|
||||
}
|
||||
|
|
@ -393,9 +459,9 @@ public class WeatherForecastTaskExec extends AbstractWeatherTask {
|
|||
*/
|
||||
private String getFormatFilesFinalStoragePath(){
|
||||
if(WeatherDataSourceEnum.PANGU.getKey().equals(weatherTask.getPredictionModel())){
|
||||
return systemStorageProperties.getPanguModelExecPath()+"/format";
|
||||
return systemStorageProperties.getPanguDataPath()+"/format";
|
||||
}else if(WeatherDataSourceEnum.GRAPHCAST.getKey().equals(weatherTask.getPredictionModel())){
|
||||
return systemStorageProperties.getGraphcastModelExecPath()+"/format";
|
||||
return systemStorageProperties.getGraphcastDataPath()+"/format";
|
||||
}
|
||||
return "";
|
||||
}
|
||||
|
|
|
|||
|
|
@ -14,7 +14,6 @@ import org.springframework.web.bind.annotation.GetMapping;
|
|||
import org.springframework.web.bind.annotation.RequestMapping;
|
||||
import org.springframework.web.bind.annotation.RequestParam;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
import java.util.Date;
|
||||
import java.util.List;
|
||||
|
||||
|
|
|
|||
|
|
@ -87,4 +87,20 @@ public class SourceRebuildTaskController{
|
|||
sourceRebuildTaskService.runTask(taskId);
|
||||
return Result.OK();
|
||||
}
|
||||
|
||||
@AutoLog(value = "设置任务置顶")
|
||||
@Operation(summary = "设置任务置顶")
|
||||
@PutMapping("setTaskTop")
|
||||
public Result<?> setTaskTop(@NotNull(message = "任务ID不能为空") Integer taskId){
|
||||
sourceRebuildTaskService.setTaskTop(taskId);
|
||||
return Result.OK();
|
||||
}
|
||||
|
||||
@AutoLog(value = "设置取消任务置顶")
|
||||
@Operation(summary = "设置取消任务置顶")
|
||||
@PutMapping("setCancelTaskTop")
|
||||
public Result<?> setCancelTaskTop(@NotNull(message = "任务ID不能为空") Integer taskId){
|
||||
sourceRebuildTaskService.setCancelTaskTop(taskId);
|
||||
return Result.OK();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -62,4 +62,16 @@ public interface SourceRebuildTaskService extends IService<SourceRebuildTask> {
|
|||
* @return
|
||||
*/
|
||||
List<SourceRebuildTaskLog> getTaskLog(Integer taskId);
|
||||
|
||||
/**
|
||||
* 设置任务置顶
|
||||
* @param taskId
|
||||
*/
|
||||
void setTaskTop(Integer taskId);
|
||||
|
||||
/**
|
||||
* 取消置顶
|
||||
* @param taskId
|
||||
*/
|
||||
void setCancelTaskTop(Integer taskId);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
|
|||
import lombok.RequiredArgsConstructor;
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
import org.jeecg.common.constant.enums.SourceRebuildTaskStatusEnum;
|
||||
import org.jeecg.common.constant.enums.TaskTopEnum;
|
||||
import org.jeecg.common.properties.SourceRebuildProperties;
|
||||
import org.jeecg.common.system.query.PageRequest;
|
||||
import org.jeecg.modules.base.entity.SourceRebuildMonitoringData;
|
||||
|
|
@ -77,8 +78,6 @@ public class SourceRebuildTaskServiceImpl extends ServiceImpl<SourceRebuildTaskM
|
|||
if (Objects.nonNull(checkNameResult)) {
|
||||
throw new RuntimeException("此任务已存在");
|
||||
}
|
||||
sourceRebuildTask.setCreateTime(LocalDateTime.now());
|
||||
sourceRebuildTask.setUpdateTime(LocalDateTime.now());
|
||||
sourceRebuildTask.setTaskStatus(SourceRebuildTaskStatusEnum.NOT_STARTED.getValue());
|
||||
this.baseMapper.insert(sourceRebuildTask);
|
||||
}
|
||||
|
|
@ -132,7 +131,6 @@ public class SourceRebuildTaskServiceImpl extends ServiceImpl<SourceRebuildTaskM
|
|||
checkIdResult.setHalflife(sourceRebuildTask.getHalflife());
|
||||
checkIdResult.setQmin(sourceRebuildTask.getQmin());
|
||||
checkIdResult.setQmax(sourceRebuildTask.getQmax());
|
||||
checkIdResult.setUpdateTime(LocalDateTime.now());
|
||||
this.updateById(checkIdResult);
|
||||
}
|
||||
|
||||
|
|
@ -192,4 +190,40 @@ public class SourceRebuildTaskServiceImpl extends ServiceImpl<SourceRebuildTaskM
|
|||
queryWrapper.select(SourceRebuildTaskLog::getCreateTime,SourceRebuildTaskLog::getLogContent);
|
||||
return sourceRebuildTaskLogMapper.selectList(queryWrapper);
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置任务置顶
|
||||
*
|
||||
* @param taskId
|
||||
*/
|
||||
@Transactional(rollbackFor = RuntimeException.class)
|
||||
@Override
|
||||
public void setTaskTop(Integer taskId) {
|
||||
SourceRebuildTask task = this.baseMapper.selectById(taskId);
|
||||
if (Objects.isNull(task)) {
|
||||
throw new RuntimeException("此任务不存在");
|
||||
}
|
||||
if (TaskTopEnum.NOT_TOP.getValue().equals(task.getTopTask())) {
|
||||
task.setTopTask(TaskTopEnum.TOP.getValue());
|
||||
this.baseMapper.updateById(task);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 取消置顶
|
||||
*
|
||||
* @param taskId
|
||||
*/
|
||||
@Transactional(rollbackFor = RuntimeException.class)
|
||||
@Override
|
||||
public void setCancelTaskTop(Integer taskId) {
|
||||
SourceRebuildTask task = this.baseMapper.selectById(taskId);
|
||||
if (Objects.isNull(task)) {
|
||||
throw new RuntimeException("此任务不存在");
|
||||
}
|
||||
if (TaskTopEnum.TOP.getValue().equals(task.getTopTask())) {
|
||||
task.setTopTask(TaskTopEnum.NOT_TOP.getValue());
|
||||
this.baseMapper.updateById(task);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -40,7 +40,7 @@ public class TaskResultDataServiceImpl implements TaskResultDataService {
|
|||
conn = new RConnection(serverProperties.getHost(),sourceRebuildProperties.getPort());
|
||||
conn.login(sourceRebuildProperties.getUsername(), sourceRebuildProperties.getPassword());
|
||||
} catch (RserveException e) {
|
||||
throw new RuntimeException("Rserve连接失败");
|
||||
throw new RuntimeException("Rserve连接失败",e);
|
||||
}
|
||||
return conn;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -99,8 +99,8 @@ public class TransportTaskController {
|
|||
return Result.OK();
|
||||
}
|
||||
|
||||
@AutoLog(value = "设置任务置顶")
|
||||
@Operation(summary = "设置任务置顶")
|
||||
@AutoLog(value = "设置取消任务置顶")
|
||||
@Operation(summary = "设置取消任务置顶")
|
||||
@PutMapping("setCancelTaskTop")
|
||||
public Result<?> setCancelTaskTop(@NotNull(message = "任务ID不能为空") Integer taskId){
|
||||
transportTaskService.setCancelTaskTop(taskId);
|
||||
|
|
|
|||
|
|
@ -107,7 +107,7 @@ public class TransportTaskServiceImpl extends ServiceImpl<TransportTaskMapper,Tr
|
|||
transportTask.setTaskType(TransportTaskTypeEnum.MANUALLY.getKey());
|
||||
}
|
||||
transportTask.setTimeConsuming(0D);
|
||||
transportTask.setTopTask(TransportTaskTopEnum.NOT_TOP.getValue());
|
||||
transportTask.setTopTask(TaskTopEnum.NOT_TOP.getValue());
|
||||
this.baseMapper.insert(transportTask);
|
||||
if (TransportTaskModeEnum.BACK_FORWARD.getKey().equals(transportTask.getTaskMode()) &&
|
||||
CollUtil.isNotEmpty(transportTask.getBackwardChild())) {
|
||||
|
|
@ -311,7 +311,7 @@ public class TransportTaskServiceImpl extends ServiceImpl<TransportTaskMapper,Tr
|
|||
if(Objects.isNull(transportTask)){
|
||||
throw new RuntimeException("此任务不存在");
|
||||
}
|
||||
transportTask.setTopTask(TransportTaskTopEnum.TOP.getValue());
|
||||
transportTask.setTopTask(TaskTopEnum.TOP.getValue());
|
||||
this.baseMapper.updateById(transportTask);
|
||||
}
|
||||
|
||||
|
|
@ -1067,7 +1067,7 @@ public class TransportTaskServiceImpl extends ServiceImpl<TransportTaskMapper,Tr
|
|||
if(Objects.isNull(transportTask)){
|
||||
throw new RuntimeException("此任务不存在");
|
||||
}
|
||||
transportTask.setTopTask(TransportTaskTopEnum.NOT_TOP.getValue());
|
||||
transportTask.setTopTask(TaskTopEnum.NOT_TOP.getValue());
|
||||
transportTask.setUpdateTime(LocalDateTime.now());
|
||||
this.baseMapper.updateById(transportTask);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -141,12 +141,6 @@ public class WeatherDataController {
|
|||
downloadT1hJob.downloadT1HFile();
|
||||
}
|
||||
|
||||
@GetMapping("handleStaticDataToDB")
|
||||
public Result<?> handleStaticDataToDB(String path,Integer dataSource){
|
||||
weatherDataService.handleStaticDataToDB(path,dataSource);
|
||||
return Result.OK();
|
||||
}
|
||||
|
||||
@AutoLog(value = "关联气象数据")
|
||||
@Operation(summary = "关联气象数据")
|
||||
@PutMapping("linkedData")
|
||||
|
|
|
|||
|
|
@ -2,7 +2,6 @@ package org.jeecg.controller;
|
|||
|
||||
import com.baomidou.mybatisplus.core.metadata.IPage;
|
||||
import io.swagger.v3.oas.annotations.Operation;
|
||||
import jakarta.validation.constraints.NotBlank;
|
||||
import jakarta.validation.constraints.NotNull;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.jeecg.common.api.vo.Result;
|
||||
|
|
@ -86,4 +85,21 @@ public class WeatherTaskController {
|
|||
public Result<?> getTaskLog(@NotNull(message = "预测任务ID不能为空") Integer taskId){
|
||||
return Result.OK(weatherTaskService.getTaskLog(taskId));
|
||||
}
|
||||
|
||||
@AutoLog(value = "设置任务置顶")
|
||||
@Operation(summary = "设置任务置顶")
|
||||
@PutMapping("setTaskTop")
|
||||
public Result<?> setTaskTop(@NotNull(message = "任务ID不能为空") Integer taskId){
|
||||
weatherTaskService.setTaskTop(taskId);
|
||||
return Result.OK();
|
||||
}
|
||||
|
||||
@AutoLog(value = "设置取消任务置顶")
|
||||
@Operation(summary = "设置取消任务置顶")
|
||||
@PutMapping("setCancelTaskTop")
|
||||
public Result<?> setCancelTaskTop(@NotNull(message = "任务ID不能为空") Integer taskId){
|
||||
weatherTaskService.setCancelTaskTop(taskId);
|
||||
return Result.OK();
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -195,24 +195,42 @@ public class DownloadT1hJob {
|
|||
.filter(Files::isRegularFile)
|
||||
.filter(path -> path.toString().toLowerCase().endsWith(WeatherFileSuffixEnum.GRIB2.getValue()))
|
||||
.map(Path::toFile)
|
||||
.map(file -> extractFileInfo(file, baseTime))
|
||||
.map(file -> {
|
||||
try {
|
||||
return extractFileInfo(file, baseTime);
|
||||
} catch (IOException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
})
|
||||
.collect(Collectors.toList());
|
||||
} catch (IOException e) {
|
||||
throw new RuntimeException("读取文件夹失败: " + folderPath, e);
|
||||
}
|
||||
}
|
||||
|
||||
private WeatherData extractFileInfo(File file, String baseTime) {
|
||||
private WeatherData extractFileInfo(File file, String baseTime) throws IOException {
|
||||
String gribFileMD5 = this.getGribFileMD5(file.getAbsolutePath());
|
||||
WeatherData data = new WeatherData();
|
||||
data.setFileName(file.getName());
|
||||
data.setFileExt(getFileExtension(file.getName()));
|
||||
data.setFilePath(file.getAbsolutePath());
|
||||
data.setDataSource(WeatherDataSourceEnum.T1H.getKey());
|
||||
data.setDataStartTime(parseStartTimeFromFileName(file.getName()));
|
||||
data.setMd5Value(gribFileMD5);
|
||||
data.setTimeBatch(baseTime);
|
||||
return data;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取GRIB文件的MD5唯一值
|
||||
*/
|
||||
private String getGribFileMD5(String filePath) throws IOException {
|
||||
try (FileInputStream fis = new FileInputStream(filePath)) {
|
||||
// 底层自动采用流式读取,内存占用极低
|
||||
return DigestUtils.md5Hex(fis);
|
||||
}
|
||||
}
|
||||
|
||||
private LocalDateTime parseStartTimeFromFileName(String fileName) {
|
||||
// 从文件名解析时间 示例:"T1H_20251029_00.nc" 这样的格式
|
||||
try {
|
||||
|
|
|
|||
|
|
@ -34,10 +34,6 @@ public interface WeatherDataService extends IService<WeatherData> {
|
|||
*/
|
||||
void delete(List<Integer> ids);
|
||||
|
||||
/**
|
||||
* 处理静态气象数据入库接口,比上传快
|
||||
*/
|
||||
void handleStaticDataToDB(String path,Integer dataSource);
|
||||
/**
|
||||
* 关联气象数据
|
||||
* @param dataType
|
||||
|
|
|
|||
|
|
@ -61,4 +61,16 @@ public interface WeatherTaskService extends IService<WeatherTask> {
|
|||
* @return
|
||||
*/
|
||||
List<WeatherTaskLog> getTaskLog(Integer taskId);
|
||||
|
||||
/**
|
||||
* 设置任务置顶
|
||||
* @param taskId
|
||||
*/
|
||||
void setTaskTop(Integer taskId);
|
||||
|
||||
/**
|
||||
* 取消置顶
|
||||
* @param taskId
|
||||
*/
|
||||
void setCancelTaskTop(Integer taskId);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -358,48 +358,6 @@ public class WeatherDataServiceImpl extends ServiceImpl<WeatherDataMapper, Weath
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理静态气象数据入库接口,比上传快
|
||||
*/
|
||||
@Transactional(rollbackFor = RuntimeException.class)
|
||||
@Override
|
||||
public void handleStaticDataToDB(String path,Integer dataSource) {
|
||||
File[] files = FileUtil.file(path).listFiles();
|
||||
for(File file : files) {
|
||||
if(file.getName().endsWith(".grib2") || file.getName().endsWith(".grib") || file.getName().endsWith(".grb2")) {
|
||||
InputStream is = null;
|
||||
try{
|
||||
is = new FileInputStream(file);
|
||||
WeatherData weatherData = new WeatherData();
|
||||
weatherData.setFileName(file.getName());
|
||||
weatherData.setFileExt(file.getName().substring(file.getName().lastIndexOf(".")+1));
|
||||
weatherData.setDataSource(dataSource);
|
||||
weatherData.setFilePath(file.getAbsolutePath());
|
||||
//获取文件数据开始日期
|
||||
String reftime = NcUtil.getReftime(file.getAbsolutePath());
|
||||
Grib2TimeReader.readValidTime(file.getAbsolutePath());
|
||||
if(StringUtils.isBlank(reftime)) {
|
||||
throw new JeecgFileUploadException("解析气象文件起始时间数据异常,此文件可能损坏");
|
||||
}
|
||||
Instant instant = Instant.parse(reftime);
|
||||
LocalDateTime utcDateTime = LocalDateTime.ofInstant(instant, ZoneId.of("UTC"));
|
||||
weatherData.setDataStartTime(utcDateTime);
|
||||
this.baseMapper.insert(weatherData);
|
||||
}catch (Exception e){
|
||||
throw new RuntimeException(e.getMessage());
|
||||
}finally {
|
||||
if(is != null){
|
||||
try {
|
||||
is.close();
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 关联气象数据
|
||||
*
|
||||
|
|
@ -433,7 +391,7 @@ public class WeatherDataServiceImpl extends ServiceImpl<WeatherDataMapper, Weath
|
|||
LocalDate localDate = null;
|
||||
if(WeatherDataSourceEnum.NCEP.getKey().equals(dataType)) {
|
||||
String dateTimeStr = file.getName().substring(0, file.getName().indexOf("."));
|
||||
DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMddHHmmss");
|
||||
DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMddHHmm");
|
||||
LocalDateTime dateTime = LocalDateTime.parse(dateTimeStr, formatter);
|
||||
localDate = dateTime.toLocalDate();
|
||||
}else if(WeatherDataSourceEnum.FNL.getKey().equals(dataType)){
|
||||
|
|
@ -460,6 +418,11 @@ public class WeatherDataServiceImpl extends ServiceImpl<WeatherDataMapper, Weath
|
|||
if (CollUtil.isNotEmpty(fileList)) {
|
||||
for (File file : fileList) {
|
||||
try {
|
||||
//获取文件数据开始日期
|
||||
LocalDateTime localDateTime = Grib2TimeReader.readValidTime(file.getAbsolutePath());
|
||||
LocalDate localDate = localDateTime.toLocalDate();
|
||||
//再次验证grib文件的有效数据时间在给定时间内,才关联保存
|
||||
if(!localDate.isBefore(startDate) && !localDate.isAfter(endDate)) {
|
||||
//如果此文件存在则无需再次新增
|
||||
String gribFileMD5 = this.getGribFileMD5(file.getAbsolutePath());
|
||||
LambdaQueryWrapper<WeatherData> queryWrapper = new LambdaQueryWrapper<>();
|
||||
|
|
@ -474,12 +437,11 @@ public class WeatherDataServiceImpl extends ServiceImpl<WeatherDataMapper, Weath
|
|||
weatherData.setFileExt(file.getName().substring(file.getName().lastIndexOf(".")+1));
|
||||
weatherData.setDataSource(dataType);
|
||||
weatherData.setFilePath(file.getAbsolutePath());
|
||||
//获取文件数据开始日期
|
||||
LocalDateTime localDateTime = Grib2TimeReader.readValidTime(file.getAbsolutePath());
|
||||
weatherData.setDataStartTime(localDateTime);
|
||||
weatherData.setMd5Value(gribFileMD5);
|
||||
this.baseMapper.insert(weatherData);
|
||||
|
||||
weatherLinkedDataLogService.create(dataType,file.getAbsolutePath()+"关联成功");
|
||||
}
|
||||
}catch (Exception e){
|
||||
log.error("关联{}气象数据文件出现错误,原因为:",file.getAbsolutePath(),e);
|
||||
weatherLinkedDataLogService.create(dataType,file.getAbsolutePath()+"关联失败,原因为:"+e.getMessage());
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
|
|||
import lombok.RequiredArgsConstructor;
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
import org.apache.logging.log4j.util.Strings;
|
||||
import org.jeecg.common.constant.enums.TaskTopEnum;
|
||||
import org.jeecg.common.constant.enums.WeatherForecastDatasourceEnum;
|
||||
import org.jeecg.common.constant.enums.WeatherTaskStatusEnum;
|
||||
import org.jeecg.common.properties.SystemStorageProperties;
|
||||
|
|
@ -31,6 +32,7 @@ import java.time.LocalDate;
|
|||
import java.time.LocalDateTime;
|
||||
import java.util.List;
|
||||
import java.util.Objects;
|
||||
import java.util.UUID;
|
||||
|
||||
/**
|
||||
* 天气预报预测任务管理
|
||||
|
|
@ -93,26 +95,15 @@ public class WeatherTaskServiceImpl extends ServiceImpl<WeatherTaskMapper, Weath
|
|||
}
|
||||
//手动获取id
|
||||
weatherTask.setTaskStatus(WeatherTaskStatusEnum.NOT_STARTED.getValue());
|
||||
this.save(weatherTask);
|
||||
if (WeatherForecastDatasourceEnum.LOCATION_FILE.getKey().equals(weatherTask.getDataSources())){
|
||||
try {
|
||||
MultipartFile file = weatherTask.getFile();
|
||||
//构造文件名称
|
||||
StringBuilder fileName = new StringBuilder();
|
||||
fileName.append(file.getOriginalFilename().substring(0,file.getOriginalFilename().lastIndexOf(".")));
|
||||
fileName.append("_"+weatherTask.getId());
|
||||
fileName.append(file.getOriginalFilename().substring(file.getOriginalFilename().lastIndexOf(".")));
|
||||
|
||||
String inputFileStoragePath = getInputFileStoragePath(file.getOriginalFilename());
|
||||
//文件保存地址
|
||||
File storageFile = new File(this.systemStorageProperties.getForecastFileTmpPath()+File.separator+fileName);
|
||||
File storageFile = new File(inputFileStoragePath);
|
||||
//如果不存在则创建
|
||||
if(!FileUtil.exist(storageFile.getParent())){
|
||||
FileUtil.mkdir(storageFile.getParent());
|
||||
}else{
|
||||
//如果重名直接删除
|
||||
if (storageFile.exists()){
|
||||
storageFile.delete();
|
||||
}
|
||||
}
|
||||
file.transferTo(storageFile);
|
||||
weatherTask.setInputFile(storageFile.getAbsolutePath());
|
||||
|
|
@ -120,9 +111,25 @@ public class WeatherTaskServiceImpl extends ServiceImpl<WeatherTaskMapper, Weath
|
|||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
this.save(weatherTask);
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理上传的inputfile
|
||||
* @param srcFileName
|
||||
* @return
|
||||
*/
|
||||
private String getInputFileStoragePath(String srcFileName){
|
||||
//构造文件名称
|
||||
StringBuilder fileName = new StringBuilder();
|
||||
fileName.append(systemStorageProperties.getForecastFileTmpPath());
|
||||
fileName.append(File.separator);
|
||||
fileName.append(UUID.randomUUID());
|
||||
fileName.append(srcFileName.substring(srcFileName.lastIndexOf(".")));
|
||||
return fileName.toString();
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取单条任务数据
|
||||
*
|
||||
|
|
@ -172,35 +179,15 @@ public class WeatherTaskServiceImpl extends ServiceImpl<WeatherTaskMapper, Weath
|
|||
if (Objects.nonNull(weatherTask.getFile())){
|
||||
try {
|
||||
MultipartFile file = weatherTask.getFile();
|
||||
//构造文件名称
|
||||
StringBuilder fileName = new StringBuilder();
|
||||
fileName.append(file.getOriginalFilename().substring(0,file.getOriginalFilename().lastIndexOf(".")));
|
||||
fileName.append("_"+weatherTask.getId());
|
||||
fileName.append(file.getOriginalFilename().substring(file.getOriginalFilename().lastIndexOf(".")));
|
||||
|
||||
String inputFileStoragePath = getInputFileStoragePath(file.getOriginalFilename());
|
||||
//文件保存地址
|
||||
File storageFile = new File(this.systemStorageProperties.getForecastFileTmpPath()+File.separator+fileName);
|
||||
File storageFile = new File(inputFileStoragePath);
|
||||
//如果不存在则创建
|
||||
if(!FileUtil.exist(storageFile.getParent())){
|
||||
FileUtil.mkdir(storageFile.getParent());
|
||||
}else{
|
||||
//如果重名直接删除
|
||||
if (storageFile.exists()){
|
||||
storageFile.delete();
|
||||
}else {
|
||||
//如果文件换了,则找到包含此记录id的,先删除再保存
|
||||
List<File> files = FileUtil.loopFiles(this.systemStorageProperties.getForecastFileTmpPath(), new FileFilter() {
|
||||
@Override
|
||||
public boolean accept(File file) {
|
||||
String flag = "_"+weatherTask.getId();
|
||||
return file.getName().contains(flag);
|
||||
}
|
||||
});
|
||||
if (CollUtil.isNotEmpty(files)) {
|
||||
for (File delFile : files) {
|
||||
delFile.delete();
|
||||
}
|
||||
}
|
||||
if(FileUtil.exist(queryResult.getInputFile())){
|
||||
FileUtil.del(queryResult.getInputFile());
|
||||
}
|
||||
}
|
||||
file.transferTo(storageFile);
|
||||
|
|
@ -210,6 +197,9 @@ public class WeatherTaskServiceImpl extends ServiceImpl<WeatherTaskMapper, Weath
|
|||
}
|
||||
}
|
||||
}else{
|
||||
if(StrUtil.isNotBlank(queryResult.getInputFile()) && FileUtil.exist(queryResult.getInputFile())){
|
||||
FileUtil.del(queryResult.getInputFile());
|
||||
}
|
||||
queryResult.setInputFile(Strings.EMPTY);
|
||||
}
|
||||
this.updateById(queryResult);
|
||||
|
|
@ -275,4 +265,38 @@ public class WeatherTaskServiceImpl extends ServiceImpl<WeatherTaskMapper, Weath
|
|||
queryWrapper.orderByAsc(WeatherTaskLog::getCreateTime);
|
||||
return this.weatherTaskLogMapper.selectList(queryWrapper);
|
||||
}
|
||||
|
||||
/**
|
||||
* 设置任务置顶
|
||||
* @param taskId
|
||||
*/
|
||||
@Transactional(rollbackFor = RuntimeException.class)
|
||||
@Override
|
||||
public void setTaskTop(Integer taskId) {
|
||||
WeatherTask weatherTask = this.baseMapper.selectById(taskId);
|
||||
if (Objects.isNull(weatherTask)) {
|
||||
throw new RuntimeException("此任务不存在");
|
||||
}
|
||||
if (TaskTopEnum.NOT_TOP.getValue().equals(weatherTask.getTopTask())) {
|
||||
weatherTask.setTopTask(TaskTopEnum.TOP.getValue());
|
||||
this.baseMapper.updateById(weatherTask);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 取消置顶
|
||||
* @param taskId
|
||||
*/
|
||||
@Transactional(rollbackFor = RuntimeException.class)
|
||||
@Override
|
||||
public void setCancelTaskTop(Integer taskId) {
|
||||
WeatherTask weatherTask = this.baseMapper.selectById(taskId);
|
||||
if (Objects.isNull(weatherTask)) {
|
||||
throw new RuntimeException("此任务不存在");
|
||||
}
|
||||
if (TaskTopEnum.TOP.getValue().equals(weatherTask.getTopTask())) {
|
||||
weatherTask.setTopTask(TaskTopEnum.NOT_TOP.getValue());
|
||||
this.baseMapper.updateById(weatherTask);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -34,4 +34,13 @@
|
|||
<version>${jeecgboot.version}</version>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
<build>
|
||||
<finalName>stas-cloud-consumer</finalName>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-maven-plugin</artifactId>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
</project>
|
||||
|
|
@ -1,5 +1,5 @@
|
|||
server:
|
||||
port: 8020
|
||||
port: 8011
|
||||
|
||||
spring:
|
||||
application:
|
||||
|
|
|
|||
|
|
@ -22,6 +22,7 @@
|
|||
<module>jeecg-large-screen-start</module>
|
||||
<module>jeecg-visual</module>
|
||||
<module>jeecg-weather-start</module>
|
||||
<module>jeecg-consumer-start</module>
|
||||
<module>jeecg-event-start</module>
|
||||
<module>jeecg-sync-start</module>
|
||||
<module>jeecg-data-analyze-start</module>
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user