panbaolin/SourceTermAnalysisSystem_java
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

气象与数据同步问题修改

Browse Source
This commit is contained in:
hekaiyu 2025-12-16 19:59:47 +08:00
parent 2ffb1df2f4
commit 0c8373ffd5
18 changed files with 103 additions and 470 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
jeecg-boot-base-core/src/main/java/org/jeecg/common/constant/enums/WeatherVariableNameEnum.java
View File

@ -7,8 +7,8 @@ public enum WeatherVariableNameEnum {
PANGU_T(WeatherDataSourceEnum.PANGU.getKey(), 0, "Temperature_height_above_ground"),
PANGU_P(WeatherDataSourceEnum.PANGU.getKey(), 1, "Pressure_msl"),
PANGU_H(WeatherDataSourceEnum.PANGU.getKey(), 2, "Specific_humidity_isobaric"),
PANGU_P(WeatherDataSourceEnum.PANGU.getKey(), 1, "Pressure_height_above_ground"),
PANGU_H(WeatherDataSourceEnum.PANGU.getKey(), 2, "Specific_humidity_height_above_ground"),
PANGU_U(WeatherDataSourceEnum.PANGU.getKey(), 3, "u-component_of_wind_height_above_ground"),
PANGU_V(WeatherDataSourceEnum.PANGU.getKey(), 4, "v-component_of_wind_height_above_ground"),
CRA40_T(WeatherDataSourceEnum.CRA40.getKey(), 0, "Temperature_isobaric"),
@ -26,11 +26,11 @@ public enum WeatherVariableNameEnum {
FNL_H(WeatherDataSourceEnum.FNL.getKey(), 2, "Relative_humidity_height_above_ground"),
FNL_U(WeatherDataSourceEnum.FNL.getKey(), 3, "u-component_of_wind_height_above_ground"),
FNL_V(WeatherDataSourceEnum.FNL.getKey(), 4, "v-component_of_wind_height_above_ground"),
T1H_T(WeatherDataSourceEnum.T1H.getKey(), 0, "t2mz"),
T1H_P(WeatherDataSourceEnum.T1H.getKey(), 1, "psz"),
T1H_H(WeatherDataSourceEnum.T1H.getKey(), 2, "rh2m"),
T1H_U(WeatherDataSourceEnum.T1H.getKey(), 3, "UGRD10m"),
T1H_V(WeatherDataSourceEnum.T1H.getKey(), 4, "VGRD10m");
T1H_T(WeatherDataSourceEnum.T1H.getKey(), 0, "Temperature_height_above_ground"),
T1H_P(WeatherDataSourceEnum.T1H.getKey(), 1, "Pressure_height_above_ground"),
T1H_H(WeatherDataSourceEnum.T1H.getKey(), 2, "Relative_humidity_height_above_ground"),
T1H_U(WeatherDataSourceEnum.T1H.getKey(), 3, "u-component_of_wind_height_above_ground"),
T1H_V(WeatherDataSourceEnum.T1H.getKey(), 4, "v-component_of_wind_height_above_ground");
private Integer type;

19
jeecg-boot-base-core/src/main/java/org/jeecg/common/properties/T1hDownloadProperties.java
View File

@ -14,6 +14,11 @@ public class T1hDownloadProperties {
*/
private String cron;
/**
* 脚本目录
*/
private String scriptPath;
/**
* python环境路径
*/
@ -25,10 +30,20 @@ public class T1hDownloadProperties {
private String downloadT1hPy;
/**
* 合并T1H的Py脚本路径
* 合并NC的Py脚本路径
*/
private String mergeT1hPy;
/**
* 格式化为grib2的Py脚本路径
*/
private String reformatPy;
/**
* lookup文件名称
*/
private String lookupFileName;
/**
* 未合并T1H的文件目录
*/
@ -47,6 +62,6 @@ public class T1hDownloadProperties {
/**
* 下载小时数
*/
private String hour;
private Integer hour;
}

1
jeecg-boot-base-core/src/main/java/org/jeecg/common/util/ExecutePyUtils.java
View File

@ -25,7 +25,6 @@ public class ExecutePyUtils {
Process process = null;
try {
System.out.println("执行命令: " + String.join(" ", processBuilder.command()));
System.out.println("工作目录: " + processBuilder.directory());
process = processBuilder.start();
// 异步处理输出流

3
jeecg-module-sync/src/main/java/org/jeecg/quartz/entity/QuartzJob.java → jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/entity/QuartzJob.java
View File

@ -1,4 +1,4 @@
package org.jeecg.quartz.entity;
package org.jeecg.modules.base.entity;
import java.io.Serializable;
@ -7,7 +7,6 @@ import org.jeecgframework.poi.excel.annotation.Excel;
import org.springframework.format.annotation.DateTimeFormat;
import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import com.fasterxml.jackson.annotation.JsonFormat;

14
jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/mapper/QuartzJobMapper.java Normal file
View File

@ -0,0 +1,14 @@
package org.jeecg.modules.base.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import org.jeecg.modules.base.entity.QuartzJob;
/**
* @Description: 定时任务在线管理
* @Author: jeecg-boot
* @Date: 2019-01-02
* @Version: V1.0
*/
public interface QuartzJobMapper extends BaseMapper<QuartzJob> {
}

300
jeecg-module-sync/src/main/java/org/jeecg/quartz/controller/QuartzJobController.java
View File

@ -1,300 +0,0 @@
package org.jeecg.quartz.controller;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.extern.slf4j.Slf4j;
import org.jeecg.common.api.vo.Result;
import org.jeecg.common.constant.CommonConstant;
import org.jeecg.common.constant.SymbolConstant;
import org.jeecg.common.system.query.QueryGenerator;
import org.jeecg.common.system.vo.LoginUser;
import org.jeecg.common.util.ImportExcelUtil;
import org.jeecg.common.util.oConvertUtils;
import org.jeecg.config.security.utils.SecureUtil;
import org.jeecg.quartz.entity.QuartzJob;
import org.jeecg.quartz.service.IQuartzJobService;
import org.jeecgframework.poi.excel.ExcelImportUtil;
import org.jeecgframework.poi.excel.def.NormalExcelConstants;
import org.jeecgframework.poi.excel.entity.ExportParams;
import org.jeecgframework.poi.excel.entity.ImportParams;
import org.jeecgframework.poi.excel.view.JeecgEntityExcelView;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;
import org.springframework.web.multipart.MultipartHttpServletRequest;
import org.springframework.web.servlet.ModelAndView;
import jakarta.servlet.http.HttpServletRequest;
import jakarta.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
/**
* @Description: 定时任务在线管理
* @Author: jeecg-boot
* @Date: 2019-01-02
* @Version:V1.0
*/
@RestController
@RequestMapping("/sys/quartzJob")
@Slf4j
@Tag(name = "定时任务接口")
public class QuartzJobController {
@Autowired
private IQuartzJobService quartzJobService;
@Autowired
private Scheduler scheduler;
/**
* 分页列表查询
*
* @param quartzJob
* @param pageNo
* @param pageSize
* @param req
* @return
*/
@RequestMapping(value = "/list", method = RequestMethod.GET)
public Result<?> queryPageList(QuartzJob quartzJob, @RequestParam(name = "pageNo", defaultValue = "1") Integer pageNo,
@RequestParam(name = "pageSize", defaultValue = "10") Integer pageSize, HttpServletRequest req) {
QueryWrapper<QuartzJob> queryWrapper = QueryGenerator.initQueryWrapper(quartzJob, req.getParameterMap());
Page<QuartzJob> page = new Page<QuartzJob>(pageNo, pageSize);
IPage<QuartzJob> pageList = quartzJobService.page(page, queryWrapper);
return Result.ok(pageList);
}
/**
* 添加定时任务
*
* @param quartzJob
* @return
*/
//@RequiresRoles("admin")
@PreAuthorize("@jps.requiresPermissions('system:quartzJob:add')")
@RequestMapping(value = "/add", method = RequestMethod.POST)
public Result<?> add(@RequestBody QuartzJob quartzJob) {
quartzJobService.saveAndScheduleJob(quartzJob);
return Result.ok("创建定时任务成功");
}
/**
* 更新定时任务
*
* @param quartzJob
* @return
*/
//@RequiresRoles("admin")
@PreAuthorize("@jps.requiresPermissions('system:quartzJob:edit')")
@RequestMapping(value = "/edit", method ={RequestMethod.PUT, RequestMethod.POST})
public Result<?> eidt(@RequestBody QuartzJob quartzJob) {
try {
quartzJobService.editAndScheduleJob(quartzJob);
} catch (SchedulerException e) {
log.error(e.getMessage(),e);
return Result.error("更新定时任务失败!");
}
return Result.ok("更新定时任务成功!");
}
/**
* 通过id删除
*
* @param id
* @return
*/
//@RequiresRoles("admin")
@PreAuthorize("@jps.requiresPermissions('system:quartzJob:delete')")
@RequestMapping(value = "/delete", method = RequestMethod.DELETE)
public Result<?> delete(@RequestParam(name = "id", required = true) String id) {
QuartzJob quartzJob = quartzJobService.getById(id);
if (quartzJob == null) {
return Result.error("未找到对应实体");
}
quartzJobService.deleteAndStopJob(quartzJob);
return Result.ok("删除成功!");
}
/**
* 批量删除
*
* @param ids
* @return
*/
//@RequiresRoles("admin")
@PreAuthorize("@jps.requiresPermissions('system:quartzJob:deleteBatch')")
@RequestMapping(value = "/deleteBatch", method = RequestMethod.DELETE)
public Result<?> deleteBatch(@RequestParam(name = "ids", required = true) String ids) {
if (ids == null || "".equals(ids.trim())) {
return Result.error("参数不识别！");
}
for (String id : Arrays.asList(ids.split(SymbolConstant.COMMA))) {
QuartzJob job = quartzJobService.getById(id);
quartzJobService.deleteAndStopJob(job);
}
return Result.ok("删除定时任务成功!");
}
/**
* 暂停定时任务
*
* @param id
* @return
*/
//@RequiresRoles("admin")
@PreAuthorize("@jps.requiresPermissions('system:quartzJob:pause')")
@GetMapping(value = "/pause")
@Operation(summary = "停止定时任务")
public Result<Object> pauseJob(@RequestParam(name = "id") String id) {
QuartzJob job = quartzJobService.getById(id);
if (job == null) {
return Result.error("定时任务不存在！");
}
quartzJobService.pause(job);
return Result.ok("停止定时任务成功");
}
/**
* 启动定时任务
*
* @param id
* @return
*/
//@RequiresRoles("admin")
@PreAuthorize("@jps.requiresPermissions('system:quartzJob:resume')")
@GetMapping(value = "/resume")
@Operation(summary = "启动定时任务")
public Result<Object> resumeJob(@RequestParam(name = "id") String id) {
QuartzJob job = quartzJobService.getById(id);
if (job == null) {
return Result.error("定时任务不存在！");
}
quartzJobService.resumeJob(job);
//scheduler.resumeJob(JobKey.jobKey(job.getJobClassName().trim()));
return Result.ok("启动定时任务成功");
}
/**
* 通过id查询
*
* @param id
* @return
*/
@RequestMapping(value = "/queryById", method = RequestMethod.GET)
public Result<?> queryById(@RequestParam(name = "id", required = true) String id) {
QuartzJob quartzJob = quartzJobService.getById(id);
return Result.ok(quartzJob);
}
/**
* 导出excel
*
* @param request
* @param quartzJob
*/
@RequestMapping(value = "/exportXls")
public ModelAndView exportXls(HttpServletRequest request, QuartzJob quartzJob) {
// Step.1 组装查询条件
QueryWrapper<QuartzJob> queryWrapper = QueryGenerator.initQueryWrapper(quartzJob, request.getParameterMap());
// 过滤选中数据
String selections = request.getParameter("selections");
if (oConvertUtils.isNotEmpty(selections)) {
List<String> selectionList = Arrays.asList(selections.split(","));
queryWrapper.in("id",selectionList);
}
// Step.2 AutoPoi 导出Excel
ModelAndView mv = new ModelAndView(new JeecgEntityExcelView());
List<QuartzJob> pageList = quartzJobService.list(queryWrapper);
// 导出文件名称
mv.addObject(NormalExcelConstants.FILE_NAME, "定时任务列表");
mv.addObject(NormalExcelConstants.CLASS, QuartzJob.class);
//获取当前登录用户
//update-begin---author:wangshuai ---date:20211227 for：[JTC-116]导出人写死了------------
LoginUser user = SecureUtil.currentUser();
mv.addObject(NormalExcelConstants.PARAMS, new ExportParams("定时任务列表数据", "导出人:"+user.getRealname(), "导出信息"));
//update-end---author:wangshuai ---date:20211227 for：[JTC-116]导出人写死了------------
mv.addObject(NormalExcelConstants.DATA_LIST, pageList);
return mv;
}
/**
* 通过excel导入数据
*
* @param request
* @param response
* @return
*/
@RequestMapping(value = "/importExcel", method = RequestMethod.POST)
public Result<?> importExcel(HttpServletRequest request, HttpServletResponse response) throws IOException {
MultipartHttpServletRequest multipartRequest = (MultipartHttpServletRequest) request;
Map<String, MultipartFile> fileMap = multipartRequest.getFileMap();
// 错误信息
List<String> errorMessage = new ArrayList<>();
int successLines = 0, errorLines = 0;
for (Map.Entry<String, MultipartFile> entity : fileMap.entrySet()) {
// 获取上传文件对象
MultipartFile file = entity.getValue();
ImportParams params = new ImportParams();
params.setTitleRows(2);
params.setHeadRows(1);
params.setNeedSave(true);
try {
List<QuartzJob> listQuartzJobs = ExcelImportUtil.importExcel(file.getInputStream(), QuartzJob.class, params);
//add-begin-author:taoyan date:20210909 for:导入定时任务，并不会被启动和调度，需要手动点击启动，才会加入调度任务中 #2986
for(QuartzJob job: listQuartzJobs){
job.setStatus(CommonConstant.STATUS_DISABLE);
}
List<String> list = ImportExcelUtil.importDateSave(listQuartzJobs, IQuartzJobService.class, errorMessage,CommonConstant.SQL_INDEX_UNIQ_JOB_CLASS_NAME);
//add-end-author:taoyan date:20210909 for:导入定时任务，并不会被启动和调度，需要手动点击启动，才会加入调度任务中 #2986
errorLines+=list.size();
successLines+=(listQuartzJobs.size()-errorLines);
} catch (Exception e) {
log.error(e.getMessage(), e);
return Result.error("文件导入失败！");
} finally {
try {
file.getInputStream().close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
return ImportExcelUtil.imporReturnRes(errorLines,successLines,errorMessage);
}
/**
* 立即执行
* @param id
* @return
*/
//@RequiresRoles("admin")
@PreAuthorize("@jps.requiresPermissions('system:quartzJob:execute')")
@GetMapping("/execute")
public Result<?> execute(@RequestParam(name = "id", required = true) String id) {
QuartzJob quartzJob = quartzJobService.getById(id);
if (quartzJob == null) {
return Result.error("未找到对应实体");
}
try {
quartzJobService.execute(quartzJob);
} catch (Exception e) {
//e.printStackTrace();
log.info("定时任务 立即执行失败>>"+e.getMessage());
return Result.error("执行失败!");
}
return Result.ok("执行成功!");
}
}

35
jeecg-module-sync/src/main/java/org/jeecg/quartz/job/AsyncJob.java
View File

@ -1,35 +0,0 @@
package org.jeecg.quartz.job;
import lombok.extern.slf4j.Slf4j;
import org.jeecg.common.util.DateUtils;
import org.quartz.*;
/**
* @Description: 同步定时任务测试
*
* 此处的同步是指 当定时任务的执行时间大于任务的时间间隔时
* 会等待第一个任务执行完成才会走第二个任务
*
*
* @author: taoyan
* @date: 2020年06月19日
*/
@PersistJobDataAfterExecution
@DisallowConcurrentExecution
@Slf4j
public class AsyncJob implements Job {
@Override
public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
log.info(" --- 同步任务调度开始 --- ");
try {
//此处模拟任务执行时间 5秒 任务表达式配置为每秒执行一次：0/1 * * * * ? *
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
//测试发现 每5秒执行一次
log.info(" --- 执行完毕，时间："+DateUtils.now()+"---");
}
}

23
jeecg-module-sync/src/main/java/org/jeecg/quartz/job/SampleJob.java
View File

@ -1,23 +0,0 @@
package org.jeecg.quartz.job;
import org.jeecg.common.util.DateUtils;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import lombok.extern.slf4j.Slf4j;
/**
* 示例不带参定时任务
*
* @Author Scott
*/
@Slf4j
public class SampleJob implements Job {
@Override
public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
log.info(" Job Execution key："+jobExecutionContext.getJobDetail().getKey());
log.info(String.format(" Jeecg-Boot 普通定时任务 SampleJob ! 时间:" + DateUtils.getTimestamp()));
}
}

32
jeecg-module-sync/src/main/java/org/jeecg/quartz/job/SampleParamJob.java
View File

@ -1,32 +0,0 @@
package org.jeecg.quartz.job;
import org.jeecg.common.util.DateUtils;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import lombok.extern.slf4j.Slf4j;
/**
* 示例带参定时任务
*
* @Author Scott
*/
@Slf4j
public class SampleParamJob implements Job {
/**
* 若参数变量名修改 QuartzJobController中也需对应修改
*/
private String parameter;
public void setParameter(String parameter) {
this.parameter = parameter;
}
@Override
public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
log.info(" Job Execution key："+jobExecutionContext.getJobDetail().getKey());
log.info( String.format("welcome %s! Jeecg-Boot 带参数定时任务 SampleParamJob ! 时间:" + DateUtils.now(), this.parameter));
}
}

25
jeecg-module-sync/src/main/java/org/jeecg/quartz/mapper/QuartzJobMapper.java
View File

@ -1,25 +0,0 @@
package org.jeecg.quartz.mapper;
import java.util.List;
import org.apache.ibatis.annotations.Param;
import org.jeecg.quartz.entity.QuartzJob;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
/**
* @Description: 定时任务在线管理
* @Author: jeecg-boot
* @Date: 2019-01-02
* @Version: V1.0
*/
public interface QuartzJobMapper extends BaseMapper<QuartzJob> {
/**
* 根据jobClassName查询
* @param jobClassName 任务类名
* @return
*/
public List<QuartzJob> findByJobClassName(@Param("jobClassName") String jobClassName);
}

9
jeecg-module-sync/src/main/java/org/jeecg/quartz/mapper/xml/QuartzJobMapper.xml
View File

@ -1,9 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="org.jeecg.quartz.mapper.QuartzJobMapper">
<!-- 根据jobClassName查询 -->
<select id="findByJobClassName" resultType="org.jeecg.quartz.entity.QuartzJob">
select * from sys_quartz_job where job_class_name = #{jobClassName}
</select>
</mapper>

11
jeecg-module-sync/src/main/java/org/jeecg/quartz/service/IQuartzJobService.java
View File

@ -1,8 +1,6 @@
package org.jeecg.quartz.service;
import java.util.List;
import org.jeecg.quartz.entity.QuartzJob;
import org.jeecg.modules.base.entity.QuartzJob;
import org.quartz.SchedulerException;
import com.baomidou.mybatisplus.extension.service.IService;
@ -15,13 +13,6 @@ import com.baomidou.mybatisplus.extension.service.IService;
*/
public interface IQuartzJobService extends IService<QuartzJob> {
/**
* 通过类名寻找定时任务
* @param jobClassName 类名
* @return List<QuartzJob>
*/
List<QuartzJob> findByJobClassName(String jobClassName);
/**
* 保存定时任务
* @param quartzJob

10
jeecg-module-sync/src/main/java/org/jeecg/quartz/service/impl/QuartzJobServiceImpl.java
View File

@ -5,8 +5,8 @@ import lombok.extern.slf4j.Slf4j;
import org.jeecg.common.constant.CommonConstant;
import org.jeecg.common.exception.JeecgBootException;
import org.jeecg.common.util.DateUtils;
import org.jeecg.quartz.entity.QuartzJob;
import org.jeecg.quartz.mapper.QuartzJobMapper;
import org.jeecg.modules.base.entity.QuartzJob;
import org.jeecg.modules.base.mapper.QuartzJobMapper;
import org.jeecg.quartz.service.IQuartzJobService;
import org.quartz.*;
import org.springframework.beans.factory.annotation.Autowired;
@ -14,7 +14,6 @@ import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.util.Date;
import java.util.List;
/**
* @Description: 定时任务在线管理
@ -35,11 +34,6 @@ public class QuartzJobServiceImpl extends ServiceImpl<QuartzJobMapper, QuartzJob
*/
private static final String JOB_TEST_GROUP = "test_group";
@Override
public List<QuartzJob> findByJobClassName(String jobClassName) {
return quartzJobMapper.findByJobClassName(jobClassName);
}
/**
* 保存&启动定时任务
*/

2
jeecg-module-sync/src/main/java/org/jeecg/taskConfig/controller/StasTaskConfigController.java
View File

@ -17,8 +17,6 @@ import org.jeecg.dataSource.service.IStasDataSourceService;
import org.jeecg.modules.base.entity.StasDataSource;
import org.jeecg.modules.base.entity.StasSyncStrategy;
import org.jeecg.modules.base.entity.StasTaskConfig;
import org.jeecg.quartz.entity.QuartzJob;
import org.jeecg.quartz.service.IQuartzJobService;
import org.jeecg.stasSyncStrategy.service.IStasSyncStrategyService;
import org.jeecg.taskConfig.service.IStasTaskConfigService;

19
jeecg-module-sync/src/main/java/org/jeecg/taskConfig/job/SyncDataJob.java
View File

@ -211,10 +211,11 @@ public class SyncDataJob implements Job {
// 获取上次同步的位置
String syncOrigin = stasSyncStrategy.getSyncOrigin();
long lastSyncedId = StringUtils.isNotBlank(syncOrigin) ? Long.parseLong(syncOrigin) : idRange.getMinId();
long currentStart = (lastSyncedId > 0 && lastSyncedId > idRange.getMinId()) ?
lastSyncedId : idRange.getMinId();
long currentEnd = currentStart + syncCount;
long currentStart = (lastSyncedId > 0 && lastSyncedId > idRange.getMinId()) ?
lastSyncedId + 1 : idRange.getMinId();
long currentEnd = currentStart + syncCount - 1;
if (currentEnd > idRange.getMaxId()) {
currentEnd = idRange.getMaxId();
StringBuilder whereClause = new StringBuilder();
@ -225,22 +226,28 @@ public class SyncDataJob implements Job {
deleteByEquals(targetConn, stasSyncStrategy, sourceDbType, whereClause.toString());
}
// 确保起始位置不超过结束位置
if (currentStart > currentEnd) {
saveSyncLog(recordId, String.format("%s表已同步到最新，无需继续同步", stasSyncStrategy.getTableName()));
return;
}
String whereClause = stasSyncStrategy.getColumnName() + " BETWEEN " + currentStart + " AND " + currentEnd;
saveSyncLog(recordId,String.format("%s表同步ID范围: %s 至 %s", stasSyncStrategy.getTableName(), currentStart, currentEnd));
saveSyncLog(recordId, String.format("%s表同步ID范围: %s 至 %s", stasSyncStrategy.getTableName(), currentStart, currentEnd));
int rowsSynced = syncDataBatch(sourceConn, targetConn, stasSyncStrategy.getSourceOwner(),
stasSyncStrategy.getTargetOwner(), stasSyncStrategy.getTableName(),
whereClause, sourceDbType, targetDbType);
saveSyncLog(recordId,String.format("%s表已同步 %s 行数据", stasSyncStrategy.getTableName(), rowsSynced));
saveSyncLog(recordId, String.format("%s表已同步 %s 行数据", stasSyncStrategy.getTableName(), rowsSynced));
saveSyncNum(recordId, stasSyncStrategy.getTableName(), rowsSynced);
// 更新同步位置
if (currentEnd > 0) {
stasSyncStrategy.setSyncOrigin(String.valueOf(currentEnd));
stasSyncStrategyMapper.updateById(stasSyncStrategy);
saveSyncLog(recordId,String.format("%s表最终同步位置已更新为: %s", stasSyncStrategy.getTableName(), currentEnd));
saveSyncLog(recordId, String.format("%s表最终同步位置已更新为: %s", stasSyncStrategy.getTableName(), currentEnd));
}
}

2
jeecg-module-sync/src/main/java/org/jeecg/taskConfig/service/impl/StasTaskConfigServiceImpl.java
View File

@ -13,7 +13,7 @@ import org.jeecg.modules.base.entity.StasSyncStrategy;
import org.jeecg.modules.base.mapper.StasSyncStrategyMapper;
import org.jeecg.modules.base.entity.StasTaskConfig;
import org.jeecg.modules.base.mapper.StasTaskConfigMapper;
import org.jeecg.quartz.entity.QuartzJob;
import org.jeecg.modules.base.entity.QuartzJob;
import org.jeecg.quartz.service.IQuartzJobService;
import org.jeecg.taskConfig.service.IStasTaskConfigService;
import org.jeecg.vo.*;

49
jeecg-module-weather/src/main/java/org/jeecg/job/DownloadT1hJob.java
View File

@ -48,8 +48,10 @@ public class DownloadT1hJob {
String baseTime = getBaseTime();
// 第一阶段：下载文件
downloadAllT1hFiles(baseTime);
// 第二阶段：合并文件
mergeT1hFiles(baseTime);
// 第二阶段：合并为NC文件
mergeT1hNCFiles(baseTime);
// 第三阶段：解析为Grib2文件
reformat(baseTime);
// 合并后删除原始文件
Arrays.stream(new File(getFullPath(t1hDownloadProperties.getT1hPath(), baseTime)).listFiles()).filter(File::isFile).forEach(File::delete);
// 更新气象文件信息
@ -79,17 +81,16 @@ public class DownloadT1hJob {
}
/**
* 合并T1H文件
* 合并为NC文件
*/
private void mergeT1hFiles(String baseTime) {
private void mergeT1hNCFiles(String baseTime) {
List<String> variables = T1hFilePrefixEnum.getAllValues();
List<String> forecastTimes = generateForecastTimes();
ProcessBuilder processBuilder = new ProcessBuilder(
t1hDownloadProperties.getPythonPath(),
getPythonScriptPath(t1hDownloadProperties.getMergeT1hPy()),
"--indir", getFullPath(t1hDownloadProperties.getT1hPath(), baseTime),
"--output_dir", getFullPath(systemStorageProperties.getT1h(), baseTime),
"--output_dir", getFullPath(t1hDownloadProperties.getT1hPath(), baseTime),
"--variables", String.join(",", variables),
"--forecast_times", String.join(",", forecastTimes),
"--base_date", baseTime
@ -98,6 +99,25 @@ public class DownloadT1hJob {
ExecutePyUtils.executePythonProcess(processBuilder, "文件合并");
}
/**
* 格式化为GRIB2文件
*/
private void reformat(String baseTime) {
ProcessBuilder processBuilder = new ProcessBuilder(
t1hDownloadProperties.getPythonPath(),
getPythonScriptPath(t1hDownloadProperties.getReformatPy()),
"--ini_dir", getPythonScriptPath(t1hDownloadProperties.getScriptPath()),
"--output_dir", getFullPath(systemStorageProperties.getT1h(), baseTime),
"--cra_dir", getFullPath(t1hDownloadProperties.getT1hPath(), baseTime),
"--start_date", getStartTime(),
"--end_date", getEndTime(),
"--var_table", getPythonScriptPath(t1hDownloadProperties.getLookupFileName())
);
ExecutePyUtils.executePythonProcess(processBuilder, "文件合并");
}
private void saveWeatherData(String baseTime){
//删除一个月前的文件
LocalDateTime oneMonthAgo = LocalDateTime.now().minusMonths(1);
@ -152,7 +172,7 @@ public class DownloadT1hJob {
*/
private List<String> generateForecastTimes() {
List<String> forecastTimes = new ArrayList<>();
int hour = Integer.parseInt(t1hDownloadProperties.getHour());
int hour = t1hDownloadProperties.getHour();
for (int i = 0; i <= hour; i += 6) {
if (i == 0) {
@ -238,4 +258,19 @@ public class DownloadT1hJob {
private String getBaseTime() {
return LocalDateTime.now().minusDays(1).format(DATE_FORMATTER) + "00";
}
/**
* 获取开始时间
*/
private String getStartTime() {
return LocalDateTime.now().minusDays(1).format(DATE_FORMATTER);
}
/**
* 获取结束时间
*/
private String getEndTime() {
Integer day = t1hDownloadProperties.getHour() / 24;
return LocalDateTime.now().plusDays(day - 2).format(DATE_FORMATTER);
}
}

5
jeecg-module-weather/src/main/java/org/jeecg/service/impl/WeatherDataServiceImpl.java
View File

@ -840,6 +840,9 @@ public class WeatherDataServiceImpl extends ServiceImpl<WeatherDataMapper, Weath
List<Double> lonData, List<Double> latData, ValueConverter converter) {
// 统一将采样率降为0.5
double originalStep = getOriginalStep(lonData, latData);
if(originalStep > 1){
throw new JeecgBootException("气象网格精度仅支持小于1°");
}
double targetStep = WeatherStepConstants.DEFAULT_STEP;
if (originalStep < targetStep - 0.001) { // 添加容差避免浮点数比较问题
@ -856,6 +859,8 @@ public class WeatherDataServiceImpl extends ServiceImpl<WeatherDataMapper, Weath
lonData = downsampleLonLat(lonData, ratio);
latData = downsampleLonLat(latData, ratio);
}else {
targetStep = originalStep;
}
List<Double> maxMinLon = getMaxMin(lonData);