潘集

中建材设备上报

@@ -20,6 +20,8 @@ public class HealthController {
     private ShzzDevicePullService shzzDevicePullService;
     @Resource
     private HnHsqDeviceReportService hnHsqDeviceReportService;
+    @Resource
+    private PjDeviceReportService pjDeviceReportService;
 
     @GetMapping("/health")
     public String health() {
@@ -45,4 +47,8 @@ public class HealthController {
         shzzDevicePullService.pullDeviceAndPushToIot();
     }
 
+    @GetMapping("/test")
+    public void test() {
+        pjDeviceReportService.reportZongJianCaiEnergy();
+    }
 }

@@ -80,8 +80,11 @@ public class PjDeviceReportService {
     @Value("${pj.third.energyFormId:t6937bc62e4332f00072a0849}")
     private String energyFormId;
 
-    @Value("${pj.third.selectEnergySql:SELECT * FROM ts_kv_dictionary;}")
-    private String selectEnergySql;
+    @Value("${pj.third.selectJinHongEnergySql:SELECT * FROM ts_kv_dictionary;}")
+    private String selectJinHongEnergySql;
+
+    @Value("${pj.third.selectZongJianCaiEnergySql:SELECT * FROM ts_kv_dictionary;}")
+    private String selectZongJianCaiEnergySql;
 
     @Resource
     private RedisTemplate<String, String> redisTemplate;
@@ -250,16 +253,16 @@ public class PjDeviceReportService {
     }
 
     /**
-     * 企业设备能耗上报
+     * 金宏能耗上报
      */
-    public void batchReportEnergyEnterprise() {
+    public void reportJinHongEnergy() {
         log.info("开始执行企业设备能耗上报任务");
         long startTime = System.currentTimeMillis();
 
         try {
             // 1. 查询需要同步的数据
             log.info("步骤1: 开始查询需要同步的设备数据");
-            List<Object> needSyncDataList = initConnectAndSelectData2();
+            List<Object> needSyncDataList = initJinHongConnectAndSelectData();
 
             if (CollectionUtils.isEmpty(needSyncDataList)) {
                 log.info("未查询到需要同步的设备数据，任务结束");
@@ -416,6 +419,162 @@ public class PjDeviceReportService {
         }
     }
 
+    /**
+     * 金宏能耗上报
+     */
+    public void reportZongJianCaiEnergy() {
+        log.info("开始执行企业设备能耗上报任务");
+        long startTime = System.currentTimeMillis();
+
+        try {
+            // 1. 查询需要同步的数据
+            log.info("步骤1: 开始查询需要同步的设备数据");
+            List<Object> needSyncDataList = initZongJianCaiConnectAndSelectData();
+
+            if (CollectionUtils.isEmpty(needSyncDataList)) {
+                log.info("未查询到需要同步的设备数据，任务结束");
+                return;
+            }
+
+            log.info("共查询到 {} 条设备数据需要同步", needSyncDataList.size());
+
+            // 2. 准备上报数据
+            log.info("步骤2: 开始准备上报数据");
+
+            CloseableHttpClient httpclient = HttpClients.createDefault();
+            HttpPost httpPost = new HttpPost(reportUrl);
+
+            // 3. 获取并设置Token
+            log.info("步骤3: 获取访问令牌");
+            String token = getTokenNoCache();
+            if (StringUtils.isBlank(token)) {
+                log.error("获取访问令牌失败，无法进行数据上报");
+                return;
+            }
+
+            httpPost.setHeader("Authorization", "Bearer " + token);
+            httpPost.setHeader("Content-Type", "application/json; charset=utf-8");
+            log.info("请求URL: {}, Token已设置", reportUrl);
+
+            // 4. 构建上报数据
+            log.info("步骤4: 构建上报数据结构");
+            List<Map<String, Object>> dataList = new ArrayList<>(needSyncDataList.size());
+            List<Map<String, Object>> saveDataList = new ArrayList<>(needSyncDataList.size());
+            int successCount = 0;
+            int errorCount = 0;
+            DeviceDataStorage storage = new DeviceDataStorage();
+            for (int i = 0; i < needSyncDataList.size(); i++) {
+                try {
+                    Object needSyncData = needSyncDataList.get(i);
+                    List<Object> dataObject = (ArrayList) needSyncData;
+
+                    if (dataObject.size() < 4) {
+                        log.warn("第 {} 条数据格式不正确，期望至少4个字段，实际: {}", i + 1, dataObject.size());
+                        errorCount++;
+                        continue;
+                    }
+
+                    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("deviceName", dataObject.get(1));
+                    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)|| "水".equals(energyType)) {
+                        data.put("energyValue", energyValue);
+                        data.put("timeSpan", "日");
+                        data.put("collectTime", toYyyyMmDd(ts));
+                    } else {
+                        log.warn("第 {} 条数据,设备:{} 物模型数据错误!", i + 1, deviceCode);
+                        continue;
+                    }
+
+                    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);
+
+                    successCount++;
+
+                    // 每100条记录输出一次进度
+                    if (successCount % 100 == 0) {
+                        log.info("已成功构建 {} 条上报数据", successCount);
+                    }
+
+                } catch (Exception e) {
+                    log.error("处理第 {} 条数据时发生异常", i + 1, e);
+                    errorCount++;
+                }
+            }
+
+            log.info("数据构建完成，成功: {} 条，失败: {} 条", successCount, errorCount);
+
+            if (CollectionUtils.isEmpty(dataList)) {
+                log.error("没有成功构建任何上报数据，任务结束");
+                return;
+            }
+
+            // 5. 构建请求体
+            log.info("步骤5: 构建请求体");
+            Map<String, Object> sendData = new HashMap<>(3);
+            sendData.put("formId", energyFormId);
+            List<String> uniqueKeyList = Arrays.asList(energyUniqueKeys.split(","));
+            sendData.put("uniqueKeys", uniqueKeyList);
+            sendData.put("datas", dataList);
+
+            String json = JSON.toJSONString(sendData);
+            log.debug("请求体JSON数据: {}", json);
+            log.info("请求体大小: {} 字符", json.length());
+
+            httpPost.setEntity(new StringEntity(json, StandardCharsets.UTF_8));
+
+            // 6. 发送请求
+            log.info("步骤6: 发送HTTP请求到第三方平台");
+            try (CloseableHttpResponse response = httpclient.execute(httpPost)) {
+                int statusCode = response.getStatusLine().getStatusCode();
+                String result = EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8);
+
+                log.info("HTTP响应状态码: {}", statusCode);
+                log.info("HTTP响应内容: {}", result);
+
+                if (statusCode == 200) {
+                    JSONObject jsonResult = JSON.parseObject(result);
+                    if (jsonResult != null && "true".equalsIgnoreCase(jsonResult.getString("success"))) {
+                        log.info("数据上报成功！共上报 {} 条设备数据", dataList.size());
+                        //记录上报数据
+                        saveReportData(saveDataList);
+                    } else {
+                        log.error("数据上报失败！响应状态异常: {}", result);
+                    }
+                } else {
+                    log.error("HTTP请求失败，状态码: {}", statusCode);
+                }
+
+            } catch (Exception e) {
+                log.error("发送HTTP请求时发生异常", e);
+            }
+
+        } finally {
+            long endTime = System.currentTimeMillis();
+            log.info("企业设备能耗上报任务执行完成，总耗时: {} 毫秒", (endTime - startTime));
+        }
+    }
+
     private void saveReportData(List<Map<String, Object>> dataList) {
         if (CollectionUtils.isEmpty(dataList)) {
             return;
@@ -654,7 +813,90 @@ public class PjDeviceReportService {
         return resultList;
     }
 
-    private List<Object> initConnectAndSelectData2() {
+    private List<Object> initJinHongConnectAndSelectData() {
+        Connection connection = null;
+        PreparedStatement statement = null;
+        ResultSet resultSet = null;
+        HikariDataSource dataSource = null;
+        List<Object> resultList = new ArrayList<>();
+
+        log.info("开始连接数据库，URL: {}", jdbcUrl);
+
+        try {
+            HikariConfig config = new HikariConfig();
+            config.setJdbcUrl(jdbcUrl);
+            config.setUsername(jdbcUserName);
+            config.setPassword(jdbcPassword);
+            config.setDriverClassName("org.postgresql.Driver");
+            config.setMaximumPoolSize(5);
+            config.setMinimumIdle(5);
+            config.setConnectionTimeout(60000);
+            config.setConnectionTestQuery("SELECT 1");
+
+            dataSource = new HikariDataSource(config);
+            log.info("Hikari连接池配置完成");
+
+            connection = dataSource.getConnection();
+            log.info("数据库连接成功");
+
+            statement = connection.prepareStatement(selectJinHongEnergySql);
+            log.info("执行SQL查询: {}", selectSql);
+
+            resultSet = statement.executeQuery();
+            ResultSetMetaData metaData = resultSet.getMetaData();
+            int columnCount = metaData.getColumnCount();
+            log.info("查询结果集元数据获取成功，共{}列", columnCount);
+
+            int rowCount = 0;
+            while (resultSet.next()) {
+                List<Object> result = new ArrayList<>(columnCount);
+                for (int index = 1; index <= columnCount; index++) {
+                    int columnType = metaData.getColumnType(index);
+                    Object value = getTypedValue(resultSet, index, columnType);
+                    result.add(value);
+                }
+                resultList.add(result);
+                rowCount++;
+
+                // 每处理1000行记录一次日志
+                if (rowCount % 1000 == 0) {
+                    log.info("已处理{}行数据", rowCount);
+                }
+            }
+
+            log.info("数据查询完成，共获取{}行数据", rowCount);
+
+        } catch (SQLException e) {
+            log.error("数据库操作异常，URL: {}, 用户名: {}", jdbcUrl, jdbcUserName, e);
+        } catch (Exception e) {
+            log.error("初始化数据库连接或查询数据时发生异常", e);
+        } finally {
+            // 释放资源
+            try {
+                if (resultSet != null) resultSet.close();
+                if (statement != null) statement.close();
+                if (connection != null) connection.close();
+                log.info("数据库连接资源已释放");
+            } catch (SQLException e) {
+                log.error("关闭数据库资源时发生异常", e);
+            }
+
+            if (dataSource != null) {
+                try {
+                    dataSource.close();
+                    log.info("HikariDataSource连接池已关闭");
+                } catch (Exception e) {
+                    log.error("关闭HikariDataSource连接池时发生异常", e);
+                }
+            }
+        }
+
+        log.info("数据库操作完成，返回{}条记录", resultList.size());
+        return resultList;
+    }
+
+
+    private List<Object> initZongJianCaiConnectAndSelectData() {
         Connection connection = null;
         PreparedStatement statement = null;
         ResultSet resultSet = null;
@@ -680,7 +922,7 @@ public class PjDeviceReportService {
             connection = dataSource.getConnection();
             log.info("数据库连接成功");
 
-            statement = connection.prepareStatement(selectEnergySql);
+            statement = connection.prepareStatement(selectZongJianCaiEnergySql);
             log.info("执行SQL查询: {}", selectSql);
 
             resultSet = statement.executeQuery();

@@ -39,7 +39,8 @@ public class PanJiZoneScheduler extends AbstractZoneScheduler {
         logStart(taskName);
         try {
             log.info("[{}] PJ pushing devices...", getZoneName());
-            pjDeviceReportService.batchReportEnergyEnterprise();
+            pjDeviceReportService.reportJinHongEnergy();
+            pjDeviceReportService.reportZongJianCaiEnergy();
             Thread.sleep(1000);
         } catch (Exception e) {
             logError(taskName, e);

@@ -157,7 +157,7 @@ pj:
                   AND de.device_profile_id = '0418b010-f01d-11f0-9cb8-e3376d1e7978';"
     energyFormId: "t69393ccde4332f00072a08e8"
     energyUniqueKeys: "entName,deviceCode,timeSpan,collectTime,energyType"
-    selectEnergySql: "SELECT
+    selectJinHongEnergySql: "SELECT
                   tko.name AS entName,
                   de.name AS deviceName,
                   dc.credentials_id AS deviceCode,
@@ -175,8 +175,28 @@ pj:
                   LEFT JOIN device_credentials dc ON de.id = dc.device_id
                 WHERE
                   de.tenant_id = '0414df80-f01d-11f0-9cb8-e3376d1e7978'
+                  and de.organization_id='12eade61-d3aa-47d6-a71b-1d1384dc4c30'
+                  AND de.device_profile_id IN ('b1845b90-6634-11f1-9cb8-e3376d1e7978', '8c195900-6634-11f1-9cb8-e3376d1e7978', '5eb07070-6634-11f1-9cb8-e3376d1e7978');"
+    selectZongJianCaiEnergySql: "SELECT
+                  tko.name AS entName,
+                  de.name AS deviceName,
+                  dc.credentials_id AS deviceCode,
+                  case tkl.key
+                  when 544 then '电'
+                  when 545 then '水'
+                  when 546 then '天然气' end AS energyType,
+                  COALESCE (tkl.dbl_v, tkl.long_v, 0) AS deviceValue,
+                  tkl.ts
+                FROM
+                  device de
+                  LEFT JOIN tk_organization tko ON de.organization_id = tko.id
+                  LEFT JOIN ts_kv_latest tkl ON de.id = tkl.entity_id
+                  AND tkl.KEY IN (544, 545, 546)
+                  LEFT JOIN device_credentials dc ON de.id = dc.device_id
+                WHERE
+                  de.tenant_id = '0414df80-f01d-11f0-9cb8-e3376d1e7978'
+                  and de.organization_id='00347cce-eb9e-4b6d-897c-ae39dbdf53c7'
                   AND de.device_profile_id IN ('b1845b90-6634-11f1-9cb8-e3376d1e7978', '8c195900-6634-11f1-9cb8-e3376d1e7978', '5eb07070-6634-11f1-9cb8-e3376d1e7978');"
-
 wuwei:
   third:
     domain: "http://60.169.172.121:1804/report-property"