工资发放重构

rc-service/rc-inner-api/other-inner/other-inner-model/src/main/java/com/red/circle/other/inner/enums/team/TeamBillCycleStatus.java

@@ -17,6 +17,11 @@ public enum TeamBillCycleStatus {
    */
   PAY_OUT,
 
+  /**
+   * 未完成: 账单周期结束但未达成结算条件.
+   */
+  UNDONE,
+
   /**
    * 已结算，账单进行了核实,并支付成功的.
    */

rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/config/TelegramProperties.java

@@ -0,0 +1,29 @@
+package com.red.circle.other.app.config;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+
+@Data
+@Component
+@ConfigurationProperties(prefix = "telegram")
+public class TelegramProperties {
+
+    private Bot bot = new Bot();
+    private Monitor monitor = new Monitor();
+
+    @Data
+    public static class Bot {
+        private String token;
+    }
+
+    @Data
+    public static class Monitor {
+        private List<String> chatIds;
+        private int alertLimitSeconds = 60;
+        private String proxyHost;
+        private Integer proxyPort;
+    }
+}

rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/listener/team/TeamBillSettleListener.java

@@ -171,12 +171,17 @@ public class TeamBillSettleListener implements MessageListener {
         .setOwnSalary(ownSalary(teamBillMemberTargets))
         .setTotalSalary(ownTotalSalary(teamBillMemberTargets));
 
-    if (CollectionUtils.isEmpty(teamMemberTargets)) {
+    if (!Boolean.TRUE.equals(isPay)) {
-        return null;
+      return settleResult;
     }
 
-    if (!Boolean.TRUE.equals(isPay)) {
+    if (CollectionUtils.isEmpty(teamMemberTargets)) {
-        return settleResult;
+        teamBillCycleService.updateSelective(new TeamBillCycle()
+                .setId(teamBillCycle.getId())
+                .setRegion(teamBillCycle.getRegion())
+                .setStatus(TeamBillCycleStatus.UNDONE)
+        );
+        return null;
     }
 
       teamBillCycleService.updateSelective(new TeamBillCycle()
@@ -186,6 +191,10 @@ public class TeamBillSettleListener implements MessageListener {
               .setSettleResult(settleResult)
       );
 
+    if (true) {
+        return settleResult;
+    }
+
     if (Objects.isNull(regionPolicyManager.getPolicyType()) ||
             Objects.equals(TeamPolicyTypeEnum.MONEY.name(), regionPolicyManager.getPolicyType())) {
       log.warn("美金入账{}", teamProfile);

rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/scheduler/MyAmazingBot.java

@@ -13,7 +13,7 @@ import org.telegram.telegrambots.meta.generics.TelegramClient;
 import java.io.File;
 
 public class MyAmazingBot implements LongPollingSingleThreadUpdateConsumer {
-    private TelegramClient telegramClient = new OkHttpTelegramClient("8087837543:AAEidUG9kRxGdflo3nemYygloXfbbBG1lH4");
+    private TelegramClient telegramClient = new OkHttpTelegramClient("8350995319:AAGwh1O2PRnBrXs0lj9ZyuhFDWUPLT1SWyk");
 
     @Override
     public void consume(Update update) {

rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/scheduler/TeamBillTask.java

@@ -19,10 +19,6 @@ import java.time.ZoneId;
 import java.time.format.DateTimeFormatter;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import lombok.AllArgsConstructor;
@@ -85,71 +81,76 @@ public class TeamBillTask {
 
     /**
      * 创建本月账单，发出结算信号.
+     * 小批量同步发送，避免 MQ 超时
      */
     private void consumeProcessPayOutBill() {
         AtomicInteger total = new AtomicInteger(0);
         AtomicInteger success = new AtomicInteger(0);
         AtomicInteger fail = new AtomicInteger(0);
 
-        ExecutorService executor = Executors.newFixedThreadPool(20);
+        final int BATCH_SIZE = 10;
+        final long BATCH_DELAY_MS = 1000;
 
-        try {
+        List<TeamProfile> currentBatch = new ArrayList<>();
-            teamProfileService.scanStatusAvailable(teamProfiles -> {
-                List<CompletableFuture<Void>> futures = new ArrayList<>();
 
-                for (TeamProfile teamProfile : teamProfiles) {
+        teamProfileService.scanStatusAvailable(teamProfiles -> {
-                    TeamBillCmd cmd = new TeamBillCmd()
+            for (TeamProfile teamProfile : teamProfiles) {
-                            .setSysOrigin(teamProfile.getSysOrigin())
+                currentBatch.add(teamProfile);
-                            .setTeamId(teamProfile.getId())
-                            .setRegion(teamProfile.getRegion())
-                            .setOperationTime(TimestampUtils.now())
-                            .setOperationBackUser(teamProfile.getCreateUser());
 
-                    teamBillCycleService.createIfAbsent(cmd);
+                if (currentBatch.size() >= BATCH_SIZE) {
+                    List<TeamProfile> batchToProcess = new ArrayList<>(currentBatch);
+                    processBatch(batchToProcess, success, fail);
+                    total.addAndGet(batchToProcess.size());
+                    currentBatch.clear();
 
-                    CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
+                    try {
-                        try {
+                        Thread.sleep(BATCH_DELAY_MS);
-                            otherMqMessage.teamBillSettle(new TeamBillSettleEvent()
+                    } catch (InterruptedException e) {
-                                    .setTeamId(teamProfile.getId()));
+                        Thread.currentThread().interrupt();
-                            success.incrementAndGet();
+                        log.warn("批次延迟被中断", e);
-                        } catch (Exception e) {
+                    }
-                            fail.incrementAndGet();
-                            log.error("发送结算消息失败: teamId={}", teamProfile.getId(), e);
-                        }
-                    }, executor);
-
-                    futures.add(future);
-                    total.incrementAndGet();
                 }
-
-                // 等待本批次完成
-                try {
-                    CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
-                            .get(30, TimeUnit.SECONDS);
-                } catch (Exception e) {
-                    log.error("批次处理超时或异常", e);
-                }
-
-                log.info("批次完成: size={}, 累计成功={}, 累计失败={}",
-                        teamProfiles.size(), success.get(), fail.get());
-
-            }, 100);
-
-        } finally {
-            executor.shutdown();
-            try {
-                if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
-                    executor.shutdownNow();
-                }
-            } catch (InterruptedException e) {
-                executor.shutdownNow();
-                Thread.currentThread().interrupt();
             }
+
+            log.info("当前进度: 累计={}, 成功={}, 失败={}", total.get(), success.get(), fail.get());
+
+        }, 100);
+
+        if (!currentBatch.isEmpty()) {
+            processBatch(new ArrayList<>(currentBatch), success, fail);
+            total.addAndGet(currentBatch.size());
         }
 
         log.info("结算消息发送完成: 总数={}, 成功={}, 失败={}",
                 total.get(), success.get(), fail.get());
     }
 
+    private void processBatch(List<TeamProfile> batch, AtomicInteger success, AtomicInteger fail) {
+        for (TeamProfile teamProfile : batch) {
+            try {
+                TeamBillCmd cmd = new TeamBillCmd()
+                        .setSysOrigin(teamProfile.getSysOrigin())
+                        .setTeamId(teamProfile.getId())
+                        .setRegion(teamProfile.getRegion())
+                        .setOperationTime(TimestampUtils.now())
+                        .setOperationBackUser(teamProfile.getCreateUser());
+
+                teamBillCycleService.createIfAbsent(cmd);
+
+                otherMqMessage.teamBillSettle(new TeamBillSettleEvent()
+                        .setTeamId(teamProfile.getId()));
+
+                success.incrementAndGet();
+
+            } catch (Exception e) {
+                fail.incrementAndGet();
+                log.error("处理团队账单失败: teamId={}", teamProfile.getId(), e);
+            }
+        }
+
+        log.info("批次完成: size={}, 当前成功={}, 当前失败={}",
+                batch.size(), success.get(), fail.get());
+    }
+
 
 }

rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/service/TelegramMonitorService.java

@@ -0,0 +1,289 @@
+package com.red.circle.other.app.service;
+
+import com.red.circle.other.app.config.TelegramProperties;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+import org.telegram.telegrambots.client.okhttp.OkHttpTelegramClient;
+import org.telegram.telegrambots.meta.api.methods.send.SendDocument;
+import org.telegram.telegrambots.meta.api.methods.send.SendMessage;
+import org.telegram.telegrambots.meta.api.methods.send.SendPhoto;
+import org.telegram.telegrambots.meta.api.objects.InputFile;
+import org.telegram.telegrambots.meta.exceptions.TelegramApiException;
+import org.telegram.telegrambots.meta.generics.TelegramClient;
+
+import javax.annotation.PostConstruct;
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.net.Proxy;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import okhttp3.OkHttpClient;
+
+/**
+ * Telegram 监控告警服务
+ */
+@Slf4j
+@Service
+public class TelegramMonitorService {
+
+    @Autowired
+    private TelegramProperties telegramProperties;
+
+    @Value("${spring.profiles.active:dev}")
+    private String env;
+
+    private TelegramClient telegramClient;
+
+    private final Map<String, Long> alertLimitCache = new ConcurrentHashMap<>();
+    private static final DateTimeFormatter DATE_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+
+    @PostConstruct
+    public void init() {
+        String botToken = telegramProperties.getBot().getToken();
+        List<String> chatIds = telegramProperties.getMonitor().getChatIds();
+        int alertLimitSeconds = telegramProperties.getMonitor().getAlertLimitSeconds();
+        String proxyHost = telegramProperties.getMonitor().getProxyHost();
+        Integer proxyPort = telegramProperties.getMonitor().getProxyPort();
+        
+        OkHttpClient.Builder builder = new OkHttpClient.Builder();
+        
+        if (proxyHost != null && proxyPort != null) {
+            builder.proxy(new Proxy(Proxy.Type.HTTP, new InetSocketAddress(proxyHost, proxyPort)));
+            log.info("Telegram 使用代理: {}:{}", proxyHost, proxyPort);
+        }
+        
+        OkHttpClient okHttpClient = builder.build();
+        telegramClient = new OkHttpTelegramClient(okHttpClient, botToken);
+        
+        log.info("Telegram 监控服务初始化完成, chatIds: {}, 限流间隔: {}秒", chatIds, alertLimitSeconds);
+    }
+
+    /**
+     * 发送系统告警
+     */
+    public void sendSystemAlert(AlertLevel level, String title, String content) {
+        String message = buildAlertMessage(level, "系统告警", title, content);
+        sendAlertWithLimit(level.name() + ":" + title, message);
+    }
+
+    /**
+     * 发送业务告警
+     */
+    public void sendBusinessAlert(AlertLevel level, String businessType, String content) {
+        String message = buildAlertMessage(level, "业务告警", businessType, content);
+        sendAlertWithLimit(level.name() + ":BUSINESS:" + businessType, message);
+    }
+
+    /**
+     * 发送定时任务执行报告
+     */
+    public void sendScheduledReport(String taskName, boolean success, String detail) {
+        AlertLevel level = success ? AlertLevel.INFO : AlertLevel.ERROR;
+        String status = success ? "✅ 执行成功" : "❌ 执行失败";
+        String message = buildAlertMessage(level, "定时任务", taskName, status + "\n" + detail);
+        sendAlertWithLimit("SCHEDULE:" + taskName, message);
+    }
+
+    /**
+     * 发送异常告警
+     */
+    public void sendExceptionAlert(String module, Exception exception, String context) {
+        StringBuilder content = new StringBuilder();
+        content.append("📌 模块：").append(module).append("\n");
+        content.append("📌 上下文：").append(context).append("\n");
+        content.append("📌 异常类型：").append(exception.getClass().getSimpleName()).append("\n");
+        content.append("📌 异常信息：").append(exception.getMessage());
+
+        String message = buildAlertMessage(AlertLevel.ERROR, "异常告警", module, content.toString());
+        sendAlertWithLimit("EXCEPTION:" + module, message);
+    }
+
+    /**
+     * 发送每日数据报表
+     */
+    public void sendDailyReport(String title, Map<String, Object> data) {
+        StringBuilder content = new StringBuilder();
+        data.forEach((key, value) -> content.append("📊 ").append(key).append(": ").append(value).append("\n"));
+
+        String message = buildAlertMessage(AlertLevel.INFO, "每日报表", title, content.toString());
+        sendToAllChats(message);
+    }
+
+    /**
+     * 发送文本消息（不限流，慎用）
+     */
+    public void sendMessage(String content) {
+        sendToAllChats(content);
+    }
+
+    /**
+     * 发送带图片的告警
+     */
+    public void sendAlertWithImage(AlertLevel level, String title, String content, String imagePath) {
+        String message = buildAlertMessage(level, "图片告警", title, content);
+        List<String> chatIds = telegramProperties.getMonitor().getChatIds();
+        for (String chatId : chatIds) {
+            sendImageToChat(chatId, imagePath, message);
+        }
+    }
+
+    /**
+     * 发送带文件的告警
+     */
+    public void sendAlertWithFile(AlertLevel level, String title, String content, String filePath) {
+        String message = buildAlertMessage(level, "文件告警", title, content);
+        List<String> chatIds = telegramProperties.getMonitor().getChatIds();
+        for (String chatId : chatIds) {
+            sendFileToChat(chatId, filePath, message);
+        }
+    }
+
+    /**
+     * 构建告警消息格式
+     */
+    private String buildAlertMessage(AlertLevel level, String category, String title, String content) {
+        String icon = level.getIcon();
+        String levelText = level.getText();
+
+        return String.format(
+                "%s 【%s】%s\n" +
+                        "━━━━━━━━━━━━━━━━\n" +
+                        "📍 服务：rc-service-other\n" +
+                        "⏰ 时间：%s\n" +
+                        "🌍 环境：%s\n" +
+                        "📋 标题：%s\n" +
+                        "━━━━━━━━━━━━━━━━\n" +
+                        "%s\n" +
+                        "━━━━━━━━━━━━━━━━",
+                icon, levelText, category,
+                LocalDateTime.now().format(DATE_FORMATTER),
+                env.toUpperCase(),
+                title,
+                content
+        );
+    }
+
+    /**
+     * 带限流的发送
+     */
+    private void sendAlertWithLimit(String alertKey, String message) {
+        long now = System.currentTimeMillis();
+        Long lastSendTime = alertLimitCache.get(alertKey);
+        int alertLimitSeconds = telegramProperties.getMonitor().getAlertLimitSeconds();
+
+        if (lastSendTime != null && (now - lastSendTime) < alertLimitSeconds * 1000L) {
+            log.warn("告警被限流跳过: {}", alertKey);
+            return;
+        }
+
+        alertLimitCache.put(alertKey, now);
+        sendToAllChats(message);
+
+        cleanExpiredCache(now);
+    }
+
+    /**
+     * 发送到所有配置的 chatId
+     */
+    private void sendToAllChats(String message) {
+        List<String> chatIds = telegramProperties.getMonitor().getChatIds();
+        for (String chatId : chatIds) {
+            sendTextToChat(chatId, message);
+        }
+    }
+
+    /**
+     * 发送文本消息到指定 chat
+     */
+    private void sendTextToChat(String chatId, String message) {
+        SendMessage sendMessage = new SendMessage(chatId, message);
+        try {
+            telegramClient.execute(sendMessage);
+            log.info("Telegram 消息发送成功: chatId={}", chatId);
+        } catch (TelegramApiException e) {
+            log.error("Telegram 消息发送失败: chatId={}, error={}", chatId, e.getMessage());
+        }
+    }
+
+    /**
+     * 发送图片到指定 chat
+     */
+    private void sendImageToChat(String chatId, String imagePath, String caption) {
+        File photoFile = new File(imagePath);
+        if (!photoFile.exists()) {
+            log.error("图片文件不存在: {}", imagePath);
+            return;
+        }
+
+        SendPhoto sendPhoto = new SendPhoto(chatId, new InputFile(photoFile));
+        sendPhoto.setCaption(caption);
+
+        try {
+            telegramClient.execute(sendPhoto);
+            log.info("Telegram 图片发送成功: chatId={}, path={}", chatId, imagePath);
+        } catch (TelegramApiException e) {
+            log.error("Telegram 图片发送失败: chatId={}, error={}", chatId, e.getMessage());
+        }
+    }
+
+    /**
+     * 发送文件到指定 chat
+     */
+    private void sendFileToChat(String chatId, String filePath, String caption) {
+        File file = new File(filePath);
+        if (!file.exists()) {
+            log.error("文件不存在: {}", filePath);
+            return;
+        }
+
+        SendDocument sendDocument = new SendDocument(chatId, new InputFile(file));
+        sendDocument.setCaption(caption);
+
+        try {
+            telegramClient.execute(sendDocument);
+            log.info("Telegram 文件发送成功: chatId={}, path={}", chatId, filePath);
+        } catch (TelegramApiException e) {
+            log.error("Telegram 文件发送失败: chatId={}, error={}", chatId, e.getMessage());
+        }
+    }
+
+    /**
+     * 清理过期的限流缓存
+     */
+    private void cleanExpiredCache(long now) {
+        int alertLimitSeconds = telegramProperties.getMonitor().getAlertLimitSeconds();
+        alertLimitCache.entrySet().removeIf(entry ->
+                (now - entry.getValue()) > alertLimitSeconds * 2000L
+        );
+    }
+
+    /**
+     * 告警级别枚举
+     */
+    public enum AlertLevel {
+        ERROR("🔴", "ERROR"),
+        WARN("🟡", "WARN"),
+        INFO("🔵", "INFO");
+
+        private final String icon;
+        private final String text;
+
+        AlertLevel(String icon, String text) {
+            this.icon = icon;
+            this.text = text;
+        }
+
+        public String getIcon() {
+            return icon;
+        }
+
+        public String getText() {
+            return text;
+        }
+    }
+}

rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/mongo/service/team/TeamBillCycleUtils.java

@@ -227,6 +227,9 @@ public final class TeamBillCycleUtils {
         }
 
           Integer policyEffectiveDay = NumUtils.isNullDefaultZero(teamPolicy.getEffectiveDay());
+          resultTarget.setLevel(teamPolicy.getLevel());
+          resultTarget.setSatisfyLevel(Boolean.TRUE);
+
           if (targetTimeHours >= policyOnlineTimeHours && effectiveDay >= policyEffectiveDay) {
               resultTarget.setMemberSalary(teamPolicy.getMemberSalary());
               resultTarget.setOwnSalary(teamPolicy.getOwnSalary());
@@ -237,8 +240,6 @@ public final class TeamBillCycleUtils {
               resultTarget.setTotalSalary(teamPolicy.getMemberSalary().multiply(new BigDecimal("0.9")).setScale(2, RoundingMode.DOWN));
           }
 
-          resultTarget.setLevel(teamPolicy.getLevel());
-          resultTarget.setSatisfyLevel(Boolean.TRUE);
           resultTarget.setPropsRewards(teamPolicy.getPropsRewards());
           return resultTarget;
       }

rc-service/rc-service-other/other-start/src/test/java/TgTest.java

@@ -0,0 +1,22 @@
+import com.red.circle.OtherServiceApplication;
+import com.red.circle.other.app.scheduler.MyAmazingBot;
+import com.red.circle.other.app.service.TelegramMonitorService;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+
+@SpringBootTest(classes = OtherServiceApplication.class)
+@RunWith(SpringRunner.class)
+public class TgTest {
+
+    @Autowired
+    private TelegramMonitorService telegramMonitorService;
+
+    @Test
+    public void test(){
+        telegramMonitorService.sendBusinessAlert(TelegramMonitorService.AlertLevel.ERROR, "TEAM_BILL", "出错啦！！！");
+    }
+
+}