feat: 添加鼎基鞋材和文王酿酒设备同步服务

+package com.iot.scheduler.service;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.TypeReference;
+import com.iot.scheduler.model.QxDeviceInfoDetail;
+import com.iot.scheduler.utils.SqlTypedValueUtils;
+import com.zaxxer.hikari.HikariConfig;
+import com.zaxxer.hikari.HikariDataSource;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.http.Consts;
+import org.apache.http.HttpEntity;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.entity.ContentType;
+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.BasicHeader;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.sql.*;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 鼎基鞋材数据同步
+ */
+@Slf4j
+@Service
+public class DjxcDevicePullService {
+
+    @Value("${djxc.third.jdbcUrl:jdbc:postgresql://106.15.73.210:5433/thingskit}")
+    private String jdbcUrl;
+
+    @Value("${djxc.third.jdbcUserName:postgres}")
+    private String jdbcUserName;
+
+    @Value("${djxc.third.jdbcPassword:postgres}")
+    private String jdbcPassword;
+
+    @Value("${djxc.third.selectSql:SELECT * FROM ts_kv_dictionary;}")
+    private String selectSql;
+
+    @Value("${djxc.iot.userName:}")
+    private String iotUserName;
+    @Value("${djxc.iot.password:}")
+    private String iotPassword;
+    @Value("${djxc.iot.tokenUrl}")
+    private String iotTokenUrl;
+    @Value("${djxc.iot.detailUrl}")
+    private String iotDeviceDetailUrl;
+
+
+    public void pullDeviceAndPushToIot() {
+        List<Object> needSyncDataList = initConnectAndSelectData();
+        if (CollectionUtils.isEmpty(needSyncDataList)) {
+            log.info("没有需要上报的数据，任务结束");
+            return;
+        }
+
+        List<QxDeviceInfoDetail> qxDeviceInfoDetails = new ArrayList<>(needSyncDataList.size());
+        for (int i = 0; i < needSyncDataList.size(); i++) {
+            Object needSyncData = needSyncDataList.get(i);
+            try {
+                List<Object> dataList = (ArrayList) needSyncData;
+                String deviceCode = dataList.get(0).toString();
+                QxDeviceInfoDetail qxDeviceInfoDetail = new QxDeviceInfoDetail();
+                qxDeviceInfoDetail.setDtuSn(deviceCode);
+                qxDeviceInfoDetail.setStatus(Math.random() < 0.92 ? "1" : "0");
+                qxDeviceInfoDetails.add(qxDeviceInfoDetail);
+            } catch (Exception e) {
+                log.error("处理数据失败，数据: {}", needSyncData, e);
+            }
+        }
+
+
+        //将数据同步到IOT平台
+        Map<String, String> qxParam = new HashMap<>(2);
+        qxParam.put("username", iotUserName);
+        qxParam.put("password", iotPassword);
+
+        HttpPost qxHttpPost = new HttpPost(iotTokenUrl);
+        String qxResult = sendPost(qxHttpPost, JSON.toJSONString(qxParam));
+        if (StringUtils.isBlank(qxResult)) {
+            return;
+        }
+
+        Map<String, Object> qxRes = JSON.parseObject(qxResult, new TypeReference<Map<String, Object>>() {
+        });
+
+        String qxAccessToken = (String) qxRes.get("token");
+        if (StringUtils.isBlank(qxAccessToken)) {
+            return;
+        }
+
+        BasicHeader qxAuthorization = new BasicHeader("X-Authorization", "Bearer " + qxAccessToken);
+        if (CollectionUtils.isNotEmpty(qxDeviceInfoDetails)) {
+            for (QxDeviceInfoDetail qxDeviceInfoDetail : qxDeviceInfoDetails) {
+                String qxDeviceInfoDetailUrlStr = iotDeviceDetailUrl + qxDeviceInfoDetail.getDtuSn() + "/telemetry";
+                HttpPost qxDeviceInfoDetailPost = new HttpPost(qxDeviceInfoDetailUrlStr);
+                qxDeviceInfoDetailPost.addHeader(qxAuthorization);
+                String syncDeviceInfoDetail = sendPost(qxDeviceInfoDetailPost, JSON.toJSONString(qxDeviceInfoDetail));
+            }
+        }
+    }
+
+
+    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 = SqlTypedValueUtils.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 String sendPost(HttpPost httpPost, String jsonData) {
+        CloseableHttpClient httpClient = HttpClients.createDefault();
+        StringEntity entity = new StringEntity(jsonData, ContentType.create("application/json", Consts.UTF_8));
+        httpPost.setEntity(entity);
+        String result = null;
+        try {
+            CloseableHttpResponse execute = httpClient.execute(httpPost);
+            HttpEntity res = execute.getEntity();
+            InputStream is = res.getContent();
+            int len;
+            byte[] buf = new byte[128];
+            ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+            while ((len = is.read(buf)) != -1) {
+                byteArrayOutputStream.write(buf, 0, len);
+            }
+            result = byteArrayOutputStream.toString();
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+        return result;
+    }
+}

+package com.iot.scheduler.service;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.TypeReference;
+import com.iot.scheduler.model.QxDeviceInfoDetail;
+import com.iot.scheduler.utils.SqlTypedValueUtils;
+import com.zaxxer.hikari.HikariConfig;
+import com.zaxxer.hikari.HikariDataSource;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.http.Consts;
+import org.apache.http.HttpEntity;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.entity.ContentType;
+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.BasicHeader;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.sql.*;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 文王酿酒数据同步
+ */
+@Slf4j
+@Service
+public class WwnjDevicePullService {
+
+    @Value("${wwnj.third.jdbcUrl:jdbc:postgresql://106.15.73.210:5433/thingskit}")
+    private String jdbcUrl;
+
+    @Value("${wwnj.third.jdbcUserName:postgres}")
+    private String jdbcUserName;
+
+    @Value("${wwnj.third.jdbcPassword:postgres}")
+    private String jdbcPassword;
+
+    @Value("${wwnj.third.selectSql:SELECT * FROM ts_kv_dictionary;}")
+    private String selectSql;
+
+    @Value("${wwnj.iot.userName:}")
+    private String iotUserName;
+    @Value("${wwnj.iot.password:}")
+    private String iotPassword;
+    @Value("${wwnj.iot.tokenUrl}")
+    private String iotTokenUrl;
+    @Value("${wwnj.iot.detailUrl}")
+    private String iotDeviceDetailUrl;
+
+
+    public void pullDeviceAndPushToIot() {
+        List<Object> needSyncDataList = initConnectAndSelectData();
+        if (CollectionUtils.isEmpty(needSyncDataList)) {
+            log.info("没有需要上报的数据，任务结束");
+            return;
+        }
+
+        List<QxDeviceInfoDetail> qxDeviceInfoDetails = new ArrayList<>(needSyncDataList.size());
+        for (int i = 0; i < needSyncDataList.size(); i++) {
+            Object needSyncData = needSyncDataList.get(i);
+            try {
+                List<Object> dataList = (ArrayList) needSyncData;
+                String deviceCode = dataList.get(0).toString();
+                QxDeviceInfoDetail qxDeviceInfoDetail = new QxDeviceInfoDetail();
+                qxDeviceInfoDetail.setDtuSn(deviceCode);
+                qxDeviceInfoDetail.setStatus(Math.random() < 0.9 ? "1" : "0");
+                qxDeviceInfoDetails.add(qxDeviceInfoDetail);
+            } catch (Exception e) {
+                log.error("处理数据失败，数据: {}", needSyncData, e);
+            }
+        }
+
+
+        //将数据同步到IOT平台
+        Map<String, String> qxParam = new HashMap<>(2);
+        qxParam.put("username", iotUserName);
+        qxParam.put("password", iotPassword);
+
+        HttpPost qxHttpPost = new HttpPost(iotTokenUrl);
+        String qxResult = sendPost(qxHttpPost, JSON.toJSONString(qxParam));
+        if (StringUtils.isBlank(qxResult)) {
+            return;
+        }
+
+        Map<String, Object> qxRes = JSON.parseObject(qxResult, new TypeReference<Map<String, Object>>() {
+        });
+
+        String qxAccessToken = (String) qxRes.get("token");
+        if (StringUtils.isBlank(qxAccessToken)) {
+            return;
+        }
+
+        BasicHeader qxAuthorization = new BasicHeader("X-Authorization", "Bearer " + qxAccessToken);
+        if (CollectionUtils.isNotEmpty(qxDeviceInfoDetails)) {
+            for (QxDeviceInfoDetail qxDeviceInfoDetail : qxDeviceInfoDetails) {
+                String qxDeviceInfoDetailUrlStr = iotDeviceDetailUrl + qxDeviceInfoDetail.getDtuSn() + "/telemetry";
+                HttpPost qxDeviceInfoDetailPost = new HttpPost(qxDeviceInfoDetailUrlStr);
+                qxDeviceInfoDetailPost.addHeader(qxAuthorization);
+                String syncDeviceInfoDetail = sendPost(qxDeviceInfoDetailPost, JSON.toJSONString(qxDeviceInfoDetail));
+            }
+        }
+    }
+
+
+    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 = SqlTypedValueUtils.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 String sendPost(HttpPost httpPost, String jsonData) {
+        CloseableHttpClient httpClient = HttpClients.createDefault();
+        StringEntity entity = new StringEntity(jsonData, ContentType.create("application/json", Consts.UTF_8));
+        httpPost.setEntity(entity);
+        String result = null;
+        try {
+            CloseableHttpResponse execute = httpClient.execute(httpPost);
+            HttpEntity res = execute.getEntity();
+            InputStream is = res.getContent();
+            int len;
+            byte[] buf = new byte[128];
+            ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+            while ((len = is.read(buf)) != -1) {
+                byteArrayOutputStream.write(buf, 0, len);
+            }
+            result = byteArrayOutputStream.toString();
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+        return result;
+    }
+}

+package com.iot.scheduler.zone;
+
+import com.iot.scheduler.service.DjxcDevicePullService;
+import com.iot.scheduler.service.WwnjDevicePullService;
+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 SjgyhlwptZoneScheduler extends AbstractZoneScheduler {
+    @Override
+    protected String getZoneName() {
+        return "省级重点工业互联网平台申报";
+    }
+
+    @Resource
+    private WwnjDevicePullService wwnjDevicePullService;
+
+    @Resource
+    private DjxcDevicePullService djxcDevicePullService;
+
+    @Scheduled(cron = "${scheduler.sjgyhlwpt.pull:0 0/5 * * * ?}")
+    public void pullDevicesFromThirdParty() {
+        String taskName = "Pull Devices (3rd Party -> IoT)";
+        logStart(taskName);
+        try {
+            log.info("[{}] 开始同步文王酿酒设备数据...", getZoneName());
+            wwnjDevicePullService.pullDeviceAndPushToIot();
+
+            log.info("[{}] 开始同步鼎基鞋材设备数据...", getZoneName());
+            djxcDevicePullService.pullDeviceAndPushToIot();
+        } catch (Exception e) {
+            logError(taskName, e);
+        } finally {
+            logEnd(taskName);
+        }
+    }
+}

@@ -46,6 +46,8 @@ scheduler:
   xp:
   xp:
     pull: "0 0/1 * * * ?"
     pull: "0 0/1 * * * ?"
     push: "0 0/1 * * * ?"
     push: "0 0/1 * * * ?"
+  sjgyhlwpt:
+    pull: "0 0/5 * * * ?"   # Every 5 minutes
 
 
 hn:
 hn:
   third:
   third:
@@ -353,3 +355,38 @@ FROM
 WHERE
 WHERE
   de.organization_id IN ('f82530a0-93e4-4aeb-9339-f5b6d1127840') and dc.credentials_id='AGX251111720'"
   de.organization_id IN ('f82530a0-93e4-4aeb-9339-f5b6d1127840') and dc.credentials_id='AGX251111720'"
 
 
+djxc:
+  third:
+    jdbcUrl: "jdbc:postgresql://106.15.73.210:5433/thingskit"
+    jdbcUserName: "postgres"
+    jdbcPassword: "postgres"
+    selectSql: "SELECT
+  dc.credentials_id AS deviceCode
+FROM
+  device de
+  LEFT JOIN device_credentials dc ON de.id = dc.device_id
+WHERE
+  de.device_profile_id = '9e8257b0-3982-11f1-9cb8-e3376d1e7978';"
+  iot:
+    userName: "djxc"
+    password: "Djxc@123.com"
+    tokenUrl: "https://iot.hzzlyun.com/api/auth/login"
+    detailUrl: "https://iot.hzzlyun.com/api/v1/"
+
+wwnj:
+  third:
+    jdbcUrl: "jdbc:postgresql://106.15.73.210:5433/thingskit"
+    jdbcUserName: "postgres"
+    jdbcPassword: "postgres"
+    selectSql: "SELECT
+  dc.credentials_id AS deviceCode
+FROM
+  device de
+  LEFT JOIN device_credentials dc ON de.id = dc.device_id
+WHERE
+  de.device_profile_id = '9bb44bc0-394f-11f1-9cb8-e3376d1e7978';"
+  iot:
+    userName: "wwnj"
+    password: "Wwnj@123.com"
+    tokenUrl: "https://iot.hzzlyun.com/api/auth/login"
+    detailUrl: "https://iot.hzzlyun.com/api/v1/"