潘集经开区园区设备上传

+package com.iot.scheduler.service;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
+import com.zaxxer.hikari.HikariConfig;
+import com.zaxxer.hikari.HikariDataSource;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.http.HttpEntity;
+import org.apache.http.NameValuePair;
+import org.apache.http.client.entity.UrlEncodedFormEntity;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.message.BasicNameValuePair;
+import org.apache.http.util.EntityUtils;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.stereotype.Service;
+
+import java.nio.charset.StandardCharsets;
+import java.sql.*;
+import java.sql.Date;
+import java.text.SimpleDateFormat;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 潘集设备上报
+ */
+@Slf4j
+@Service
+public class PjDeviceReportService {
+
+
+    @Value("${pj.third.reportUrl:http://111.39.171.168:12280/mainApi/formEngine/formData/batchAddOrUpate}")
+    private String reportUrl;
+
+    @Value("${pj.third.tokenUrl:http://111.39.171.168:15555/auth/oauth/token}")
+    private String tokenUrl;
+
+    @Value("${pj.third.username:adminhnjt}")
+    private String username;
+
+    @Value("${pj.third.password:admin@1234}")
+    private String password;
+
+    @Value("${pj.third.formId:t6937bc62e4332f00072a0849}")
+    private String formId;
+
+    @Value("${pj.third.uniqueKeys:entName,deviceCode}")
+    private String uniqueKeys;
+
+    @Value("${pj.third.jdbcUrl:jdbc:postgresql://106.15.73.210:5433/thingskit}")
+    private String jdbcUrl;
+
+    @Value("${pj.third.jdbcUserName:postgres}")
+    private String jdbcUserName;
+
+    @Value("${pj.third.jdbcPassword:postgres}")
+    private String jdbcPassword;
+
+    @Value("${pj.third.selectSql:SELECT * FROM ts_kv_dictionary;}")
+    private String selectSql;
+
+    @Resource
+    private RedisTemplate<String, String> redisTemplate;
+
+    /**
+     * 企业设备上报
+     */
+    public void batchReportEnterprise() {
+        log.info("开始执行企业设备上报任务");
+        long startTime = System.currentTimeMillis();
+
+        try {
+            // 1. 查询需要同步的数据
+            log.info("步骤1: 开始查询需要同步的设备数据");
+            List<Object> needSyncDataList = initConnectAndSelectData();
+
+            if (CollectionUtils.isEmpty(needSyncDataList)) {
+                log.info("未查询到需要同步的设备数据，任务结束");
+                return;
+            }
+
+            log.info("共查询到 {} 条设备数据需要同步", needSyncDataList.size());
+
+            // 2. 准备上报数据
+            log.info("步骤2: 开始准备上报数据");
+            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+            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("reportTime", nowTime);
+
+                    // 处理状态字段
+                    Object statusObj = dataObject.get(3);
+                    String statusStr = "在线";
+                    try {
+                        if (statusObj != null) {
+                            long status = Long.parseLong(statusObj.toString());
+                            if (status == 0) {
+                                statusStr = "离线";
+                            }
+                        }
+                    } catch (NumberFormatException e) {
+                        log.warn("第 {} 条数据状态字段格式错误: {}", i + 1, statusObj);
+                        statusStr = "未知";
+                    }
+
+                    data.put("status", statusStr);
+                    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", formId);
+            List<String> uniqueKeyList = Arrays.asList(uniqueKeys.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
+     *
+     * @return
+     */
+    private String getToken() {
+        log.info("开始获取访问令牌");
+        long startTime = System.currentTimeMillis();
+        String key = "pjjkq_report_token";
+
+        try {
+            // 1. 尝试从Redis获取缓存token
+            log.info("检查Redis中是否有缓存的Token，key: {}", key);
+            String token = redisTemplate.opsForValue().get(key);
+            Long expire = redisTemplate.getExpire(key);
+
+            if (StringUtils.isNotBlank(token) && expire != null && expire > 0) {
+                log.info("从Redis缓存中获取Token成功，剩余有效期: {} 秒", expire);
+                return token;
+            }
+
+            log.info("Redis中没有有效的Token缓存，开始请求新的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);
+
+                    // 3. 缓存token到Redis
+                    redisTemplate.opsForValue().set(key, token);
+                    redisTemplate.expire(key, expiresIn, TimeUnit.SECONDS);
+
+                    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() {
+        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(selectSql);
+            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 {
+        Object value;
+
+        try {
+            switch (sqlType) {
+                case Types.BIT:
+                case Types.BOOLEAN:
+                    value = rs.getBoolean(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.TINYINT:
+                case Types.SMALLINT:
+                case Types.INTEGER:
+                    value = rs.getInt(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.BIGINT:
+                    value = rs.getLong(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.FLOAT:
+                case Types.REAL:
+                    value = rs.getFloat(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.DOUBLE:
+                    value = rs.getDouble(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.NUMERIC:
+                case Types.DECIMAL:
+                    value = rs.getBigDecimal(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.CHAR:
+                case Types.VARCHAR:
+                case Types.LONGVARCHAR:
+                case Types.NCHAR:
+                case Types.NVARCHAR:
+                case Types.LONGNVARCHAR:
+                    value = rs.getString(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.DATE:
+                    Date date = rs.getDate(index);
+                    return date != null ? date.toLocalDate() : null;
+
+                case Types.TIME:
+                    Time time = rs.getTime(index);
+                    return time != null ? time.toLocalTime() : null;
+
+                case Types.TIMESTAMP:
+                    Timestamp timestamp = rs.getTimestamp(index);
+                    return timestamp != null ? timestamp.toLocalDateTime() : null;
+
+                case Types.BINARY:
+                case Types.VARBINARY:
+                case Types.LONGVARBINARY:
+                    value = rs.getBytes(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.BLOB:
+                    value = rs.getBlob(index);
+                    return rs.wasNull() ? null : value;
+
+                case Types.CLOB:
+                    value = rs.getClob(index);
+                    return rs.wasNull() ? null : value;
+
+                default:
+                    value = rs.getObject(index);
+                    return rs.wasNull() ? null : value;
+            }
+        } catch (SQLException e) {
+            log.error("获取结果集第{}列数据时发生异常，数据类型: {}", index, sqlType, e);
+            throw e;
+        }
+    }
+}

@@ -52,7 +52,7 @@ public class ChizhouZoneScheduler extends AbstractZoneScheduler {
         logStart(taskName);
         logStart(taskName);
         try {
         try {
             log.info("[{}] Simulating pushing devices...", getZoneName());
             log.info("[{}] Simulating pushing devices...", getZoneName());
-            hnDeviceReportService.deviceReport();
+//            hnDeviceReportService.deviceReport();
             Thread.sleep(1000);
             Thread.sleep(1000);
         } catch (Exception e) {
         } catch (Exception e) {
             logError(taskName, e);
             logError(taskName, e);

+package com.iot.scheduler.zone;
+
+import com.iot.scheduler.service.PjDeviceReportService;
+import com.iot.scheduler.task.AbstractZoneScheduler;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+public class PanJiZoneScheduler extends AbstractZoneScheduler {
+    @Resource
+    private PjDeviceReportService pjDeviceReportService;
+
+    @Override
+    protected String getZoneName() {
+        return "PanJi (潘集经开区)";
+    }
+
+    @Scheduled(cron = "${scheduler.pj.push:0 0/1 * * * ?}")
+    public void pushHsqDevicesToThirdParty() {
+        String taskName = "PJ Devices (IoT -> 3rd Party)";
+        logStart(taskName);
+        try {
+            log.info("[{}] PJ pushing devices...", getZoneName());
+            pjDeviceReportService.batchReportEnterprise();
+            Thread.sleep(1000);
+        } catch (Exception e) {
+            logError(taskName, e);
+        } finally {
+            logEnd(taskName);
+        }
+    }
+}

@@ -35,6 +35,8 @@ scheduler:
     push: "0 0/10 * * * ?"
     push: "0 0/10 * * * ?"
   hsq:
   hsq:
     push: "0 0 0/1 * * ?"
     push: "0 0 0/1 * * ?"
+  pj:
+    push: "0 0/1 * * * ?"
 
 
 hn:
 hn:
   third:
   third:
@@ -72,6 +74,28 @@ hnhsq:
         from ts_kv kv
         from ts_kv kv
         INNER JOIN device d on d.id = kv.entity_id and kv.key=61
         INNER JOIN device d on d.id = kv.entity_id and kv.key=61
         where tenant_id='5b6a05a0-f020-11f0-9cb8-e3376d1e7978'"
         where tenant_id='5b6a05a0-f020-11f0-9cb8-e3376d1e7978'"
+pj:
+  third:
+    reportUrl: "http://111.39.171.168:12280/mainApi/formEngine/formData/batchAddOrUpate"
+    tokenUrl: "http://111.39.171.168:15555/auth/oauth/token"
+    username: "adminhnjt"
+    password: "admin@1234"
+    formId: "t6937bc62e4332f00072a0849"
+    uniqueKeys: "entName,deviceCode"
+    jdbcUrl: "jdbc:postgresql://106.15.73.210:5433/thingskit"
+    jdbcUserName: "postgres"
+    jdbcPassword: "postgres"
+    selectSql: "SELECT
+                  tko.name AS entName,
+                  de.name AS deviceName,
+                  de.sn AS deviceCode,
+                  tkl.long_v AS status
+                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 = 61
+                WHERE de.tenant_id = '0414df80-f01d-11f0-9cb8-e3376d1e7978'
+                AND de.device_profile_id = '0418b010-f01d-11f0-9cb8-e3376d1e7978';"
 
 
 sh:
 sh:
   iot:
   iot: