|
|
|
|
@ -1,19 +1,25 @@
|
|
|
|
|
package cn.iocoder.yudao.module.iot.service.device;
|
|
|
|
|
|
|
|
|
|
import com.baomidou.dynamic.datasource.annotation.DS;
|
|
|
|
|
import com.fasterxml.jackson.core.JsonProcessingException;
|
|
|
|
|
import com.fasterxml.jackson.core.type.TypeReference;
|
|
|
|
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
|
|
|
|
import com.taosdata.jdbc.utils.BlobUtil;
|
|
|
|
|
import org.apache.commons.codec.DecoderException;
|
|
|
|
|
import org.apache.commons.codec.binary.Hex;
|
|
|
|
|
import org.springframework.dao.EmptyResultDataAccessException;
|
|
|
|
|
import org.springframework.jdbc.core.JdbcTemplate;
|
|
|
|
|
import org.springframework.jdbc.core.RowMapper;
|
|
|
|
|
import org.springframework.stereotype.Service;
|
|
|
|
|
|
|
|
|
|
import javax.annotation.Resource;
|
|
|
|
|
import java.nio.charset.StandardCharsets;
|
|
|
|
|
import java.sql.Blob;
|
|
|
|
|
import java.sql.ResultSet;
|
|
|
|
|
import java.sql.SQLException;
|
|
|
|
|
import java.sql.Timestamp;
|
|
|
|
|
import java.text.SimpleDateFormat;
|
|
|
|
|
import java.util.ArrayList;
|
|
|
|
|
import java.util.HashMap;
|
|
|
|
|
import java.util.List;
|
|
|
|
|
import java.util.Map;
|
|
|
|
|
import java.util.*;
|
|
|
|
|
|
|
|
|
|
@Service
|
|
|
|
|
public class TDengineService {
|
|
|
|
|
@ -40,7 +46,7 @@ public class TDengineService {
|
|
|
|
|
// 3. 创建超级表
|
|
|
|
|
String createSuperTableSQL = "CREATE STABLE IF NOT EXISTS device_data (" +
|
|
|
|
|
"ts TIMESTAMP, " +
|
|
|
|
|
"query_data NCHAR(2048)" +
|
|
|
|
|
"query_data BLOB" +
|
|
|
|
|
") TAGS (device_id BIGINT)";
|
|
|
|
|
jdbcTemplate.execute(createSuperTableSQL);
|
|
|
|
|
|
|
|
|
|
@ -82,8 +88,44 @@ public class TDengineService {
|
|
|
|
|
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
|
|
|
|
|
String timestampStr = sdf.format(ts);
|
|
|
|
|
result.put("timestamp", timestampStr);
|
|
|
|
|
String queryData = rs.getString("query_data");
|
|
|
|
|
result.put("queryData", queryData);
|
|
|
|
|
|
|
|
|
|
byte[] blob = rs.getBytes("query_data");
|
|
|
|
|
if (blob != null) {
|
|
|
|
|
String jsonStr = new String(blob, StandardCharsets.UTF_8);
|
|
|
|
|
|
|
|
|
|
try {
|
|
|
|
|
// 1. 先去除外层的双引号（如果存在）
|
|
|
|
|
String trimmed = jsonStr.trim();
|
|
|
|
|
if (trimmed.startsWith("\"") && trimmed.endsWith("\"")) {
|
|
|
|
|
trimmed = trimmed.substring(1, trimmed.length() - 1);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// 2. 检查是否是十六进制字符串
|
|
|
|
|
if (isHexString(trimmed)) {
|
|
|
|
|
// 3. 十六进制解码
|
|
|
|
|
String decodedJson = hexToString(trimmed);
|
|
|
|
|
result.put("queryData", decodedJson);
|
|
|
|
|
} else {
|
|
|
|
|
// 如果不是十六进制，尝试直接解析
|
|
|
|
|
ObjectMapper objectMapper = new ObjectMapper();
|
|
|
|
|
List<Map<String, Object>> queryData = objectMapper.readValue(
|
|
|
|
|
trimmed,
|
|
|
|
|
new TypeReference<List<Map<String, Object>>>() {}
|
|
|
|
|
);
|
|
|
|
|
result.put("queryData", queryData);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
} catch (Exception e) {
|
|
|
|
|
System.err.println("解析JSON失败: " + e.getMessage());
|
|
|
|
|
result.put("queryData", new ArrayList<>());
|
|
|
|
|
result.put("rawData", jsonStr);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
} else {
|
|
|
|
|
result.put("queryData", new ArrayList<>());
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
result.put("deviceId", id);
|
|
|
|
|
result.put("tableName", tableName);
|
|
|
|
|
return result;
|
|
|
|
|
@ -112,12 +154,11 @@ public class TDengineService {
|
|
|
|
|
/**
|
|
|
|
|
* 向指定设备插入带时间戳的数据
|
|
|
|
|
* @param id 设备ID
|
|
|
|
|
* @param queryData 查询数据
|
|
|
|
|
* @param timestamp 时间戳
|
|
|
|
|
* @return 插入是否成功
|
|
|
|
|
*/
|
|
|
|
|
@DS("tdengine")
|
|
|
|
|
public boolean insertDeviceData(Long id, String queryData, Timestamp timestamp) {
|
|
|
|
|
public boolean insertDeviceData(Long id, String jsonString, Timestamp timestamp) {
|
|
|
|
|
try {
|
|
|
|
|
// 确保使用正确的数据库
|
|
|
|
|
jdbcTemplate.execute("USE besure");
|
|
|
|
|
@ -125,9 +166,12 @@ public class TDengineService {
|
|
|
|
|
String tableName = "d_" + id;
|
|
|
|
|
String sql = "INSERT INTO " + tableName + " (ts, query_data) VALUES (?, ?)";
|
|
|
|
|
|
|
|
|
|
int affectedRows = jdbcTemplate.update(sql, timestamp, queryData);
|
|
|
|
|
System.out.println("向设备" + id + "插入数据成功，时间戳: " + timestamp);
|
|
|
|
|
return affectedRows > 0;
|
|
|
|
|
return jdbcTemplate.update(sql, ps -> {
|
|
|
|
|
ps.setTimestamp(1, timestamp);
|
|
|
|
|
// JSON字符串转byte数组
|
|
|
|
|
byte[] blobData = jsonString.getBytes(StandardCharsets.UTF_8);
|
|
|
|
|
ps.setBytes(2, blobData);
|
|
|
|
|
}) > 0;
|
|
|
|
|
} catch (Exception e) {
|
|
|
|
|
System.out.println("向设备" + id + "插入数据时发生异常: " + e.getMessage());
|
|
|
|
|
return false;
|
|
|
|
|
@ -140,24 +184,202 @@ public class TDengineService {
|
|
|
|
|
* @return 设备数据列表，按时间戳倒序排列
|
|
|
|
|
*/
|
|
|
|
|
@DS("tdengine")
|
|
|
|
|
public List<Map<String, Object>> getAllDeviceDataOrderByTimeDesc(Long id) {
|
|
|
|
|
public List<Map<String, Object>> getNewestDeviceDataOrderByTimeDesc(Long id) {
|
|
|
|
|
String tableName = "d_" + id;
|
|
|
|
|
String sql = "SELECT ts, query_data FROM besure." + tableName + " ORDER BY ts DESC";
|
|
|
|
|
String sql = "SELECT ts, query_data FROM besure." + tableName + " ORDER BY ts DESC LIMIT 1";
|
|
|
|
|
|
|
|
|
|
try {
|
|
|
|
|
return jdbcTemplate.query(sql, new RowMapper<Map<String, Object>>() {
|
|
|
|
|
return Collections.singletonList(jdbcTemplate.queryForObject(sql, new RowMapper<Map<String, Object>>() {
|
|
|
|
|
@Override
|
|
|
|
|
public Map<String, Object> mapRow(ResultSet rs, int rowNum) throws SQLException {
|
|
|
|
|
Map<String, Object> result = new HashMap<>();
|
|
|
|
|
result.put("timestamp", rs.getTimestamp("ts"));
|
|
|
|
|
result.put("queryData", rs.getString("query_data"));
|
|
|
|
|
result.put("deviceId", id);
|
|
|
|
|
|
|
|
|
|
byte[] blob = rs.getBytes("query_data");
|
|
|
|
|
if (blob != null) {
|
|
|
|
|
String jsonStr = new String(blob, StandardCharsets.UTF_8);
|
|
|
|
|
|
|
|
|
|
try {
|
|
|
|
|
// 1. 先去除外层的双引号（如果存在）
|
|
|
|
|
String trimmed = jsonStr.trim();
|
|
|
|
|
if (trimmed.startsWith("\"") && trimmed.endsWith("\"")) {
|
|
|
|
|
trimmed = trimmed.substring(1, trimmed.length() - 1);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// 2. 检查是否是十六进制字符串
|
|
|
|
|
if (isHexString(trimmed)) {
|
|
|
|
|
// 3. 十六进制解码
|
|
|
|
|
String decodedJson = hexToString(trimmed);
|
|
|
|
|
|
|
|
|
|
result.put("queryData", decodedJson);
|
|
|
|
|
} else {
|
|
|
|
|
// 如果不是十六进制，尝试直接解析
|
|
|
|
|
ObjectMapper objectMapper = new ObjectMapper();
|
|
|
|
|
List<Map<String, Object>> queryData = objectMapper.readValue(
|
|
|
|
|
trimmed,
|
|
|
|
|
new TypeReference<List<Map<String, Object>>>() {}
|
|
|
|
|
);
|
|
|
|
|
result.put("queryData", queryData);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
} catch (Exception e) {
|
|
|
|
|
System.err.println("解析JSON失败: " + e.getMessage());
|
|
|
|
|
result.put("queryData", new ArrayList<>());
|
|
|
|
|
result.put("rawData", jsonStr);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
} else {
|
|
|
|
|
result.put("queryData", new ArrayList<>());
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
return result;
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
}));
|
|
|
|
|
} catch (EmptyResultDataAccessException e) {
|
|
|
|
|
return Collections.singletonList(createEmptyResult(id));
|
|
|
|
|
} catch (Exception e) {
|
|
|
|
|
System.out.println("查询设备" + id + "的全部数据时发生异常: " + e.getMessage());
|
|
|
|
|
System.err.println("查询设备" + id + "的最新数据时发生异常: " + e.getMessage());
|
|
|
|
|
e.printStackTrace();
|
|
|
|
|
return Collections.singletonList(createEmptyResult(id));
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// 检查是否是十六进制字符串
|
|
|
|
|
private boolean isHexString(String str) {
|
|
|
|
|
if (str == null || str.trim().isEmpty()) {
|
|
|
|
|
return false;
|
|
|
|
|
}
|
|
|
|
|
String s = str.trim();
|
|
|
|
|
// 检查是否只包含十六进制字符（0-9, a-f, A-F）
|
|
|
|
|
return s.matches("^[0-9a-fA-F]+$");
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// 十六进制解码
|
|
|
|
|
private String hexToString(String hex) {
|
|
|
|
|
try {
|
|
|
|
|
// 使用 Apache Commons Codec
|
|
|
|
|
byte[] bytes = Hex.decodeHex(hex);
|
|
|
|
|
return new String(bytes, StandardCharsets.UTF_8);
|
|
|
|
|
} catch (DecoderException e) {
|
|
|
|
|
throw new RuntimeException("十六进制解码失败: " + e.getMessage(), e);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
private List<Map<String, Object>> parseJsonData(byte[] blob) {
|
|
|
|
|
if (blob == null || blob.length == 0) {
|
|
|
|
|
return new ArrayList<>();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
try {
|
|
|
|
|
String jsonStr = new String(blob, StandardCharsets.UTF_8);
|
|
|
|
|
|
|
|
|
|
// 检查是否是有效的JSON
|
|
|
|
|
if (jsonStr.trim().startsWith("[") && jsonStr.trim().endsWith("]")) {
|
|
|
|
|
ObjectMapper objectMapper = new ObjectMapper();
|
|
|
|
|
return objectMapper.readValue(jsonStr,
|
|
|
|
|
new TypeReference<List<Map<String, Object>>>() {});
|
|
|
|
|
} else {
|
|
|
|
|
// 可能是字符串化的JSON，尝试去除引号
|
|
|
|
|
jsonStr = jsonStr.trim();
|
|
|
|
|
if (jsonStr.startsWith("\"") && jsonStr.endsWith("\"")) {
|
|
|
|
|
jsonStr = jsonStr.substring(1, jsonStr.length() - 1);
|
|
|
|
|
return new ObjectMapper().readValue(jsonStr,
|
|
|
|
|
new TypeReference<List<Map<String, Object>>>() {});
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
} catch (Exception e) {
|
|
|
|
|
System.err.println("解析JSON数据失败: " + e.getMessage());
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
return new ArrayList<>();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
private Map<String, Object> createEmptyResult(Long deviceId) {
|
|
|
|
|
Map<String, Object> result = new HashMap<>();
|
|
|
|
|
result.put("deviceId", deviceId);
|
|
|
|
|
result.put("queryData", new ArrayList<>());
|
|
|
|
|
result.put("timestamp", null);
|
|
|
|
|
return result;
|
|
|
|
|
}
|
|
|
|
|
/**
|
|
|
|
|
* 查询指定设备表的全部数据并按时间倒序排序
|
|
|
|
|
* @param id 设备ID
|
|
|
|
|
* @return 设备数据列表，按时间戳倒序排列
|
|
|
|
|
*/
|
|
|
|
|
@DS("tdengine")
|
|
|
|
|
public List<Map<String, Object>> getstDeviceDataOrderByTimeDesc(Long id,String collectionStartTime, String collectionEndTime) {
|
|
|
|
|
String tableName = "d_" + id;
|
|
|
|
|
StringBuilder sqlBuilder = new StringBuilder();
|
|
|
|
|
List<Object> params = new ArrayList<>();
|
|
|
|
|
|
|
|
|
|
sqlBuilder.append("SELECT ts, query_data FROM besure.").append(tableName).append(" WHERE 1=1");
|
|
|
|
|
|
|
|
|
|
if (collectionStartTime != null) {
|
|
|
|
|
// 直接将时间字符串拼接到SQL中
|
|
|
|
|
sqlBuilder.append(" AND ts >= '").append(collectionStartTime).append("'");
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if (collectionEndTime != null) {
|
|
|
|
|
sqlBuilder.append(" AND ts <= '").append(collectionEndTime).append("'");
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
sqlBuilder.append(" ORDER BY ts DESC");
|
|
|
|
|
|
|
|
|
|
try {
|
|
|
|
|
return jdbcTemplate.query(sqlBuilder.toString(), new RowMapper<Map<String, Object>>() {
|
|
|
|
|
@Override
|
|
|
|
|
public Map<String, Object> mapRow(ResultSet rs, int rowNum) throws SQLException {
|
|
|
|
|
Map<String, Object> result = new HashMap<>();
|
|
|
|
|
result.put("timestamp", rs.getTimestamp("ts"));
|
|
|
|
|
result.put("deviceId", id);
|
|
|
|
|
|
|
|
|
|
byte[] blob = rs.getBytes("query_data");
|
|
|
|
|
if (blob != null) {
|
|
|
|
|
String jsonStr = new String(blob, StandardCharsets.UTF_8);
|
|
|
|
|
|
|
|
|
|
try {
|
|
|
|
|
// 1. 先去除外层的双引号（如果存在）
|
|
|
|
|
String trimmed = jsonStr.trim();
|
|
|
|
|
if (trimmed.startsWith("\"") && trimmed.endsWith("\"")) {
|
|
|
|
|
trimmed = trimmed.substring(1, trimmed.length() - 1);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// 2. 检查是否是十六进制字符串
|
|
|
|
|
if (isHexString(trimmed)) {
|
|
|
|
|
// 3. 十六进制解码
|
|
|
|
|
String decodedJson = hexToString(trimmed);
|
|
|
|
|
|
|
|
|
|
result.put("queryData", decodedJson);
|
|
|
|
|
} else {
|
|
|
|
|
// 如果不是十六进制，尝试直接解析
|
|
|
|
|
ObjectMapper objectMapper = new ObjectMapper();
|
|
|
|
|
List<Map<String, Object>> queryData = objectMapper.readValue(
|
|
|
|
|
trimmed,
|
|
|
|
|
new TypeReference<List<Map<String, Object>>>() {}
|
|
|
|
|
);
|
|
|
|
|
result.put("queryData", queryData);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
} catch (Exception e) {
|
|
|
|
|
System.err.println("解析JSON失败: " + e.getMessage());
|
|
|
|
|
result.put("queryData", new ArrayList<>());
|
|
|
|
|
result.put("rawData", jsonStr);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
} else {
|
|
|
|
|
result.put("queryData", new ArrayList<>());
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
return result;
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
} catch (EmptyResultDataAccessException e) {
|
|
|
|
|
return Collections.singletonList(createEmptyResult(id));
|
|
|
|
|
} catch (Exception e) {
|
|
|
|
|
System.err.println("查询设备" + id + "的最新数据时发生异常: " + e.getMessage());
|
|
|
|
|
e.printStackTrace();
|
|
|
|
|
return Collections.singletonList(createEmptyResult(id));
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
HuangHuiKang
1 month ago