Merge remote-tracking branch 'origin/Payment-Configuration' into Payment-Configuration

fs-company/src/main/java/com/fs/app/controller/im/FsImMsgSendLogController.java

@@ -1,5 +1,6 @@
 package com.fs.app.controller.im;
 
+import com.fs.app.service.param.BatchSendImTextRequest;
 import com.fs.app.service.param.FsImMsgSendLogRequest;
 import com.fs.app.service.param.FsImMsgSendLogResponse;
 import com.fs.common.core.controller.BaseController;
@@ -11,9 +12,12 @@ import com.fs.common.utils.poi.ExcelUtil;
 import com.fs.framework.security.LoginUser;
 import com.fs.framework.service.TokenService;
 import com.fs.im.domain.FsImMsgSendLog;
+import com.fs.im.dto.OpenImResponseDTO;
 import com.fs.im.service.IFsImMsgSendLogService;
+import com.fs.im.service.OpenIMService;
 import com.github.pagehelper.PageHelper;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.validation.annotation.Validated;
 import org.springframework.web.bind.annotation.*;
 
 import javax.servlet.http.HttpServletResponse;
@@ -30,6 +34,9 @@ public class FsImMsgSendLogController extends BaseController {
     @Autowired
     private IFsImMsgSendLogService fsImMsgSendLogService;
 
+    @Autowired
+    private OpenIMService openIMService;
+
     @Autowired
     private TokenService tokenService;
 
@@ -65,4 +72,29 @@ public class FsImMsgSendLogController extends BaseController {
         request.setCompanyId(loginUser.getCompany().getCompanyId());
         return fsImMsgSendLogService.getFsImMsgSendStatistics(request);
     }
+
+    /**
+     * 批量发送文本消息
+     */
+
+    @PostMapping("/batchSendText")
+    public R batchSendText(@Validated @RequestBody BatchSendImTextRequest request) {
+        LoginUser loginUser = tokenService.getLoginUser(ServletUtils.getRequest());
+        request.setSenderId("C"+loginUser.getUser().getUserId());
+        if(request.getUserIds().size() > 100){
+            return R.error("一次最多只能发送100个用户");
+        }
+        // 前面加一个U
+        List<String> userIds = request.getUserIds();
+        userIds.replaceAll(userId -> "U"+userId);
+        request.setUserIds(userIds);
+        request.setPushDesc(request.getPushDesc() == null ? request.getTextContent() : request.getPushDesc());
+        OpenImResponseDTO openImResponseDTO =openIMService.batchSendTextMessage( request.getSenderId(), request.getUserIds(), request.getTextContent(), request.getPushTitle(), request.getPushDesc(), request.getEx());
+        if(openImResponseDTO.getErrCode() == 0){
+            return R.ok("发送成功");
+        }else {
+            return R.error(openImResponseDTO.getErrMsg());
+        }
+    }
+
 }

fs-qw-task/src/main/java/com/fs/app/task/CourseWatchLogScheduler.java

