package com.fs.sop.service.impl;


import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.fs.common.annotation.DataSource;
import com.fs.common.core.domain.R;
import com.fs.common.core.redis.RedisCache;
import com.fs.common.enums.DataSourceType;
import com.fs.common.utils.StringUtils;
import com.fs.company.domain.CompanyUser;
import com.fs.company.domain.CompanyUserDelayTime;
import com.fs.company.mapper.CompanyUserDelayTimeMapper;
import com.fs.company.service.ICompanyUserService;
import com.fs.course.config.CourseConfig;
import com.fs.course.domain.FsCourseWatchLog;
import com.fs.course.service.IFsCourseWatchLogService;
import com.fs.his.domain.FsUser;
import com.fs.his.mapper.FsUserMapper;
import com.fs.fastGpt.service.IFastGptChatSessionService;
import com.fs.qw.domain.QwApiSopLogToken;
import com.fs.qw.domain.QwExternalContact;
import com.fs.qw.domain.QwUser;
import com.fs.qw.mapper.QwExternalContactMapper;
import com.fs.qw.mapper.QwUserMapper;
import com.fs.qw.param.SopMsgParam;
import com.fs.qw.result.QwExternalContactByQwResult;
import com.fs.qw.service.IQwApiSopLogTokenService;
import com.fs.qw.service.impl.AsyncSopTestService;
import com.fs.qw.service.impl.QwExternalContactServiceImpl;
import com.fs.qw.vo.QwSopTempSetting;
import com.fs.qwApi.Result.QwAddMsgTemplateResult;
import com.fs.qwApi.Result.QwGroupmsgSendResult;
import com.fs.qwApi.Result.QwUploadImgResult;
import com.fs.qwApi.Result.SendItemResult;
import com.fs.qwApi.param.QwExternalContactHParam;
import com.fs.qwApi.param.QwGetGroupmsgSendParam;
import com.fs.qwApi.param.QwMsgTemplateSop;
import com.fs.qwApi.service.QwApiService;
import com.fs.sop.domain.QwSopLogs;
import com.fs.sop.mapper.QwSopLogsMapper;
import com.fs.sop.mapper.SopUserLogsInfoMapper;
import com.fs.sop.params.GetQwSopLogsByJsApiParam;
import com.fs.sop.params.QwSopLogsParam;
import com.fs.sop.params.SendSopParamDetailsC;
import com.fs.sop.service.IQwSopLogsService;
import com.fs.sop.vo.QwSopLogsDoSendListTVO;
import com.fs.sop.vo.QwSopLogsListCVO;
import com.fs.system.service.ISysConfigService;
import com.fs.voice.utils.StringUtil;
import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.stream.Collector;
import java.util.stream.Collectors;

/**
 * 企业微信SOP  定时任务Service业务层处理
 *
 * @author fs
 * @date 2024-08-05
 */
@Service
public class QwSopLogsServiceImpl extends ServiceImpl<QwSopLogsMapper, QwSopLogs> implements IQwSopLogsService
{

    private static final Logger logger = LoggerFactory.getLogger(QwSopLogsServiceImpl.class);
    public static final String FS_USER_BLACKLIST_REMARK = "该客户已经被管理员拉黑";

    /** 发群/群公告：目标为群聊，不按单个客户拉黑拦截 */
    private static boolean shouldSkipBlacklistCheck(Integer sendType) {
        return sendType != null && (sendType == 6 || sendType == 21);
    }
    @Autowired
    private RedisCache redisCache;
    @Autowired
    private QwSopLogsMapper qwSopLogsMapper;
    @Autowired
    private CompanyUserDelayTimeMapper companyUserDelayTimeMapper;

    @Autowired
    private ISysConfigService configService;

    @Autowired
    private QwApiService qwApiService;

    @Autowired
    private IQwSopLogsService qwSopLogsService;

    @Autowired
    private QwExternalContactServiceImpl qwExternalContactService;

    @Autowired
    @Lazy
    private IFsCourseWatchLogService watchLogService;

    @Autowired
    private QwUserMapper qwUserMapper;

    @Autowired
    private QwExternalContactMapper qwExternalContactMapper;

    @Autowired
    private SopUserLogsInfoMapper sopUserLogsInfoMapper;

    private AsyncSopTestService asyncSopTestService;

    @Autowired
    private IQwApiSopLogTokenService sopLogTokenService;

    @Autowired
    public void setAsyncSopTestService(@Lazy AsyncSopTestService asyncSopTestService){
        this.asyncSopTestService = asyncSopTestService;
    }


    @Lazy
    @Autowired
    private IFastGptChatSessionService fastGptChatSessionService;

    @Autowired
    private ICompanyUserService companyUserService;

    @Autowired
    private FsUserMapper fsUserMapper;


    /**
     * 查询企业微信SOP  定时任务
     *
     * @param id 企业微信SOP  定时任务主键
     * @return 企业微信SOP  定时任务
     */
    @Override
    public QwSopLogs selectQwSopLogsById(String id)
    {
        return qwSopLogsMapper.selectQwSopLogsById(id);
    }

    /**
     * 查询企业微信SOP  定时任务列表
     *
     * @param QwSopLogs 企业微信SOP  定时任务
     * @return 企业微信SOP  定时任务
     */
    @Override
    public List<QwSopLogs> selectQwSopLogsList(QwSopLogs QwSopLogs)
    {
        return qwSopLogsMapper.selectQwSopLogsList(QwSopLogs);
    }

    /**
     * 获取单条规则 的 执行记录历史
     */
    @Override
    public List<QwSopLogsListCVO> selectQwSopLogsListBySopId(QwSopLogsParam param) {
        if (param == null || param.getType() == null) {
            return Collections.emptyList();
        }

        switch (param.getType()) {
            // 个微
            case 1:
                return qwSopLogsMapper.selectQwSopLogsListByGwSopId(param);
            // 企微
            case 2:
                return handleQwSopLogs(param);
            default:
                return Collections.emptyList();
        }
    }

    private List<QwSopLogsListCVO> handleQwSopLogs(QwSopLogsParam param) {
        if (StringUtils.isNotEmpty(param.getPhone())) {
            try {
                CompanyUser companyUser = companyUserService.selectCompanyUserByPhone(param.getPhone());
                if (companyUser == null) {
                    return Collections.emptyList();
                }
                QwUser qwUser = new QwUser();
                qwUser.setCompanyUserId(companyUser.getUserId());
                List<QwUser> qwUsers = qwUserMapper.selectQwUserList(qwUser);
                if (CollectionUtils.isNotEmpty(qwUsers)){
                    List<String> qwUserIds = qwUsers.stream().map(QwUser::getQwUserId).collect(Collectors.toList());
                    param.setQwUseridByList(qwUserIds);
                }
            } catch (Exception e) {
                logger.error("根据手机号查询企业用户出现异常，手机号: {}, 异常信息: {}", param.getPhone(), e.getMessage(), e);
                return Collections.emptyList();
            }
        }
        return qwSopLogsMapper.selectQwSopLogsListByQwSopId(param);
    }


    /**
     * 新增企业微信SOP  定时任务
     *
     * @param QwSopLogs 企业微信SOP  定时任务
     * @return 结果
     */
    @Override
    public int insertQwSopLogs(QwSopLogs QwSopLogs)
    {
        return  qwSopLogsMapper.insertQwSopLogs(QwSopLogs);
    }

    /**
     * 修改企业微信SOP  定时任务
     *
     * @param QwSopLogs 企业微信SOP  定时任务
     * @return 结果
     */
    @Override
    public int updateQwSopLogs(QwSopLogs QwSopLogs)
    {
        return qwSopLogsMapper.updateQwSopLogs(QwSopLogs);
    }

    @Override
    public int editCourseQwSopLogs(String[] ids) {
        return qwSopLogsMapper.editCourseQwSopLogs(ids);
    }

    /**
     * 批量删除企业微信SOP  定时任务
     *
     * @param ids 需要删除的企业微信SOP  定时任务主键
     * @return 结果
     */
    @Override
    public int deleteQwSopLogsByIds(String[] ids)
    {
        return qwSopLogsMapper.deleteQwSopLogsByIds(ids);
    }

    /**
     * 删除企业微信SOP  定时任务信息
     *
     * @param id 企业微信SOP  定时任务主键
     * @return 结果
     */
    @Override
    public int deleteQwSopLogsById(String id)
    {
        return qwSopLogsMapper.deleteQwSopLogsById(id);
    }

