潘集设备上报

@@ -2,11 +2,13 @@ package com.iot.scheduler.service;
 
 
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
 import com.alibaba.fastjson.JSONObject;
+import com.iot.scheduler.utils.DeviceDataStorage;
 import com.zaxxer.hikari.HikariConfig;
 import com.zaxxer.hikari.HikariConfig;
 import com.zaxxer.hikari.HikariDataSource;
 import com.zaxxer.hikari.HikariDataSource;
 import jakarta.annotation.Resource;
 import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.SerializationUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.http.HttpEntity;
 import org.apache.http.HttpEntity;
 import org.apache.http.NameValuePair;
 import org.apache.http.NameValuePair;
@@ -22,10 +24,15 @@ import org.springframework.beans.factory.annotation.Value;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Service;
 import org.springframework.stereotype.Service;
 
 
+import java.io.Serializable;
 import java.nio.charset.StandardCharsets;
 import java.nio.charset.StandardCharsets;
 import java.sql.*;
 import java.sql.*;
 import java.sql.Date;
 import java.sql.Date;
 import java.text.SimpleDateFormat;
 import java.text.SimpleDateFormat;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
 import java.util.*;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeUnit;
 
 
@@ -79,6 +86,25 @@ public class PjDeviceReportService {
     @Resource
     @Resource
     private RedisTemplate<String, String> redisTemplate;
     private RedisTemplate<String, String> redisTemplate;
 
 
+    private static final DateTimeFormatter YYYY_MM_DD = DateTimeFormatter.ofPattern("yyyy-MM-dd");
+    private static final DateTimeFormatter YYYY_MM = DateTimeFormatter.ofPattern("yyyy-MM");
+
+    /**
+     * 时间戳（毫秒） -> yyyy-MM-dd
+     */
+    private static String toYyyyMmDd(long timestampMillis) {
+        return LocalDateTime.ofInstant(Instant.ofEpochMilli(timestampMillis), ZoneId.systemDefault())
+                .format(YYYY_MM_DD);
+    }
+
+    /**
+     * 时间戳（毫秒） -> yyyy-MM
+     */
+    private static String toYyyyMm(long timestampMillis) {
+        return LocalDateTime.ofInstant(Instant.ofEpochMilli(timestampMillis), ZoneId.systemDefault())
+                .format(YYYY_MM);
+    }
+
     /**
     /**
      * 企业设备上报
      * 企业设备上报
      */
      */
@@ -244,16 +270,13 @@ public class PjDeviceReportService {
 
 
             // 2. 准备上报数据
             // 2. 准备上报数据
             log.info("步骤2: 开始准备上报数据");
             log.info("步骤2: 开始准备上报数据");
-            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd");
-            String nowTime = simpleDateFormat.format(new java.util.Date());
-            log.info("上报时间: {}", nowTime);
 
 
             CloseableHttpClient httpclient = HttpClients.createDefault();
             CloseableHttpClient httpclient = HttpClients.createDefault();
             HttpPost httpPost = new HttpPost(reportUrl);
             HttpPost httpPost = new HttpPost(reportUrl);
 
 
             // 3. 获取并设置Token
             // 3. 获取并设置Token
             log.info("步骤3: 获取访问令牌");
             log.info("步骤3: 获取访问令牌");
-            String token = getToken();
+            String token = getTokenNoCache();
             if (StringUtils.isBlank(token)) {
             if (StringUtils.isBlank(token)) {
                 log.error("获取访问令牌失败，无法进行数据上报");
                 log.error("获取访问令牌失败，无法进行数据上报");
                 return;
                 return;
@@ -266,9 +289,10 @@ public class PjDeviceReportService {
             // 4. 构建上报数据
             // 4. 构建上报数据
             log.info("步骤4: 构建上报数据结构");
             log.info("步骤4: 构建上报数据结构");
             List<Map<String, Object>> dataList = new ArrayList<>(needSyncDataList.size());
             List<Map<String, Object>> dataList = new ArrayList<>(needSyncDataList.size());
+            List<Map<String, Object>> saveDataList = new ArrayList<>(needSyncDataList.size());
             int successCount = 0;
             int successCount = 0;
             int errorCount = 0;
             int errorCount = 0;
-
+            DeviceDataStorage storage = new DeviceDataStorage();
             for (int i = 0; i < needSyncDataList.size(); i++) {
             for (int i = 0; i < needSyncDataList.size(); i++) {
                 try {
                 try {
                     Object needSyncData = needSyncDataList.get(i);
                     Object needSyncData = needSyncDataList.get(i);
@@ -281,14 +305,51 @@ public class PjDeviceReportService {
                     }
                     }
 
 
                     Map<String, Object> data = new HashMap<>();
                     Map<String, Object> data = new HashMap<>();
+                    String deviceCode = (String) dataObject.get(2);
+                    String energyType = (String) dataObject.get(3);
+
                     data.put("entName", dataObject.get(0));
                     data.put("entName", dataObject.get(0));
                     data.put("deviceName", dataObject.get(1));
                     data.put("deviceName", dataObject.get(1));
-                    data.put("deviceCode", dataObject.get(2));
-                    data.put("energyType", dataObject.get(3));
-                    data.put("energyValue", dataObject.get(4));
-                    data.put("timeSpan", "日");
-                    data.put("collectTime", nowTime);
+                    data.put("deviceCode", deviceCode);
+                    data.put("energyType", energyType);
+                    Long ts = (Long) dataObject.get(5);
+                    DeviceDataStorage.DeviceRecord lastRecord = storage.getLastRecord(deviceCode);
+                    if (lastRecord == null) {
+                        log.warn("第 {} 条数据,设备:{} 未找到上一期上报数据", i + 1, deviceCode);
+                        continue;
+                    }
+                    if (lastRecord.getTimestamp() >= ts) {
+                        log.warn("第 {} 条数据,设备:{} 数据未更新无需上报", i + 1, deviceCode);
+                        continue;
+                    }
+                    Double energyValue = toDouble(dataObject.get(4));
+
+                    if ("电".equals(energyType) || "天然气".equals(energyType)) {
+                        data.put("energyValue", energyValue);
+                        data.put("timeSpan", "日");
+                        data.put("collectTime", toYyyyMmDd(ts));
+                    } else if ("水".equals(energyType)) {
+                        Double reportValue = energyValue - lastRecord.getValue();
+                        if (reportValue <= 0) {
+                            log.warn("第 {} 条数据,设备:{} 水表抄表记录异常", i + 1, deviceCode);
+                        }
+                        data.put("energyValue", reportValue);
+                        data.put("timeSpan", "月");
+                        data.put("collectTime", toYyyyMm(ts));
+                    } else {
+                        log.warn("第 {} 条数据,设备:{} 物模型数据错误!", i + 1, deviceCode);
+                        continue;
+                    }
+
                     dataList.add(data);
                     dataList.add(data);
+                    Map<String, Object> saveData = (Map<String, Object>) SerializationUtils.clone((Serializable) data);
+                    saveData.put("ts",ts);
+                    saveData.put("orignValue",energyValue);
+                    saveDataList.add(saveData);
+
+                    //处理历史数据放开
+//                    saveReportData(Collections.singletonList(saveData));
+
                     successCount++;
                     successCount++;
 
 
                     // 每100条记录输出一次进度
                     // 每100条记录输出一次进度
@@ -336,6 +397,8 @@ public class PjDeviceReportService {
                     JSONObject jsonResult = JSON.parseObject(result);
                     JSONObject jsonResult = JSON.parseObject(result);
                     if (jsonResult != null && "true".equalsIgnoreCase(jsonResult.getString("success"))) {
                     if (jsonResult != null && "true".equalsIgnoreCase(jsonResult.getString("success"))) {
                         log.info("数据上报成功！共上报 {} 条设备数据", dataList.size());
                         log.info("数据上报成功！共上报 {} 条设备数据", dataList.size());
+                        //记录上报数据
+                       saveReportData(saveDataList);
                     } else {
                     } else {
                         log.error("数据上报失败！响应状态异常: {}", result);
                         log.error("数据上报失败！响应状态异常: {}", result);
                     }
                     }
@@ -353,6 +416,19 @@ public class PjDeviceReportService {
         }
         }
     }
     }
 
 
+    private void saveReportData(List<Map<String, Object>> dataList) {
+        if (CollectionUtils.isEmpty(dataList)) {
+            return;
+        }
+        DeviceDataStorage storage = new DeviceDataStorage();
+        for (Map<String, Object> data : dataList) {
+            String deviceCode = (String) data.get("deviceCode");
+            Long ts = (Long) data.get("ts");
+            Double orignValue = (Double) data.get("orignValue");
+            storage.updateData(deviceCode, ts, orignValue);
+        }
+    }
+
     /**
     /**
      * 获取token
      * 获取token
      *
      *
@@ -432,6 +508,70 @@ public class PjDeviceReportService {
         }
         }
     }
     }
 
 
+    /**
+     * 获取token
+     *
+     * @return
+     */
+    private String getTokenNoCache() {
+        log.info("开始获取访问令牌");
+        long startTime = System.currentTimeMillis();
+        String key = "pjjkq_report_token";
+
+        try {
+            String token;
+            // 2. 请求新的token
+            CloseableHttpClient httpclient = HttpClients.createDefault();
+            HttpPost httpPost = new HttpPost(tokenUrl);
+            httpPost.setHeader("Authorization", "Basic bWljcm8tcG9ydGFsOlBAc3MxMjM0");
+            httpPost.setHeader("Content-Type", "application/x-www-form-urlencoded; charset=utf-8");
+
+            List<NameValuePair> params = new ArrayList<>();
+            params.add(new BasicNameValuePair("grant_type", "password"));
+            params.add(new BasicNameValuePair("username", username));
+            params.add(new BasicNameValuePair("password", password)); // 密码脱敏
+
+            log.info("Token请求URL: {}", tokenUrl);
+            log.info("请求参数 - grant_type: password, username: {}", username);
+
+            httpPost.setEntity(new UrlEncodedFormEntity(params, StandardCharsets.UTF_8));
+
+            try (CloseableHttpResponse response = httpclient.execute(httpPost)) {
+                int statusCode = response.getStatusLine().getStatusCode();
+                log.info("Token接口响应状态码: {}", statusCode);
+
+                HttpEntity responseEntity = response.getEntity();
+                String result = EntityUtils.toString(responseEntity);
+                log.debug("Token接口原始响应: {}", result);
+
+                EntityUtils.consume(responseEntity);
+
+                if (statusCode == 200) {
+                    JSONObject json = JSON.parseObject(result);
+                    token = json.getString("access_token");
+                    int expiresIn = json.getIntValue("expires_in");
+
+                    log.info("成功获取Token，有效期: {} 秒", expiresIn);
+
+                    log.info("Token已缓存到Redis，key: {}，有效期: {} 秒", key, expiresIn);
+
+                    return token;
+                } else {
+                    log.error("获取Token失败，状态码: {}, 响应: {}", statusCode, result);
+                    return null;
+                }
+
+            } catch (Exception e) {
+                log.error("调用Token接口时发生异常", e);
+                return null;
+            }
+
+        } finally {
+            long endTime = System.currentTimeMillis();
+            log.info("获取Token操作完成，耗时: {} 毫秒", (endTime - startTime));
+        }
+    }
+
     private List<Object> initConnectAndSelectData() {
     private List<Object> initConnectAndSelectData() {
         Connection connection = null;
         Connection connection = null;
         PreparedStatement statement = null;
         PreparedStatement statement = null;
@@ -674,4 +814,24 @@ public class PjDeviceReportService {
             throw e;
             throw e;
         }
         }
     }
     }
+
+    public static Double toDouble(Object obj) {
+        if (obj == null) return null;
+        if (obj instanceof Double) {
+            return (Double) obj;
+        } else if (obj instanceof Long) {
+            return ((Long) obj).doubleValue();
+        } else if (obj instanceof Integer) {
+            return ((Integer) obj).doubleValue();
+        } else if (obj instanceof Number) {
+            return ((Number) obj).doubleValue();
+        } else if (obj instanceof String) {
+            try {
+                return Double.parseDouble((String) obj);
+            } catch (NumberFormatException e) {
+                return null;
+            }
+        }
+        return null;
+    }
 }
 }

+package com.iot.scheduler.utils;
+
+import java.io.*;
+import java.nio.file.*;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+
+public class DeviceDataStorage {
+
+    private static final String DATA_DIR = "/thingskit/iot-scheduler/reportData"; // 存储目录
+    private static final DateTimeFormatter TIMESTAMP_FORMAT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+
+    static {
+        // 确保存储目录存在
+        try {
+            Files.createDirectories(Paths.get(DATA_DIR));
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+    }
+
+    /**
+     * 更新设备数据（追加一条记录）
+     *
+     * @param deviceCode 设备编码
+     * @param timestamp  时间戳（毫秒）
+     * @param value      实时值
+     */
+    public synchronized void updateData(String deviceCode, long timestamp, double value) {
+        String fileName = getFileName(deviceCode);
+        String line = formatDataLine(timestamp, value) + System.lineSeparator();
+        try (FileWriter fw = new FileWriter(fileName, true);
+             BufferedWriter bw = new BufferedWriter(fw)) {
+            bw.write(line);
+        } catch (IOException e) {
+            System.err.println("写入设备数据失败: " + deviceCode);
+            e.printStackTrace();
+        }
+    }
+
+    /**
+     * 查询设备上次存储的数据（最新一条）
+     *
+     * @param deviceCode 设备编码
+     * @return DeviceRecord 对象，若无数据则返回 null
+     */
+    public DeviceRecord getLastRecord(String deviceCode) {
+        String fileName = getFileName(deviceCode);
+        File file = new File(fileName);
+        if (!file.exists()) {
+            return null;
+        }
+
+        String lastLine = null;
+        try (BufferedReader br = new BufferedReader(new FileReader(file))) {
+            String line;
+            while ((line = br.readLine()) != null) {
+                if (!line.trim().isEmpty()) {
+                    lastLine = line;
+                }
+            }
+        } catch (IOException e) {
+            System.err.println("读取设备数据失败: " + deviceCode);
+            e.printStackTrace();
+            return null;
+        }
+
+        if (lastLine == null) {
+            return null;
+        }
+        return parseRecord(lastLine);
+    }
+
+    /**
+     * 根据设备编码和时间戳查询指定时间的数据（精确匹配）
+     *
+     * @param deviceCode 设备编码
+     * @param timestamp  时间戳（毫秒）
+     * @return DeviceRecord 对象，若不存在则返回 null
+     */
+    public DeviceRecord getRecordByTimestamp(String deviceCode, long timestamp) {
+        String fileName = getFileName(deviceCode);
+        File file = new File(fileName);
+        if (!file.exists()) {
+            return null;
+        }
+
+        try (BufferedReader br = new BufferedReader(new FileReader(file))) {
+            String line;
+            while ((line = br.readLine()) != null) {
+                if (line.trim().isEmpty()) continue;
+                DeviceRecord record = parseRecord(line);
+                if (record != null && record.getTimestamp() == timestamp) {
+                    return record;
+                }
+            }
+        } catch (IOException e) {
+            System.err.println("读取设备数据失败: " + deviceCode);
+            e.printStackTrace();
+        }
+        return null;
+    }
+
+    // 获取设备文件路径
+    private String getFileName(String deviceCode) {
+        return DATA_DIR + File.separator + deviceCode + ".txt";
+    }
+
+    // 格式化存储行: 时间戳 | 数值
+    private String formatDataLine(long timestamp, double value) {
+        String timeStr = formatTimestamp(timestamp);
+        return timeStr + " | " + value + " | " + timestamp;
+    }
+
+    // 将时间戳转为可读格式
+    private String formatTimestamp(long timestamp) {
+        LocalDateTime dateTime = LocalDateTime.ofInstant(
+                java.time.Instant.ofEpochMilli(timestamp),
+                ZoneId.systemDefault()
+        );
+        return dateTime.format(TIMESTAMP_FORMAT);
+    }
+
+    // 解析记录行
+    private DeviceRecord parseRecord(String line) {
+        String[] parts = line.split("\\|");
+        if (parts.length < 3) {
+            // 兼容旧格式
+            if (parts.length == 2) {
+                try {
+                    String timestampStr = parts[0].trim();
+                    double value = Double.parseDouble(parts[1].trim());
+                    LocalDateTime ldt = LocalDateTime.parse(timestampStr, TIMESTAMP_FORMAT);
+                    long timestamp = ldt.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
+                    return new DeviceRecord(timestamp, value);
+                } catch (Exception e) {
+                    return null;
+                }
+            }
+            return null;
+        }
+
+        try {
+            long timestamp = Long.parseLong(parts[2].trim());
+            double value = Double.parseDouble(parts[1].trim());
+            return new DeviceRecord(timestamp, value);
+        } catch (NumberFormatException e) {
+            return null;
+        }
+    }
+
+    // 记录实体类
+    public static class DeviceRecord {
+        private final long timestamp; // 毫秒时间戳
+        private final double value;
+
+        public DeviceRecord(long timestamp, double value) {
+            this.timestamp = timestamp;
+            this.value = value;
+        }
+
+        public long getTimestamp() {
+            return timestamp;
+        }
+
+        public double getValue() {
+            return value;
+        }
+
+        @Override
+        public String toString() {
+            LocalDateTime dateTime = LocalDateTime.ofInstant(
+                    java.time.Instant.ofEpochMilli(timestamp),
+                    ZoneId.systemDefault()
+            );
+            return "DeviceRecord{" +
+                    "timestamp=" + timestamp +
+                    " (" + dateTime.format(TIMESTAMP_FORMAT) + ")" +
+                    ", value=" + value +
+                    '}';
+        }
+    }
+
+
+    public static void main(String[] args) {
+        DeviceDataStorage storage = new DeviceDataStorage();
+
+        // 模拟不同时间点的数据（时间戳作为参数传入）
+        long time1 = 1718352000000L; // 2024-06-14 16:00:00
+        long time2 = 1718438400000L; // 2024-06-15 16:00:00
+        long time3 = 1718524800000L; // 2024-06-16 16:00:00
+
+        // 更新设备数据（传入时间戳）
+        storage.updateData("device_001", time1, 23.5);
+        storage.updateData("device_001", time2, 24.1);
+        storage.updateData("device_001", time3, 25.3);
+
+        storage.updateData("device_002", time1, 68.3);
+        storage.updateData("device_002", time2, 69.1);
+
+        // 查询上次存储的数据
+        DeviceDataStorage.DeviceRecord lastRecord = storage.getLastRecord("device_001");
+        System.out.println("device_001 最后一条数据: " + lastRecord);
+
+        // 根据具体时间戳查询
+        DeviceDataStorage.DeviceRecord record = storage.getRecordByTimestamp("device_001", time2);
+        System.out.println("device_001 指定时间的数据: " + record);
+
+        // 查询不存在的设备
+        DeviceDataStorage.DeviceRecord notExist = storage.getLastRecord("device_999");
+        System.out.println("device_999: " + (notExist == null ? "无数据" : notExist));
+    }
+
+}

@@ -165,7 +165,8 @@ pj:
                   when 544 then '电'
                   when 544 then '电'
                   when 545 then '水' 
                   when 545 then '水' 
                   when 546 then '天然气' end AS energyType,
                   when 546 then '天然气' end AS energyType,
-                  COALESCE (tkl.dbl_v, tkl.long_v, 0) AS deviceValue
+                  COALESCE (tkl.dbl_v, tkl.long_v, 0) AS deviceValue,
+                  tkl.ts
                 FROM
                 FROM
                   device de
                   device de
                   LEFT JOIN tk_organization tko ON de.organization_id = tko.id
                   LEFT JOIN tk_organization tko ON de.organization_id = tko.id