diff --git a/rc-service/rc-inner-api/other-inner/other-inner-model/src/main/java/com/red/circle/other/inner/enums/team/TeamBillCycleStatus.java b/rc-service/rc-inner-api/other-inner/other-inner-model/src/main/java/com/red/circle/other/inner/enums/team/TeamBillCycleStatus.java index 3bb7c6cf..3986c61f 100644 --- a/rc-service/rc-inner-api/other-inner/other-inner-model/src/main/java/com/red/circle/other/inner/enums/team/TeamBillCycleStatus.java +++ b/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, + /** * 已结算，账单进行了核实,并支付成功的. */ diff --git a/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/config/TelegramProperties.java b/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/config/TelegramProperties.java new file mode 100644 index 00000000..895a40bd --- /dev/null +++ b/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 chatIds; + private int alertLimitSeconds = 60; + private String proxyHost; + private Integer proxyPort; + } +} diff --git a/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/listener/team/TeamBillSettleListener.java b/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/listener/team/TeamBillSettleListener.java index bdd5f97a..0d44ffe0 100644 --- a/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/listener/team/TeamBillSettleListener.java +++ b/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)) { - return null; + if (!Boolean.TRUE.equals(isPay)) { + return settleResult; } - if (!Boolean.TRUE.equals(isPay)) { - return settleResult; + if (CollectionUtils.isEmpty(teamMemberTargets)) { + 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); diff --git a/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/scheduler/MyAmazingBot.java b/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/scheduler/MyAmazingBot.java index 75e1b9fa..fbecd29c 100644 --- a/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/scheduler/MyAmazingBot.java +++ b/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) { diff --git a/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/scheduler/TeamBillTask.java b/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/scheduler/TeamBillTask.java index 6b353db8..b08a9beb 100644 --- a/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/scheduler/TeamBillTask.java +++ b/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 { - teamProfileService.scanStatusAvailable(teamProfiles -> { - List> futures = new ArrayList<>(); + List currentBatch = new ArrayList<>(); - for (TeamProfile teamProfile : teamProfiles) { - TeamBillCmd cmd = new TeamBillCmd() - .setSysOrigin(teamProfile.getSysOrigin()) - .setTeamId(teamProfile.getId()) - .setRegion(teamProfile.getRegion()) - .setOperationTime(TimestampUtils.now()) - .setOperationBackUser(teamProfile.getCreateUser()); + teamProfileService.scanStatusAvailable(teamProfiles -> { + for (TeamProfile teamProfile : teamProfiles) { + currentBatch.add(teamProfile); - teamBillCycleService.createIfAbsent(cmd); + if (currentBatch.size() >= BATCH_SIZE) { + List batchToProcess = new ArrayList<>(currentBatch); + processBatch(batchToProcess, success, fail); + total.addAndGet(batchToProcess.size()); + currentBatch.clear(); - CompletableFuture future = CompletableFuture.runAsync(() -> { - try { - otherMqMessage.teamBillSettle(new TeamBillSettleEvent() - .setTeamId(teamProfile.getId())); - success.incrementAndGet(); - } catch (Exception e) { - fail.incrementAndGet(); - log.error("发送结算消息失败: teamId={}", teamProfile.getId(), e); - } - }, executor); - - futures.add(future); - total.incrementAndGet(); + try { + Thread.sleep(BATCH_DELAY_MS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + log.warn("批次延迟被中断", e); + } } - - // 等待本批次完成 - 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 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()); + } + } diff --git a/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/service/TelegramMonitorService.java b/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/service/TelegramMonitorService.java new file mode 100644 index 00000000..4ad2172d --- /dev/null +++ b/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 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 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 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 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 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 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; + } + } +} diff --git a/rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/mongo/service/team/TeamBillCycleUtils.java b/rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/mongo/service/team/TeamBillCycleUtils.java index 2adebcb4..7f6b4df4 100644 --- a/rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/mongo/service/team/TeamBillCycleUtils.java +++ b/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; } diff --git a/rc-service/rc-service-other/other-start/src/test/java/TgTest.java b/rc-service/rc-service-other/other-start/src/test/java/TgTest.java new file mode 100644 index 00000000..7722a548 --- /dev/null +++ b/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", "出错啦！！！"); + } + +}