|
|
@@ -4,6 +4,7 @@ import com.baomidou.dynamic.datasource.annotation.DS;
|
|
|
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
|
|
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
|
|
|
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
|
|
|
+import com.usky.common.core.bean.ApiResult;
|
|
|
import com.usky.common.core.exception.BusinessException;
|
|
|
import com.usky.common.mybatis.core.AbstractCrudService;
|
|
|
import com.usky.demo.domain.*;
|
|
|
@@ -481,36 +482,49 @@ public class QueryTdengineDataServiceImpl extends AbstractCrudService<QueryTdeng
|
|
|
}
|
|
|
|
|
|
@Override
|
|
|
- public DeviceCsvImportResultVO importDeviceCsv(String deviceUuid, MultipartFile file) {
|
|
|
+ public ApiResult<DeviceCsvImportResultVO> importDeviceCsv(String deviceUuid, MultipartFile file) {
|
|
|
if (StringUtils.isBlank(deviceUuid)) {
|
|
|
- throw new BusinessException("设备UUID不能为空");
|
|
|
+ log.error("设备UUID不能为空");
|
|
|
+ return ApiResult.error("设备UUID不能为空");
|
|
|
}
|
|
|
if (!DEVICE_UUID_PATTERN.matcher(deviceUuid.trim()).matches()) {
|
|
|
- throw new BusinessException("设备UUID格式无效");
|
|
|
+ log.error("设备UUID格式无效, deviceUuid=" + deviceUuid);
|
|
|
+ return ApiResult.error("设备UUID格式无效");
|
|
|
}
|
|
|
if (file == null || file.isEmpty()) {
|
|
|
- throw new BusinessException("导入文件不能为空");
|
|
|
+ log.error("导入文件不能为空, deviceUuid=" + deviceUuid);
|
|
|
+ return ApiResult.error("导入文件不能为空");
|
|
|
}
|
|
|
String originalFilename = file.getOriginalFilename();
|
|
|
if (StringUtils.isBlank(originalFilename) || !originalFilename.toLowerCase(Locale.ROOT).endsWith(".csv")) {
|
|
|
- throw new BusinessException("仅支持 CSV 文件");
|
|
|
+ log.error("仅支持 CSV 文件, deviceUuid=" + deviceUuid + ", filename=" + originalFilename);
|
|
|
+ return ApiResult.error("仅支持 CSV 文件");
|
|
|
}
|
|
|
|
|
|
String normalizedUuid = deviceUuid.trim();
|
|
|
String tableName = buildDeviceSubTableName(normalizedUuid);
|
|
|
try {
|
|
|
List<TsdbCsvRow> rows = parseCsvRows(file);
|
|
|
- batchInsertRows(tableName, rows);
|
|
|
+ if (rows == null) {
|
|
|
+ return ApiResult.error("CSV 解析失败");
|
|
|
+ }
|
|
|
+ int importedRows = batchInsertRows(tableName, rows);
|
|
|
+ if (importedRows == 0) {
|
|
|
+ log.error("CSV 无有效可导入数据行, deviceUuid=" + normalizedUuid + ", tableName=" + tableName);
|
|
|
+ return ApiResult.error("CSV 无有效可导入数据行");
|
|
|
+ }
|
|
|
|
|
|
DeviceCsvImportResultVO result = new DeviceCsvImportResultVO();
|
|
|
result.setDeviceUuid(normalizedUuid);
|
|
|
result.setTableName(tableName);
|
|
|
- result.setImportedRows(rows.size());
|
|
|
- return result;
|
|
|
+ result.setImportedRows(importedRows);
|
|
|
+ return ApiResult.success(result);
|
|
|
} catch (IOException ex) {
|
|
|
- throw new BusinessException("解析 CSV 文件失败: " + ex.getMessage());
|
|
|
+ log.error("解析 CSV 文件失败, deviceUuid=" + normalizedUuid + ", tableName=" + tableName + ", error=" + ex.getMessage());
|
|
|
+ return ApiResult.error("解析 CSV 文件失败: " + ex.getMessage());
|
|
|
} catch (SQLException ex) {
|
|
|
- throw new BusinessException("时序数据导入失败: " + ex.getMessage());
|
|
|
+ log.error("时序数据导入失败, deviceUuid=" + normalizedUuid + ", tableName=" + tableName + ", error=" + ex.getMessage());
|
|
|
+ return ApiResult.error("时序数据导入失败: " + ex.getMessage());
|
|
|
}
|
|
|
}
|
|
|
|
|
|
@@ -533,27 +547,42 @@ public class QueryTdengineDataServiceImpl extends AbstractCrudService<QueryTdeng
|
|
|
if (!headerResolved) {
|
|
|
if (isHeaderRow(parts)) {
|
|
|
tsIndex = findColumnIndex(parts, TS_HEADER_ALIASES, "ts");
|
|
|
+ if (tsIndex < 0) {
|
|
|
+ return null;
|
|
|
+ }
|
|
|
pIndex = findColumnIndex(parts, P_HEADER_ALIASES, "p");
|
|
|
+ if (pIndex < 0) {
|
|
|
+ return null;
|
|
|
+ }
|
|
|
headerResolved = true;
|
|
|
continue;
|
|
|
}
|
|
|
headerResolved = true;
|
|
|
}
|
|
|
if (parts.length <= Math.max(tsIndex, pIndex)) {
|
|
|
- throw new BusinessException("CSV 第 " + lineNo + " 行列数不足");
|
|
|
+ log.error("CSV 第 " + lineNo + " 行列数不足");
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ Long tsMillis = parseTimestamp(parts[tsIndex], lineNo);
|
|
|
+ if (tsMillis == null) {
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ Double pValue = parseMetricValue(parts[pIndex], lineNo);
|
|
|
+ if (pValue == null) {
|
|
|
+ return null;
|
|
|
}
|
|
|
- rows.add(new TsdbCsvRow(
|
|
|
- parseTimestamp(parts[tsIndex], lineNo),
|
|
|
- parseMetricValue(parts[pIndex], lineNo)));
|
|
|
+ rows.add(new TsdbCsvRow(tsMillis, pValue));
|
|
|
}
|
|
|
}
|
|
|
if (rows.isEmpty()) {
|
|
|
- throw new BusinessException("CSV 无有效数据行");
|
|
|
+ log.error("CSV 无有效数据行");
|
|
|
+ return null;
|
|
|
}
|
|
|
return rows;
|
|
|
}
|
|
|
|
|
|
- private void batchInsertRows(String tableName, List<TsdbCsvRow> rows) throws SQLException {
|
|
|
+ private int batchInsertRows(String tableName, List<TsdbCsvRow> rows) throws SQLException {
|
|
|
+ int importedRows = 0;
|
|
|
try (Connection connection = dataSource.getConnection();
|
|
|
Statement statement = connection.createStatement()) {
|
|
|
for (int start = 0; start < rows.size(); start += IMPORT_BATCH_SIZE) {
|
|
|
@@ -561,16 +590,30 @@ public class QueryTdengineDataServiceImpl extends AbstractCrudService<QueryTdeng
|
|
|
StringBuilder sql = new StringBuilder("INSERT INTO ")
|
|
|
.append(tableName)
|
|
|
.append(" (ts, p) VALUES ");
|
|
|
+ int batchCount = 0;
|
|
|
for (int i = start; i < end; i++) {
|
|
|
TsdbCsvRow row = rows.get(i);
|
|
|
- if (i > start) {
|
|
|
+ if (!isImportableRow(row)) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ if (batchCount > 0) {
|
|
|
sql.append(' ');
|
|
|
}
|
|
|
sql.append('(').append(row.tsMillis).append(',').append(row.pValue).append(')');
|
|
|
+ batchCount++;
|
|
|
+ }
|
|
|
+ if (batchCount == 0) {
|
|
|
+ continue;
|
|
|
}
|
|
|
statement.execute(sql.toString());
|
|
|
+ importedRows += batchCount;
|
|
|
}
|
|
|
}
|
|
|
+ return importedRows;
|
|
|
+ }
|
|
|
+
|
|
|
+ private static boolean isImportableRow(TsdbCsvRow row) {
|
|
|
+ return row.tsMillis > 0 && Double.isFinite(row.pValue);
|
|
|
}
|
|
|
|
|
|
private static String[] splitCsvLine(String line) {
|
|
|
@@ -589,19 +632,21 @@ public class QueryTdengineDataServiceImpl extends AbstractCrudService<QueryTdeng
|
|
|
return false;
|
|
|
}
|
|
|
|
|
|
- private static int findColumnIndex(String[] headers, List<String> aliases, String columnLabel) {
|
|
|
+ private int findColumnIndex(String[] headers, List<String> aliases, String columnLabel) {
|
|
|
for (int i = 0; i < headers.length; i++) {
|
|
|
if (aliases.contains(headers[i].trim().toLowerCase(Locale.ROOT))) {
|
|
|
return i;
|
|
|
}
|
|
|
}
|
|
|
- throw new BusinessException("CSV 缺少 " + columnLabel + " 列");
|
|
|
+ log.error("CSV 缺少 " + columnLabel + " 列");
|
|
|
+ return -1;
|
|
|
}
|
|
|
|
|
|
- private static long parseTimestamp(String rawValue, int lineNo) {
|
|
|
+ private Long parseTimestamp(String rawValue, int lineNo) {
|
|
|
String value = rawValue == null ? "" : rawValue.trim();
|
|
|
if (value.isEmpty()) {
|
|
|
- throw new BusinessException("CSV 第 " + lineNo + " 行 ts 为空");
|
|
|
+ log.error("CSV 第 " + lineNo + " 行 ts 为空");
|
|
|
+ return null;
|
|
|
}
|
|
|
if (value.matches("\\d+")) {
|
|
|
long epoch = Long.parseLong(value);
|
|
|
@@ -618,19 +663,22 @@ public class QueryTdengineDataServiceImpl extends AbstractCrudService<QueryTdeng
|
|
|
try {
|
|
|
return new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss").parse(value).getTime();
|
|
|
} catch (ParseException ex) {
|
|
|
- throw new BusinessException("CSV 第 " + lineNo + " 行 ts 格式无效: " + value);
|
|
|
+ log.error("CSV 第 " + lineNo + " 行 ts 格式无效: " + value);
|
|
|
+ return null;
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- private static double parseMetricValue(String rawValue, int lineNo) {
|
|
|
+ private Double parseMetricValue(String rawValue, int lineNo) {
|
|
|
String value = rawValue == null ? "" : rawValue.trim();
|
|
|
if (value.isEmpty()) {
|
|
|
- throw new BusinessException("CSV 第 " + lineNo + " 行 p 为空");
|
|
|
+ log.error("CSV 第 " + lineNo + " 行 p 为空");
|
|
|
+ return null;
|
|
|
}
|
|
|
try {
|
|
|
return new BigDecimal(value).doubleValue();
|
|
|
} catch (NumberFormatException ex) {
|
|
|
- throw new BusinessException("CSV 第 " + lineNo + " 行 p 不是有效数值: " + value);
|
|
|
+ log.error("CSV 第 " + lineNo + " 行 p 不是有效数值: " + value);
|
|
|
+ return null;
|
|
|
}
|
|
|
}
|
|
|
|