lowflow/flow-im/flow-im-entity/src/main/java/com/flow/job/SendUskyJob.java

package com.flow.job;

import com.flow.entity.Notify;
import com.flow.mapstruct.NotifyRowMapper;
import dm.jdbc.filter.stat.json.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Arrays;
import java.util.List;

/**
 *
 * @author fyc
 * @email yuchuan.fu@chinausky.com
 * @date 2025/5/14
 */

@Component
@Slf4j
public class SendUskyJob {

    private static final DateTimeFormatter FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
    private static final String USKY_URL = "https://gateway.usky.cn/prod-api/system/mceReceive/addMce";
    private static final String USKY_TEST_URL = "http://192.168.10.165:13200/offline-api/system/mceReceive/addMce";

    @Autowired
    private JdbcTemplate jdbcTemplate;

    @Autowired
    private RestTemplate restTemplate;

    // 每30秒执行一次
    @Scheduled(fixedRate = 30000)
    public void executeTask() {
        // 记录任务开始时间
        LocalDateTime startTime = LocalDateTime.now();
        log.info("执行消息同步定时任务，开始时间：" + startTime.format(FORMATTER));

        // 一小时内的消息
        List<Notify> notifies = unSendUsky(startTime.minusHours(1), startTime);
        int notifySize = notifies.size();
        log.info("需要同步的消息数量为：" + notifySize);

        if (!notifies.isEmpty()) {
            sendUsky(notifies);
            updateNotify(notifies);
        }

        // 记录任务结束时间
        LocalDateTime endTime = LocalDateTime.now();
        log.info("执行消息同步定时任务，结束时间：" + endTime.format(FORMATTER));
    }

    // 查询未发送的消息
    public List<Notify> unSendUsky(LocalDateTime startTime, LocalDateTime endTime) {
        String sql = "SELECT * FROM sys_notify WHERE is_send_usky = 0 AND receiving_time BETWEEN ? AND ?";
        return jdbcTemplate.query(
                sql,
                new Object[]{startTime, endTime},
                new NotifyRowMapper()
        );
    }

    // 发送消息
    @Async
    public void sendUsky(List<Notify> unSend) {
        int successCount = 0;
        int failCount = 0;
        for (Notify send : unSend) {
            JSONObject sendJson = new JSONObject()
                    .put("infoTypeName", "message_type")
                    .put("infoType", 3)
                    .put("infoTitle", "OA审批")
                    .put("infoContent", send.getContent() + "。 url：" + send.getUrl())
                    .put("userName", send.getSender())
                    .put("userNames", Arrays.asList(send.getReceiver()))
                    .put("id", send.getId());

            String sendJsonString = sendJson.toString();
            log.info("准备发送 {} 的消息：{} ", send.getSender(), sendJsonString);

            try {
                // restTemplate.postForObject(USKY_TEST_URL, sendJsonString, String.class);
                restTemplate.postForObject(USKY_URL, sendJsonString, String.class);
                log.info("消息发送成功，ID: " + send.getId());
                successCount++;
            } catch (Exception e) {
                log.error("消息发送失败，ID: " + send.getId(), e);
                failCount++;
            }
        }
        log.info("消息发送完成，成功数量：" + successCount + "，失败数量：" + failCount);
    }

    // 更新已发送消息同步状态
    public void updateNotify(List<Notify> unSend) {
        for (Notify send : unSend) {
            String sql = "UPDATE sys_notify SET is_send_usky = 1 WHERE id = ?";
            jdbcTemplate.update(sql, send.getId());
        }
        log.info("更新消息状态完成，共更新 {} 条消息", unSend.size());
    }
}