diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/DeviceController.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/DeviceController.java index d5f971a58..0d7cef904 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/DeviceController.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/DeviceController.java @@ -385,8 +385,7 @@ public class DeviceController { @RequestParam(value = "updateSupport", required = false, defaultValue = "false") Boolean updateSupport, @RequestParam("deviceId") Long deviceId) throws Exception { List list = ExcelUtils.read(file, DeviceAttributeImportExcelVO.class); - deviceService.importDeviceAttributeList(list, updateSupport, deviceId); - return success("导入成功"); + return success(deviceService.importDeviceAttributeList(list, updateSupport, deviceId)); } @GetMapping("/devicePointList") diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/devicecontactmodel/DeviceContactModelMapper.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/devicecontactmodel/DeviceContactModelMapper.java index bb61d4f14..a5aa51704 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/devicecontactmodel/DeviceContactModelMapper.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/devicecontactmodel/DeviceContactModelMapper.java @@ -68,8 +68,17 @@ public interface DeviceContactModelMapper extends BaseMapperX selectAttributeNameAndUnitById(@Param("id") Long id); + default List selectListByDeviceIdAndAttributeCodes(Long deviceId, Collection attributeCodes) { + if (deviceId == null || org.springframework.util.CollectionUtils.isEmpty(attributeCodes)) { + return Collections.emptyList(); + } + return selectList(new LambdaQueryWrapperX() + .eq(DeviceContactModelDO::getDeviceId, deviceId) + .in(DeviceContactModelDO::getAttributeCode, attributeCodes)); + } + default List selectListById(DeviceContactModelPageReqVO reqVO){ return selectList(new LambdaQueryWrapperX().eqIfPresent(DeviceContactModelDO::getDeviceId, reqVO.getDeviceId())); }; -} \ No newline at end of file +} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/framework/mqtt/consumer/MqttDataHandler.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/framework/mqtt/consumer/MqttDataHandler.java index 3a1f3295d..389418019 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/framework/mqtt/consumer/MqttDataHandler.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/framework/mqtt/consumer/MqttDataHandler.java @@ -483,7 +483,7 @@ public class MqttDataHandler extends SuperConsumer { attributeCode, processedValue, JSON.toJSONString(pointRulesRespVO)); // 执行匹配失败后的逻辑 - handleMatchedFailureRule(devicePointRulesDO, pointRulesRespVO, processedValue, device, attributeCode); + // handleMatchedFailureRule(devicePointRulesDO, pointRulesRespVO, processedValue, device, attributeCode); } } diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/DeviceService.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/DeviceService.java index 86bcc22c8..7d47c0671 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/DeviceService.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/DeviceService.java @@ -167,5 +167,5 @@ public interface DeviceService { Map historyAnalyse(Long deviceId, String collectionStartTime, String collectionEndTime, List attributeCodes); - void importDeviceAttributeList(List list, Boolean updateSupport, Long deviceId); + String importDeviceAttributeList(List list, Boolean updateSupport, Long deviceId); } diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/DeviceServiceImpl.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/DeviceServiceImpl.java index 5def5f1ae..446faace5 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/DeviceServiceImpl.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/DeviceServiceImpl.java @@ -51,7 +51,6 @@ import cn.iocoder.yudao.module.iot.framework.mqtt.consumer.IMqttservice; import cn.iocoder.yudao.module.iot.util.MapListStatsCalculator; import com.alibaba.fastjson.JSON; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; -import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; import com.baomidou.mybatisplus.core.toolkit.CollectionUtils; import com.baomidou.mybatisplus.core.toolkit.Wrappers; @@ -62,6 +61,7 @@ import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang.math.NumberUtils; import org.apache.commons.lang3.StringUtils; import org.eclipse.paho.client.mqttv3.MqttException; +import org.springframework.dao.DuplicateKeyException; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.stereotype.Service; @@ -1180,36 +1180,156 @@ public class DeviceServiceImpl implements DeviceService { } @Override - public void importDeviceAttributeList(List list, Boolean updateSupport, Long deviceId) { + public String importDeviceAttributeList(List list, Boolean updateSupport, Long deviceId) { // 1.1 参数校验 if (CollUtil.isEmpty(list)) { throw exception(IMPORT_LIST_IS_EMPTY); } - // 1.2 查询主单位数据 - QueryWrapper wrapper = new QueryWrapper<>(); - wrapper.eq("primary_flag", "Y"); - List erpProductUnitDOS = productUnitMapper.selectList(wrapper); - Map> map = erpProductUnitDOS.stream().filter(Objects::nonNull) - .collect(Collectors.groupingBy( - ErpProductUnitDO::getName, - Collectors.toList() - )); - List erpDeviceContactDOs = new ArrayList<>(); - list.stream().forEach(importDeviceAttribute -> { - /* if(importDeviceAttribute.getDataUnit()!=null){ - List erpProductUnitDOS1 = map.get(importDeviceAttribute.getPrimaryName()); - if (CollUtil.isEmpty(erpProductUnitDOS1)) { - importDeviceAttribute.setPrimaryId(null); - } else { - importDeviceAttribute.setPrimaryId(erpProductUnitDOS1.get(0).getId()); - } - }*/ + Set importAttributeCodes = list.stream() + .filter(Objects::nonNull) + .map(DeviceAttributeImportExcelVO::getAttributeCode) + .map(StringUtils::trimToNull) + .filter(Objects::nonNull) + .collect(Collectors.toCollection(LinkedHashSet::new)); + + Set existsAttributeCodes = deviceContactModelMapper + .selectListByDeviceIdAndAttributeCodes(deviceId, importAttributeCodes) + .stream() + .map(DeviceContactModelDO::getAttributeCode) + .filter(Objects::nonNull) + .collect(Collectors.toSet()); + + Set acceptedAttributeCodes = new HashSet<>(existsAttributeCodes); + List insertList = new ArrayList<>(list.size()); + int invalidCount = 0; + int duplicateCount = 0; + + for (DeviceAttributeImportExcelVO importDeviceAttribute : list) { + if (importDeviceAttribute == null) { + invalidCount++; + continue; + } + + String attributeCode = StringUtils.trimToNull(importDeviceAttribute.getAttributeCode()); + String attributeName = StringUtils.trimToNull(importDeviceAttribute.getAttributeName()); + if (attributeCode == null || attributeName == null) { + invalidCount++; + continue; + } + + try { + tdengineService.validateColumnName(attributeCode); + } catch (Exception e) { + invalidCount++; + log.warn("Skip invalid imported device attribute column: deviceId={}, attributeCode={}, reason={}", + deviceId, attributeCode, e.getMessage()); + continue; + } + + if (!acceptedAttributeCodes.add(attributeCode)) { + duplicateCount++; + continue; + } + DeviceContactModelDO deviceAttribute = BeanUtils.toBean(importDeviceAttribute, DeviceContactModelDO.class); deviceAttribute.setDeviceId(deviceId); - erpDeviceContactDOs.add(deviceAttribute); - }); - CollUtil.isNotEmpty(erpDeviceContactDOs);deviceContactModelMapper.insertBatch(erpDeviceContactDOs); + deviceAttribute.setAttributeCode(attributeCode); + deviceAttribute.setAttributeName(attributeName); + insertList.add(deviceAttribute); + } + + ImportInsertResult insertResult = insertDeviceAttributes(insertList, existsAttributeCodes); + duplicateCount += insertResult.getDuplicateCount(); + int tdengineFailureCount = addTdengineColumns(deviceId, insertResult.getInsertedList()); + + return String.format("\u5bfc\u5165\u5b8c\u6210\uff1a\u6210\u529f%d\u6761\uff0c\u8fc7\u6ee4%d\u6761\uff08\u5df2\u5b58\u5728/\u91cd\u590d%d\u6761\uff0c\u4e0d\u7b26\u5408\u89c4\u5219%d\u6761\uff0c\u63d2\u5165\u5931\u8d25%d\u6761\uff09\uff0cTDengine\u5217\u540c\u6b65\u5931\u8d25%d\u6761", + insertResult.getInsertedList().size(), + duplicateCount + invalidCount + insertResult.getFailureCount(), + duplicateCount, + invalidCount, + insertResult.getFailureCount(), + tdengineFailureCount); + } + + private ImportInsertResult insertDeviceAttributes(List insertList, Set preExistingAttributeCodes) { + ImportInsertResult result = new ImportInsertResult(); + if (CollUtil.isEmpty(insertList)) { + return result; + } + + try { + deviceContactModelMapper.insertBatch(insertList, 1000); + result.getInsertedList().addAll(insertList); + return result; + } catch (DuplicateKeyException e) { + log.warn("Batch import device attributes has concurrent duplicates, fallback to row insert: size={}", insertList.size()); + } catch (Exception e) { + log.warn("Batch import device attributes failed, fallback to row insert: size={}, reason={}", insertList.size(), e.getMessage()); + } + + Map currentAttributeMap = deviceContactModelMapper + .selectListByDeviceIdAndAttributeCodes(insertList.get(0).getDeviceId(), + insertList.stream() + .map(DeviceContactModelDO::getAttributeCode) + .collect(Collectors.toSet())) + .stream() + .filter(item -> !preExistingAttributeCodes.contains(item.getAttributeCode())) + .collect(Collectors.toMap(DeviceContactModelDO::getAttributeCode, Function.identity(), (first, second) -> first)); + result.getInsertedList().addAll(currentAttributeMap.values()); + + for (DeviceContactModelDO item : insertList) { + if (currentAttributeMap.containsKey(item.getAttributeCode())) { + continue; + } + try { + deviceContactModelMapper.insert(item); + result.getInsertedList().add(item); + } catch (DuplicateKeyException e) { + result.increaseDuplicateCount(); + log.warn("Skip concurrent duplicate imported device attribute: deviceId={}, attributeCode={}", + item.getDeviceId(), item.getAttributeCode()); + } catch (Exception e) { + result.increaseFailureCount(); + log.warn("Imported device attribute insert failed: deviceId={}, attributeCode={}, reason={}", + item.getDeviceId(), item.getAttributeCode(), e.getMessage()); + } + } + return result; + } + + private int addTdengineColumns(Long deviceId, List insertedList) { + if (CollUtil.isEmpty(insertedList)) { + return 0; + } + + return tdengineService.AddTDDatabaseColumns(deviceId, insertedList); + } + + private static class ImportInsertResult { + private final List insertedList = new ArrayList<>(); + private int duplicateCount; + private int failureCount; + + public List getInsertedList() { + return insertedList; + } + + public int getDuplicateCount() { + return duplicateCount; + } + + public void increaseDuplicateCount() { + duplicateCount++; + } + + public int getFailureCount() { + return failureCount; + } + + public void increaseFailureCount() { + failureCount++; + } } diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/TDengineService.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/TDengineService.java index df6140a3f..e2654f0b6 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/TDengineService.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/TDengineService.java @@ -862,6 +862,57 @@ public class TDengineService { } } + @DS("tdengine") + public int AddTDDatabaseColumns(Long deviceId, List deviceContactModels) { + if (deviceId == null || CollectionUtils.isEmpty(deviceContactModels)) { + return 0; + } + + String tableName = "besure_server.d_" + deviceId; + Set existingColumns = getTDColumnNames(tableName); + int failureCount = 0; + + for (DeviceContactModelDO deviceContactModel : deviceContactModels) { + if (deviceContactModel == null || StrUtil.isBlank(deviceContactModel.getAttributeCode())) { + continue; + } + + String attributeCode = deviceContactModel.getAttributeCode(); + if (existingColumns.contains(attributeCode.toLowerCase())) { + continue; + } + + try { + AddTDDatabaseColumn(deviceId, deviceContactModel); + existingColumns.add(attributeCode.toLowerCase()); + } catch (Exception e) { + if (columnExists(deviceId, attributeCode)) { + existingColumns.add(attributeCode.toLowerCase()); + continue; + } + failureCount++; + log.warn("Batch add TDengine column failed: table={}, column={}, reason={}", + tableName, attributeCode, e.getMessage()); + } + } + return failureCount; + } + + private Set getTDColumnNames(String tableName) { + try { + List> columns = jdbcTemplate.queryForList("DESC " + tableName); + return columns.stream() + .map(column -> column.get("Field")) + .filter(Objects::nonNull) + .map(Object::toString) + .map(String::toLowerCase) + .collect(Collectors.toSet()); + } catch (Exception e) { + log.warn("Query TDengine columns failed: table={}, reason={}", tableName, e.getMessage()); + return new HashSet<>(); + } + } + /**