    /**
     *  检索执行符合条件的定时任务
     */
    @Override
    public void checkQwSopLogs() {

        List<QwSopLogs> qwSopLogs = qwSopLogsMapper.checkQwSopLogs();

        //发送消息组件表
        QwMsgTemplateSop templateSop= new QwMsgTemplateSop();

        templateSop.setChatType("single");
        templateSop.setAllowSelect(false);
        List<QwMsgTemplateSop.Attachment> attachmentLists=new ArrayList<>();

        // 创建一个 Map，用于分组：key 是 qwUserid + sendTime + sopId，value 是对应的 QwSopLogs 列表
        Map<String, List<QwSopLogs>> groupedLogs = qwSopLogs.stream()
                .collect(Collectors.groupingBy(log -> log.getQwUserid() + "|"
                        + log.getSendTime() + "|"
                        + log.getSopId()+"|"+log.getCorpId()));

        // 遍历分组数据
        for (Map.Entry<String, List<QwSopLogs>> entry : groupedLogs.entrySet()) {
            String key = entry.getKey();
            // 分组的 key（qwUserid|sendTime|sopId）
            List<QwSopLogs> logs = entry.getValue(); // 分组对应的日志列表

            // 从 key 中提取 qwUserid
            String qwUserid = key.split("\\|")[0];
            String corpId  = key.split("\\|")[3];



            // 收集 externalUserId 的去重集合
            Set<String> externalUserIds = logs.stream()
                    .map(QwSopLogs::getExternalUserId)
                    .filter(Objects::nonNull) // 过滤 null 值
                    .collect(Collectors.toSet());
            List<String> externalUserIdsCollect = new ArrayList<>(externalUserIds);


            // 收集唯一的 contentJson
            Set<String> uniqueContentJsons = logs.stream()
                    .map(QwSopLogs::getContentJson)
                    .filter(Objects::nonNull) // 过滤 null 值
                    .collect(Collectors.toSet());

            List<String> uniqueContentJsonList = new ArrayList<>(uniqueContentJsons);

            // 获取 QwSopLogs 的 id 列表
            List<String> ids = logs.stream()
                    .map(QwSopLogs::getId)
                    .collect(Collectors.toList());
            List<String> collectIds= new ArrayList<>(ids);



            QwSopTempSetting.Content content = JSON.parseObject(uniqueContentJsonList.get(0), QwSopTempSetting.Content.class);

            List<QwSopTempSetting.Content.Setting> setting = content.getSetting();
            setting.forEach(set -> {
                String contentType = set.getContentType();
                switch (contentType){
                    //文本
                    case "1":
                        templateSop.setTextContent(set.getValue());
                        break;
                    //图片
                    case "2":
                        if(!StringUtil.strIsNullOrEmpty(set.getImgUrl())){

                            QwUploadImgResult uploadimg = null;
                            try {

                                uploadimg = qwApiService.uploadimgs(set.getImgUrl(), corpId);

                                if (uploadimg.getErrcode()!=0){
                                    logger.error("SOP企业微信接口图片上传失败:{}", uploadimg.getErrMsg());
                                }else {
                                    QwMsgTemplateSop.Attachment attachment=new QwMsgTemplateSop.Attachment();
                                    attachment.setType(2);
                                    attachment.setImagePicUrl(uploadimg.getUrl());
                                    attachmentLists.add(attachment);
                                }



                            } catch (Exception e) {
                                logger.error("SOP企业微信接口图片上传失败",e);
                            }

                        }

                        break;
                    //链接
                    case "3":
                        if(!StringUtil.strIsNullOrEmpty(set.getLinkImageUrl())){
                            try {
                                QwMsgTemplateSop.Attachment attachment=new QwMsgTemplateSop.Attachment();
                                attachment.setType(3);
                                attachment.setLinkTitle(set.getLinkTitle());
                                attachment.setLinkPicurl(set.getLinkImageUrl());
                                attachment.setLinkDesc(set.getLinkDescribe());
                                attachment.setLinkUrl(set.getLinkUrl());
                                attachmentLists.add(attachment);

                            } catch (Exception e) {
                                logger.error("SOP企业微信接口链接图片上传失败",e);
                            }

                        }

                        break;
                }

            });

            templateSop.setSender(qwUserid);
            templateSop.setAttachments(attachmentLists);

            // 设置批次大小，假设每批最多处理10000个用户
            int batchSize = 10000;
            int totalSize = externalUserIdsCollect.size();
            int startIndex = 0;

            // 按批次处理
            while (startIndex < totalSize) {
                // 获取当前批次的用户列表
                int endIndex = Math.min(startIndex + batchSize, totalSize);
                List<String> currentBatch = externalUserIdsCollect.subList(startIndex, endIndex);

                // 创建新的 templateSop 对象，并设置当前批次的 externalUseridList
                templateSop.setExternalUseridList(currentBatch);

                // 调用 API 发送消息模板
                QwAddMsgTemplateResult qwAddMsgTemplateResult = qwApiService.addMsgTemplateBySop(templateSop, corpId);

                // 批量更新
                if (qwAddMsgTemplateResult.getErrCode() == 0 || qwAddMsgTemplateResult.getErrCode() == 41063) {
                    qwSopLogsMapper.updateQwSopLogsBycollectIds(collectIds, 1L, qwAddMsgTemplateResult.getMsgId());
                } else {
                    qwSopLogsMapper.updateQwSopLogsBycollectIds(collectIds, 0L, qwAddMsgTemplateResult.getMsgId());
                }

                // 更新 startIndex 继续处理下一批次
                startIndex = endIndex;
            }

//            templateSop.setSender(qwUserid);
//            templateSop.setExternalUseridList(externalUserIdsCollect);
//            templateSop.setAttachments(attachmentLists);
//
//            QwAddMsgTemplateResult qwAddMsgTemplateResult = qwApiService.addMsgTemplateBySop(templateSop, corpId);
//
//            //批量更新
//            if (qwAddMsgTemplateResult.getErrCode() == 0 || qwAddMsgTemplateResult.getErrCode()==41063) {
//                qwSopLogsMapper.updateQwSopLogsBycollectIds(collectIds,1L,qwAddMsgTemplateResult.getMsgId());
//            }else {
//                qwSopLogsMapper.updateQwSopLogsBycollectIds(collectIds,0L,qwAddMsgTemplateResult.getMsgId());
//            }

        }

    }

