504-Software-Development-Group
/
WeatherService
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

后台管理新增一键下载并导入的接口

Browse Source
refactor
shiyi 9 months ago
parent 443a226fd7
commit 17a35e9ab2
13 changed files with 259 additions and 175 deletions
Show Stats Download Patch File Download Diff File

10
weather-service/src/main/java/com/htfp/weather/download/BaseDataDownloader.java
Unescape Escape View File

@ -25,22 +25,22 @@ public abstract class BaseDataDownloader {
/** /**
* 获取所有下载文件的相关信息 * 获取所有下载文件的相关信息
*/ */
public abstract List<FileInfo> getFilesInfo(); public abstract List<DownLoadFileInfo> getFilesInfo();
/** /**
* 单个文件下载 * 单个文件下载
* @param fileInfo: 目标文件的下载信息 * @param downLoadFileInfo: 目标文件的下载信息
* @return 下载成功/失败 * @return 下载成功/失败
* @throws IOException * @throws IOException
*/ */
public abstract FileInfo download(FileInfo fileInfo) throws IOException; public abstract DownLoadFileInfo download(DownLoadFileInfo downLoadFileInfo) throws IOException;
/** /**
* 下载所有目标文件 * 下载所有目标文件
* @param fileInfoList * @param downLoadFileInfoList
* @return * @return
*/ */
public abstract List<FileInfo> downloadAll(List<FileInfo> fileInfoList); public abstract List<DownLoadFileInfo> downloadAll(List<DownLoadFileInfo> downLoadFileInfoList);
} }

4
weather-service/src/main/java/com/htfp/weather/download/FileInfo.java → weather-service/src/main/java/com/htfp/weather/download/DownLoadFileInfo.java
Unescape Escape View File

