|
|
@@ -23,12 +23,21 @@ import org.springframework.beans.factory.annotation.Autowired;
|
|
|
import org.springframework.cache.annotation.Cacheable;
|
|
|
import org.springframework.jdbc.core.JdbcTemplate;
|
|
|
import org.springframework.stereotype.Service;
|
|
|
+import org.springframework.web.multipart.MultipartFile;
|
|
|
|
|
|
import javax.sql.DataSource;
|
|
|
+import java.io.BufferedReader;
|
|
|
+import java.io.IOException;
|
|
|
+import java.io.InputStreamReader;
|
|
|
import java.math.BigDecimal;
|
|
|
+import java.nio.charset.StandardCharsets;
|
|
|
import java.sql.*;
|
|
|
import java.text.ParseException;
|
|
|
import java.text.SimpleDateFormat;
|
|
|
+import java.time.LocalDateTime;
|
|
|
+import java.time.ZoneId;
|
|
|
+import java.time.format.DateTimeFormatter;
|
|
|
+import java.time.format.DateTimeParseException;
|
|
|
import java.util.*;
|
|
|
import java.util.regex.Pattern;
|
|
|
import java.util.stream.Collectors;
|
|
|
@@ -47,6 +56,16 @@ public class QueryTdengineDataServiceImpl extends AbstractCrudService<QueryTdeng
|
|
|
|
|
|
private static final Pattern SQL_IDENTIFIER_PATTERN = Pattern.compile("^[a-zA-Z_][a-zA-Z0-9_]*$");
|
|
|
|
|
|
+ private static final Pattern DEVICE_UUID_PATTERN = Pattern.compile("^[a-zA-Z0-9_-]+$");
|
|
|
+
|
|
|
+ private static final int IMPORT_BATCH_SIZE = 500;
|
|
|
+
|
|
|
+ private static final DateTimeFormatter TS_FORMAT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
|
|
|
+
|
|
|
+ private static final List<String> TS_HEADER_ALIASES = Arrays.asList("ts", "timestamp", "time");
|
|
|
+
|
|
|
+ private static final List<String> P_HEADER_ALIASES = Arrays.asList("p", "power", "active_power", "value");
|
|
|
+
|
|
|
@Autowired
|
|
|
private TsdbUtils tsdbUtils;
|
|
|
@Autowired
|
|
|
@@ -460,4 +479,172 @@ public class QueryTdengineDataServiceImpl extends AbstractCrudService<QueryTdeng
|
|
|
throw new BusinessException(paramName + " 格式非法，仅允许字母、数字与下划线，且不能以数字开头");
|
|
|
}
|
|
|
}
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public DeviceCsvImportResultVO importDeviceCsv(String deviceUuid, MultipartFile file) {
|
|
|
+ if (StringUtils.isBlank(deviceUuid)) {
|
|
|
+ throw new BusinessException("设备UUID不能为空");
|
|
|
+ }
|
|
|
+ if (!DEVICE_UUID_PATTERN.matcher(deviceUuid.trim()).matches()) {
|
|
|
+ throw new BusinessException("设备UUID格式无效");
|
|
|
+ }
|
|
|
+ if (file == null || file.isEmpty()) {
|
|
|
+ throw new BusinessException("导入文件不能为空");
|
|
|
+ }
|
|
|
+ String originalFilename = file.getOriginalFilename();
|
|
|
+ if (StringUtils.isBlank(originalFilename) || !originalFilename.toLowerCase(Locale.ROOT).endsWith(".csv")) {
|
|
|
+ throw new BusinessException("仅支持 CSV 文件");
|
|
|
+ }
|
|
|
+
|
|
|
+ String normalizedUuid = deviceUuid.trim();
|
|
|
+ String tableName = buildDeviceSubTableName(normalizedUuid);
|
|
|
+ try {
|
|
|
+ List<TsdbCsvRow> rows = parseCsvRows(file);
|
|
|
+ batchInsertRows(tableName, rows);
|
|
|
+
|
|
|
+ DeviceCsvImportResultVO result = new DeviceCsvImportResultVO();
|
|
|
+ result.setDeviceUuid(normalizedUuid);
|
|
|
+ result.setTableName(tableName);
|
|
|
+ result.setImportedRows(rows.size());
|
|
|
+ return result;
|
|
|
+ } catch (IOException ex) {
|
|
|
+ throw new BusinessException("解析 CSV 文件失败: " + ex.getMessage());
|
|
|
+ } catch (SQLException ex) {
|
|
|
+ throw new BusinessException("时序数据导入失败: " + ex.getMessage());
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private List<TsdbCsvRow> parseCsvRows(MultipartFile file) throws IOException {
|
|
|
+ List<TsdbCsvRow> rows = new ArrayList<>();
|
|
|
+ try (BufferedReader reader = new BufferedReader(
|
|
|
+ new InputStreamReader(file.getInputStream(), StandardCharsets.UTF_8))) {
|
|
|
+ String line;
|
|
|
+ int tsIndex = 0;
|
|
|
+ int pIndex = 1;
|
|
|
+ boolean headerResolved = false;
|
|
|
+ int lineNo = 0;
|
|
|
+ while ((line = reader.readLine()) != null) {
|
|
|
+ lineNo++;
|
|
|
+ line = line.trim();
|
|
|
+ if (line.isEmpty()) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ String[] parts = splitCsvLine(line);
|
|
|
+ if (!headerResolved) {
|
|
|
+ if (isHeaderRow(parts)) {
|
|
|
+ tsIndex = findColumnIndex(parts, TS_HEADER_ALIASES, "ts");
|
|
|
+ pIndex = findColumnIndex(parts, P_HEADER_ALIASES, "p");
|
|
|
+ headerResolved = true;
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ headerResolved = true;
|
|
|
+ }
|
|
|
+ if (parts.length <= Math.max(tsIndex, pIndex)) {
|
|
|
+ throw new BusinessException("CSV 第 " + lineNo + " 行列数不足");
|
|
|
+ }
|
|
|
+ rows.add(new TsdbCsvRow(
|
|
|
+ parseTimestamp(parts[tsIndex], lineNo),
|
|
|
+ parseMetricValue(parts[pIndex], lineNo)));
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (rows.isEmpty()) {
|
|
|
+ throw new BusinessException("CSV 无有效数据行");
|
|
|
+ }
|
|
|
+ return rows;
|
|
|
+ }
|
|
|
+
|
|
|
+ private void batchInsertRows(String tableName, List<TsdbCsvRow> rows) throws SQLException {
|
|
|
+ try (Connection connection = dataSource.getConnection();
|
|
|
+ Statement statement = connection.createStatement()) {
|
|
|
+ for (int start = 0; start < rows.size(); start += IMPORT_BATCH_SIZE) {
|
|
|
+ int end = Math.min(start + IMPORT_BATCH_SIZE, rows.size());
|
|
|
+ StringBuilder sql = new StringBuilder("INSERT INTO ")
|
|
|
+ .append(tableName)
|
|
|
+ .append(" (ts, p) VALUES ");
|
|
|
+ for (int i = start; i < end; i++) {
|
|
|
+ TsdbCsvRow row = rows.get(i);
|
|
|
+ if (i > start) {
|
|
|
+ sql.append(' ');
|
|
|
+ }
|
|
|
+ sql.append('(').append(row.tsMillis).append(',').append(row.pValue).append(')');
|
|
|
+ }
|
|
|
+ statement.execute(sql.toString());
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private static String[] splitCsvLine(String line) {
|
|
|
+ return Arrays.stream(line.split(",", -1))
|
|
|
+ .map(String::trim)
|
|
|
+ .toArray(String[]::new);
|
|
|
+ }
|
|
|
+
|
|
|
+ private static boolean isHeaderRow(String[] parts) {
|
|
|
+ for (String part : parts) {
|
|
|
+ String normalized = part.trim().toLowerCase(Locale.ROOT);
|
|
|
+ if (TS_HEADER_ALIASES.contains(normalized) || P_HEADER_ALIASES.contains(normalized)) {
|
|
|
+ return true;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+
|
|
|
+ private static 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 + " 列");
|
|
|
+ }
|
|
|
+
|
|
|
+ private static long parseTimestamp(String rawValue, int lineNo) {
|
|
|
+ String value = rawValue == null ? "" : rawValue.trim();
|
|
|
+ if (value.isEmpty()) {
|
|
|
+ throw new BusinessException("CSV 第 " + lineNo + " 行 ts 为空");
|
|
|
+ }
|
|
|
+ if (value.matches("\\d+")) {
|
|
|
+ long epoch = Long.parseLong(value);
|
|
|
+ return value.length() <= 10 ? epoch * 1000L : epoch;
|
|
|
+ }
|
|
|
+ try {
|
|
|
+ return LocalDateTime.parse(value, TS_FORMAT)
|
|
|
+ .atZone(ZoneId.systemDefault())
|
|
|
+ .toInstant()
|
|
|
+ .toEpochMilli();
|
|
|
+ } catch (DateTimeParseException ignored) {
|
|
|
+ // fall through
|
|
|
+ }
|
|
|
+ try {
|
|
|
+ return new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss").parse(value).getTime();
|
|
|
+ } catch (ParseException ex) {
|
|
|
+ throw new BusinessException("CSV 第 " + lineNo + " 行 ts 格式无效: " + value);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private static double parseMetricValue(String rawValue, int lineNo) {
|
|
|
+ String value = rawValue == null ? "" : rawValue.trim();
|
|
|
+ if (value.isEmpty()) {
|
|
|
+ throw new BusinessException("CSV 第 " + lineNo + " 行 p 为空");
|
|
|
+ }
|
|
|
+ try {
|
|
|
+ return new BigDecimal(value).doubleValue();
|
|
|
+ } catch (NumberFormatException ex) {
|
|
|
+ throw new BusinessException("CSV 第 " + lineNo + " 行 p 不是有效数值: " + value);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private static String buildDeviceSubTableName(String deviceUuid) {
|
|
|
+ return "_" + deviceUuid;
|
|
|
+ }
|
|
|
+
|
|
|
+ private static final class TsdbCsvRow {
|
|
|
+ private final long tsMillis;
|
|
|
+ private final double pValue;
|
|
|
+
|
|
|
+ private TsdbCsvRow(long tsMillis, double pValue) {
|
|
|
+ this.tsMillis = tsMillis;
|
|
|
+ this.pValue = pValue;
|
|
|
+ }
|
|
|
+ }
|
|
|
}
|