    /**
     *  检查企微官方群发的发送结果
     *  未成功的更新为群发助手发送
     */
    @Override
    public void qwSopLogsResult() {
        logger.info("开始执行企业微信群发消息结果查询任务");
        long startTime = System.currentTimeMillis();

        List<QwSopLogs> qwSopLogs = qwSopLogsMapper.selectSopLogsByCreateCorpMassSendResult();
        if (qwSopLogs.isEmpty()) {
            logger.info("没有需要查询结果的群发消息记录");
            return;
        }

        logger.info("需要处理的群发消息记录数量: {}", qwSopLogs.size());

        // 按企业ID分组处理，避免并发请求超过企业微信接口限制
        Map<String, List<QwSopLogs>> corpGroupedLogs = qwSopLogs.stream()
                .collect(Collectors.groupingBy(QwSopLogs::getCorpId));

        logger.info("按企业ID分组后，共有{}个企业需要处理", corpGroupedLogs.size());

        // 创建线程池，只对企业进行多线程处理
        int threadCount = Math.min(10, Runtime.getRuntime().availableProcessors());
        ExecutorService executorService = Executors.newFixedThreadPool(threadCount);
        CountDownLatch mainLatch = new CountDownLatch(corpGroupedLogs.size());

        // 对每个企业创建一个任务
        for (Map.Entry<String, List<QwSopLogs>> entry : corpGroupedLogs.entrySet()) {
            String corpId = entry.getKey();
            List<QwSopLogs> corpLogs = entry.getValue();

            executorService.submit(() -> {
                try {
                    logger.info("开始处理企业ID: {}，记录数量: {}", corpId, corpLogs.size());
                    List<QwSopLogs> updateList = new ArrayList<>();

                    // 同步处理每条记录
                    for (QwSopLogs sopLog : corpLogs) {
                        try {
                            // 添加随机延迟，避免请求过于集中
                            Thread.sleep(100 + new Random().nextInt(100));

                            QwGetGroupmsgSendParam qwGetGroupmsgSendParam = new QwGetGroupmsgSendParam();
                            qwGetGroupmsgSendParam.setMsgid(sopLog.getMsgId());
                            qwGetGroupmsgSendParam.setUserid(sopLog.getQwUserid());

                            QwGroupmsgSendResult groupmsgSendResult = qwApiService.getGroupmsgSendResult(qwGetGroupmsgSendParam, sopLog.getCorpId());

                            if (groupmsgSendResult.getErrCode() == 0) {
                                List<SendItemResult> sendList = groupmsgSendResult.getSendList();
                                for (SendItemResult itemResult : sendList) {
                                    QwSopLogs logCopy = new QwSopLogs();
                                    logCopy.setId(sopLog.getId());

                                    if (itemResult.getStatus() != 1 && itemResult.getStatus() != 0) {
                                        logCopy.setSendType(2);
                                        logCopy.setSendStatus(3L);
                                        logCopy.setRemark("补发");
                                        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
                                        LocalDateTime currentTime = LocalDateTime.now();
                                        String newTimeString = currentTime.format(formatter);
                                        logCopy.setSendTime(newTimeString);
                                        logCopy.setSort(3);
                                    } else {

                                        logCopy.setReceivingStatus(Long.valueOf(itemResult.getStatus()));

                                        if (itemResult.getStatus() == 0) {
                                            logCopy.setSendStatus(5L);
                                            logCopy.setRemark("员工未发送,已作废");
                                        }

                                    }
                                    updateList.add(logCopy);
                                }
                            } else if (groupmsgSendResult.getErrCode() == 45033 || groupmsgSendResult.getErrCode() == 40001) {
                                // 接口调用频率超过限制或不合法的access_token
                                logger.warn("企业微信接口调用频率超限，企业ID: {}，错误码: {}，错误信息: {}，MsgId: {}",
                                        sopLog.getCorpId(), groupmsgSendResult.getErrCode(),
                                        groupmsgSendResult.getErrMsg(), sopLog.getMsgId());
                                // 增加更长的延迟后重试
                                Thread.sleep(2000 + new Random().nextInt(1000));

                                // 重新尝试请求
                                groupmsgSendResult = qwApiService.getGroupmsgSendResult(qwGetGroupmsgSendParam, sopLog.getCorpId());

                                if (groupmsgSendResult.getErrCode() == 0) {
                                    List<SendItemResult> sendList = groupmsgSendResult.getSendList();
                                    for (SendItemResult itemResult : sendList) {
                                        QwSopLogs logCopy = new QwSopLogs();
                                        logCopy.setId(sopLog.getId());

                                        if (itemResult.getStatus() != 1  && itemResult.getStatus() != 0 ) {
                                            logCopy.setSendType(2);
                                            logCopy.setSendStatus(3L);
                                            logCopy.setSort(3);
                                            logCopy.setRemark("补发");
                                            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
                                            LocalDateTime currentTime = LocalDateTime.now();
                                            String newTimeString = currentTime.format(formatter);
                                            logCopy.setSendTime(newTimeString);
                                        }else {
                                            logCopy.setReceivingStatus(Long.valueOf(itemResult.getStatus()));
                                            if (itemResult.getStatus() == 0) {
                                                logCopy.setSendStatus(5L);
                                                logCopy.setRemark("员工未发送,已作废");
                                            }
                                        }
                                        updateList.add(logCopy);
                                    }
                                } else {
                                    logger.error("重试后仍获取群发消息结果失败，企业ID: {}，错误码: {}，错误信息: {}，MsgId: {}",
                                            sopLog.getCorpId(), groupmsgSendResult.getErrCode(),
                                            groupmsgSendResult.getErrMsg(), sopLog.getMsgId());
                                }
                            } else {
                                logger.warn("获取群发消息结果失败，企业ID: {}，错误码: {}，错误信息: {}，MsgId: {}",
                                        sopLog.getCorpId(), groupmsgSendResult.getErrCode(),
                                        groupmsgSendResult.getErrMsg(), sopLog.getMsgId());
                            }

                            // 每处理50条记录，批量更新一次数据库
                            if (updateList.size() >= 50) {
                                batchUpdateDatabase(updateList);
                                updateList.clear();
                            }
                        } catch (Exception e) {
                            logger.error("处理群发消息结果异常，企业ID: {}，MsgId: {}", sopLog.getCorpId(), sopLog.getMsgId(), e);
                        }
                    }

                    // 处理剩余的更新
                    if (!updateList.isEmpty()) {
                        batchUpdateDatabase(updateList);
                    }

                    logger.info("企业ID: {} 处理完成", corpId);
                } catch (Exception e) {
                    logger.error("处理企业群发消息结果异常，企业ID: {}", corpId, e);
                } finally {
                    mainLatch.countDown();
                }
            });
        }

        try {
            // 等待所有企业处理完成
            boolean completed = mainLatch.await(30, TimeUnit.MINUTES);
            if (!completed) {
                logger.warn("群发消息结果查询任务未在规定时间内完成");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            logger.error("等待群发消息结果查询任务完成时被中断", e);
        } finally {
            executorService.shutdown();
        }

        long endTime = System.currentTimeMillis();
        logger.info("企业微信群发消息结果查询任务完成，处理记录总数: {}，总耗时: {}ms",
                qwSopLogs.size(), (endTime - startTime));
    }

    @Override
    public void qwSopLogsResultNew() {


        logger.info("开始执行企业微信群发消息结果查询任务");
        long startTime = System.currentTimeMillis();

        List<QwSopLogs> qwSopLogsList = qwSopLogsMapper.selectSopLogsByCreateCorpMassSendResult();
        if (qwSopLogsList.isEmpty()) {
            return ;
        }

        // 分组处理
        Map<String, List<QwSopLogs>> grouped = qwSopLogsList.stream().collect(
                Collectors.groupingBy(log -> log.getQwUserid() + "|" + log.getCorpId() + "|" + log.getMsgId())
        );

        // 线程安全的数据结构
        List<QwSopLogs> allUpdates = Collections.synchronizedList(new ArrayList<>());
        Queue<Map.Entry<String, List<QwSopLogs>>> taskQueue = new ConcurrentLinkedQueue<>(grouped.entrySet());
        Queue<Map.Entry<String, List<QwSopLogs>>> apiFailedTasks = new ConcurrentLinkedQueue<>();
        ExecutorService executor = Executors.newFixedThreadPool(10);

        AtomicInteger totalGroupsProcessed = new AtomicInteger(0); // 处理的分组数
        AtomicInteger totalRecordsUpdated = new AtomicInteger(0);  // 更新的记录数

        int batchSize = 300; // 每300个组批量更新一次
        int maxRetries = 3;   // 最大重试次数

        // 处理所有任务（包括重试）
        for (int retryCount = 0; retryCount <= maxRetries; retryCount++) {
            int currentBatchCount = 0;

            while (!taskQueue.isEmpty()) {
                int currentBatchSize = Math.min(batchSize - currentBatchCount, taskQueue.size());
                CountDownLatch batchLatch = new CountDownLatch(currentBatchSize);

                // 处理当前批次任务
                for (int i = 0; i < currentBatchSize; i++) {
                    Map.Entry<String, List<QwSopLogs>> entry = taskQueue.poll();
                    executor.submit(() -> {
                        try {
                            // 处理单个分组（传入apiFailedTasks）
                            List<QwSopLogs> updates = processSingleGroup(entry, apiFailedTasks);
                            if (updates != null) {
                                synchronized (allUpdates) {
                                    allUpdates.addAll(updates);
                                }
                                // 统计更新记录数
                                totalRecordsUpdated.addAndGet(updates.size());
                            }

                            // 统计处理的分组数
                            totalGroupsProcessed.incrementAndGet();
                        } finally {
                            batchLatch.countDown();
                        }
                    });
                }

                // 等待当前批次完成
                try {
                    batchLatch.await();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }

                currentBatchCount += currentBatchSize;

                // 每处理1000个组或任务队列空时更新数据库
                if (currentBatchCount >= batchSize || taskQueue.isEmpty()) {
                    synchronized (allUpdates) {
                        if (!allUpdates.isEmpty()) {
                            batchUpdateDatabase(new ArrayList<>(allUpdates));
                            logger.info("每处理300个组或任务队列空时更新数据库："+new ArrayList<>(allUpdates).size());
                            allUpdates.clear();
                        }

                    }
                    currentBatchCount = 0;
                }
            }

            // 准备下一轮重试
            if (retryCount < maxRetries) {
                taskQueue.addAll(apiFailedTasks);
                apiFailedTasks.clear();
            }
        }

        // 最终更新剩余数据
        synchronized (allUpdates) {
            if (!allUpdates.isEmpty()) {
                logger.info("最终更新剩余数据："+new ArrayList<>(allUpdates).size());
                batchUpdateDatabase(new ArrayList<>(allUpdates));
            }
        }

        executor.shutdown();
        try {
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                executor.shutdownNow();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }

        long endTime = System.currentTimeMillis();
        logger.info("成功处理分组数: " + totalGroupsProcessed.get());
        logger.info("企业微信群发消息结果查询任务完成-更新记录总数:{},总耗时:{} " , totalRecordsUpdated.get(),(endTime - startTime));
    }

    private List<QwSopLogs> processSingleGroup(Map.Entry<String, List<QwSopLogs>> entry, Queue<Map.Entry<String, List<QwSopLogs>>> apiFailedTasks) {

        String key = entry.getKey();
        List<QwSopLogs> corpLogs = entry.getValue();
        String[] keys = key.split("\\|");
        String qwUserid = keys[0];
        String corpId = keys[1];
        String msgID = keys[2];

        QwGetGroupmsgSendParam param = new QwGetGroupmsgSendParam();
        param.setMsgid(msgID);
        param.setUserid(qwUserid);
        param.setLimit(1000);

        List<QwSopLogs> groupUpdates = new ArrayList<>();
        String nextCursor = null;
        boolean apiSuccess = true;

        do {
            param.setCursor(nextCursor);
            QwGroupmsgSendResult result = fetchWithRetry(param, corpId); // 重试3次

            // API调用失败处理
            if (result == null || result.getErrCode() != 0) {
                apiSuccess = false;
                break;
            }

            // 处理当前页结果
            List<QwSopLogs> pageUpdates = processPageResult(result, corpLogs, corpId, msgID);
            groupUpdates.addAll(pageUpdates);
            nextCursor = result.getNextCursor();

        } while (nextCursor != null && !nextCursor.isEmpty());

        // API调用失败时返回null，将任务加入失败队列
        if (!apiSuccess) {
            apiFailedTasks.offer(entry);
            return null;
        }

        return groupUpdates;
    }

    private QwGroupmsgSendResult fetchWithRetry(QwGetGroupmsgSendParam param, String corpId) {
        int retryCount = 0;
        while (retryCount <= 3) {
            QwGroupmsgSendResult result = qwApiService.getGroupmsgSendResult(param, corpId);

            // 请求失败情况
            if (result == null) {
                retryCount++;
                sleepWithJitter(2000);
                continue;
            }

            // 成功情况
            if (result.getErrCode() == 0) {
                return result;
            }

            // 需要重试的错误码
            if (result.getErrCode() == 45033 || result.getErrCode() == -1) {
                retryCount++;
                sleepWithJitter(2000 + new Random().nextInt(1000));
                continue;
            }

            // 其他错误码不重试
            return result;
        }
        return null; // 超过重试次数
    }

    private void sleepWithJitter(int baseMillis) {
        try {
            int sleepTime = baseMillis + (new Random().nextInt(1000));
            Thread.sleep(sleepTime);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    private List<QwSopLogs> processPageResult(QwGroupmsgSendResult result, List<QwSopLogs> logs, String corpId, String msgId) {
        Map<String, SendItemResult> sendMap = result.getSendList().stream()
                .collect(Collectors.toMap(
                        r -> r.getUserId() + "_" + r.getExternalUserId() + "_" + corpId + "_" + msgId,
                        Function.identity(),
                        (a, b) -> a
                ));

        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        String now = LocalDateTime.now().format(formatter);
        List<QwSopLogs> matchedLogs = new ArrayList<>();

        for (QwSopLogs log : logs) {
            String logKey = log.getQwUserid() + "_" + log.getExternalUserId() + "_" + log.getCorpId() + "_" + msgId;
            SendItemResult matched = sendMap.get(logKey);
            if (matched == null) continue;

            switch (matched.getStatus()) {
                case 0:
                    log.setSendStatus(5L);
                    log.setRemark("员工未在规定时间发送");
                    break;
                case 1:
                    log.setSendStatus(1L);
                    log.setReceivingStatus(1L);
                    break;
                case 2:
                    log.setSendType(2);
                    log.setSendStatus(3L);
                    log.setRemark("因客户不是好友导致发送失败,补发");
                    log.setReceivingStatus(0L);
                    log.setSendTime(now);
                    log.setSort(30000001);
                    break;
                case 3:
                    log.setSendType(2);
                    log.setSendStatus(3L);
                    log.setRemark("客户已经收到其他群发消息,补发");
                    log.setReceivingStatus(0L);
                    log.setSendTime(now);
                    log.setSort(30000001);
                    break;
            }
            matchedLogs.add(log);
        }
        return matchedLogs;
    }

//    @Override
//    public void qwSopLogsResultNew() {
//
//        logger.info("开始执行企业微信群发消息结果查询任务");
//        long startTime = System.currentTimeMillis();
//
//        List<QwSopLogs> qwSopLogsList = qwSopLogsMapper.selectSopLogsByCreateCorpMassSendResult();
//        if (qwSopLogsList.isEmpty()) {
//            logger.info("没有需要查询结果的群发消息记录");
//            return;
//        }
//
//
//        Map<String, List<QwSopLogs>> grouped = qwSopLogsList.stream().collect(
//                Collectors.groupingBy(log -> log.getQwUserid() + "|" + log.getCorpId() + "|" + log.getMsgId())
//        );
//        for (Map.Entry<String, List<QwSopLogs>> entry : grouped.entrySet()) {
//            String key = entry.getKey();
//            List<QwSopLogs> corpLogs = entry.getValue();
//
//            String[] keys = key.split("\\|");
//            String qwUserid = keys[0];
//            String corpId = keys[1];
//            String msgID = keys[2];
//
//            QwGetGroupmsgSendParam param = new QwGetGroupmsgSendParam();
//            param.setMsgid(msgID);
//            param.setUserid(qwUserid);
//            param.setLimit(1000);
//
//            fetchAndProcessAllPages(param, corpId, corpLogs, msgID);
//
//        }
//
//        long endTime = System.currentTimeMillis();
//        logger.info("企业微信群发消息结果查询任务完成，处理记录总数: {}，总耗时: {}ms",
//                qwSopLogsList.size(), (endTime - startTime));
//
//    }
//    private void fetchAndProcessAllPages(QwGetGroupmsgSendParam param, String corpId, List<QwSopLogs> logs, String msgId) {
//        String nextCursor = null;
//
//        do {
//            param.setCursor(nextCursor);
//            QwGroupmsgSendResult result = qwApiService.getGroupmsgSendResult(param, corpId);
//
//            if (result == null) {
//                logger.error("接口调用失败: {}", param);
//                return;
//            }
//
//            if (result.getErrCode() == 45033) {
//                try {
//                    Thread.sleep(2000 + new Random().nextInt(1000));
//                    result = qwApiService.getGroupmsgSendResult(param, corpId);
//                } catch (InterruptedException e) {
//                    Thread.currentThread().interrupt();
//                    logger.error("线程中断", e);
//                    return;
//                }
//            }
//
//            if (result.getErrCode() != 0) {
//                logger.error("查询失败: {}, errCode: {}, errMsg: {}", param, result.getErrCode(), result.getErrMsg());
//                return;
//            }
//
//            processPageResult(result, logs, corpId, msgId);
//            nextCursor = result.getNextCursor();
//
//        } while (nextCursor != null && !nextCursor.isEmpty());
//    }
//    private void processPageResult(QwGroupmsgSendResult result, List<QwSopLogs> logs, String corpId, String msgId) {
//        Map<String, SendItemResult> sendMap = result.getSendList().stream()
//                .collect(Collectors.toMap(
//                        r -> r.getUserId() + "_" + r.getExternalUserId() + "_" + corpId + "_" + msgId,
//                        Function.identity(),
//                        (a, b) -> a  // 如果重复，保留第一个
//                ));
//
//        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
//        String now = LocalDateTime.now().format(formatter);
//
//        // 只处理匹配得上的记录
//        List<QwSopLogs> matchedLogs = new ArrayList<>();
//
//        for (QwSopLogs log : logs) {
//            String logKey = log.getQwUserid() + "_" + log.getExternalUserId() + "_" + log.getCorpId() + "_" + msgId;
//            SendItemResult matched = sendMap.get(logKey);
//
//            if (matched != null) {
//
//                switch (matched.getStatus()) {
//                    case 0:
//                        log.setSendStatus(5L);
//                        log.setRemark("员工未在规定时间发送");
//                        break;
//                    case 1:
//                        log.setSendStatus(1L);
//                        log.setReceivingStatus(1L);
//                        break;
//                    case 2:
//                        log.setSendType(2);
//                        log.setSendStatus(3L);
//                        log.setRemark("因客户不是好友导致发送失败,补发");
//                        log.setReceivingStatus(0L);
//                        log.setSendTime(now);
//                        log.setSort(30000001);
//                    case 3:
//                        log.setSendType(2);
//                        log.setSendStatus(3L);
//                        log.setRemark("客户已经收到其他群发消息,补发");
//                        log.setReceivingStatus(0L);
//                        log.setSendTime(now);
//                        log.setSort(30000001);
//                        break;
//                    default:
//                        break;
//                }
//
//                matchedLogs.add(log);
//            }
//        }
//
//        if (!matchedLogs.isEmpty()) {
//            batchUpdateDatabase(matchedLogs);
//        }
//    }


    /**
     * 批量更新数据库
     * @param updateList 需要更新的记录列表
     */
    private void batchUpdateDatabase(List<QwSopLogs> updateList) {
        int updateBatchSize = 500;
        for (int i = 0; i < updateList.size(); i += updateBatchSize) {
            int endIndex = Math.min(i + updateBatchSize, updateList.size());
            List<QwSopLogs> batch = updateList.subList(i, endIndex);
            try {
                qwSopLogsMapper.batchUpdateStatusByQwResult(batch);
                logger.info("批量修改 sopLogs 成功，修改数量: " + batch.size());
            } catch (Exception e) {
                logger.error("批量修改 sopLogs 失败", e);
            }
        }
    }

    @Override
    public int updateQwSopLogsByWatchLogType(String id,String remark) {
        return qwSopLogsMapper.updateQwSopLogsByWatchLogType(id,remark);
    }

    @Override
    public List<QwSopLogsDoSendListTVO> selectQwSopLogsDoSendList(String qwUserId,String corpId) {
        return qwSopLogsMapper.selectQwSopLogsDoSendList(qwUserId,corpId);
    }

    @Override
    public int updateQwSopLogsBySendStatus() {
        return qwSopLogsMapper.updateQwSopLogsBySendStatus();
    }

    @Override
    public int updateQwSopLogsSendType(QwSopLogs qwSopLogs) {
       return qwSopLogsMapper.updateQwSopLogs(qwSopLogs);
    }

    @Override
    public List<QwSopLogs> selectQwSopLogsListVO(SopMsgParam param) {
        return qwSopLogsMapper.selectQwSopLogsListVO(param);
    }

    @Override
    public SendSopParamDetailsC getQwSopLogsByJsApi(GetQwSopLogsByJsApiParam param) {
        //写入企业微信 在线状态 10分钟 通过订阅方式，如果10分钟没有请求此接口，调用一个LOGIN方法
//        redisCache.setCacheObject("qwActive:"+param.getCorpId()+":"+param.getQwUserId(),1,30 ,TimeUnit.MINUTES);

        // 封装消息体
        SendSopParamDetailsC sopParamDetailsC = new SendSopParamDetailsC();
        // 查询员工信息的id
        QwUser qwUser = qwExternalContactService.getQwUserByRedis(param.getCorpId().trim(),param.getQwUserId().trim());
        // 如果发送方式不等于侧边栏跳过
        if(qwUser.getSendMsgType() != 0){
            return sopParamDetailsC;
        }

        logger.info("主动获取发送信息:"+param);


//        sendDelayTime(sopParamDetailsC,param);
        // 使用线程安全的集合
        List<QwSopLogsDoSendListTVO> sendJsApiList = Collections.synchronizedList(new ArrayList<>());
        try {

            List<QwSopLogsDoSendListTVO> logsByJsApi = qwSopLogsMapper.getQwSopLogsByJsApiAll(param);


//            // 优先返回 sendType == 8 的第一条记录
//            List<QwSopLogsDoSendListTVO> result = logsByJsApi.stream()
//                    .filter(log -> log.getSendType() == 8)
//                    .findFirst()
//                    .map(Collections::singletonList) // 单元素不可变 List
//                    .orElse(logsByJsApi); // 如果没有匹配项，返回原列表
            // 优先返回 sendType == 8 的第一条记录
            List<QwSopLogsDoSendListTVO> result = logsByJsApi.stream()
                    .collect(Collector.of(
                            () -> new Object() {
                                QwSopLogsDoSendListTVO type8 = null;
                                List<QwSopLogsDoSendListTVO> type9List = new ArrayList<>();
                                List<QwSopLogsDoSendListTVO> otherList = new ArrayList<>();
                            },
                            (acc, log) -> {
                                if (acc.type8 == null && log.getSendType() == 8) {
                                    acc.type8 = log;
                                } else if (acc.type8 == null) {
                                    if (log.getSendType() == 9) {
                                        acc.type9List.add(log);
                                    } else {
                                        acc.otherList.add(log);
                                    }
                                }
                            },
                            (acc1, acc2) -> { throw new UnsupportedOperationException(); },
                            acc -> {
                                if (acc.type8 != null) {
                                    return Collections.singletonList(acc.type8);
                                } else if (!acc.type9List.isEmpty()) {
                                    List<QwSopLogsDoSendListTVO> combined = new ArrayList<>(acc.type9List.size() + acc.otherList.size());
                                    combined.addAll(acc.type9List);
                                    combined.addAll(acc.otherList);
                                    return combined;
                                }
                                return logsByJsApi;
                            }
                    ));



            Long qwId=qwUser.getId();

            sopParamDetailsC.setAppKey(qwUser.getAppKey());
            sopParamDetailsC.setIsAuto(qwUser.getIsAuto());

            // 客户的信息
            QwExternalContactHParam contactHParam = new QwExternalContactHParam();
            contactHParam.setUserId(param.getQwUserId().trim());
            contactHParam.setExternalUserId(param.getExternalUserId().trim());
            contactHParam.setCorpId(param.getCorpId().trim());
            Long qwExternalContactId = qwExternalContactMapper.getQwExternalContactId(contactHParam);

            for (QwSopLogsDoSendListTVO log : result) {

                try {
                    if (!shouldSkipBlacklistCheck(log.getSendType())
                            && log.getFsUserId() != null && isFsUserBlacklisted(log.getFsUserId())) {
                        qwSopLogsService.updateQwSopLogsByWatchLogType(log.getId(), FS_USER_BLACKLIST_REMARK);
                        continue;
                    }
                    if (log.getSendType()==8){
                        //(AI消息不做任何判断 直接入)切 只取了一条
                        sendJsApiList.add(log);

                    }else {
                        switch (log.getSendType()) {
                            case 3:
                            case 7:
                                try {
                                    // 使用现代的日期时间 API
                                    LocalDateTime sendTime = log.getSendTime().toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime();
                                    LocalDateTime expiryDateTime = sendTime.plusHours(12);
                                    LocalDateTime now = LocalDateTime.now();

                                    // 判断是否过期
                                    if (now.isAfter(expiryDateTime)) {
                                        // 作废消息
                                        qwSopLogsService.updateQwSopLogsByWatchLogType(log.getId(), "已过期，不发送");
                                    } else {
                                        sendJsApiList.add(log);
                                    }
                                } catch (Exception e) {
                                    // 记录错误日志
                                    logger.error("Error processing log in logsByJsApiOver: {}", log.getId() + ":" + e);
                                }
                                break;
                            case 6:
                            case 9:
                            case 10:
                                sendJsApiList.add(log);
                                break;
                            default:
                                QwSopTempSetting.Content content = JSON.parseObject(log.getContentJson(), QwSopTempSetting.Content.class);
                                List<QwSopTempSetting.Content.Setting> setting = content.getSetting().stream().filter(e -> "9".equals(e.getContentType())).collect(Collectors.toList());

                                //有app的异步推送消息
                                if (!setting.isEmpty()) {
//                                    asyncSopTestService.asyncSendMsgBySopAppLinkNormal(setting, param.getExternalId());
                                    asyncSopTestService.asyncSendMsgBySopAppLinkNormalIM(setting, param.getCorpId(),qwUser.getCompanyUserId(),log.getFsUserId());
                                }

                                if (log.getExpiryTime() == null) {
                                    // 作废消息
                                    qwSopLogsService.updateQwSopLogsByWatchLogType(log.getId(), "SOP任务被删除");
                                } else {
                                    Integer expiryTime = log.getExpiryTime();
                                    LocalDateTime sendTime = log.getSendTime().toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime();
                                    LocalDateTime expiryDateTime = sendTime.plusHours(expiryTime);
                                    LocalDateTime now = LocalDateTime.now();

                                    // 判断是否过期
                                    if (now.isAfter(expiryDateTime)) {
                                        // 作废消息
                                        qwSopLogsService.updateQwSopLogsByWatchLogType(log.getId(), "已过期，不发送");
                                    } else {
                                        switch (content.getType()) {
                                            case 1:
                                                // 普通消息，不做判断，加入发送列表
                                                sendJsApiList.add(log);
                                                break;
                                            case 2:
                                                // 课程消息，进行复杂的条件判断
                                                FsCourseWatchLog watchLog = watchLogService.getWatchCourseLogVideoBySop(
                                                        Long.valueOf(content.getVideoId()),
                                                        String.valueOf(qwId),
                                                        qwExternalContactId
                                                );

                                                Integer courseType = content.getCourseType();
                                                String logId = log.getId();

                                                if (content.getCourseType() == null) {
                                                    qwSopLogsService.updateQwSopLogsByWatchLogType(logId, "模板未选消息类型，不发送");
                                                }
                                                if (watchLog != null) {

                                                    //新逻辑
                                                    if (isCourseTypeValid(courseType, watchLog.getLogType())) {
                                                        // 加入到发送列表中
                                                        sendJsApiList.add(log);
                                                    } else {
                                                        // 作废消息
                                                        qwSopLogsService.updateQwSopLogsByWatchLogType(logId, "看课状态未满足，不发送");
                                                    }


                                                } else {
                                                    // 没有观看记录，只发普通消息
                                                    if (courseType == 0) {
                                                        sendJsApiList.add(log);
                                                    } else {
                                                        qwSopLogsService.updateQwSopLogsByWatchLogType(logId, "无观看记录，不发送");
                                                    }
                                                }
                                                break;
                                            case 3:
                                                // 订单消息，不做判断，加入发送列表
                                                sendJsApiList.add(log);
                                                break;
                                            case 4:
                                                // Ai消息，不加入序列单独处理
                                                try {
                                                    logger.info("Ai触达消息 : {}", log.getId());
                                                    //  fastGptChatSessionService.sendAiTouch(log,param.getCorpId().trim());
                                                } catch (Exception e) {
                                                    logger.error("Ai消息 : {}", content.getType());
                                                }
                                                break;
                                            default:
                                                // 未知类型，记录警告
                                                logger.error("Unknown content type logsByJsApi: {}", content.getType());
                                                break;
                                        }
                                    }
                                }
                                break;
                        }
                    }
                } catch (Exception e) {
                    // 记录错误日志
                    logger.error("Error processing log in logsByJsApi: {}", log.getId()+" : " + e);
                }
            }
            //获取外部联系人
            QwSopLogs qwSopLogsByJsApiNext = qwSopLogsMapper.getQwSopLogsByJsApiNext(param);

            if (qwSopLogsByJsApiNext != null && !StringUtil.strIsNullOrEmpty(qwSopLogsByJsApiNext.getExternalUserId())){
                //过滤非外部联系人
                try{
                    List<String> exIds=redisCache.getCacheObject("qwUserExtIds:"+param.getCorpId()+":"+param.getQwUserId());
                    if(StringUtils.isNotEmpty(exIds)&& !exIds.isEmpty()){
                        //查询外部联系人是否存在，如果不存在，返回下一个外部联系人
                        if(!exIds.contains(qwSopLogsByJsApiNext.getExternalUserId())){
                            //更新此外部联系人所有待发记录为作废
                            param.setExternalUserId(qwSopLogsByJsApiNext.getExternalUserId());
                            qwSopLogsMapper.updateQwSopLogsByDeleted(param);
                            //获取下一个外部联系人 最大重试10次
                            for(int i=0;i<10;i++){
                                qwSopLogsByJsApiNext = qwSopLogsMapper.getQwSopLogsByJsApiNext(param);
                                if(qwSopLogsByJsApiNext!=null&&StringUtils.isNotEmpty(qwSopLogsByJsApiNext.getExternalUserId())){
                                    //判断非外部联系人
                                    if(!exIds.contains(qwSopLogsByJsApiNext.getExternalUserId())){
                                        param.setExternalUserId(qwSopLogsByJsApiNext.getExternalUserId());
                                        qwSopLogsMapper.updateQwSopLogsByDeleted(param);
                                    }
                                    else{
                                        break;
                                    }
                                }
                                else {
                                    break;
                                }
                            }
                        }
                    }
                }
                catch (Exception e){
                }
                sopParamDetailsC.setNextExternalUserId(qwSopLogsByJsApiNext.getExternalUserId().trim());
                sopParamDetailsC.setNextExternalId(qwSopLogsByJsApiNext.getExternalId());
                sopParamDetailsC.setType(qwSopLogsByJsApiNext.getSendType() == 6 ? 1 : 0);
            }

            sopParamDetailsC.setSendList(new ArrayList<>(sendJsApiList));

        }catch (Exception e){
            e.printStackTrace();
            logger.error("主动获取发送信息失败：-"+e);
        }

        return sopParamDetailsC;
    }

    /**
     * 查询是否存在自定义延时时间并赋值
     *
     * @param sopParamDetailsC
     * @param param
     */
    private void sendDelayTime(SendSopParamDetailsC sopParamDetailsC, GetQwSopLogsByJsApiParam param) {
        QwUser qwUser  = qwUserMapper.selectQwUserByQwUserId(param.getQwUserId());
        CompanyUserDelayTime companyUserDelayTime = companyUserDelayTimeMapper.selectUserDelayTimeByCompanyUserId(qwUser.getCompanyUserId(),qwUser.getCompanyId());
        if (ObjectUtil.isEmpty(companyUserDelayTime)){
            String json = configService.selectConfigByKey("course.config");
            CourseConfig config = JSONUtil.toBean(json, CourseConfig.class);
            sopParamDetailsC.setSendDelayTime(config.getSendDelayTime());
        }else {
            sopParamDetailsC.setSendDelayTime(companyUserDelayTime.getSendDelayTime());
        }
    }

    // 逻辑封装方法
    public static boolean isCourseTypeValid(int courseType, Integer logType) {
        switch (courseType) {
            case 0:
                return true;  // courseType == 0 直接返回 true
            case 1:
                return logType==3;  // logType == "3" 时返回 true
            case 2:
                return logType==4;  // logType == "4" 时返回 true
            case 3:
                return logType==2;  // logType == "2" 时返回 true
            case 4:
                return logType==3 || logType==4;  // logType == "3" 或 "4" 时返回 true
            default:
                return false;  // 其他情况返回 false
        }
    }



//    @Override
//    public SendSopParamDetailsC getQwSopLogsByJsApi(GetQwSopLogsByJsApiParam param) {
//
//        // 封装消息体
//        SendSopParamDetailsC sopParamDetailsC = new SendSopParamDetailsC();
//
//        // 使用线程安全的集合
//        List<QwSopLogsDoSendListTVO> sendJsApiList = Collections.synchronizedList(new ArrayList<>());
//        List<QwSopLogsDoSendListTVO> sendJsApiListOver = Collections.synchronizedList(new ArrayList<>());
//
//        // 创建一个自定义的线程池，根据实际需求调整线程数
//        ExecutorService executor = Executors.newFixedThreadPool(4);
//
//        try {
//            // Step 1: 批量更新已过期的记录
//            List<QwSopLogsDoSendListTVO> logsByJsApiNotExtId = qwSopLogsMapper.getQwSopLogsByJsApiNotExtId(param);
//            if (!logsByJsApiNotExtId.isEmpty()) {
//                processAndInsertQwSopLogs(logsByJsApiNotExtId);
//            }
//
//            // Step 2: 并行处理 logsByJsApiOver 和 logsByJsApi
//            CompletableFuture<Void> futureOver = CompletableFuture.runAsync(() -> {
//                // 查询当前客户的完课消息
//                List<QwSopLogsDoSendListTVO> logsByJsApiOver = qwSopLogsMapper.getQwSopLogsByJsApiOver(param);
//                for (QwSopLogsDoSendListTVO log : logsByJsApiOver) {
//                    try {
//                        // 使用现代的日期时间 API
//                        LocalDateTime sendTime = log.getSendTime().toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime();
//                        LocalDateTime expiryDateTime = sendTime.plusHours(2);
//                        LocalDateTime now = LocalDateTime.now();
//
//                        // 判断是否过期
//                        if (now.isAfter(expiryDateTime)) {
//                            // 作废消息
//                            qwSopLogsService.updateQwSopLogsByWatchLogType(log.getId(), "已过期，不发送");
//                        } else {
//                            sendJsApiListOver.add(log);
//                        }
//                    } catch (Exception e) {
//                        // 记录错误日志
//                        logger.error("Error processing log in logsByJsApiOver: {}", log.getId()+":"+e);
//                    }
//                }
//            }, executor).exceptionally(ex -> {
//                // 记录异步任务中的异常
//                logger.error("处理已经完课消息出错logsByJsApi：", ex);
//                return null;
//            });
//
//            //处理当前联系人的消息 是否符合条件
//            CompletableFuture<Void> futureApi = CompletableFuture.runAsync(() -> {
//
//                List<QwSopLogsDoSendListTVO> sendJsApiListFor = getSendJsApiListFor(param.getCorpId(),param.getExternalUserId(), param.getQwUserId());
//                sendJsApiList.addAll(sendJsApiListFor);
//
//            }, executor).exceptionally(ex -> {
//                // 记录异步任务中的异常
//                logger.error("Exception in futureApi logsByJsApi: "+ex);
//                return null;
//            });
//
//
//            //处理下一个联系人的消息 是否符合条件
//            CompletableFuture<Void> futureApiNext = CompletableFuture.runAsync(() -> {
//
//                String nextContactId = processNextContacts(param);
//                sopParamDetailsC.setNextExternalUserId(nextContactId);
//
//            }, executor).exceptionally(ex -> {
//                // 记录异步任务中的异常
//                logger.error("Exception in futureApiNext logsByJsApi: "+ex);
//                return null;
//            });
//
//            // 等待所有异步任务完成
//            CompletableFuture.allOf(futureOver, futureApi,futureApiNext).join();
//
//
//            if (sendJsApiList.size() > 3) {
//                List<QwSopLogsDoSendListTVO> list = new ArrayList<>(sendJsApiList.subList(0, 3));
//                if (!sendJsApiListOver.isEmpty()) {
//                    list.add(sendJsApiListOver.get(0));
//                }
//                sopParamDetailsC.setSendList(list);
//            } else {
//                if (!sendJsApiListOver.isEmpty()) {
//                    sendJsApiList.add(sendJsApiListOver.get(0));
//                }
//                sopParamDetailsC.setSendList(new ArrayList<>(sendJsApiList));
//            }
//
//            // Step 4: 批量更新推送记录状态
//            if (!sendJsApiList.isEmpty()) {
//                processAndInsertQwSopLogsByTakeRecordsList(sopParamDetailsC.getSendList());
//            }
//        }catch (Exception e){
//            logger.error("主动获取发送信息失败：-"+param);
//        }finally {
//            // 关闭线程池
//            executor.shutdown();
//        }
//
//        return sopParamDetailsC;
//    }

//    private String getQwSopLogsByJsApiNext(GetQwSopLogsByJsApiParam param) {
//
//        String nextExternalUserId = qwSopLogsMapper.getQwSopLogsByJsApiNext(param);
//        if (nextExternalUserId==null&& nextExternalUserId.isEmpty()){
//            return null;
//        }
//        List<QwSopLogsDoSendListTVO> nextSendList = getSendJsApiListFor(
//                param.getCorpId(),
//                nextExternalUserId,
//                param.getQwUserId()
//        );
//        if (!nextSendList.isEmpty()) {
//            // 找到一个有可发送内容的联系人，设置 nextExternalUserId
//            return nextExternalUserId;
//        }else {
//            param.setExternalUserId(nextExternalUserId);
//            return   getQwSopLogsByJsApiNext(param);
//        }
//    }
    /**
    * 获取下一个待发送的联系人
    */
//    public String processNextContacts(GetQwSopLogsByJsApiParam param) {
//
//        // 首先，获取可能的下一个外部联系人ID
//        String nextExternalUserId = qwSopLogsMapper.getQwSopLogsByJsApiNext(param);
//
//        // 开一个循环，不断往后找
//        while (StringUtils.isNotBlank(nextExternalUserId)) {
//            // 1) 获取下一个联系人可发送列表
//            List<QwSopLogsDoSendListTVO> nextSendList = getSendJsApiListFor(
//                    param.getCorpId(),
//                    nextExternalUserId,
//                    param.getQwUserId()
//            );
//
//            // 2) 判断是否有可发送内容
//            if (!nextSendList.isEmpty()) {
//                // 找到一个有可发送内容的联系人，设置 nextExternalUserId
//                return nextExternalUserId;
//            }
//
//            // 如果为空，则继续往下找“下一个”的“下一个”...
//            param.setExternalUserId(nextExternalUserId);
//            nextExternalUserId = qwSopLogsMapper.getQwSopLogsByJsApiNext(param);
//        }
//
//
//        return null;
//    }


    /**
    * 获取联系人待发送列表
    */
    public List<QwSopLogsDoSendListTVO> getSendJsApiListFor(String corpId,
                                                            String externalUserId,
                                                            String qwUserId) {

        GetQwSopLogsByJsApiParam param=new GetQwSopLogsByJsApiParam();
            param.setCorpId(corpId);
            param.setExternalUserId(externalUserId);
            param.setQwUserId(qwUserId);

        List<QwSopLogsDoSendListTVO> sendJsApiList = new ArrayList<>();

        // 查询当前客户的待发送记录
        List<QwSopLogsDoSendListTVO> logsByJsApi = qwSopLogsMapper.getQwSopLogsByJsApi(param);

        // 查询员工信息的id
        Long qwId = qwUserMapper.selectQwUserByQwUserIdAndCorpIdToId(param.getQwUserId(), param.getCorpId());

        // 客户的信息
        QwExternalContactHParam contactHParam = new QwExternalContactHParam();
        contactHParam.setUserId(param.getQwUserId());
        contactHParam.setExternalUserId(param.getExternalUserId());
        contactHParam.setCorpId(param.getCorpId());

        Long qwExternalContactId = qwExternalContactMapper.getQwExternalContactId(contactHParam);

        for (QwSopLogsDoSendListTVO log : logsByJsApi) {

            try {
                QwSopTempSetting.Content content = JSON.parseObject(log.getContentJson(), QwSopTempSetting.Content.class);

                if (log.getExpiryTime() == null) {
                    // 作废消息
                    qwSopLogsService.updateQwSopLogsByWatchLogType(log.getId(), "SOP任务被删除");
                }
                else {
                    Integer expiryTime = log.getExpiryTime();
                    LocalDateTime sendTime = log.getSendTime().toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime();
                    LocalDateTime expiryDateTime = sendTime.plusHours(expiryTime);
                    LocalDateTime now = LocalDateTime.now();

                    // 判断是否过期
                    if (now.isAfter(expiryDateTime)) {
                        // 作废消息
                        qwSopLogsService.updateQwSopLogsByWatchLogType(log.getId(), "已过期，不发送");
                    } else {
                        switch (content.getType()) {
                            case 1:
                                // 普通消息，不做判断，加入发送列表
                                sendJsApiList.add(log);
                                break;
                            case 2:
                                // 课程消息，进行复杂的条件判断
                                FsCourseWatchLog watchLog = watchLogService.getWatchCourseLogVideoBySop(
                                        Long.valueOf(content.getVideoId()),
                                        String.valueOf(qwId),
                                        qwExternalContactId
                                );

                                Integer courseType = content.getCourseType();
                                String logId = log.getId();

                                if (watchLog != null) {

                                    //新逻辑
                                    if (isCourseTypeValid(courseType, watchLog.getLogType())) {
                                        // 加入到发送列表中
                                        sendJsApiList.add(log);
                                    } else {
                                        // 作废消息
                                        qwSopLogsService.updateQwSopLogsByWatchLogType(logId, "看课状态未满足，不发送");
                                    }

                                } else {
                                    // 没有观看记录，只发普通消息
                                    if (courseType == 0) {
                                        sendJsApiList.add(log);
                                    } else {
                                        qwSopLogsService.updateQwSopLogsByWatchLogType(logId, "无观看记录，不发送");
                                    }
                                }
                                break;
                            case 3:
                                // 订单消息，不做判断，加入发送列表
                                sendJsApiList.add(log);
                                break;
                            case 4:
                                // Ai消息，不加入序列单独处理
                                try {
                                    logger.info("Ai触达消息 : {}", log.getId());
                                    fastGptChatSessionService.sendAiTouch(log,param.getCorpId().trim());
                                }catch (Exception e){
                                    logger.error("Ai消息 : {}", content.getType());
                                }
                                break;
                            default:
                                // 未知类型，记录警告
                                logger.error("Unknown content type logsByJsApi: {}", content.getType());
                                break;
                        }
                    }
                }
            } catch (Exception e) {
                // 记录错误日志
                logger.error("Error processing log in logsByJsApi: {}", log.getId()+" : " + e);
            }
        }

        return sendJsApiList;
    }

    @Override
    public QwExternalContactByQwResult getExternalContactByAppKey(String appKey) {

        return  qwExternalContactMapper.getExternalContactByAppKey(appKey);
    }

    //
    @Override
    public R deleteQwSopLogsByJsApi(GetQwSopLogsByJsApiParam param) {

        try {
            logger.info("删除或拉黑-处理营期-"+param.getQwUserId()+"|"+param.getCorpId()+"|"+param.getExternalUserId());

            //修改客户状态
            QwExternalContact contact = new QwExternalContact();
            contact.setStatus(7);
            contact.setUserId(param.getQwUserId());
            contact.setCorpId(param.getCorpId());
            contact.setExternalUserId(param.getExternalUserId());
            qwExternalContactMapper.updateQwExternalContactByUseridBlock(contact);


            //删除营期
            sopUserLogsInfoMapper.deleteByQwUserIdAndCorpIdToContactId(param.getQwUserId(), param.getCorpId(), param.getExternalUserId());

            //删除待发送记录
            qwSopLogsMapper.updateQwSopLogsByDeleted(param);


            return  R.ok();
        }catch (Exception e){
            return R.error();
        }


    }

    public void batchInsertQwSopLogs(List<QwSopLogs> logsToInsert) {
        if(logsToInsert == null || logsToInsert.isEmpty()) return;
        applyBlacklistCheck(logsToInsert);
        qwSopLogsMapper.batchInsertQwSopLogs(logsToInsert);
    }

    @Override
    public void applyBlacklistCheck(QwSopLogs sopLogs) {
        if (sopLogs == null || sopLogs.getFsUserId() == null || shouldSkipBlacklistCheck(sopLogs.getSendType())) {
            return;
        }
        if (isFsUserBlacklisted(sopLogs.getFsUserId())) {
            markSopLogAsBlacklisted(sopLogs);
        }
    }

    @Override
    public void applyBlacklistCheck(List<QwSopLogs> sopLogs) {
        if (sopLogs == null || sopLogs.isEmpty()) {
            return;
        }
        Set<Long> fsUserIds = sopLogs.stream()
                .map(QwSopLogs::getFsUserId)
                .filter(Objects::nonNull)
                .collect(Collectors.toSet());
        if (fsUserIds.isEmpty()) {
            return;
        }
        Set<Long> blacklistedUserIds = new HashSet<>(fsUserMapper.selectBlacklistedUserIds(fsUserIds));
        if (blacklistedUserIds.isEmpty()) {
            return;
        }
        for (QwSopLogs sopLog : sopLogs) {
            if (!shouldSkipBlacklistCheck(sopLog.getSendType())
                    && sopLog.getFsUserId() != null && blacklistedUserIds.contains(sopLog.getFsUserId())) {
                markSopLogAsBlacklisted(sopLog);
            }
        }
    }

    @Override
    public void cancelPendingSopLogsByBlacklistedFsUserId(Long fsUserId) {
        if (fsUserId == null) {
            return;
        }
        qwSopLogsMapper.updateQwSopLogsByBlacklistedFsUserId(fsUserId, FS_USER_BLACKLIST_REMARK);
    }

    private boolean isFsUserBlacklisted(Long fsUserId) {
        if (fsUserId == null) {
            return false;
        }
        FsUser fsUser = fsUserMapper.selectFsUserByUserId(fsUserId);
        return fsUser != null && fsUser.getManageStatus() != null && fsUser.getManageStatus() == 2;
    }

    private void markSopLogAsBlacklisted(QwSopLogs sopLogs) {
        sopLogs.setSendStatus(5L);
        sopLogs.setReceivingStatus(4L);
        sopLogs.setRemark(FS_USER_BLACKLIST_REMARK);
    }

    @Override
    public List<QwSopLogsDoSendListTVO> expiredMessagesByQwSopLogs() {
        return qwSopLogsMapper.getExpiredMessagesByQwSopLogs();
    }

    @Override
    @DataSource(DataSourceType.SOP)
    public void updateBatch(List<QwSopLogs> collect) {
        super.updateBatchById(collect);
    }

    @Override
    public void countQwApiAopLogToken(String dateStr) {
        List<QwApiSopLogToken> qwApiSopLogTokens = qwSopLogsMapper.countQwApiAopLogToken(dateStr);
        sopLogTokenService.saveBatch(qwApiSopLogTokens,500);
    }

    @Override
    public void deleteQwSopLogsLabel(QwSopLogs qwSopLogs) {
        qwSopLogsMapper.deleteQwSopLogsLabel(qwSopLogs);
    }


    // 定义一个方法来批量处理插入逻辑，支持每 500 条数据一次的批量插入
    private void processAndInsertQwSopLogs(List<QwSopLogsDoSendListTVO> logsByJsApiNotExtId) {
        // 定义批量插入的大小
        int batchSize = 500;

        // 循环处理外部用户 ID，每次处理批量大小的子集
        for (int i = 0; i < logsByJsApiNotExtId.size(); i += batchSize) {

            int endIndex = Math.min(i + batchSize, logsByJsApiNotExtId.size());
            List<QwSopLogsDoSendListTVO> batchList = logsByJsApiNotExtId.subList(i, endIndex);  // 获取当前批次的子集

            // 直接使用批次数据进行批量更新，不需要额外的 List
            try {
                qwSopLogsMapper.batchUpdateQwSopLogsBySendTime(batchList);
            } catch (Exception e) {
                // 记录异常日志，方便后续排查问题
                logger.error("批量更新数据时发生异常，处理的批次起始索引为: " + i, e);
            }
        }
    }

    // 定义一个方法来批量处理插入逻辑，支持每 500 条数据一次的批量插入
    private void processAndInsertQwSopLogsByTakeRecordsList(List<QwSopLogsDoSendListTVO> logsByJsApi ) {
        // 定义批量插入的大小
        int batchSize = 500;

        // 循环处理外部用户 ID，每次处理批量大小的子集
        for (int i = 0; i < logsByJsApi.size(); i += batchSize) {

            int endIndex = Math.min(i + batchSize, logsByJsApi.size());
            List<QwSopLogsDoSendListTVO> batchList = logsByJsApi.subList(i, endIndex);  // 获取当前批次的子集

            // 直接使用批次数据进行批量更新，不需要额外的 List
            try {
                qwSopLogsMapper.batchUpdateQwSopLogsByTakeRecordsList(batchList);
            } catch (Exception e) {
                // 记录异常日志，方便后续排查问题
                logger.error("批量更新数据时发生异常，处理的批次起始索引为: " + i, e);
            }
        }
    }

    /**
     * 创建企业群发消息
     * 该方法负责处理企业微信的群发消息创建和发送
     */
    @Override
    public void createCorpMassSending(String date)  {

        long startTime = System.currentTimeMillis();
        logger.info("开始执行企业微信群发消息创建任务");

        List<QwSopLogs> qwSopLogs = qwSopLogsMapper.createCorpMassSending(date);
        if (qwSopLogs.isEmpty()) {
            logger.error("zyp \n【企微官方群发记录为空】");
            return;
        }

        Map<String, List<QwSopLogs>> groupedLogs = new HashMap<>();
        for (QwSopLogs log : qwSopLogs) {
            String key = log.getQwUserid() + "|" + log.getSendTime() + "|" + log.getSopId() + "|" + log.getCorpId();
            groupedLogs.computeIfAbsent(key, k -> new ArrayList<>()).add(log);
        }

        int threadCount = Math.min(10, Runtime.getRuntime().availableProcessors() + 1);
        ExecutorService executorService = Executors.newFixedThreadPool(threadCount);
        ExecutorService messageExecutorService = Executors.newFixedThreadPool(20);
        List<QwSopLogs> updateList = Collections.synchronizedList(new ArrayList<>());

        CountDownLatch latch = new CountDownLatch(groupedLogs.size());

        try {
            for (Map.Entry<String, List<QwSopLogs>> entry : groupedLogs.entrySet()) {
                String key = entry.getKey();
                List<QwSopLogs> logs = entry.getValue();
                String[] keys = key.split("\\|");
                String qwUserid = keys[0];
                String corpId = keys[3];

                QwUser qwUser = qwExternalContactService.getQwUserByRedis(corpId.trim(), qwUserid.trim());
                if (qwUser != null && qwUser.getIsDel() == 0) {
                    executorService.submit(() -> {
                        try {
                            Map<String, List<QwSopLogs>> userLogsMap = new HashMap<>();
                            for (QwSopLogs log : logs) {
                                userLogsMap.computeIfAbsent(log.getExternalUserId(), k -> new ArrayList<>()).add(log);
                            }

                            CountDownLatch messageLatch = new CountDownLatch(userLogsMap.size());
                            for (Map.Entry<String, List<QwSopLogs>> userEntry : userLogsMap.entrySet()) {
                                String externalUserId = userEntry.getKey();
                                List<QwSopLogs> userLogs = userEntry.getValue();

                                messageExecutorService.submit(() -> {
                                    try {
                                        QwMsgTemplateSop templateSop = new QwMsgTemplateSop();
                                        templateSop.setChatType("single");
                                        templateSop.setAllowSelect(false);
                                        templateSop.setSender(qwUserid);
                                        templateSop.setExternalUseridList(Collections.singletonList(externalUserId));

                                        List<QwMsgTemplateSop.Attachment> attachments = new ArrayList<>();
                                        boolean hasError = false;
                                        for (QwSopLogs log : userLogs) {
                                            try {
                                                QwSopTempSetting.Content content = JSON.parseObject(log.getContentJson(), QwSopTempSetting.Content.class);
                                                if (content == null || content.getSetting() == null) continue;
                                                Long courseId = content.getCourseId();
                                                for (QwSopTempSetting.Content.Setting set : content.getSetting()) {
                                                    processContent(set, corpId, templateSop, attachments, courseId);
                                                }
                                            } catch (Exception e) {
                                                logger.error("消息内容解析失败，logId：{},{},{}", log.getId(), e,key);
                                                hasError = true;
                                            }
                                        }
                                        if (!hasError && (!attachments.isEmpty() || templateSop.getTextContent() != null)) {
                                            templateSop.setAttachments(attachments);
                                            try {
                                                QwAddMsgTemplateResult result = qwApiService.addMsgTemplateBySop(templateSop, corpId);
                                                if (result == null) {
                                                    logger.error("【官方群发-单链】企微接口无响应, corpId={}, qwUserid={}, externalUserId={}, logId={}",
                                                            corpId, qwUserid, externalUserId, userLogs.get(0).getId());
                                                    for (QwSopLogs log : userLogs) {
                                                        log.setSendStatus(3L);
                                                        log.setRemark("企微接口无响应");
                                                        updateList.add(log);
                                                    }
                                                } else if (result.getErrCode() == 0 || result.getErrCode() == 41063){
                                                    for (QwSopLogs log : userLogs) {
                                                        log.setSendStatus(1L);
                                                        log.setMsgId(result.getMsgId());
                                                        updateList.add(log);
                                                    }
                                                }else {

                                                    for (QwSopLogs log : userLogs) {
                                                        log.setSendType(2);
                                                        log.setSendStatus(3L);
                                                        log.setRemark("官方有误,sop补发");
                                                        log.setReceivingStatus(0L);
                                                        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
                                                        LocalDateTime currentTime = LocalDateTime.now();
                                                        String newTimeString = currentTime.format(formatter);
                                                        log.setSendTime(newTimeString);
                                                        log.setSort(30000001);
                                                        updateList.add(log);
                                                    }

                                                    logger.error("企业微信接口-消息发送失败-进入sop补偿，corpId：{}，errCode：{}，errMsg：{},key:{},{}", corpId, result.getErrCode(), result.getErrMsg(),key,externalUserId);
                                                }
                                            } catch (Exception e) {
                                                logger.error("消息发送失败，user：{},{},{}", externalUserId, e,key);
                                                for (QwSopLogs log : userLogs) {
                                                    log.setSendStatus(0L);
                                                    log.setRemark("信息异常");
                                                    updateList.add(log);
                                                }
                                            }
                                        }

                                    } finally {
                                        logger.info("执行结束-messageLatch-countDown："+updateList.size());
                                        messageLatch.countDown();
                                    }
                                });

                            }

                            logger.info("messageExecutorService-updateList总量："+updateList.size());

                            // 等待所有消息发送完成
                            messageLatch.await();

                        } catch (InterruptedException e) {
                            logger.info("messageExecutorService-Thread.currentThread().interrupt()："+updateList.size());
                            Thread.currentThread().interrupt();
                        } finally {
                            logger.info("finally-latch.countDown:"+updateList.size());
                            latch.countDown();
                        }
                    });
                } else {
                    logger.error("员工信息无效-不存在或被删除，corpId:{}, userId:{}", corpId, qwUserid);

                    latch.countDown(); // 确保每个分组都减少计数
                }
            }

            latch.await(); // 等待所有分组提交的任务完成
            logger.info("关闭线程池并等待任务完成："+updateList.size());
            // 关闭线程池并等待任务完成
            executorService.shutdown();
            messageExecutorService.shutdown();
            if (!executorService.awaitTermination(300, TimeUnit.SECONDS)) {
                logger.error("ExecutorService未完全关闭");
            }
            if (!messageExecutorService.awaitTermination(300, TimeUnit.SECONDS)) {
                logger.error("MessageExecutorService未完全关闭");
            }

            // 5. 同步块生成快照（终极防护），创建快照避免并发修改
            List<QwSopLogs> batchList;
            synchronized (updateList) { // 加锁确保无并发修改
                batchList = new ArrayList<>(updateList);
            }


            logger.info("批量修改总数: {}", batchList.size());

            if (!batchList.isEmpty()){
                int batchSize = 1000;
                for (int i = 0; i < batchList.size(); i += batchSize) {
                    int end = Math.min(i + batchSize, batchList.size());
                    List<QwSopLogs> subList = batchList.subList(i, end);
                    qwSopLogsMapper.batchUpdateStatus(subList);
                }
            }


        } catch (InterruptedException e) {
            logger.error("线程中断异常", e);
            Thread.currentThread().interrupt();
        } finally {
            long endTime = System.currentTimeMillis();
            logger.info("任务完成，耗时: {} 毫秒", endTime - startTime);
        }
    }

    // 处理不同类型的内容
//    private void processContent(QwSopTempSetting.Content.Setting set, String corpId,
//                                QwMsgTemplateSop templateSop, List<QwMsgTemplateSop.Attachment> attachments
//                                ,Long courseId,CourseConfig config) {
//        switch (set.getContentType()) {
//            case "1":
//                templateSop.setTextContent(set.getValue());
//                break;
//            case "2":
//                handleImageAttachment(set, corpId, attachments);
//                break;
//            case "3":
//                handleLinkAttachment(set, corpId, attachments);
//                break;
//            case "4":
//                handleMiniProgramAttachment(set, corpId, attachments,courseId,config);
//                break;
//        }
//    }
//
//    // 处理图片附件
//    private void handleImageAttachment(QwSopTempSetting.Content.Setting set, String corpId,
//                                       List<QwMsgTemplateSop.Attachment> attachments) {
//        if (StringUtils.isNotEmpty(set.getImgUrl())) {
//            try {
//                QwUploadImgResult result = qwApiService.uploadimgs(set.getImgUrl(), corpId);
//                if (result.getErrcode() == 0) {
//                    QwMsgTemplateSop.Attachment attachment = new QwMsgTemplateSop.Attachment();
//                    attachment.setType(2);
//                    attachment.setImagePicUrl(result.getUrl());
//                    attachments.add(attachment);
//                }
//            } catch (Exception e) {
//                logger.error("图片上传失败", e);
//            }
//        }
//    }
//
//    // 处理链接附件
//    private void handleLinkAttachment(QwSopTempSetting.Content.Setting set, String corpId,
//                                      List<QwMsgTemplateSop.Attachment> attachments) {
//        if (StringUtils.isNotEmpty(set.getLinkImageUrl())) {
//            try {
////                QwUploadImgResult result = qwApiService.uploadimgs(set.getLinkImageUrl(), corpId);
////                if (result.getErrcode() == 0) {
//                    QwMsgTemplateSop.Attachment attachment = new QwMsgTemplateSop.Attachment();
//                    attachment.setType(3);
//                    attachment.setLinkTitle(set.getLinkTitle());
//                    attachment.setLinkPicurl(set.getLinkImageUrl());
//                    attachment.setLinkDesc(set.getLinkDescribe());
//                    attachment.setLinkUrl(set.getLinkUrl());
//                    attachments.add(attachment);
////                }
//            } catch (Exception e) {
//                logger.error("链接图片上传失败", e);
//            }
//        }
//    }
//
//    private void handleMiniProgramAttachment(QwSopTempSetting.Content.Setting set, String corpId,
//                                             List<QwMsgTemplateSop.Attachment> attachments,
//                                             Long courseId,CourseConfig config) {
//        if (StringUtils.isNotEmpty(set.getMiniprogramPage())) {
//            try {
//                String key = String.format("miniprogram:%s:%s", corpId, courseId);
//                String mediaId = redisCache.getCacheObject(key);
//
//                if (StringUtils.isNotEmpty(mediaId)) {
//
//                    QwMsgTemplateSop.Attachment attachment = new QwMsgTemplateSop.Attachment();
//                    attachment.setType(4);
////                    attachment.setMiniProgramTitle(set.getMiniprogramTitle());
//                    //强制限制
//                    String title = set.getMiniprogramTitle() != null ? set.getMiniprogramTitle() : "";
//                    int maxLength = 20;
//                    attachment.setMiniProgramTitle(title.length() > maxLength ? title.substring(0, maxLength) : title);
//
//                    attachment.setMiniProgramPicMediaId(mediaId);
//
//                    if (StringUtil.strIsNullOrEmpty(config.getMiniprogramAppid())){
//                        logger.error("小程序配置为空，设置成固定的默认值。");
//                        attachment.setMiniProgramAppId("wxc84c6f789ba7f176");
//                    }else {
//                        attachment.setMiniProgramAppId(config.getMiniprogramAppid());
//                    }
//
//                    attachment.setMiniProgramPage(set.getMiniprogramPage());
//                    attachments.add(attachment);
//                } else {
//                    logger.error("未找到小程序图片mediaId, corpId:{}, courseId:{}", corpId, courseId);
//                }
//            } catch (Exception e) {
//                logger.error("获取小程序图片mediaId失败", e);
//            }
//        }
//    }

    @Override
    public void createCorpMassSendingByUserLogs(String taskStartTime,String taskEndTime) {

        long startTime = System.currentTimeMillis();
        logger.info("开始执行企业微信群发消息创建任务");

        List<QwSopLogs> qwSopLogsList = qwSopLogsMapper.selectSopLogsByCreateCorpMassSending(taskStartTime,taskEndTime);
        if (qwSopLogsList.isEmpty()) {
            logger.error("zyp \n【企微官方群发记录为空】");
            return;
        }

        Map<String, List<QwSopLogs>> grouped = qwSopLogsList.stream().collect(
                Collectors.groupingBy(log -> log.getQwUserid() + "|" + log.getCorpId() + "|" + log.getSopId() + "|" + log.getUserLogsId())
        );

        int threadCount = Math.min(10, Runtime.getRuntime().availableProcessors() + 1);
        ExecutorService executor = Executors.newFixedThreadPool(threadCount);
        Queue<QwSopLogs> updateQueue = new ConcurrentLinkedQueue<>();

        try {
            // 并行提交所有分组
            List<CompletableFuture<Void>> futures = grouped.values().stream()
                    .map(group -> CompletableFuture.runAsync(() -> handleGroup(group, updateQueue), executor))
                    .collect(Collectors.toList());

            // 等待所有任务完成
            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
        } catch (Exception e) {
            logger.error("批量并发执行异常", e);
        } finally {
            executor.shutdown();
            try {
                if (!executor.awaitTermination(300, TimeUnit.SECONDS)) {
                    logger.error("Executor 未完全关闭");
                }
            } catch (InterruptedException ie) {
                Thread.currentThread().interrupt();
            }
        }

        // 批量更新发送状态
        batchUpdate(updateQueue);

        long endTime = System.currentTimeMillis();
        logger.info("企业微信群发任务完成，总耗时：{} 毫秒，更新记录数：{}",
                endTime - startTime, updateQueue.size());

    }

    private void handleGroup(List<QwSopLogs> logsGroup, Queue<QwSopLogs> updateQueue) {

        try {
            QwSopLogs first = logsGroup.get(0);
            String corpId = first.getCorpId() == null ? "" : first.getCorpId().trim();
            if (StringUtils.isEmpty(corpId)) {
                logger.error("corpId 为空，无法创建企微群发");
                logsGroup.forEach(log -> {
                    log.setSendStatus(5L);
                    log.setRemark("corpId 为空");
                    updateQueue.add(log);
                });
                return;
            }
            String rawSender = first.getQwUserid();
            String qwUserid = StringUtils.isEmpty(rawSender) ? null : rawSender.trim();
            if (StringUtils.isEmpty(qwUserid) || "null".equalsIgnoreCase(qwUserid)) {
                logger.error("发送员工未配置，corpId:{}", corpId);
                logsGroup.forEach(log -> {
                    log.setSendStatus(5L);
                    log.setRemark("未配置发送员工");
                    updateQueue.add(log);
                });
                return;
            }
            QwUser qwUser = qwExternalContactService.getQwUserByRedis(corpId, qwUserid);
            if (qwUser == null || qwUser.getIsDel() != 0) {
                logger.error("员工信息无效-不存在或被删除，corpId:{}，userId:{}", corpId, qwUserid);
                logsGroup.forEach(log -> {
                    log.setSendStatus(5L);
                    log.setRemark("员工信息无效");
                    updateQueue.add(log);
                });
                return;
            }

            String firstKey = corpId + "|" + qwUserid;

            // 提取内容与目标客户列表
            String contentJson = logsGroup.get(0).getContentJson();
            QwSopTempSetting.Content content = JSON.parseObject(contentJson, QwSopTempSetting.Content.class);
            if (content == null || content.getSetting() == null || content.getSetting().isEmpty()) {
                logger.warn("消息内容为空或格式异常，key={}，json={} ", firstKey, contentJson);
                logsGroup.forEach(log -> {
                    log.setSendStatus(3L);
                    log.setRemark("消息内容为空或格式异常");
                    updateQueue.add(log);
                });
                return;
            }

            QwMsgTemplateSop template = new QwMsgTemplateSop();
            template.setChatType("single");
            template.setAllowSelect(false);
            template.setSender(qwUserid);

            List<String> externalUserIds = logsGroup.stream()
                    .map(QwSopLogs::getExternalUserId)
                    .filter(Objects::nonNull)
                    .map(String::trim)
                    .filter(s -> !s.isEmpty())
                    .collect(Collectors.toList());
            template.setExternalUseridList(externalUserIds);

            List<QwMsgTemplateSop.Attachment> attachments = new ArrayList<>();
            // 解析并填充消息体
            for (QwSopTempSetting.Content.Setting set : content.getSetting()) {
                processContent(set, corpId, template, attachments, content.getCourseId());
            }
            template.setAttachments(attachments);

            // 调用企业微信接口
            QwAddMsgTemplateResult result = qwApiService.addMsgTemplateBySop(template, corpId);
            if (result == null) {
                logger.error("【官方群发-营期】企微接口无响应, corpId={}, qwUserid={}, sopId={}, 客户数={}, logId={}",
                        corpId, qwUserid, first.getSopId(), externalUserIds.size(), first.getId());
                logsGroup.forEach(log -> {
                    log.setSendStatus(3L);
                    log.setRemark("企微接口无响应");
                    updateQueue.add(log);
                });
                return;
            }

            int errCode = result.getErrCode();
            String errMsg = result.getErrMsg();
            Integer nowSort = (errCode == 0 || errCode == 41063) ? 1 : 3;
            String remark;
            if (errCode == 0 || errCode == 41063) {
                remark = null;
            } else if (errCode == 45033) {
                remark = "官方接口达到上限补发";
            } else {
                remark = "官方有误,sop补发";
            }

            LocalDateTime now = LocalDateTime.now();
            String sendTime = now.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));

            // 收集更新项
            for (QwSopLogs log : logsGroup) {
                if (errCode == 0 || errCode == 41063) {
                    log.setSendStatus(1L);
                    log.setMsgId(result.getMsgId());
                } else {
                    log.setSendType(2);
                    log.setSendStatus(3L);
                    log.setRemark(remark);
                    log.setReceivingStatus(0L);
                    log.setSendTime(sendTime);
                    log.setSort(30000001);
                }
                updateQueue.add(log);
            }

            if (errCode != 0 && errCode != 41063) {
                logger.error("企业微信接口-消息发送失败-进入sop补偿，corpId：{}，errCode：{}，errMsg：{}",
                        corpId, errCode, errMsg);
            }

        } catch (Exception e) {
            QwSopLogs first = logsGroup.isEmpty() ? null : logsGroup.get(0);
            logger.error("【官方群发-营期】处理分组异常, corpId={}, qwUserid={}, sopId={}, 记录数={}, logId={}, 原因={}",
                    first != null ? first.getCorpId() : null,
                    first != null ? first.getQwUserid() : null,
                    first != null ? first.getSopId() : null,
                    logsGroup.size(),
                    first != null ? first.getId() : null,
                    e.getMessage(), e);
        }
    }

    /**
     * 批量更新发送状态
     */
    private void batchUpdate(Collection<QwSopLogs> toUpdate) {
        if (toUpdate.isEmpty()) return;
        logger.info("开始批量更新，记录数量：{}", toUpdate.size());
        List<QwSopLogs> list = new ArrayList<>(toUpdate);
        int batchSize = 500;
        for (int i = 0; i < list.size(); i += batchSize) {
            int end = Math.min(i + batchSize, list.size());
            List<QwSopLogs> subList = list.subList(i, end);
            qwSopLogsMapper.batchUpdateStatus(subList);
        }
    }

    // 处理不同类型的内容
    private void processContent(QwSopTempSetting.Content.Setting set, String corpId,
                                QwMsgTemplateSop templateSop, List<QwMsgTemplateSop.Attachment> attachments,Long courseId) {
        // 发送直播卡片兼容处理
        if(set.getContentType() != null && "12".equals(set.getContentType())) set.setContentType("4");
        switch (set.getContentType()) {
            case "1":
                templateSop.setTextContent(set.getValue());
                break;
            case "2":
                handleImageAttachment(set, corpId, attachments);
                break;
            case "3":
                handleLinkAttachment(set, corpId, attachments);
                break;
            case "4":
                handleMiniProgramAttachment(set, corpId, attachments,courseId);
                break;
        }
    }

    // 处理图片附件
    private void handleImageAttachment(QwSopTempSetting.Content.Setting set, String corpId,
                                       List<QwMsgTemplateSop.Attachment> attachments) {
        if (StringUtils.isNotEmpty(set.getImgUrl())) {
            try {
                QwUploadImgResult result = qwApiService.uploadimgs(set.getImgUrl(), corpId);
                if (result.getErrcode() == 0) {
                    QwMsgTemplateSop.Attachment attachment = new QwMsgTemplateSop.Attachment();
                    attachment.setType(2);
                    attachment.setImagePicUrl(result.getUrl());
                    attachments.add(attachment);
                }
            } catch (Exception e) {
                logger.error("图片上传失败", e);
            }
        }
    }

    // 处理链接附件
    private void handleLinkAttachment(QwSopTempSetting.Content.Setting set, String corpId,
                                      List<QwMsgTemplateSop.Attachment> attachments) {
        if (StringUtils.isNotEmpty(set.getLinkImageUrl())) {
            try {
//                QwUploadImgResult result = qwApiService.uploadimgs(set.getLinkImageUrl(), corpId);
//                if (result.getErrcode() == 0) {
                QwMsgTemplateSop.Attachment attachment = new QwMsgTemplateSop.Attachment();
                attachment.setType(3);
                attachment.setLinkTitle(set.getLinkTitle());
                attachment.setLinkPicurl(set.getLinkImageUrl());
                attachment.setLinkDesc(set.getLinkDescribe());
                attachment.setLinkUrl(set.getLinkUrl());
                attachments.add(attachment);
//                }
            } catch (Exception e) {
                logger.error("链接图片上传失败", e);
            }
        }
    }

    private void handleMiniProgramAttachment(QwSopTempSetting.Content.Setting set, String corpId, List<QwMsgTemplateSop.Attachment> attachments,Long courseId) {
        if (StringUtils.isNotEmpty(set.getMiniprogramPage())) {
            try {
                String key = String.format("miniprogram:%s:%s", corpId, courseId);
                String mediaId = redisCache.getCacheObject(key);

                if (StringUtils.isNotEmpty(mediaId)) {

                    QwMsgTemplateSop.Attachment attachment = new QwMsgTemplateSop.Attachment();
                    attachment.setType(4);
//                    attachment.setMiniProgramTitle(set.getMiniprogramTitle());
                    //强制限制
                    String title = set.getMiniprogramTitle() != null ? set.getMiniprogramTitle() : "";
                    int maxLength = 20;
                    attachment.setMiniProgramTitle(title.length() > maxLength ? title.substring(0, maxLength) : title);
                    attachment.setMiniProgramPicMediaId(mediaId);
                    attachment.setMiniProgramAppId(set.getMiniprogramAppid());
                    attachment.setMiniProgramPage(set.getMiniprogramPage());
                    attachments.add(attachment);
                } else {
                    logger.error("未找到小程序图片mediaId, corpId:{}, courseId:{}", corpId, courseId);
                }
            } catch (Exception e) {
                logger.error("获取小程序图片mediaId失败", e);
            }
        }
    }


}