@ -9,7 +9,7 @@ import lombok.ToString;
@Data @Data
@ToString @ToString
public class FileInfo { public class DownLoadFileInfo {
/**保存文件名*/ /**保存文件名*/
private String fileName; private String fileName;
/**数据起报时间*/ /**数据起报时间*/
@ -26,7 +26,7 @@ public class FileInfo {
private String savePath; private String savePath;
private boolean isDownloadSuccess; private boolean isDownloadSuccess;
public FileInfo() { public DownLoadFileInfo() {
} }
} }

66
weather-service/src/main/java/com/htfp/weather/download/gfs/GfsDownloader.java
Unescape Escape View File

@ -1,7 +1,7 @@
package com.htfp.weather.download.gfs; package com.htfp.weather.download.gfs;
import com.htfp.weather.download.BaseDataDownloader; import com.htfp.weather.download.BaseDataDownloader;
import com.htfp.weather.download.FileInfo; import com.htfp.weather.download.DownLoadFileInfo;
import com.htfp.weather.info.Constant; import com.htfp.weather.info.Constant;
import com.htfp.weather.utils.DateTimeUtils; import com.htfp.weather.utils.DateTimeUtils;
import com.htfp.weather.utils.HttpClientUtils; import com.htfp.weather.utils.HttpClientUtils;
@ -26,7 +26,7 @@ import java.util.concurrent.Future;
/** /**
* @author shiyi * @author shiyi
* @ * @Description GFS数据下载类, 目前初始时间默认为调用初始化方法的时间，不允许设置
*/ */
@Slf4j @Slf4j
@Component("gfsDownloader") @Component("gfsDownloader")
@ -62,33 +62,33 @@ public class GfsDownloader extends BaseDataDownloader {
} }
@Override @Override
public FileInfo download(FileInfo fileInfo) { public DownLoadFileInfo download(DownLoadFileInfo downLoadFileInfo) {
return download0(fileInfo, 2); return download0(downLoadFileInfo, 2);
} }
private FileInfo download0(FileInfo fileInfo, int retryNum) { private DownLoadFileInfo download0(DownLoadFileInfo downLoadFileInfo, int retryNum) {
String url = fileInfo.getUrl(); String url = downLoadFileInfo.getUrl();
File destDir = new File(fileInfo.getSavePath()); File destDir = new File(downLoadFileInfo.getSavePath());
File fileOut = new File(destDir, fileInfo.getFileName()); File fileOut = new File(destDir, downLoadFileInfo.getFileName());
log.info("[GFS Download] 文件下载中，保存至 {}", fileOut); log.info("[GFS Download] 文件下载中，保存至 {}", fileOut);
try { try {
// DONE: 改用okhttp FIXME 2024/6/8: 如果连接是https，jar包启动下载会报错 javax.net.ssl.SSLException: Received fatal alert: internal_error // DONE: 改用okhttp FIXME 2024/6/8: 如果连接是https，jar包启动下载会报错 javax.net.ssl.SSLException: Received fatal alert: internal_error
HttpClientUtils.downloadFileByUrl(url, fileOut.getPath()); HttpClientUtils.downloadFileByUrl(url, fileOut.getPath());
log.info("[GFS Download] 文件下载成功: {}", fileOut); log.info("[GFS Download] 文件下载成功: {}", fileOut);
fileInfo.setDownloadSuccess(fileValid(fileOut.getAbsolutePath())); downLoadFileInfo.setDownloadSuccess(fileValid(fileOut.getAbsolutePath()));
} catch (Exception e) { } catch (Exception e) {
// DONE 2024/5/24: 文件服务器在外网不稳定，增加重试机制 // DONE 2024/5/24: 文件服务器在外网不稳定，增加重试机制
fileInfo.setDownloadSuccess(false); downLoadFileInfo.setDownloadSuccess(false);
if (retryNum > 0) { if (retryNum > 0) {
log.error("[GFS Download] 文件下载失败，重试中: {}", fileOut); log.error("[GFS Download] 文件下载失败，重试中: {}", fileOut);
return download0(fileInfo, retryNum - 1); return download0(downLoadFileInfo, retryNum - 1);
} else { } else {
e.printStackTrace(); e.printStackTrace();
log.error("[GFS Download] 文件下载失败: {}", fileOut); log.error("[GFS Download] 文件下载失败: {}", fileOut);
} }
} }
return fileInfo; return downLoadFileInfo;
} }
/** 简单校验验证文件完整性，并生成索引文件*/ /** 简单校验验证文件完整性，并生成索引文件*/
@ -101,14 +101,14 @@ public class GfsDownloader extends BaseDataDownloader {
} }
} }
@Override @Override
public List<FileInfo> downloadAll(List<FileInfo> fileInfoList) { public List<DownLoadFileInfo> downloadAll(List<DownLoadFileInfo> downLoadFileInfoList) {
log.info("[GFS Download] 下载任务启动，共 {} 个文件", fileInfoList.size()); log.info("[GFS Download] 下载任务启动，共 {} 个文件", downLoadFileInfoList.size());
List<Future<FileInfo>> futures = new ArrayList<>(); List<Future<DownLoadFileInfo>> futures = new ArrayList<>();
List<FileInfo> finishList = new ArrayList<>(); List<DownLoadFileInfo> finishList = new ArrayList<>();
for (FileInfo fileInfo : fileInfoList) { for (DownLoadFileInfo downLoadFileInfo : downLoadFileInfoList) {
futures.add(executorService.submit(() -> download(fileInfo))); futures.add(executorService.submit(() -> download(downLoadFileInfo)));
} }
for (Future<FileInfo> future : futures) { for (Future<DownLoadFileInfo> future : futures) {
try { try {
finishList.add(future.get()); finishList.add(future.get());
} catch (InterruptedException | ExecutionException e) { } catch (InterruptedException | ExecutionException e) {
@ -120,7 +120,7 @@ public class GfsDownloader extends BaseDataDownloader {
} }
@Override @Override
public List<FileInfo> getFilesInfo() { public List<DownLoadFileInfo> getFilesInfo() {
// 时间选项填充 // 时间选项填充
String lonlatBoxStr = null; String lonlatBoxStr = null;
String levelsStr = null; String levelsStr = null;
@ -145,28 +145,28 @@ public class GfsDownloader extends BaseDataDownloader {
// 分辨率为3h，整数截断；包括目标时刻本身 // 分辨率为3h，整数截断；包括目标时刻本身
final int nFiles = 1 + gfsDataConfig.getDuration() / forecastStep; final int nFiles = 1 + gfsDataConfig.getDuration() / forecastStep;
// 存储文件信息 // 存储文件信息
List<FileInfo> fileInfoList = new ArrayList<>(nFiles); List<DownLoadFileInfo> downLoadFileInfoList = new ArrayList<>(nFiles);
for (int i = 0; i < nFiles; i++) { for (int i = 0; i < nFiles; i++) {
FileInfo fileInfo = new FileInfo(); DownLoadFileInfo downLoadFileInfo = new DownLoadFileInfo();
int forecastHour = i * forecastStep + hourDiff; int forecastHour = i * forecastStep + hourDiff;
fileInfo.setForecastHour(forecastHour); downLoadFileInfo.setForecastHour(forecastHour);
String baseURL = getBaseURL(forecastHour, levelsStr, variablesStr, lonlatBoxStr); String baseURL = getBaseURL(forecastHour, levelsStr, variablesStr, lonlatBoxStr);
fileInfo.setUrl(baseURL); downLoadFileInfo.setUrl(baseURL);
fileInfo.setRefTimeStr(refTimeStr); downLoadFileInfo.setRefTimeStr(refTimeStr);
fileInfo.setForecastUTCTimeStr( downLoadFileInfo.setForecastUTCTimeStr(
this.getRefTime().plusHours(fileInfo.getForecastHour()) this.getRefTime().plusHours(downLoadFileInfo.getForecastHour())
.format(DateTimeFormatter.ofPattern(Constant.UTC_TIME_STRING)) .format(DateTimeFormatter.ofPattern(Constant.UTC_TIME_STRING))
); );
fileInfo.setForecastBJTimeStr( downLoadFileInfo.setForecastBJTimeStr(
DateTimeUtils.getLocalZoneDateTime(this.getRefTime().plusHours(fileInfo.getForecastHour())) DateTimeUtils.getLocalZoneDateTime(this.getRefTime().plusHours(downLoadFileInfo.getForecastHour()))
.format(DateTimeFormatter.ofPattern(Constant.BJT_TIME_STRING)) .format(DateTimeFormatter.ofPattern(Constant.BJT_TIME_STRING))
); );
fileInfo.setFileName(String.format("%s-from-%s+%d.grib2", fileInfo.getForecastBJTimeStr(), fileInfo.getRefTimeStr(), forecastHour)); downLoadFileInfo.setFileName(String.format("%s-from-%s+%d.grib2", downLoadFileInfo.getForecastBJTimeStr(), downLoadFileInfo.getRefTimeStr(), forecastHour));
fileInfo.setSavePath(savePath); downLoadFileInfo.setSavePath(savePath);
fileInfoList.add(fileInfo); downLoadFileInfoList.add(downLoadFileInfo);
} }
return fileInfoList; return downLoadFileInfoList;
} }

4
weather-service/src/main/java/com/htfp/weather/download/gfs/GfsVariableIsobaricEnum.java
Unescape Escape View File

@ -43,8 +43,8 @@ public enum GfsVariableIsobaricEnum {
public static String getGfsVariableName(GfsVariableIsobaricEnum gfsVariableIsobaricEnum) { public static String getGfsVariableName(GfsVariableIsobaricEnum gfsVariableIsobaricEnum) {
return variableName.get(gfsVariableIsobaricEnum.nameInApi); return variableName.get(gfsVariableIsobaricEnum.nameInApi);
} }
public static String getGfsVariableName(String tableVariableName) { public static String getGfsVariableName(String nameInApi) {
return variableName.get(tableVariableName); return variableName.get(nameInApi);
} }
public static String getVariableNameInApi(String nameInFile) { public static String getVariableNameInApi(String nameInFile) {

115
weather-service/src/main/java/com/htfp/weather/griddata/operation/GfsDataImport.java
Unescape Escape View File

@ -5,7 +5,6 @@ import com.aliyun.tablestore.grid.model.GridDataSetMeta;
import com.aliyun.tablestore.grid.model.StoreOptions; import com.aliyun.tablestore.grid.model.StoreOptions;
import com.aliyun.tablestore.grid.model.grid.Grid2D; import com.aliyun.tablestore.grid.model.grid.Grid2D;
import com.htfp.weather.download.FileInfo;
import com.htfp.weather.download.gfs.GfsVariableHeightEnum; import com.htfp.weather.download.gfs.GfsVariableHeightEnum;
import com.htfp.weather.download.gfs.GfsVariableIsobaricEnum; import com.htfp.weather.download.gfs.GfsVariableIsobaricEnum;
import com.htfp.weather.griddata.common.TableConfig; import com.htfp.weather.griddata.common.TableConfig;
@ -33,7 +32,6 @@ import java.time.format.DateTimeFormatter;
import java.util.*; import java.util.*;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.stream.Collectors; import java.util.stream.Collectors;
/** /**
@ -52,7 +50,7 @@ public class GfsDataImport extends BaseTableOperation {
private final ExecutorService executorService = Executors.newFixedThreadPool(5); private final ExecutorService executorService = Executors.newFixedThreadPool(5);
/** /**
* init meta data to table store. * 表格元数据初始化
* *
* @param dataSetID * @param dataSetID
* @param dataType * @param dataType
@ -81,7 +79,7 @@ public class GfsDataImport extends BaseTableOperation {
} }
/** /**
* update meta and set status to DONE when data import finished. * 导入完成后，更新meta状态
* *
* @param meta * @param meta
* @return * @return
@ -92,31 +90,34 @@ public class GfsDataImport extends BaseTableOperation {
return meta; return meta;
} }
/**
* 导入完成后，设置meta状态
*/
public GridDataSetMeta putMeta(GridDataSetMeta meta) throws Exception { public GridDataSetMeta putMeta(GridDataSetMeta meta) throws Exception {
tableStoreGrid.putDataSetMeta(meta); tableStoreGrid.putDataSetMeta(meta);
return meta; return meta;
} }
/** 导入指定起报时间的所有文件*/ /** 导入指定起报时间的所有文件*/
public List<ImportResult> importData(OffsetDateTime refTime) throws Exception { public synchronized List<ImportFileInfo> importData(OffsetDateTime refTime) throws Exception {
importing = true; importing = true;
long start = System.currentTimeMillis(); try {
List<String> fileList = getFiles(refTime); long start = System.currentTimeMillis();
if (CollectionUtils.isEmpty(fileList)) { List<String> fileList = getFiles(refTime);
throw new AppException(ErrorCode.NO_NC_OR_GRIB_FILES); if (CollectionUtils.isEmpty(fileList)) {
} throw new AppException(ErrorCode.NO_NC_OR_GRIB_FILES);
log.info("[tablestore] 数据导入开始, refTime = {}...", refTime); }
// datasetId和起报时间绑定 log.info("[tablestore] 数据导入开始, refTime = {}...", refTime);
String dataSetId = refTime.format(DateTimeFormatter.ofPattern(Constant.DATA_SET_ID_FORMAT_STRING)); // datasetId和起报时间绑定
List<String> fileVariables = getFileVariables(tableConfig.variableList); String dataSetId = refTime.format(DateTimeFormatter.ofPattern(Constant.DATA_SET_ID_FORMAT_STRING));
int[] shape = new int[]{tableConfig.timeSizeMax, tableConfig.levSize, tableConfig.latSize, tableConfig.lonSize}; List<String> fileVariables = getFileVariables(tableConfig.variableList);
GridDataSetMeta meta = initMeta(dataSetId, tableConfig.dataType, fileVariables, shape); int[] shape = new int[]{tableConfig.timeSizeMax, tableConfig.levSize, tableConfig.latSize, tableConfig.lonSize};
List<String> forecastHours = new ArrayList<>(); // 所有有效预报时效坐标记录到数据库属性中 GridDataSetMeta meta = initMeta(dataSetId, tableConfig.dataType, fileVariables, shape);
List<String> forecastHours = new ArrayList<>(); // 所有有效预报时效坐标记录到数据库属性中
// todo 2024/5/13: 待优化，用于数据库的索引，必须连续，因此必须保证前一个时刻成功导入后才能导入下一个时刻，数据量大的时候导入时间较长 // todo 2024/5/13: 待优化，用于数据库的索引，必须连续，因此必须保证前一个时刻成功导入后才能导入下一个时刻，数据量大的时候导入时间较长
List<ImportResult> finishedList; List<ImportFileInfo> finishedList;
// List<Future<ImportResult>> futures = new ArrayList<>(); // List<Future<ImportResult>> futures = new ArrayList<>();
finishedList = new ArrayList<>(); finishedList = new ArrayList<>();
for (int i = 0; i < fileList.size(); i++) { for (int i = 0; i < fileList.size(); i++) {
String file = fileList.get(i); String file = fileList.get(i);
@ -124,37 +125,39 @@ public class GfsDataImport extends BaseTableOperation {
int forecastHour = getForecastHourFromFilename(file); int forecastHour = getForecastHourFromFilename(file);
// 2024/5/13:使用多线程，并使用Future来获取结果，但是实际瓶颈可能是网络带宽而非计算 // 2024/5/13:使用多线程，并使用Future来获取结果，但是实际瓶颈可能是网络带宽而非计算
// futures.add(executorService.submit(() -> importFromNcFile(meta, file, iTime,forecastHour))); // futures.add(executorService.submit(() -> importFromNcFile(meta, file, iTime,forecastHour)));
ImportResult importResult = importFromNcFile(meta, file, iTime, forecastHour); ImportFileInfo importFileInfo = importFromNcFile(meta, file, iTime, forecastHour);
finishedList.add(importResult); finishedList.add(importFileInfo);
if (importResult.isSuccess()) { if (importFileInfo.isSuccess()) {
forecastHours.add(String.valueOf(importResult.getForecastHour())); forecastHours.add(String.valueOf(importFileInfo.getForecastHour()));
} }
} }
// for (Future<ImportResult> future : futures) { // for (Future<ImportResult> future : futures) {
// ImportResult importResult = future.get(); // ImportResult importResult = future.get();
// finishedList.add(importResult); // finishedList.add(importResult);
// forecastHours.add(String.valueOf(importResult.getForecastHour())); // forecastHours.add(String.valueOf(importResult.getForecastHour()));
// if (importResult.isSuccess()) { // if (importResult.isSuccess()) {
// forecastHours.add(String.valueOf(importResult.getForecastHour())); // forecastHours.add(String.valueOf(importResult.getForecastHour()));
// } // }
// } // }
List<ImportResult> failedList = finishedList.stream().filter(result -> !result.isSuccess()).collect(Collectors.toList()); List<ImportFileInfo> failedList = finishedList.stream().filter(result -> !result.isSuccess()).collect(Collectors.toList());
if (!failedList.isEmpty()) { if (!failedList.isEmpty()) {
log.warn("存在气象数据导入数据库失败，失败文件列表： {} ", failedList); log.warn("存在气象数据导入数据库失败，失败文件列表： {} ", failedList);
meta.addAttribute("status", "WRONG"); meta.addAttribute("status", "WRONG");
} else { } else {
meta.addAttribute("status", "DONE"); meta.addAttribute("status", "DONE");
}
long end = System.currentTimeMillis();
log.info("[tablestore] 数据导入完成, 耗时: {} s, forecastHours: {}", (end - start)/1000., forecastHours);
meta.setForecastHours(forecastHours);
meta.addAttribute("reference_time", refTime.toInstant().toEpochMilli());
putMeta(meta);
gfsDataFetcher.lastGridDataSetMeta = meta;
log.info("[tablestore]: 更新最新的数据元信息: {}", meta);
return finishedList;
} finally {
importing = false;
} }
long end = System.currentTimeMillis();
log.info("[tablestore] 数据导入完成, 耗时: {} s, forecastHours: {}", (end - start)/1000., forecastHours);
meta.setForecastHours(forecastHours);
meta.addAttribute("reference_time", refTime.toInstant().toEpochMilli());
putMeta(meta);
gfsDataFetcher.lastGridDataSetMeta = meta;
log.info("[tablestore]: 更新最新的数据元信息: {}", meta);
importing = false;
return finishedList;
} }
/** /**
@ -164,7 +167,7 @@ public class GfsDataImport extends BaseTableOperation {
* @param iTime 文件序号 * @param iTime 文件序号
* @param forecastHour 预报时效（相对于起报时间的小时数） * @param forecastHour 预报时效（相对于起报时间的小时数）
*/ */
public ImportResult importFromNcFile(GridDataSetMeta meta, String file, int iTime, int forecastHour) { public ImportFileInfo importFromNcFile(GridDataSetMeta meta, String file, int iTime, int forecastHour) {
try (NetcdfFile ncFile = NetcdfFiles.open(file)) { try (NetcdfFile ncFile = NetcdfFiles.open(file)) {
GridDataWriter writer = tableStoreGrid.getDataWriter(meta); GridDataWriter writer = tableStoreGrid.getDataWriter(meta);
@ -176,6 +179,7 @@ public class GfsDataImport extends BaseTableOperation {
int ysize = meta.getySize(); int ysize = meta.getySize();
for (String variableName : meta.getVariables()) { for (String variableName : meta.getVariables()) {
// 风速在最后单独处理
if ("Wind_speed_gust_surface".equals(variableName) || "Wind_speed_isobaric".equals(variableName) || "Wind_direction_isobaric".equals(variableName) if ("Wind_speed_gust_surface".equals(variableName) || "Wind_speed_isobaric".equals(variableName) || "Wind_direction_isobaric".equals(variableName)
|| "Wind_speed_height_above_ground".equals(variableName) || "Wind_direction_height_above_ground".equals(variableName)) { || "Wind_speed_height_above_ground".equals(variableName) || "Wind_direction_height_above_ground".equals(variableName)) {
continue; continue;
@ -183,7 +187,7 @@ public class GfsDataImport extends BaseTableOperation {
Variable variable = ncFile.findVariable(variableName); Variable variable = ncFile.findVariable(variableName);
if (variable != null) { if (variable != null) {
int shapeLength = variable.getShape().length; int shapeLength = variable.getShape().length;
if (shapeLength== 4) { if (shapeLength== 4) { // 存在垂直坐标，即变量随高度变化
// 气压坐标和高度坐标保持各自的索引 // 气压坐标和高度坐标保持各自的索引
int zSize = variable.getShape(1); int zSize = variable.getShape(1);
for (int z = 0; z < zSize; z++) { for (int z = 0; z < zSize; z++) {
@ -195,7 +199,7 @@ public class GfsDataImport extends BaseTableOperation {
writer.writeGrid2D(variableName, iTime, z, grid2D); writer.writeGrid2D(variableName, iTime, z, grid2D);
} }
} else if (shapeLength == 3) { } else if (shapeLength == 3) {
// DONE 2024/5/24: 导入不含高度坐标的数据 // DONE 2024/5/24: 导入不含高度坐标的数据，如降水量，阵风风速
Array array = variable.read(new int[]{0, 0, 0}, new int[]{1,xsize, ysize}); Array array = variable.read(new int[]{0, 0, 0}, new int[]{1,xsize, ysize});
transferUnit(variableName, array); transferUnit(variableName, array);
Grid2D grid2D = new Grid2D(array.getDataAsByteBuffer(), variable.getDataType(), Grid2D grid2D = new Grid2D(array.getDataAsByteBuffer(), variable.getDataType(),
@ -210,12 +214,13 @@ public class GfsDataImport extends BaseTableOperation {
importWind(writer, meta, ncFile, iTime); importWind(writer, meta, ncFile, iTime);
} catch (Exception e) { } catch (Exception e) {
log.error("[tablestore] 导入文件数据失败,_t={}: {}", forecastHour, file, e); log.error("[tablestore] 导入文件数据失败,_t={}: {}", forecastHour, file, e);
return new ImportResult(false, file, forecastHour, iTime); return new ImportFileInfo(false, file, forecastHour, iTime);
} }
log.info("[tablestore] 导入文件数据成功,_t={}: {}", forecastHour, file); log.info("[tablestore] 导入文件数据成功,_t={}: {}", forecastHour, file);
return new ImportResult(true, file, forecastHour, iTime); return new ImportFileInfo(true, file, forecastHour, iTime);
} }
/**单位转换*/
private void transferUnit(String variableName, Array array) { private void transferUnit(String variableName, Array array) {
if (GfsVariableIsobaricEnum.TEMP.getNameInFile().equals(variableName)) { if (GfsVariableIsobaricEnum.TEMP.getNameInFile().equals(variableName)) {
MeteoUtils.kelvin2Celsius(array); MeteoUtils.kelvin2Celsius(array);
@ -322,13 +327,13 @@ public class GfsDataImport extends BaseTableOperation {
} }
@Data @Data
public static class ImportResult { public static class ImportFileInfo {
private boolean success; private boolean success;
private String file; private String file;
private int forecastHour; private int forecastHour;
private int iTime; private int iTime;
public ImportResult(boolean success, String file, int forecastHour, int iTime) { public ImportFileInfo(boolean success, String file, int forecastHour, int iTime) {
this.success = success; this.success = success;
this.file = file; this.file = file;
this.forecastHour = forecastHour; this.forecastHour = forecastHour;

31
weather-service/src/main/java/com/htfp/weather/schedule/GridDataProcessor.java
Unescape Escape View File

@ -1,11 +1,10 @@
package com.htfp.weather.schedule; package com.htfp.weather.schedule;
import com.htfp.weather.download.FileInfo; import com.htfp.weather.download.DownLoadFileInfo;
import com.htfp.weather.download.gfs.GfsDataConfig; import com.htfp.weather.download.gfs.GfsDataConfig;
import com.htfp.weather.download.gfs.GfsDownloader; import com.htfp.weather.download.gfs.GfsDownloader;
import com.htfp.weather.griddata.operation.GfsDataFetcher;
import com.htfp.weather.griddata.operation.GfsDataImport; import com.htfp.weather.griddata.operation.GfsDataImport;
import com.htfp.weather.griddata.operation.GfsDataImport.ImportResult; import com.htfp.weather.griddata.operation.GfsDataImport.ImportFileInfo;
import com.htfp.weather.info.Constant; import com.htfp.weather.info.Constant;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FileUtils; import org.apache.commons.io.FileUtils;
@ -60,16 +59,13 @@ public class GridDataProcessor {
clearExpiredData(); clearExpiredData();
} }
public boolean download() { private boolean download() {
gfsDownloader.iniTimeSetting(); gfsDownloader.iniTimeSetting();
try { try {
log.info("[schedule-start] 开始下载气象数据"); log.info("[schedule-start] 开始下载气象数据");
List<FileInfo> fileInfoList = gfsDownloader.getFilesInfo(); List<DownLoadFileInfo> downLoadFileInfoList = gfsDownloader.getFilesInfo();
// if (test){ List<DownLoadFileInfo> finishedList = gfsDownloader.downloadAll(downLoadFileInfoList);
// throw new RuntimeException("测试异常"); List<DownLoadFileInfo> failedList = finishedList.stream().filter(fileInfo -> !fileInfo.isDownloadSuccess()).collect(Collectors.toList());
// }
List<FileInfo> finishedList = gfsDownloader.downloadAll(fileInfoList);
List<FileInfo> failedList = finishedList.stream().filter(fileInfo -> !fileInfo.isDownloadSuccess()).collect(Collectors.toList());
if (!failedList.isEmpty()) { if (!failedList.isEmpty()) {
log.error("[schedule] 下载失败文件列表： {}", failedList); log.error("[schedule] 下载失败文件列表： {}", failedList);
downloadFailedReport(gfsDownloader.getRefTimeStr(), finishedList, failedList); downloadFailedReport(gfsDownloader.getRefTimeStr(), finishedList, failedList);
@ -84,16 +80,13 @@ public class GridDataProcessor {
} }
} }
public boolean importToTableStore(){ private boolean importToTableStore(){
try { try {
OffsetDateTime refTime = gfsDownloader.getRefTime(); OffsetDateTime refTime = gfsDownloader.getRefTime();
log.info("[schedule-start] 开始将数据导入数据库， 起报时间: {}", refTime); log.info("[schedule-start] 开始将数据导入数据库， 起报时间: {}", refTime);
// if (test){ List<ImportFileInfo> importFileInfos = gfsDataImport.importData(refTime);
// throw new RuntimeException("测试异常"); List<ImportFileInfo> finishedList = importFileInfos.stream().filter(result -> result.isSuccess()).collect(Collectors.toList());
// } List<ImportFileInfo> failedList = importFileInfos.stream().filter(result -> !result.isSuccess()).collect(Collectors.toList());
List<ImportResult> importResults = gfsDataImport.importData(refTime);
List<ImportResult> finishedList = importResults.stream().filter(result -> result.isSuccess()).collect(Collectors.toList());
List<ImportResult> failedList = importResults.stream().filter(result -> !result.isSuccess()).collect(Collectors.toList());
if (!failedList.isEmpty()) { if (!failedList.isEmpty()) {
log.error("[schedule-end] 气象数据导入数据库失败, 文件列表： {} ", failedList); log.error("[schedule-end] 气象数据导入数据库失败, 文件列表： {} ", failedList);
importFailedReport(gfsDownloader.getRefTimeStr(), finishedList, failedList); importFailedReport(gfsDownloader.getRefTimeStr(), finishedList, failedList);
@ -132,7 +125,7 @@ public class GridDataProcessor {
log.info("[schedule-end] 过期数据清理结束"); log.info("[schedule-end] 过期数据清理结束");
} }
private void downloadFailedReport(String refTime, List<FileInfo> finishedList, List<FileInfo> failedList) { private void downloadFailedReport(String refTime, List<DownLoadFileInfo> finishedList, List<DownLoadFileInfo> failedList) {
// 邮件 // 邮件
Mail mail = new Mail(); Mail mail = new Mail();
mail.setSubject("【航天飞鹏-气象服务】数据下载失败通知"); mail.setSubject("【航天飞鹏-气象服务】数据下载失败通知");
@ -185,7 +178,7 @@ public class GridDataProcessor {
} }
} }
private void importFailedReport(String refTime, List<ImportResult> finishedList, List<ImportResult> failedList) { private void importFailedReport(String refTime, List<ImportFileInfo> finishedList, List<ImportFileInfo> failedList) {
Mail mail = new Mail(); Mail mail = new Mail();
mail.setSubject("【航天飞鹏-气象服务】数据入库失败通知"); mail.setSubject("【航天飞鹏-气象服务】数据入库失败通知");
mail.setTos(mailReceiver); mail.setTos(mailReceiver);

80
weather-service/src/main/java/com/htfp/weather/web/controller/ConfigController.java
Unescape Escape View File

@ -1,20 +1,23 @@
package com.htfp.weather.web.controller; package com.htfp.weather.web.controller;
import com.htfp.weather.download.FileInfo; import com.htfp.weather.download.DownLoadFileInfo;
import com.htfp.weather.download.gfs.GfsDataConfig; import com.htfp.weather.download.gfs.GfsDataConfig;
import com.htfp.weather.download.gfs.GfsDownloader;
import com.htfp.weather.griddata.common.TableConfig; import com.htfp.weather.griddata.common.TableConfig;
import com.htfp.weather.griddata.operation.GfsDataImport; import com.htfp.weather.griddata.operation.GfsDataImport;
import com.htfp.weather.griddata.operation.GfsDataImport.ImportResult; import com.htfp.weather.griddata.operation.GfsDataImport.ImportFileInfo;
import com.htfp.weather.info.Constant; import com.htfp.weather.info.Constant;
import com.htfp.weather.utils.DateTimeUtils; import com.htfp.weather.utils.DateTimeUtils;
import com.htfp.weather.web.exception.AppException; import com.htfp.weather.web.exception.AppException;
import com.htfp.weather.web.exception.ErrorCode; import com.htfp.weather.web.exception.ErrorCode;
import com.htfp.weather.web.pojo.request.GfsConfigUpdate; import com.htfp.weather.web.pojo.request.GfsConfigUpdate;
import com.htfp.weather.web.pojo.request.TableConfigUpdate; import com.htfp.weather.web.pojo.request.TableConfigUpdate;
import com.htfp.weather.web.pojo.response.DownloadResult;
import com.htfp.weather.web.pojo.response.ImportResult;
import com.htfp.weather.web.pojo.response.Result; import com.htfp.weather.web.pojo.response.Result;
import com.htfp.weather.web.service.FileService; import com.htfp.weather.web.service.FileService;
import com.htfp.weather.web.service.GfsDataServiceImpl;
import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.StringUtils;
import org.jetbrains.annotations.NotNull;
import org.springframework.util.CollectionUtils; import org.springframework.util.CollectionUtils;
import org.springframework.validation.annotation.Validated; import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.*; import org.springframework.web.bind.annotation.*;
@ -23,11 +26,11 @@ import javax.annotation.Resource;
import java.io.File; import java.io.File;
import java.io.IOException; import java.io.IOException;
import java.time.OffsetDateTime; import java.time.OffsetDateTime;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter; import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.stream.Collectors;
/** /**
* @Author : shiyi * @Author : shiyi
@ -43,11 +46,11 @@ public class ConfigController {
@Resource @Resource
GfsDataConfig gfsDataConfig; GfsDataConfig gfsDataConfig;
@Resource @Resource
GfsDownloader gfsDownloader;
@Resource
GfsDataImport gfsDataImport; GfsDataImport gfsDataImport;
@Resource @Resource
FileService fileService; FileService fileService;
@Resource
GfsDataServiceImpl gfsDataService;
private final String SECRET = "htfpweather"; private final String SECRET = "htfpweather";
private void validSecret(String secret) { private void validSecret(String secret) {
@ -121,14 +124,8 @@ public class ConfigController {
validSecret(secret); validSecret(secret);
// List<FileInfo> allCompleted = false; // List<FileInfo> allCompleted = false;
try { try {
gfsDownloader.iniTimeSetting(); DownloadResult data = gfsDataService.downloadNowAllFile();
List<FileInfo> fileInfoList = gfsDownloader.getFilesInfo(); List<DownLoadFileInfo> failedList = data.getFailedList();
List<FileInfo> finishedList = gfsDownloader.downloadAll(fileInfoList);
List<FileInfo> successList = finishedList.stream().filter(FileInfo::isDownloadSuccess).collect(Collectors.toList());
List<FileInfo> failedList = finishedList.stream().filter(fileInfo -> !fileInfo.isDownloadSuccess()).collect(Collectors.toList());
Map<String, List<FileInfo>> data = new HashMap<>();
data.put("successList", successList);
data.put("failedList", failedList);
if (CollectionUtils.isEmpty(failedList)) { if (CollectionUtils.isEmpty(failedList)) {
return Result.success(data); return Result.success(data);
} else { } else {
@ -147,24 +144,8 @@ public class ConfigController {
String secret = params.get("secret"); String secret = params.get("secret");
validSecret(secret); validSecret(secret);
OffsetDateTime offsetDateTime = OffsetDateTime.parse(params.get("time"), Constant.API_TIME_FORMATTER); OffsetDateTime offsetDateTime = OffsetDateTime.parse(params.get("time"), Constant.API_TIME_FORMATTER);
String targetUtcStr = DateTimeUtils.getUTCDateTime(offsetDateTime).format(DateTimeFormatter.ofPattern(Constant.UTC_TIME_STRING)); DownLoadFileInfo downLoadFileInfo = gfsDataService.downloadSingleFile(offsetDateTime);
try { return Result.success(downLoadFileInfo);
gfsDownloader.iniTimeSetting();
List<FileInfo> fileInfoList = gfsDownloader.getFilesInfo(); // TODO 2024/6/12: 可以优化，不用生成24小时所有的文件信息
for (FileInfo fileInfo : fileInfoList) {
if (fileInfo.getForecastUTCTimeStr().equals(targetUtcStr)) {
FileInfo downloadResult = gfsDownloader.download(fileInfo);
if (downloadResult.isDownloadSuccess()) {
return Result.success(downloadResult);
} else {
return Result.error(ErrorCode.DOWNLOAD_ERROR);
}
}
}
return Result.error(ErrorCode.QUERY_TIME_ERROR, ": 请选择当前时间之后24小时范围内的时刻");
} catch (AppException e) {
return Result.error(ErrorCode.DOWNLOAD_START_ERROR, e.getMessage());
}
} }
/** /**
@ -175,17 +156,34 @@ public class ConfigController {
String secret = params.get("secret"); String secret = params.get("secret");
validSecret(secret); validSecret(secret);
OffsetDateTime time = OffsetDateTime.parse(params.get("time"), Constant.API_TIME_FORMATTER); OffsetDateTime time = OffsetDateTime.parse(params.get("time"), Constant.API_TIME_FORMATTER);
return getImportResponse(time);
}
/**
* 下载并导入
*/
@PostMapping("/downloadAndImport")
public Result downloadAndImport(@RequestBody Map<String, String> params) {
Result downloadResponse = downloadAllFiles(params);
if (downloadResponse.isSuccess()) {
DownloadResult data = (DownloadResult)downloadResponse.getData();
String refTimeStr = data.getSuccessList().get(0).getRefTimeStr();
OffsetDateTime time = OffsetDateTime.of(LocalDateTime.parse(refTimeStr, DateTimeFormatter.ofPattern(Constant.UTC_TIME_STRING)), ZoneOffset.UTC);
return getImportResponse(time);
} else {
return downloadResponse;
}
}
@NotNull
private Result getImportResponse(OffsetDateTime time) {
try { try {
List<ImportResult> finishedList = gfsDataImport.importData(time); ImportResult importResult = gfsDataService.importData(time);
List<ImportResult> successList = finishedList.stream().filter(result -> result.isSuccess()).collect(Collectors.toList()); List<ImportFileInfo> failedList = importResult.getFailedList();
List<ImportResult> failedList = finishedList.stream().filter(result -> !result.isSuccess()).collect(Collectors.toList());
Map<String, List<ImportResult>> data = new HashMap<>();
data.put("successList", successList);
data.put("failedList", failedList);
if (CollectionUtils.isEmpty(failedList)) { if (CollectionUtils.isEmpty(failedList)) {
return Result.success(data); return Result.success(importResult);
} else { } else {
return new Result(false, ErrorCode.TABLESTORE_IMPORT_ERROR.getCode(), ErrorCode.TABLESTORE_IMPORT_ERROR.getMsg() + ": 数据导入未全部完成", data); return new Result(false, ErrorCode.TABLESTORE_IMPORT_ERROR.getCode(), ErrorCode.TABLESTORE_IMPORT_ERROR.getMsg() + ": 数据导入未全部完成", importResult);
} }
} catch (AppException e) { } catch (AppException e) {
return Result.error(e.getErrorCode()); return Result.error(e.getErrorCode());

17
weather-service/src/main/java/com/htfp/weather/web/pojo/response/DownloadResult.java
Unescape Escape View File

@ -0,0 +1,17 @@
package com.htfp.weather.web.pojo.response;
import com.htfp.weather.download.DownLoadFileInfo;
import lombok.Data;
import java.util.List;
/**
* @Author : shiyi
* @Date : 2024/7/3 11:37
* @Description :
*/
@Data
public class DownloadResult {
List<DownLoadFileInfo> successList;
List<DownLoadFileInfo> failedList;
}

18
weather-service/src/main/java/com/htfp/weather/web/pojo/response/ImportResult.java
Unescape Escape View File

@ -0,0 +1,18 @@
package com.htfp.weather.web.pojo.response;
import com.htfp.weather.griddata.operation.GfsDataImport;
import com.htfp.weather.griddata.operation.GfsDataImport.ImportFileInfo;
import lombok.Data;
import java.util.List;
/**
* @Author : shiyi
* @Date : 2024/7/3 11:38
* @Description :
*/
@Data
public class ImportResult {
List<ImportFileInfo> successList;
List<ImportFileInfo> failedList;
}

8
weather-service/src/main/java/com/htfp/weather/web/service/FileService.java
Unescape Escape View File

@ -14,7 +14,7 @@ import java.util.List;
/** /**
* @Author : shiyi * @Author : shiyi
* @Date : 2024/6/12 11:25 * @Date : 2024/6/12 11:25
* @Description : 文件系统管理 * @Description : 本地文件管理
*/ */
@Component @Component
public class FileService { public class FileService {
@ -57,13 +57,7 @@ public class FileService {
} }
return subDirList; return subDirList;
} }
public void uploadFile(String filePath){
// 文件上传逻辑
}
public void downloadFile(String filePath){
// 文件下载逻辑
}
public void deleteFile(File file) throws IOException { public void deleteFile(File file) throws IOException {
if (file == null || !file.exists()) { if (file == null || !file.exists()) {

59
weather-service/src/main/java/com/htfp/weather/web/service/GfsDataServiceImpl.java
Unescape Escape View File

@ -4,12 +4,15 @@ import com.aliyun.tablestore.grid.consts.AttributionEnum;
import com.aliyun.tablestore.grid.model.GridDataSet; import com.aliyun.tablestore.grid.model.GridDataSet;
import com.aliyun.tablestore.grid.model.GridDataSetMeta; import com.aliyun.tablestore.grid.model.GridDataSetMeta;
import com.aliyun.tablestore.grid.model.grid.Grid4D; import com.aliyun.tablestore.grid.model.grid.Grid4D;
import com.htfp.weather.download.DownLoadFileInfo;
import com.htfp.weather.download.gfs.GfsDownloader;
import com.htfp.weather.download.gfs.GfsLevelsEnum; import com.htfp.weather.download.gfs.GfsLevelsEnum;
import com.htfp.weather.download.gfs.GfsVariableHeightEnum; import com.htfp.weather.download.gfs.GfsVariableHeightEnum;
import com.htfp.weather.download.gfs.GfsVariableIsobaricEnum; import com.htfp.weather.download.gfs.GfsVariableIsobaricEnum;
import com.htfp.weather.griddata.common.TableConfig; import com.htfp.weather.griddata.common.TableConfig;
import com.htfp.weather.griddata.common.ValueRange; import com.htfp.weather.griddata.common.ValueRange;
import com.htfp.weather.griddata.operation.GfsDataFetcher; import com.htfp.weather.griddata.operation.GfsDataFetcher;
import com.htfp.weather.griddata.operation.GfsDataImport;
import com.htfp.weather.info.Constant; import com.htfp.weather.info.Constant;
import com.htfp.weather.utils.DateTimeUtils; import com.htfp.weather.utils.DateTimeUtils;
import com.htfp.weather.utils.MeteoUtils; import com.htfp.weather.utils.MeteoUtils;
@ -27,7 +30,9 @@ import java.lang.reflect.Method;
import java.time.Instant; import java.time.Instant;
import java.time.OffsetDateTime; import java.time.OffsetDateTime;
import java.time.ZoneOffset; import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.*; import java.util.*;
import java.util.stream.Collectors;
/** /**
* @Author : shiyi * @Author : shiyi
@ -41,9 +46,60 @@ public class GfsDataServiceImpl implements IDataService {
GfsDataFetcher gfsDataFetcher; GfsDataFetcher gfsDataFetcher;
@Resource @Resource
TableConfig tableConfig; TableConfig tableConfig;
@Resource
GfsDownloader gfsDownloader;
@Resource
GfsDataImport gfsDataImport;
// 下载和导入数据
public DownloadResult downloadNowAllFile() {
gfsDownloader.iniTimeSetting();
List<DownLoadFileInfo> downLoadFileInfoList = gfsDownloader.getFilesInfo();
List<DownLoadFileInfo> finishedList = gfsDownloader.downloadAll(downLoadFileInfoList);
List<DownLoadFileInfo> successList = finishedList.stream().filter(DownLoadFileInfo::isDownloadSuccess).collect(Collectors.toList());
List<DownLoadFileInfo> failedList = finishedList.stream().filter(fileInfo -> !fileInfo.isDownloadSuccess()).collect(Collectors.toList());
DownloadResult downloadResult = new DownloadResult();
downloadResult.setSuccessList(successList);
downloadResult.setFailedList(failedList);
return downloadResult;
}
public DownLoadFileInfo downloadSingleFile(OffsetDateTime offsetDateTime) {
String targetUtcStr = DateTimeUtils.getUTCDateTime(offsetDateTime).format(DateTimeFormatter.ofPattern(Constant.UTC_TIME_STRING));
try {
gfsDownloader.iniTimeSetting();
List<DownLoadFileInfo> downLoadFileInfoList = gfsDownloader.getFilesInfo(); // TODO 2024/6/12: 可以优化，不用生成24小时所有的文件信息
for (DownLoadFileInfo downLoadFileInfo : downLoadFileInfoList) {
if (downLoadFileInfo.getForecastUTCTimeStr().equals(targetUtcStr)) {
DownLoadFileInfo downloadResult = gfsDownloader.download(downLoadFileInfo);
if (downloadResult.isDownloadSuccess()) {
return downloadResult;
} else {
throw new AppException(ErrorCode.DOWNLOAD_ERROR);
}
}
}
throw new AppException(ErrorCode.QUERY_TIME_ERROR, ": 请选择当前时间之后24小时范围内的时刻");
} catch (Exception e) {
throw new AppException(ErrorCode.DOWNLOAD_START_ERROR, e.getMessage());
}
}
public ImportResult importData(OffsetDateTime time) throws Exception {
List<GfsDataImport.ImportFileInfo> finishedList = gfsDataImport.importData(time);
List<GfsDataImport.ImportFileInfo> successList = finishedList.stream().filter(result -> result.isSuccess()).collect(Collectors.toList());
List<GfsDataImport.ImportFileInfo> failedList = finishedList.stream().filter(result -> !result.isSuccess()).collect(Collectors.toList());
ImportResult importResult = new ImportResult();
importResult.setFailedList(failedList);
importResult.setSuccessList(successList);
return importResult;
}
public void downloadAndImportData() {
}
// 从tablestore获取数据
/** /**
* 获取当前时刻的天气状况 * 获取当前时刻的天气状况
* *
@ -447,8 +503,7 @@ public class GfsDataServiceImpl implements IDataService {
return result; return result;
} }
@NotNull private String getVariableNameInApi(GfsLevelsEnum levelFlag, Map.Entry<String, Grid4D> variable) {
private static String getVariableNameInApi(GfsLevelsEnum levelFlag, Map.Entry<String, Grid4D> variable) {
String variableNameInFile = variable.getKey(); String variableNameInFile = variable.getKey();
String variableNameInApi; String variableNameInApi;
if (GfsLevelsEnum.SURFACE.equals(levelFlag)) { if (GfsLevelsEnum.SURFACE.equals(levelFlag)) {

8
weather-service/src/main/java/com/htfp/weather/web/service/surfaceapi/CmaServiceImpl.java
Unescape Escape View File

@ -56,8 +56,12 @@ public class CmaServiceImpl implements ISurfaceDataService{
return getSurfaceWarningByCounty(countyId); return getSurfaceWarningByCounty(countyId);
} }
public List<SurfaceWeatherWarning> getSurfaceWarningByCounty(String countyCode) throws Exception { public List<SurfaceWeatherWarning> getSurfaceWarningByCounty(String countyCode) {
return warningCache.get(countyCode); if (warningCache.containsKey(countyCode)) {
return warningCache.get(countyCode);
} else {
return new ArrayList<>();
}
// else { // else {
// // 如果缓存中没有，则先查一遍 // // 如果缓存中没有，则先查一遍
// updateCmaWarning(countyCode); // updateCmaWarning(countyCode);

14
weather-service/src/test/java/com/htfp/weather/download/GfsDownloaderTest.java
Unescape Escape View File

@ -28,24 +28,24 @@ class GfsDownloaderTest {
@Test @Test
void download() { void download() {
gfsDownloader.iniTimeSetting(); gfsDownloader.iniTimeSetting();
List<FileInfo> fileInfoList = gfsDownloader.getFilesInfo(); List<DownLoadFileInfo> downLoadFileInfoList = gfsDownloader.getFilesInfo();
gfsDownloader.download(fileInfoList.get(0)); gfsDownloader.download(downLoadFileInfoList.get(0));
} }
@Test @Test
void downloadAll() { void downloadAll() {
gfsDownloader.iniTimeSetting(); gfsDownloader.iniTimeSetting();
List<FileInfo> fileInfoList = gfsDownloader.getFilesInfo(); List<DownLoadFileInfo> downLoadFileInfoList = gfsDownloader.getFilesInfo();
gfsDownloader.downloadAll(fileInfoList); gfsDownloader.downloadAll(downLoadFileInfoList);
} }
@Test @Test
void getFilesInfo() { void getFilesInfo() {
gfsDownloader.iniTimeSetting(); gfsDownloader.iniTimeSetting();
List<FileInfo> fileInfoList = gfsDownloader.getFilesInfo(); List<DownLoadFileInfo> downLoadFileInfoList = gfsDownloader.getFilesInfo();
for (FileInfo fileInfo :fileInfoList) { for (DownLoadFileInfo downLoadFileInfo : downLoadFileInfoList) {
System.out.println(fileInfo); System.out.println(downLoadFileInfo);
} }
} }

Write Preview
Loading…
Cancel
Save