@@ -83,7 +83,7 @@ public class CourseWatchLogScheduler {
 //    }
 
 
-    @Scheduled(fixedRate = 60000) // 每分钟执行一次
+//    @Scheduled(fixedRate = 60000) // 每分钟执行一次
     public void checkWatchStatus() {
         // 尝试设置标志为 true，表示任务开始执行
         if (!isRunning1.compareAndSet(false, true)) {
@@ -127,7 +127,7 @@ public class CourseWatchLogScheduler {
     /**
      * 创建完课消息
      */
-    @Scheduled(fixedRate = 300000) // 每五分钟执行一次
+//    @Scheduled(fixedRate = 300000) // 每五分钟执行一次
     public void createCourseFinishMsg() {
         // 尝试设置标志为 true，表示任务开始执行
         if (!isRunning3.compareAndSet(false, true)) {
@@ -157,7 +157,7 @@ public class CourseWatchLogScheduler {
     /**
      * 每天删除过期短链
      */
-    @Scheduled(cron = "0 0 0 * * ?")  // 0点0分0秒执行
+//    @Scheduled(cron = "0 0 0 * * ?")  // 0点0分0秒执行
     public void delCourseExpireLink() {
         try {
             log.info("删除过期短链 - 定时任务开始");
@@ -178,8 +178,8 @@ public class CourseWatchLogScheduler {
         }
         try {
             log.info("WXH5-检查会员看课中任务执行>>>>>>>>>>>>");
-            courseWatchLogService.scheduleUpdateDurationToDatabase();
-            courseWatchLogService.checkFsUserWatchStatus();
+            courseWatchLogService.scheduleBatchUpdateToDatabase();
+            courseWatchLogService.checkWatchStatus();
             log.info("WXH5-检查会员看课中任务执行完成>>>>>>>>>>>>");
         }catch (Exception e) {
             log.error("WXH5-检查会员看课中任务执行完成 - 定时任务执行失败", e);

fs-service/src/main/java/com/fs/app/service/param/BatchSendImTextRequest.java

@@ -0,0 +1,55 @@
+package com.fs.app.service.param;
+
+import lombok.Data;
+import javax.validation.constraints.NotBlank;
+import javax.validation.constraints.NotEmpty;
+import java.io.Serializable;
+import java.util.List;
+
+/**
+ * 批量发送IM文本消息请求参数
+ */
+@Data
+public class BatchSendImTextRequest implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    private String SenderId;
+
+    /**
+     * 接收用户ID列表
+     */
+    @NotEmpty(message = "接收用户ID列表不能为空")
+    private List<String> userIds;
+
+    /**
+     * 发送的文本内容
+     */
+    @NotBlank(message = "发送内容不能为空")
+    private String textContent;
+
+    /**
+     * 发送人昵称（可选，不填则使用默认）
+     */
+    private String senderNickname;
+
+    /**
+     * 发送人头像（可选，不填则使用默认）
+     */
+    private String senderFaceURL;
+
+    /**
+     * 离线推送标题
+     */
+    private String pushTitle;
+
+    /**
+     * 离线推送描述
+     */
+    private String pushDesc;
+
+    /**
+     * 扩展字段（可选）
+     */
+    private String ex;
+}

fs-service/src/main/java/com/fs/course/service/impl/FsCourseWatchLogServiceImpl.java

@@ -399,6 +399,218 @@ public class FsCourseWatchLogServiceImpl extends ServiceImpl<FsCourseWatchLogMap
         return baseMapper.countByMap(params);
     }
 
+
+    @Autowired
+    private RedisKeyScanner redisKeyScanner;
+    @Override
+    public void scheduleBatchUpdateToDatabase() {
+        try {
+            log.info("开始更新看课时长，检查完课>>>>>>");
+
+            // 读取所有的key
+            Set<String> keys = redisKeyScanner.scanMatchKey("h5wxuser:watch:duration:*");
+            log.info("共扫描到 {} 个待处理键", keys.size());
+
+            // 如果keys为空，直接返回
+            if (CollectionUtils.isEmpty(keys)) {
+                return;
+            }
+
+            // 读取看课配置
+            String json = configService.selectConfigByKey("course.config");
+            CourseConfig config = JSONUtil.toBean(json, CourseConfig.class);
+
+            // 创建线程安全的集合
+            List<FsCourseWatchLog> logs = Collections.synchronizedList(new ArrayList<>());
+
+            // 创建线程池（根据服务器配置调整线程数）
+            int threadCount = Math.min(Runtime.getRuntime().availableProcessors() * 2, 8);
+            ExecutorService executorService = Executors.newFixedThreadPool(threadCount);
+
+            // 将keys分成多个批次，每个线程处理一批
+            List<List<String>> keyBatches = partitionKeys(new ArrayList<>(keys), threadCount * 10);
+
+            log.info("开始多线程处理，共 {} 个批次", keyBatches.size());
+
+            int type = 1; //检查完课任务
+
+            // 创建所有任务
+            List<CompletableFuture<Void>> futures = new ArrayList<>();
+            for (List<String> batchKeys : keyBatches) {
+                CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
+                    processKeyBatch(batchKeys, config, logs,type);
+                }, executorService);
+                futures.add(future);
+            }
+
+            // 等待所有任务完成
+            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
+
+            // 关闭线程池
+            executorService.shutdown();
+
+            // 批量更新到数据库
+            batchUpdateFsUserCourseWatchLog(logs, 100);
+
+        } catch (Exception e) {
+            log.error("定时任务执行失败 scheduleBatchUpdateToDatabase", e);
+        }
+    }
+
+    @Override
+    public void checkWatchStatus() {
+        try {
+            log.info("开始更新看课中断记录>>>>>");
+
+            // 读取所有的key
+            Set<String> keys = redisKeyScanner.scanMatchKey("h5wxuser:watch:heartbeat:*");
+            log.info("共扫描到 {} 个待处理键", keys.size());
+
+            // 如果keys为空，直接返回
+            if (CollectionUtils.isEmpty(keys)) {
+                return;
+            }
+
+            // 读取看课配置
+            String json = configService.selectConfigByKey("course.config");
+            CourseConfig config = JSONUtil.toBean(json, CourseConfig.class);
+
+            // 创建线程安全的集合
+            List<FsCourseWatchLog> logs = Collections.synchronizedList(new ArrayList<>());
+
+            // 创建线程池（根据服务器配置调整线程数）
+            int threadCount = Math.min(Runtime.getRuntime().availableProcessors() * 2, 8);
+            ExecutorService executorService = Executors.newFixedThreadPool(threadCount);
+
+            // 将keys分成多个批次，每个线程处理一批
+            List<List<String>> keyBatches = partitionKeys(new ArrayList<>(keys), threadCount * 10);
+
+            log.info("开始多线程处理，共 {} 个批次", keyBatches.size());
+
+            int type = 2; //检查看课中断
+
+            // 创建所有任务
+            List<CompletableFuture<Void>> futures = new ArrayList<>();
+            for (List<String> batchKeys : keyBatches) {
+                CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
+                    processKeyBatch(batchKeys, config, logs, type);
+                }, executorService);
+                futures.add(future);
+            }
+
+            // 等待所有任务完成
+            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
+
+            // 关闭线程池
+            executorService.shutdown();
+
+            // 批量更新到数据库
+            batchUpdateFsUserCourseWatchLog(logs, 100);
+
+        } catch (Exception e) {
+            log.error("定时任务执行失败checkWatchStatus", e);
+        }
+    }
+
+    /**
+     * 处理一个key批次
+     */
+    private void processKeyBatch(List<String> batchKeys, CourseConfig config, List<FsCourseWatchLog> logs,int type) {
+        LocalDateTime now = LocalDateTime.now();
+        for (String key : batchKeys) {
+            try {
+                // 取key中数据
+                Long userId = null;
+                Long videoId = null;
+                Long companyUserId = null;
+                try {
+                    String[] parts = key.split(":");
+                    userId = Long.parseLong(parts[3]);
+                    videoId = Long.parseLong(parts[4]);
+                    companyUserId = Long.parseLong(parts[5]);
+                } catch (Exception e) {
+                    log.error("key中id为null:{}", key);
+                    continue;
+                }
+                FsCourseWatchLog watchLog = new FsCourseWatchLog();
+                //检查完课
+                if (type==1){
+                    String durationStr = redisCache.getCacheObject(key);
+                    if (com.fs.common.utils.StringUtils.isEmpty(durationStr)) {
+                        redisCache.deleteObject(key);
+                        log.error("key中数据为null:{}", key);
+                        continue;  // 如果 Redis 中没有记录，跳过
+                    }
+                    Long duration = Long.valueOf(durationStr);
+
+                    watchLog.setDuration(duration);
+
+                    // 取对应视频的时长
+                    Long videoDuration;
+                    try {
+                        videoDuration = getFsUserVideoDuration(videoId);
+                    } catch (Exception e) {
+                        log.error("视频时长为空：{}", videoId);
+                        continue;
+                    }
+
+                    if (videoDuration != null && videoDuration != 0) {
+                        // 判断是否完课
+                        long percentage = (duration * 100 / videoDuration);
+                        if (percentage >= config.getAnswerRate()) {
+                            watchLog.setLogType(2); // 设置状态为"已完成"
+                            watchLog.setFinishTime(new Date());
+                            String heartbeatKey = "h5wxuser:watch:heartbeat:" + userId+ ":" + videoId + ":" + companyUserId;
+                            // 完课删除心跳记录
+                            redisCache.deleteObject(heartbeatKey);
+                            // 完课删除看课时长记录
+                            redisCache.deleteObject(key);
+                        }
+                    }
+                }else{
+                    //检查看课中断
+                    // 获取最后心跳时间
+                    String lastHeartbeatStr = redisCache.getCacheObject(key);
+                    if (com.fs.common.utils.StringUtils.isEmpty(lastHeartbeatStr)) {
+                        redisCache.deleteObject(key);
+                        continue; // 如果 Redis 中没有记录，跳过
+                    }
+                    LocalDateTime lastHeartbeatTime = LocalDateTime.parse(lastHeartbeatStr);
+                    Duration duration = Duration.between(lastHeartbeatTime, now);
+                    // 如果超过一分钟没有心跳，标记为“观看中断”
+                    if (duration.getSeconds() >= 60) {
+                        watchLog.setLogType(4);
+                        // 从 Redis 中删除该记录
+                        redisCache.deleteObject(key);
+                    } else {
+                        watchLog.setLogType(1);
+                    }
+                }
+                watchLog.setVideoId(videoId);
+                watchLog.setUserId(userId);
+                watchLog.setCompanyUserId(companyUserId);
+
+
+                // 线程安全地添加到集合
+                logs.add(watchLog);
+
+            } catch (Exception e) {
+                log.error("处理key {} 时发生异常: {}", key, e.getMessage());
+            }
+        }
+    }
+
+    /**
+     * 将keys分成多个批次
+     */
+    private List<List<String>> partitionKeys(List<String> keys, int batchSize) {
+        List<List<String>> batches = new ArrayList<>();
+        for (int i = 0; i < keys.size(); i += batchSize) {
+            int end = Math.min(i + batchSize, keys.size());
+            batches.add(keys.subList(i, end));
+        }
+        return batches;
+    }
     @Override
     public void scheduleUpdateDurationToDatabase() {
         log.info("WXH5-开始更新会员看课时长，检查完课>>>>>>");
@@ -625,31 +837,37 @@ public class FsCourseWatchLogServiceImpl extends ServiceImpl<FsCourseWatchLogMap
 
 
     public void batchUpdateFsUserCourseWatchLog(List<FsCourseWatchLog> logs, int batchSize) {
-        if (logs == null || logs.isEmpty()) {
-            log.info("待更新的日志列表为空，无需处理");
-            return;
-        }
+        try {
+            // 记录开始时间
+            long startTime = System.currentTimeMillis();
+            log.info("开始批量更新日志，总日志数量: {}", logs == null ? 0 : logs.size());
 
-        // 记录总日志数量
-        log.info("开始批量更新日志，总日志数量: {}", logs.size());
 
-        // 分批处理
-        for (int i = 0; i < logs.size(); i += batchSize) {
-            int end = Math.min(i + batchSize, logs.size());
-            List<FsCourseWatchLog> batchList = logs.subList(i, end);
+            if (logs == null || logs.isEmpty()) {
+                log.info("待更新的日志列表为空，无需处理");
+                return;
+            }
 
-            // 记录当前批次的数量
-            log.info("正在更新第 {} 批日志，数量: {}", (i / batchSize) + 1, batchList.size());
+            // 分批处理
+            for (int i = 0; i < logs.size(); i += batchSize) {
+                int end = Math.min(i + batchSize, logs.size());
+                List<FsCourseWatchLog> batchList = logs.subList(i, end);
+                // 执行批量更新
+                fsCourseWatchLogMapper.batchUpdateFsUserWatchLog(batchList);
+            }
 
-            // 执行批量更新
-            fsCourseWatchLogMapper.batchUpdateFsUserWatchLog(batchList);
 
-            // 记录当前批次更新完成
-            log.info("第 {} 批日志更新完成，数量: {}", (i / batchSize) + 1, batchList.size());
-        }
+            // 计算总耗时
+            long totalCost = System.currentTimeMillis() - startTime;
+            log.info("所有日志更新完成，总数量: {}, 总耗时: {}ms, 平均速度: {}/s",
+                    logs.size(),
+                    totalCost,
+                    totalCost > 0 ? String.format("%.2f", logs.size() * 1000.0 / totalCost) : "∞");
 
-        // 记录全部更新完成
-        log.info("所有日志更新完成，总日志数量: {}", logs.size());
+        }finally {
+            assert logs != null;
+            logs.clear(); // 提前释放引用 进入GC
+        }
     }
 
     @Override
@@ -835,144 +1053,144 @@ public class FsCourseWatchLogServiceImpl extends ServiceImpl<FsCourseWatchLogMap
     @Autowired
     private ISysConfigService configService;
 
-    @Override
-    public void scheduleBatchUpdateToDatabase() {
-        log.info("开始更新看课时长，检查完课>>>>>>");
-        //读取所有的key
-        Collection<String> keys = redisCache.keys("h5user:watch:duration:*");
-        //读取看课配置
-        String json = configService.selectConfigByKey("course.config");
-        CourseConfig config = JSONUtil.toBean(json, CourseConfig.class);
-
-        List<FsCourseWatchLog> logs = new ArrayList<>();
-        List<FsCourseWatchLog> finishedLogs = new ArrayList<>();
-        for (String key : keys) {
-            //取key中数据
-            Long qwUserId=null;
-            Long videoId=null;
-            Long externalId=null;
-            try {
-                String[] parts = key.split(":");
-                qwUserId = Long.parseLong(parts[3]);
-                externalId = Long.parseLong(parts[4]);
-                videoId = Long.parseLong(parts[5]);
-            }catch (Exception e){
-                log.error("key中id为null:{}", key);
-                continue;
-            }
-            String durationStr = redisCache.getCacheObject(key);
-            if (com.fs.common.utils.StringUtils.isEmpty(durationStr)) {
-                redisCache.deleteObject(key);
-                log.error("key中数据为null:{}", key);
-                continue;  // 如果 Redis 中没有记录，跳过
-            }
-            Long duration = Long.valueOf(durationStr);
-
-            FsCourseWatchLog watchLog = new FsCourseWatchLog();
-            watchLog.setVideoId(videoId);
-            watchLog.setQwUserId(qwUserId);
-            watchLog.setQwExternalContactId(externalId);
-            watchLog.setDuration(duration);
-
-            //取对应视频的时长
-            Long videoDuration = 0L;
-            try {
-                videoDuration = getVideoDuration(videoId);
-            }catch (Exception e){
-                log.error("视频时长识别错误:{}", key);
-                continue;
-            }
-
-            if (videoDuration != null && videoDuration != 0) {
-                boolean complete = false;
-                // 判断百分比
-                if(config.getCompletionMode() == 1 && config.getAnswerRate() != null){
-                    long percentage = (duration * 100 / videoDuration);
-                    complete = percentage >= config.getAnswerRate();
-                }
-                // 判断分钟数
-                if(config.getCompletionMode() == 2 && config.getMinutesNum() != null){
-                    int i = config.getMinutesNum() * 60;
-                    complete = videoDuration > i;
-                }
-                //判断是否完课
-                if (complete) {
-                    watchLog.setLogType(2); // 设置状态为“已完成”
-                    watchLog.setFinishTime(new Date());
-                    String heartbeatKey ="h5user:watch:heartbeat:" + qwUserId+ ":" + externalId + ":" + videoId;
-                    // 完课删除心跳记录
-                    redisCache.deleteObject(heartbeatKey);
-                    // 完课删除看课时长记录
-                    redisCache.deleteObject(key);
-
-                    finishedLogs.add(watchLog);
-                }
-            }
-            //集合中增加
-            logs.add(watchLog);
-        }
-
-        batchUpdateFsCourseWatchLog(logs,100);
-
-        // 完课打标签
-        if(CollectionUtils.isNotEmpty(finishedLogs)){
-            fsTagUpdateService.onCourseWatchFinishedBatch(finishedLogs);
-        }
-    }
-
-    @Override
-    public void checkWatchStatus() {
-        log.info("开始更新看课中断记录>>>>>");
-        // 从 Redis 中获取所有正在看课的用户记录
-        Collection<String> keys = redisCache.keys("h5user:watch:heartbeat:*");
-        LocalDateTime now = LocalDateTime.now();
-        List<FsCourseWatchLog> logs = new ArrayList<>();
-
-        List<FsCourseWatchLog> watchingLogs = new ArrayList<>();
-        for (String key : keys) {
-            FsCourseWatchLog watchLog = new FsCourseWatchLog();
-            //取key中数据
-            Long qwUserId=null;
-            Long videoId=null;
-            Long externalId=null;
-            try {
-                String[] parts = key.split(":");
-                qwUserId = Long.parseLong(parts[3]);
-                externalId = Long.parseLong(parts[4]);
-                videoId = Long.parseLong(parts[5]);
-            }catch (Exception e){
-                log.error("key中id为null:{}", key);
-                continue;
-            }
-            // 获取最后心跳时间
-            String lastHeartbeatStr = redisCache.getCacheObject(key);
-            if (com.fs.common.utils.StringUtils.isEmpty(lastHeartbeatStr)) {
-                redisCache.deleteObject(key);
-                continue; // 如果 Redis 中没有记录，跳过
-            }
-            LocalDateTime lastHeartbeatTime = LocalDateTime.parse(lastHeartbeatStr);
-            Duration duration = Duration.between(lastHeartbeatTime, now);
-
-            watchLog.setVideoId(videoId);
-            watchLog.setQwUserId(qwUserId);
-            watchLog.setQwExternalContactId(externalId);
-            // 如果超过一分钟没有心跳，标记为“观看中断”
-            if (duration.getSeconds() >= 60) {
-                watchLog.setLogType(4);
-                // 从 Redis 中删除该记录
-                redisCache.deleteObject(key);
-            }else {
-                watchLog.setLogType(1);
-                watchingLogs.add(watchLog);
-            }
-            logs.add(watchLog);
-        }
-        batchUpdateFsCourseWatchLog(logs,100);
-
-        if(CollectionUtils.isNotEmpty(watchingLogs)){
-            fsTagUpdateService.onCourseWatchingBatch(watchingLogs);
-        }
-    }
+//    @Override
+//    public void scheduleBatchUpdateToDatabase() {
+//        log.info("开始更新看课时长，检查完课>>>>>>");
+//        //读取所有的key
+//        Collection<String> keys = redisCache.keys("h5user:watch:duration:*");
+//        //读取看课配置
+//        String json = configService.selectConfigByKey("course.config");
+//        CourseConfig config = JSONUtil.toBean(json, CourseConfig.class);
+//
+//        List<FsCourseWatchLog> logs = new ArrayList<>();
+//        List<FsCourseWatchLog> finishedLogs = new ArrayList<>();
+//        for (String key : keys) {
+//            //取key中数据
+//            Long qwUserId=null;
+//            Long videoId=null;
+//            Long externalId=null;
+//            try {
+//                String[] parts = key.split(":");
+//                qwUserId = Long.parseLong(parts[3]);
+//                externalId = Long.parseLong(parts[4]);
+//                videoId = Long.parseLong(parts[5]);
+//            }catch (Exception e){
+//                log.error("key中id为null:{}", key);
+//                continue;
+//            }
+//            String durationStr = redisCache.getCacheObject(key);
+//            if (com.fs.common.utils.StringUtils.isEmpty(durationStr)) {
+//                redisCache.deleteObject(key);
+//                log.error("key中数据为null:{}", key);
+//                continue;  // 如果 Redis 中没有记录，跳过
+//            }
+//            Long duration = Long.valueOf(durationStr);
+//
+//            FsCourseWatchLog watchLog = new FsCourseWatchLog();
+//            watchLog.setVideoId(videoId);
+//            watchLog.setQwUserId(qwUserId);
+//            watchLog.setQwExternalContactId(externalId);
+//            watchLog.setDuration(duration);
+//
+//            //取对应视频的时长
+//            Long videoDuration = 0L;
+//            try {
+//                videoDuration = getVideoDuration(videoId);
+//            }catch (Exception e){
+//                log.error("视频时长识别错误:{}", key);
+//                continue;
+//            }
+//
+//            if (videoDuration != null && videoDuration != 0) {
+//                boolean complete = false;
+//                // 判断百分比
+//                if(config.getCompletionMode() == 1 && config.getAnswerRate() != null){
+//                    long percentage = (duration * 100 / videoDuration);
+//                    complete = percentage >= config.getAnswerRate();
+//                }
+//                // 判断分钟数
+//                if(config.getCompletionMode() == 2 && config.getMinutesNum() != null){
+//                    int i = config.getMinutesNum() * 60;
+//                    complete = videoDuration > i;
+//                }
+//                //判断是否完课
+//                if (complete) {
+//                    watchLog.setLogType(2); // 设置状态为“已完成”
+//                    watchLog.setFinishTime(new Date());
+//                    String heartbeatKey ="h5user:watch:heartbeat:" + qwUserId+ ":" + externalId + ":" + videoId;
+//                    // 完课删除心跳记录
+//                    redisCache.deleteObject(heartbeatKey);
+//                    // 完课删除看课时长记录
+//                    redisCache.deleteObject(key);
+//
+//                    finishedLogs.add(watchLog);
+//                }
+//            }
+//            //集合中增加
+//            logs.add(watchLog);
+//        }
+//
+//        batchUpdateFsCourseWatchLog(logs,100);
+//
+//        // 完课打标签
+//        if(CollectionUtils.isNotEmpty(finishedLogs)){
+//            fsTagUpdateService.onCourseWatchFinishedBatch(finishedLogs);
+//        }
+//    }
+
+//    @Override
+//    public void checkWatchStatus() {
+//        log.info("开始更新看课中断记录>>>>>");
+//        // 从 Redis 中获取所有正在看课的用户记录
+//        Collection<String> keys = redisCache.keys("h5user:watch:heartbeat:*");
+//        LocalDateTime now = LocalDateTime.now();
+//        List<FsCourseWatchLog> logs = new ArrayList<>();
+//
+//        List<FsCourseWatchLog> watchingLogs = new ArrayList<>();
+//        for (String key : keys) {
+//            FsCourseWatchLog watchLog = new FsCourseWatchLog();
+//            //取key中数据
+//            Long qwUserId=null;
+//            Long videoId=null;
+//            Long externalId=null;
+//            try {
+//                String[] parts = key.split(":");
+//                qwUserId = Long.parseLong(parts[3]);
+//                externalId = Long.parseLong(parts[4]);
+//                videoId = Long.parseLong(parts[5]);
+//            }catch (Exception e){
+//                log.error("key中id为null:{}", key);
+//                continue;
+//            }
+//            // 获取最后心跳时间
+//            String lastHeartbeatStr = redisCache.getCacheObject(key);
+//            if (com.fs.common.utils.StringUtils.isEmpty(lastHeartbeatStr)) {
+//                redisCache.deleteObject(key);
+//                continue; // 如果 Redis 中没有记录，跳过
+//            }
+//            LocalDateTime lastHeartbeatTime = LocalDateTime.parse(lastHeartbeatStr);
+//            Duration duration = Duration.between(lastHeartbeatTime, now);
+//
+//            watchLog.setVideoId(videoId);
+//            watchLog.setQwUserId(qwUserId);
+//            watchLog.setQwExternalContactId(externalId);
+//            // 如果超过一分钟没有心跳，标记为“观看中断”
+//            if (duration.getSeconds() >= 60) {
+//                watchLog.setLogType(4);
+//                // 从 Redis 中删除该记录
+//                redisCache.deleteObject(key);
+//            }else {
+//                watchLog.setLogType(1);
+//                watchingLogs.add(watchLog);
+//            }
+//            logs.add(watchLog);
+//        }
+//        batchUpdateFsCourseWatchLog(logs,100);
+//
+//        if(CollectionUtils.isNotEmpty(watchingLogs)){
+//            fsTagUpdateService.onCourseWatchingBatch(watchingLogs);
+//        }
+//    }
 
     @Override
     public List<FsCourseUserStatisticsListVO> selectFsCourseUserStatisticsListVO(FsCourseUserStatisticsListParam param) {

fs-service/src/main/java/com/fs/im/service/OpenIMService.java

@@ -116,4 +116,16 @@ public interface OpenIMService {
     void sendSystemMsgAsync(String userId, String text, String title);
 
     OpenImResponseDTO getUserInfo(String id);
+
+    /**
+     * 批量发送纯文本消息（不存库，仅发送）
+     * @param senderId 发送人ID
+     * @param receiverIds 接收人ID列表
+     * @param textContent 文本内容
+     * @param pushTitle 离线推送标题（可选）
+     * @param pushDesc 离线推送描述（可选）
+     * @param ex 扩展字段（可选）
+     * @return 发送结果
+     */
+     OpenImResponseDTO batchSendTextMessage(String senderId, List<String> receiverIds, String textContent, String pushTitle, String pushDesc, String ex);
 }

fs-service/src/main/java/com/fs/im/service/impl/OpenIMServiceImpl.java

@@ -1978,4 +1978,74 @@ public class OpenIMServiceImpl implements OpenIMService {
             OpenImResponseDTO responseDTO= JSONUtil.toBean(body,OpenImResponseDTO.class);
             return responseDTO;
     }
+
+
+    /**
+     * 批量发送纯文本消息（不存库，仅发送）
+     * @param senderId 发送人ID
+     * @param receiverIds 接收人ID列表
+     * @param textContent 文本内容
+     * @param pushTitle 离线推送标题（可选）
+     * @param pushDesc 离线推送描述（可选）
+     * @param ex 扩展字段（可选）
+     * @return 发送结果
+     */
+    @Override
+    public OpenImResponseDTO batchSendTextMessage(String senderId, List<String> receiverIds, String textContent,
+                                                  String pushTitle, String pushDesc, String ex) {
+        try {
+            log.info("批量发送文本消息 - 发送人: {}, 接收人数: {}, 内容: {}", senderId, receiverIds.size(), textContent);
+
+            // 构建批量发送消息DTO
+            OpenImBatchMsgDTO batchMsgDTO = new OpenImBatchMsgDTO();
+
+            // 必填字段
+            batchMsgDTO.setSendID(senderId);
+            batchMsgDTO.setRecvIDs(receiverIds);
+            batchMsgDTO.setContentType(101); // 101表示文本消息
+            batchMsgDTO.setSessionType(1);   // 1表示单聊
+
+            // 设置消息内容
+            OpenImBatchMsgDTO.Content content = new OpenImBatchMsgDTO.Content();
+            content.setContent(textContent);
+            batchMsgDTO.setContent(content);
+
+
+            // 可选字段 - 离线推送
+            if (StringUtils.isNotEmpty(pushTitle) || StringUtils.isNotEmpty(pushDesc)) {
+                OpenImBatchMsgDTO.OfflinePushInfo pushInfo = new OpenImBatchMsgDTO.OfflinePushInfo();
+                pushInfo.setTitle(StringUtils.isNotEmpty(pushTitle) ? pushTitle : "新消息");
+                pushInfo.setDesc(StringUtils.isNotEmpty(pushDesc) ? pushDesc : textContent);
+                batchMsgDTO.setOfflinePushInfo(pushInfo);
+            }
+
+            // 可选字段 - 扩展信息
+            if (StringUtils.isNotEmpty(ex)) {
+                batchMsgDTO.setEx(ex);
+            }
+
+            // 设置默认值
+            batchMsgDTO.setIsOnlineOnly(false);       // 非仅在线发送
+            batchMsgDTO.setNotOfflinePush(false);     // 允许离线推送
+            batchMsgDTO.setSendTime(System.currentTimeMillis()); // 当前时间
+
+            log.info("批量发送文本消息请求: {}", JSON.toJSONString(batchMsgDTO));
+
+            // 调用批量发送接口
+            OpenImResponseDTO response = this.openIMBatchSendMsg(batchMsgDTO);
+
+            log.info("批量发送文本消息结果 - errCode: {}, errMsg: {}",
+                    response != null ? response.getErrCode() : "null",
+                    response != null ? response.getErrMsg() : "null");
+
+            return response;
+
+        } catch (Exception e) {
+            log.error("批量发送文本消息异常 - 发送人: {}, 接收人数: {}", senderId, receiverIds.size(), e);
+            OpenImResponseDTO errorResponse = new OpenImResponseDTO();
+            errorResponse.setErrCode(500);
+            errorResponse.setErrMsg("发送失败: " + e.getMessage());
+            return errorResponse;
+        }
+    }
 }