feat(api): 添加批量插入通信数据功能

- 在CsCommunicateFeignClient中增加insertionList接口方法
- 实现批量插入的fallback处理逻辑
- 在ICsCommunicateService中添加insertionList服务接口
- 完成InfluxdbCsCommunicateServiceImpl中的批量插入实现
- 优化数据处理逻辑，支持运行状态转换和CT/PT计算
- 添加数据排序功能，按项目名、设备名、测点名排序
- 在控制器中增加批量插入数据的REST接口

cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/controller/data/PqsCommunicateController.java

@@ -75,4 +75,13 @@ public class PqsCommunicateController extends BaseController {
         return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
     }
 
+    @OperateInfo(info = LogEnum.BUSINESS_COMMON,operateType = OperateType.ADD)
+    @PostMapping("/insertionList")
+    @ApiOperation("批量插入数据")
+    public HttpResult<String> insertionList(@RequestBody List<PqsCommunicateDto> list) {
+        String methodDescribe = getMethodDescribe("insertionList");
+        pqsCommunicateCvtQuery.insertionList(list);
+        return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
+    }
+
 }

cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/service/ICsCommunicateService.java

@@ -45,4 +45,6 @@ public interface ICsCommunicateService {
     List<PqsCommunicateDto> getRawDataEnd(LineCountEvaluateParam lineParam);
 
     void insertion(PqsCommunicateDto pqsCommunicateDto);
+
+    void insertionList(List<PqsCommunicateDto> list);
 }

cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/service/impl/InfluxdbCsCommunicateServiceImpl.java

@@ -27,6 +27,7 @@ import java.time.ZoneId;
 import java.time.format.DateTimeFormatter;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 
 /**
@@ -321,6 +322,52 @@ public class InfluxdbCsCommunicateServiceImpl implements ICsCommunicateService {
        }
     }
 
+    @Override
+    public void insertionList(List<PqsCommunicateDto> list) {
+        if (CollectionUtils.isEmpty(list)) {
+            return;
+        }
+        Map<String, List<PqsCommunicateDto>> groupByDevId = list.stream().collect(Collectors.groupingBy(PqsCommunicateDto::getDevId));
+        List<String> lineProtocols = new ArrayList<>();
+        groupByDevId.forEach((ndid, dtoList) -> {
+            LambdaQueryWrapper<CsEquipmentDeliveryPO> lambdaQueryWrapper = new LambdaQueryWrapper<>();
+            lambdaQueryWrapper.eq(CsEquipmentDeliveryPO::getNdid, ndid).ne(CsEquipmentDeliveryPO::getRunStatus, 0);
+            CsEquipmentDeliveryPO po = csEquipmentDeliveryMapper.selectOne(lambdaQueryWrapper);
+            if (po == null) {
+                log.warn("未找到ndid={}对应的装置信息，跳过", ndid);
+                return;
+            }
+
+            InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(PqsCommunicate.class);
+            influxQueryWrapper.eq(PqsCommunicate::getDevId, po.getId()).timeDesc().limit(1);
+            List<PqsCommunicate> pqsCommunicates = pqsCommunicateMapper.selectByQueryWrapper(influxQueryWrapper);
+
+            Integer lastType = CollectionUtils.isEmpty(pqsCommunicates) ? null : pqsCommunicates.get(0).getType();
+
+            for (PqsCommunicateDto dto : dtoList) {
+                if (lastType != null && Objects.equals(lastType, dto.getType())) {
+                    continue;
+                }
+                Map<String, String> tags = new HashMap<>();
+                tags.put("dev_id", po.getId());
+                Map<String, Object> fields = new HashMap<>();
+                fields.put("type", dto.getType());
+                fields.put("description", dto.getDescription());
+                long time = LocalDateTime.parse(dto.getTime(), DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN))
+                        .atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
+                Point point = influxDbUtils.pointBuilder("pqs_communicate", time, TimeUnit.MILLISECONDS, tags, fields);
+                BatchPoints batchPoints = BatchPoints.database(influxDbUtils.getDbName()).retentionPolicy("")
+                        .consistency(InfluxDB.ConsistencyLevel.ALL).build();
+                batchPoints.point(point);
+                lineProtocols.add(batchPoints.lineProtocol());
+                lastType = dto.getType();
+            }
+        });
+        if (!lineProtocols.isEmpty()) {
+            influxDbUtils.batchInsert(influxDbUtils.getDbName(), "", InfluxDB.ConsistencyLevel.ALL, TimeUnit.MILLISECONDS, lineProtocols);
+        }
+    }
+
 
     public List<PqsCommunicate> getPqsCommunicateData(LineCountEvaluateParam lineParam) {
         InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(PqsCommunicate.class);