潘集设备上报

@@ -67,6 +67,15 @@ public class PjDeviceReportService {
     @Value("${pj.third.selectSql:SELECT * FROM ts_kv_dictionary;}")
     @Value("${pj.third.selectSql:SELECT * FROM ts_kv_dictionary;}")
     private String selectSql;
     private String selectSql;
 
 
+    @Value("${pj.third.energyUniqueKeys:entName,deviceCode}")
+    private String energyUniqueKeys;
+
+    @Value("${pj.third.energyFormId:t6937bc62e4332f00072a0849}")
+    private String energyFormId;
+
+    @Value("${pj.third.selectEnergySql:SELECT * FROM ts_kv_dictionary;}")
+    private String selectEnergySql;
+
     @Resource
     @Resource
     private RedisTemplate<String, String> redisTemplate;
     private RedisTemplate<String, String> redisTemplate;
 
 
@@ -215,6 +224,136 @@ public class PjDeviceReportService {
     }
     }
 
 
     /**
     /**
+     * 企业设备能耗上报
+     */
+    public void batchReportEnergyEnterprise() {
+        log.info("开始执行企业设备能耗上报任务");
+        long startTime = System.currentTimeMillis();
+
+        try {
+            // 1. 查询需要同步的数据
+            log.info("步骤1: 开始查询需要同步的设备数据");
+            List<Object> needSyncDataList = initConnectAndSelectData2();
+
+            if (CollectionUtils.isEmpty(needSyncDataList)) {
+                log.info("未查询到需要同步的设备数据，任务结束");
+                return;
+            }
+
+            log.info("共查询到 {} 条设备数据需要同步", needSyncDataList.size());
+
+            // 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();
+            HttpPost httpPost = new HttpPost(reportUrl);
+
+            // 3. 获取并设置Token
+            log.info("步骤3: 获取访问令牌");
+            String token = getToken();
+            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());
+            int successCount = 0;
+            int errorCount = 0;
+
+            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<>();
+                    data.put("entName", dataObject.get(0));
+                    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);
+                    dataList.add(data);
+                    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());
+                    } 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));
+        }
+    }
+
+    /**
      * 获取token
      * 获取token
      *
      *
      * @return
      * @return
@@ -375,6 +514,88 @@ public class PjDeviceReportService {
         return resultList;
         return resultList;
     }
     }
 
 
+    private List<Object> initConnectAndSelectData2() {
+        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(selectEnergySql);
+            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 Object getTypedValue(ResultSet rs, int index, int sqlType) throws SQLException {
     private Object getTypedValue(ResultSet rs, int index, int sqlType) throws SQLException {
         Object value;
         Object value;
 
 

@@ -32,4 +32,19 @@ public class PanJiZoneScheduler extends AbstractZoneScheduler {
             logEnd(taskName);
             logEnd(taskName);
         }
         }
     }
     }
+
+    @Scheduled(cron = "${scheduler.pj.energyPush:0 0 0 * * ?}")
+    public void pushPjEnergyDevicesToThirdParty() {
+        String taskName = "PJ Devices (IoT -> 3rd Party)";
+        logStart(taskName);
+        try {
+            log.info("[{}] PJ pushing devices...", getZoneName());
+            pjDeviceReportService.batchReportEnergyEnterprise();
+            Thread.sleep(1000);
+        } catch (Exception e) {
+            logError(taskName, e);
+        } finally {
+            logEnd(taskName);
+        }
+    }
 }
 }

@@ -37,6 +37,7 @@ scheduler:
     push: "0 0 0/1 * * ?"
     push: "0 0 0/1 * * ?"
   pj:
   pj:
     push: "0 0/10 * * * ?"
     push: "0 0/10 * * * ?"
+    energyPush: "0 0 0 * * ?"
   jmdq:
   jmdq:
     pull: "0 0/10 * * * ?"
     pull: "0 0/10 * * * ?"
     push: "0 0/10 * * * ?"
     push: "0 0/10 * * * ?"
@@ -154,6 +155,26 @@ pj:
                 WHERE
                 WHERE
                   de.tenant_id = '0414df80-f01d-11f0-9cb8-e3376d1e7978'
                   de.tenant_id = '0414df80-f01d-11f0-9cb8-e3376d1e7978'
                   AND de.device_profile_id = '0418b010-f01d-11f0-9cb8-e3376d1e7978';"
                   AND de.device_profile_id = '0418b010-f01d-11f0-9cb8-e3376d1e7978';"
+    energyFormId: "t69393ccde4332f00072a08e8"
+    energyUniqueKeys: "entName,deviceCode,timeSpan,collectTime,energyType"
+    selectEnergySql: "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
+                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.device_profile_id IN ('b1845b90-6634-11f1-9cb8-e3376d1e7978', '8c195900-6634-11f1-9cb8-e3376d1e7978', '5eb07070-6634-11f1-9cb8-e3376d1e7978');"
 
 
 wuwei:
 wuwei:
   third:
   third: