modify sone codes

04.系统编码/01.xxl-job/xxl-job-executor/src/main/java/com/ping/chuan/ahpmsdp/xxljobexecutor/dao/IDataRepository.java

@@ -14,4 +14,5 @@ import com.ping.chuan.ahpmsdp.xxljobexecutor.model.dao.DealInfo;
  */
 public interface IDataRepository {
     void insert(String keyspace, String tableName, Date initialTime, Date forecastTime, int timeEffect, Map<String, List<List<BigDecimal>>> columns);
+    void insert(String keyspace, String tableName, Date initialTime, Date forecastTime, String station, Map<String, byte[]> columns, int maxCount);
 }

04.系统编码/01.xxl-job/xxl-job-executor/src/main/java/com/ping/chuan/ahpmsdp/xxljobexecutor/dao/impl/DataRepository.java

@@ -38,17 +38,32 @@ public class DataRepository implements IDataRepository {
             return;
         }
 
-        groupInsert(keyspace, tableName, targetColumns, targetValues);
+        groupInsert(keyspace, tableName, targetColumns, targetValues, maxInsertCount, 2);
     }
 
-    private void groupInsert(String keyspace, String tableName, List<String> columns, List<Object> values){
-        int len = CalcUtil.calcClip(columns.size() - 2, maxInsertCount);
+    @Override
+    public void insert(String keyspace, String tableName, Date initialTime, Date forecastTime, String station, Map<String, byte[]> columns, int maxCount) {
+        List<String> targetColumns = prepareRadarColumnNames(columns.keySet());
+        List<Object> targetValues = prepareColumnValues(initialTime, forecastTime, station, columns.values());
+
+        if (targetColumns.size() <= maxCount){
+            Insert builder = QueryBuilder.insertInto(keyspace, tableName)
+                    .values(targetColumns, targetValues);
+            cassandraTemplate.getSession().execute(builder);
+            return;
+        }
+
+        groupInsert(keyspace, tableName, targetColumns, targetValues, maxCount, 3);
+    }
+
+    private void groupInsert(String keyspace, String tableName, List<String> columns, List<Object> values, int maxCount, int limit){
+        int len = CalcUtil.calcClip(columns.size() - limit, maxCount);
         for(int index = 0; index < len; index ++){
-            int startIndex = (index) * maxInsertCount + 2;
-            List<String> targetColumns = columns.stream().limit(2).collect(Collectors.toList());
-            targetColumns.addAll(columns.stream().skip(startIndex).limit(maxInsertCount).collect(Collectors.toList()));
-            List<Object> targetValues = values.stream().limit(2).collect(Collectors.toList());
-            targetValues.addAll(values.stream().skip(startIndex).limit(maxInsertCount).collect(Collectors.toList()));
+            int startIndex = (index) * maxCount + limit;
+            List<String> targetColumns = columns.stream().limit(limit).collect(Collectors.toList());
+            targetColumns.addAll(columns.stream().skip(startIndex).limit(maxCount).collect(Collectors.toList()));
+            List<Object> targetValues = values.stream().limit(limit).collect(Collectors.toList());
+            targetValues.addAll(values.stream().skip(startIndex).limit(maxCount).collect(Collectors.toList()));
             Insert builder = QueryBuilder.insertInto(keyspace, tableName)
                     .values(targetColumns, targetValues);
             cassandraTemplate.getSession().execute(builder);
@@ -65,6 +80,16 @@ public class DataRepository implements IDataRepository {
         return columns;
     }
 
+    private List<String> prepareRadarColumnNames(Set<String> columnSet){
+        List<String> columns = new ArrayList<>();
+        columns.add("initial_time");
+        columns.add("forecast_time");
+        columns.add("station");
+        columns.add("create_time");
+        columns.addAll(columnSet);
+        return columns;
+    }
+
     private List<Object> prepareColumnValues(Date initialTime, Date forecastTime, int timeEffect, Collection<List<List<BigDecimal>>> values){
         List<Object> columns = new ArrayList<>();
         columns.add(initialTime);
@@ -74,4 +99,14 @@ public class DataRepository implements IDataRepository {
         columns.addAll(values);
         return columns;
     }
+
+    private List<Object> prepareColumnValues(Date initialTime, Date forecastTime, String station, Collection<byte[]> values){
+        List<Object> columns = new ArrayList<>();
+        columns.add(initialTime);
+        columns.add(forecastTime);
+        columns.add(station);
+        columns.add(new Date());
+        columns.addAll(values);
+        return columns;
+    }
 }

04.系统编码/01.xxl-job/xxl-job-executor/src/main/java/com/ping/chuan/ahpmsdp/xxljobexecutor/model/domain/RadarBlock.java

@@ -0,0 +1,24 @@
+package com.ping.chuan.ahpmsdp.xxljobexecutor.model.domain;
+
+import java.util.List;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import com.xiaowuler.radar.core.weather.domain.CutConfiguration;
+import com.xiaowuler.radar.core.weather.domain.MomentBlock;
+import com.xiaowuler.radar.core.weather.domain.Radial;
+import com.xiaowuler.radar.core.weather.domain.RadialHeader;
+
+/**
+ * @describe: leida
+ * @author: xiaowuler
+ * @createTime: 2022-03-18 16:57
+ */
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class RadarBlock {
+    private CutConfiguration cutConfiguration;
+    private List<Radial> radials;
+}

04.系统编码/01.xxl-job/xxl-job-executor/src/main/java/com/ping/chuan/ahpmsdp/xxljobexecutor/service/impl/RadarService.java

@@ -1,14 +1,28 @@
 package com.ping.chuan.ahpmsdp.xxljobexecutor.service.impl;
 
+import com.ping.chuan.ahpmsdp.xxljobexecutor.dao.IDataRepository;
+import com.ping.chuan.ahpmsdp.xxljobexecutor.dao.IDealInfoRepository;
+import com.ping.chuan.ahpmsdp.xxljobexecutor.model.constant.StateConstant;
+import com.ping.chuan.ahpmsdp.xxljobexecutor.model.constant.TimeConstant;
 import com.ping.chuan.ahpmsdp.xxljobexecutor.model.dao.DealInfo;
 import com.ping.chuan.ahpmsdp.xxljobexecutor.model.dao.MetaInfo;
 import com.ping.chuan.ahpmsdp.xxljobexecutor.model.domain.Coordinate;
+import com.ping.chuan.ahpmsdp.xxljobexecutor.model.domain.RadarBlock;
 import com.ping.chuan.ahpmsdp.xxljobexecutor.service.IRadarService;
+import com.ping.chuan.ahpmsdp.xxljobexecutor.util.GZipUtils;
+
 import org.springframework.stereotype.Service;
 
-import java.util.Map;
+import java.io.IOException;
+import java.text.SimpleDateFormat;
+import java.util.*;
+import java.util.stream.Collectors;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.xiaowuler.radar.core.exceptions.RadarReadException;
+import com.xiaowuler.radar.core.weather.domain.CommonBlock;
+import com.xiaowuler.radar.core.weather.domain.CutConfiguration;
+import com.xiaowuler.radar.core.weather.domain.Radial;
 import com.xiaowuler.radar.core.weather.resolver.RadarReader;
 
 /**
@@ -16,16 +30,51 @@ import com.xiaowuler.radar.core.weather.resolver.RadarReader;
  */
 @Service
 public class RadarService implements IRadarService {
+
+    private static final ObjectMapper mapper = new ObjectMapper();
+
+    private final IDataRepository dataRepository;
+    private final IDealInfoRepository dealInfoRepository;
+    public RadarService(IDataRepository dataRepository, IDealInfoRepository dealInfoRepository){
+        this.dataRepository = dataRepository;
+        this.dealInfoRepository = dealInfoRepository;
+    }
+
     @Override
     public String read(Map.Entry<DealInfo, Map.Entry<MetaInfo, Coordinate>> entry) {
+        SimpleDateFormat sdf = new SimpleDateFormat(TimeConstant.YYYYMM_STR);
+        String keyspace = "data_%s".formatted(sdf.format( entry.getKey().getInitialTime()));
+        String error = null;
         try {
             RadarReader radarReader = new RadarReader();
             radarReader.read(entry.getKey().getLocalPath());
 
+            Map<String, byte[]> columns = toColumns(radarReader.getRadarBlock().getCommonBlock());
+            toColumns(columns, radarReader.getRadarBlock().getCommonBlock().getCutConfigurations(), radarReader.getRadarBlock().getRadialBlock().getRadials());
+            dataRepository.insert(keyspace, entry.getValue().getKey().toTableName(), entry.getKey().getInitialTime(), entry.getKey().getInitialTime(), radarReader.getRadarBlock().getCommonBlock().getSiteConfiguration().getSiteCode(), columns, 2);
+            entry.getKey().setState(StateConstant.SUCCESS);
+        } catch (RadarReadException | IOException e) {
+            entry.getKey().setState(StateConstant.FAILED);
+            entry.getKey().setMessage(e.getMessage());
+            error = e.getMessage();
+        }
+        dealInfoRepository.insert(keyspace, entry.getKey());
+        return Objects.isNull(error) ? "文件解析成功" : "文件解析失败，%s".formatted(error);
+    }
+
+    private Map<String, byte[]> toColumns(CommonBlock commonBlock) throws IOException {
+        return new LinkedHashMap<>(){{
+            put("site_configuration", GZipUtils.compress(mapper.writeValueAsBytes(commonBlock.getSiteConfiguration())));
+            put("task_configuration", GZipUtils.compress(mapper.writeValueAsBytes(commonBlock.getTaskConfiguration())));
+            put("generic_header", GZipUtils.compress(mapper.writeValueAsBytes(commonBlock.getGenericHeader())));
+        }};
+    }
 
-        } catch (RadarReadException e) {
-            e.printStackTrace();
+    private void toColumns(Map<String, byte[]> columns, List<CutConfiguration> cutConfigurations, List<Radial> radials) throws IOException {
+        for(int index = 0, len = cutConfigurations.size(); index < len; index++){
+            int num = index + 1;
+            List<Radial> targetRadial = radials.stream().filter(r -> r.getRadialHeader().getElevationNumber() == num).collect(Collectors.toList());
+            columns.put("col_%s".formatted(index), GZipUtils.compress(mapper.writeValueAsBytes(new RadarBlock(cutConfigurations.get(index), targetRadial))));
         }
-        return null;
     }
 }

04.系统编码/03.radar-resolver/radar-core/src/main/java/com/xiaowuler/radar/core/weather/resolver/RadarReader.java

@@ -6,6 +6,7 @@ import java.util.List;
 import java.util.Objects;
 import java.util.stream.Collectors;
 
+import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
 import com.xiaowuler.radar.core.custom.impl.CustomRadarInputStream;
 import com.xiaowuler.radar.core.exceptions.RadarReadException;
@@ -17,6 +18,7 @@ import com.xiaowuler.radar.core.weather.domain.*;
  * @createTime: 2022-03-09 11:09
  */
 @Slf4j
+@Data
 public class RadarReader implements Serializable {
 
     private static final double DEG2RAD = Math.PI / 180;

04.系统编码/03.radar-resolver/radar-core/target/maven-archiver/pom.properties

@@ -1,5 +1,5 @@
 #Generated by Maven
-#Fri Mar 18 15:26:32 CST 2022
+#Fri Mar 18 17:07:47 CST 2022
 groupId=com.xiaowuler
 artifactId=radar-core
 version=1.0.0-SNAPSHOT

04.系统编码/03.radar-resolver/radar-core/target/maven-status/maven-compiler-plugin/compile/default-compile/inputFiles.lst

@@ -1,19 +1,19 @@
-C:\Project\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\domain\RadarBlock.java
-C:\Project\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\domain\SiteConfiguration.java
-C:\Project\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\domain\Radial.java
-C:\Project\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\custom\ICustomInputStream.java
-C:\Project\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\domain\GenericHeader.java
-C:\Project\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\domain\CommonBlock.java
-C:\Project\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\util\ByteUtils.java
-C:\Project\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\domain\RadialBlock.java
-C:\Project\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\domain\MomentHeader.java
-C:\Project\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\resolver\RadarReader.java
-C:\Project\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\enums\RadarSuffixEnum.java
-C:\Project\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\domain\RadialHeader.java
-C:\Project\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\domain\CutConfiguration.java
-C:\Project\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\custom\impl\CustomRadarInputStream.java
-C:\Project\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\domain\MomentData.java
-C:\Project\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\custom\impl\CustomerDataOutputStream.java
-C:\Project\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\domain\TaskConfiguration.java
-C:\Project\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\exceptions\RadarReadException.java
-C:\Project\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\domain\MomentBlock.java
+E:\Projects\2021\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\exceptions\RadarReadException.java
+E:\Projects\2021\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\custom\ICustomInputStream.java
+E:\Projects\2021\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\custom\impl\CustomRadarInputStream.java
+E:\Projects\2021\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\domain\MomentBlock.java
+E:\Projects\2021\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\util\ByteUtils.java
+E:\Projects\2021\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\domain\Radial.java
+E:\Projects\2021\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\domain\SiteConfiguration.java
+E:\Projects\2021\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\domain\CommonBlock.java
+E:\Projects\2021\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\custom\impl\CustomerDataOutputStream.java
+E:\Projects\2021\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\domain\CutConfiguration.java
+E:\Projects\2021\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\enums\RadarSuffixEnum.java
+E:\Projects\2021\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\domain\TaskConfiguration.java
+E:\Projects\2021\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\domain\RadarBlock.java
+E:\Projects\2021\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\domain\RadialBlock.java
+E:\Projects\2021\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\domain\RadialHeader.java
+E:\Projects\2021\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\domain\GenericHeader.java
+E:\Projects\2021\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\resolver\RadarReader.java
+E:\Projects\2021\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\domain\MomentHeader.java
+E:\Projects\2021\AnhuiPublicMeteorologicalServiceDataPlatform\04.系统编码\03.radar-resolver\radar-core\src\main\java\com\xiaowuler\radar\core\weather\domain\MomentData.java