package com.pcloud.book.personalstage.biz.impl;

import com.pcloud.book.base.exception.BookBizException;
import com.pcloud.book.consumer.wechatgroup.WechatGroupConsr;
import com.pcloud.book.keywords.enums.ReplyTypeEnum;
import com.pcloud.book.mq.delay.DelayMessageSender;
import com.pcloud.book.pcloudkeyword.dao.PcloudRobotClassifyDao;
import com.pcloud.book.pcloudkeyword.dao.PcloudRobotDao;
import com.pcloud.book.pcloudkeyword.entity.PcloudRobot;
import com.pcloud.book.pcloudkeyword.entity.PcloudRobotClassify;
import com.pcloud.book.personalstage.biz.PersonalStageBiz;
import com.pcloud.book.personalstage.biz.PersonalStageJumpBiz;
import com.pcloud.book.personalstage.check.PersonalStageCheck;
import com.pcloud.book.personalstage.constant.PersonalStageConstant;
import com.pcloud.book.personalstage.dao.PersonalStageDao;
import com.pcloud.book.personalstage.dao.PersonalStageJumpDao;
import com.pcloud.book.personalstage.dao.PersonalStageJumpKeywordDao;
import com.pcloud.book.personalstage.dao.PersonalStageProgressMessageDao;
import com.pcloud.book.personalstage.dao.PersonalStageReplyDao;
import com.pcloud.book.personalstage.dao.PersonalStageReplyItemDao;
import com.pcloud.book.personalstage.dao.PersonalStageUserDao;
import com.pcloud.book.personalstage.dao.PersonalStageWakeupDao;
import com.pcloud.book.personalstage.dto.FusingFinishDelayDTO;
import com.pcloud.book.personalstage.dto.PersonalStageDTO;
import com.pcloud.book.personalstage.dto.PersonalStageJumpKeywordDto;
import com.pcloud.book.personalstage.dto.PersonalStageProgressDTO;
import com.pcloud.book.personalstage.dto.WakeupDelayDTO;
import com.pcloud.book.personalstage.entity.PersonalStage;
import com.pcloud.book.personalstage.entity.PersonalStageProgressMessage;
import com.pcloud.book.personalstage.entity.PersonalStageReply;
import com.pcloud.book.personalstage.entity.PersonalStageReplyItem;
import com.pcloud.book.personalstage.entity.PersonalStageUser;
import com.pcloud.book.personalstage.entity.PersonalStageWakeup;
import com.pcloud.book.personalstage.enums.JumpTypeEnum;
import com.pcloud.book.personalstage.enums.PersonalStageUserStateEnum;
import com.pcloud.book.personalstage.enums.StageReplyRelevEnum;
import com.pcloud.book.personalstage.utils.CacheUtils;
import com.pcloud.common.core.aspect.ParamLog;
import com.pcloud.common.core.mq.DelayQueueDTO;
import com.pcloud.common.utils.DateUtils;
import com.pcloud.common.utils.ListUtils;
import com.pcloud.common.utils.cache.redis.JedisClusterUtils;
import com.pcloud.common.utils.httpclient.UrlUtils;
import com.pcloud.common.utils.string.StringUtil;
import com.pcloud.wechatgroup.group.dto.GroupUserDTO;
import com.sdk.wxgroup.SendMessageTypeEnum;
import com.sdk.wxgroup.SendPicMessageVO;
import com.sdk.wxgroup.SendTextMessageVO;
import com.sdk.wxgroup.WxGroupSDK;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import com.pcloud.common.exceptions.BizException;
import com.pcloud.common.page.PageBean;
import com.pcloud.common.page.PageParam;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;

import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import java.util.stream.Collectors;

@Component("personalStageBiz")
public class PersonalStageBizImpl implements PersonalStageBiz {

    private static final Logger LOGGER = LoggerFactory.getLogger(PersonalStageBizImpl.class);

    @Autowired
    private PersonalStageDao personalStageDao;
    @Autowired
    private PersonalStageReplyDao personalStageReplyDao;
    @Autowired
    private PersonalStageReplyItemDao personalStageReplyItemDao;
    @Autowired
    private PersonalStageWakeupDao personalStageWakeupDao;
    @Autowired
    private PersonalStageCheck personalStageCheck;
    @Autowired
    private PersonalStageJumpDao personalStageJumpDao;
    @Autowired
    private PcloudRobotClassifyDao pcloudRobotClassifyDao;
    @Autowired
    private PcloudRobotDao pcloudRobotDao;
    @Autowired
    private PersonalStageUserDao personalStageUserDao;
    @Autowired
    private DelayMessageSender delayMessageSender;
    @Autowired
    private PersonalStageJumpKeywordDao personalStageJumpKeywordDao;
    @Autowired
    private PersonalStageJumpBiz personalStageJumpBiz;
    @Autowired
    private WechatGroupConsr wechatGroupConsr;
    @Autowired
    private PersonalStageProgressMessageDao personalStageProgressMessageDao;

    @Value("${wechat.group.link.prefix}")
    private String wechatLinkPrefix;


    @Transactional(rollbackFor = Exception.class)
    @ParamLog("新增阶段")
    @Override
    public Long createPersonalStage(PersonalStage personalStage) {
        personalStageCheck.createPersonalStageParamCheck(personalStage);
        Integer maxSeqNum = personalStageDao.getMaxSeqNum();
        personalStage.setSeqNum(maxSeqNum+1);
        personalStageDao.insert(personalStage);
        addWakeAndReply(personalStage);
        return personalStage.getId();
    }

    @Transactional(rollbackFor = Exception.class)
    @ParamLog("修改阶段")
    @Override
    public void updatePersonalStage(PersonalStage personalStage) {
        personalStageCheck.updatePersonalStageParamCheck(personalStage);
        personalStageDao.update(personalStage);
        deleteAddWakeAndReply(personalStage.getId());
        addWakeAndReply(personalStage);
    }

    @Transactional(rollbackFor = Exception.class)
    @ParamLog("删除阶段")
    @Override
    public void deletePersonalStage(Long id) {
        if (id==null){
            throw new BookBizException(BookBizException.PARAM_IS_NULL,"id不能为空！");
        }
        //校验是否是某个阶段的下一个阶段
        Integer count = personalStageJumpDao.getCountByPersonalStageId(id);
        if (count>0){
            throw new BookBizException(BookBizException.PARAM_IS_NULL,"该阶段为某个阶段的下一阶段，不能删除！");
        }
        personalStageDao.deleteById(id);
        deleteAddWakeAndReply(id);
    }

    @ParamLog("获取定制化阶段")
    @Override
    public PersonalStage getPersonalStage(Long id) {
        PersonalStage personalStage = personalStageDao.getById(id);
        if (personalStage==null){
            throw new BookBizException(BookBizException.ERROR,"没有该记录");
        }
        List<PersonalStageReply> replies = personalStageReplyDao.getListByPersonalStageId(id);
        Map<Long,List<PersonalStageReply>> wakeupReplyMap=new HashMap<>();
        if (!ListUtils.isEmpty(replies)){
            List<Long> replyIds=replies.stream().map(PersonalStageReply::getId).collect(Collectors.toList());
            List<PersonalStageReplyItem> replyItems = personalStageReplyItemDao.getListByReplyIds(replyIds);
            if (!ListUtils.isEmpty(replyItems)){
                Map<Long,List<PersonalStageReplyItem>> itemMap=replyItems.stream().collect(Collectors.groupingBy(PersonalStageReplyItem::getPersonalStageReplyId));
                for (PersonalStageReply reply:replies){
                    reply.setPersonalStageReplyItems(itemMap.get(reply.getId()));
                }
            }
            personalStage.setStageNotKeywordReplies(replies.stream().filter(s->StageReplyRelevEnum.STAGE_NOT_KEYWORD.value.equals(s.getRelevanceType())).collect(Collectors.toList()));
            personalStage.setFusingNotKeywordReplies(replies.stream().filter(s->StageReplyRelevEnum.FUSING_NOT_KEYWORD.value.equals(s.getRelevanceType())).collect(Collectors.toList()));
            List<PersonalStageReply> wakeupReplies=replies.stream().filter(s->StageReplyRelevEnum.WAKEUP.value.equals(s.getRelevanceType())).collect(Collectors.toList());
            wakeupReplyMap=wakeupReplies.stream().collect(Collectors.groupingBy(PersonalStageReply::getRelevanceId));
        }
        List<PersonalStageWakeup> wakeups = personalStageWakeupDao.getListByPersonalStageId(id);
        for (PersonalStageWakeup wakeup:wakeups){
            wakeup.setWakeupReplies(wakeupReplyMap.get(wakeup.getId()));
        }
        personalStage.setPersonalStageWakeups(wakeups);
        return personalStage;
    }

    @Transactional(rollbackFor = Exception.class)
    @ParamLog("修改排序值")
    @Override
    public void updateSeqNum(PersonalStage personalStage) {
        if (personalStage==null||personalStage.getId()==null||personalStage.getSeqNum()==null){
            throw new BookBizException(BookBizException.PARAM_IS_NULL,"参数有误！");
        }
        personalStageDao.updateSeqNum(personalStage.getId(),personalStage.getSeqNum());
    }

    @ParamLog("删除唤醒和回复")
    private void deleteAddWakeAndReply(Long personalStageId) {
        List<Long> replyIds = personalStageReplyDao.getIdsByPersonalStageId(personalStageId);
        personalStageReplyDao.deleteByIds(replyIds);
        personalStageReplyItemDao.deleteByReplyIds(replyIds);
        personalStageWakeupDao.deleteByPersonalStageId(personalStageId);
    }

    @ParamLog("新增唤醒和回复")
    private void addWakeAndReply(PersonalStage personalStage) {
        Long personalStageId = personalStage.getId();
        List<PersonalStageReply> allReply=new ArrayList<>();
        if (!ListUtils.isEmpty(personalStage.getPersonalStageWakeups())){
            for (PersonalStageWakeup wakeup:personalStage.getPersonalStageWakeups()){
                wakeup.setPersonalStageId(personalStageId);
            }
            personalStageWakeupDao.batchInsert(personalStage.getPersonalStageWakeups());
            for (PersonalStageWakeup wakeup:personalStage.getPersonalStageWakeups()){
                Long wakeupId=wakeup.getId();
                for (PersonalStageReply reply:wakeup.getWakeupReplies()){
                    reply.setRelevanceType(StageReplyRelevEnum.WAKEUP.value);
                    reply.setRelevanceId(wakeupId);
                    reply.setPersonalStageId(personalStageId);
                }
                allReply.addAll(wakeup.getWakeupReplies());
            }
        }
        for (PersonalStageReply reply:personalStage.getStageNotKeywordReplies()){
            reply.setRelevanceType(StageReplyRelevEnum.STAGE_NOT_KEYWORD.value);
            reply.setPersonalStageId(personalStageId);
            reply.setRelevanceId(personalStageId);
        }
        allReply.addAll(personalStage.getStageNotKeywordReplies());
        for (PersonalStageReply reply:personalStage.getFusingNotKeywordReplies()){
            reply.setRelevanceType(StageReplyRelevEnum.FUSING_NOT_KEYWORD.value);
            reply.setPersonalStageId(personalStageId);
            reply.setRelevanceId(personalStageId);
        }
        allReply.addAll(personalStage.getFusingNotKeywordReplies());
        personalStageReplyDao.batchInsert(allReply);
        List<PersonalStageReplyItem> allItems=new ArrayList<>();
        for (PersonalStageReply reply:allReply){
            Long replyId=reply.getId();
            for (PersonalStageReplyItem item:reply.getPersonalStageReplyItems()){
                item.setPersonalStageReplyId(replyId);
            }
            allItems.addAll(reply.getPersonalStageReplyItems());
        }
        personalStageReplyItemDao.batchInsert(allItems);
    }

    /**
     * 获取阶段列表
     */
    @Override
    public PageBean getPersonalStageList(Long robotClassifyId, PageParam pageParam) throws BizException {
        Map<String, Object> paramMap = new HashMap<>();
        paramMap.put("robotClassifyId", robotClassifyId);
        return personalStageDao.listPage(pageParam, paramMap, "getPersonalStageList");
    }

    /**
     * 获取阶段列表
     */
    @Override
    public List<PersonalStageDTO> getPersonalStageListNoPage(Long robotClassifyId) throws BizException {
        return personalStageDao.getPersonalStageList(robotClassifyId);
    }

    @Transactional(rollbackFor = Exception.class)
    @ParamLog("发送关键词时处理用户定制化阶段")
    @Override
    public void userSendPersonalStage(String userWxId,String ip,Integer code,String robotId, String content) {
        Boolean hasRecord;
        PersonalStageUser last = personalStageUserDao.getLast(userWxId, robotId, null);
        PersonalStage personalStage;
        if (last!=null){
            hasRecord=true;
            personalStage=personalStageDao.getById(last.getPersonalStageId());
        }else {
            hasRecord=false;
            PcloudRobot pcloudRobot = pcloudRobotDao.getByWxId(robotId);
            Long robotClassifyId=pcloudRobot.getRobotType().longValue();
            personalStage = personalStageDao.getFirstStage(robotClassifyId);
        }
        Boolean iskeyword;
        if (StringUtil.isEmpty(content)){
            iskeyword=false;
        }else {
            //关键词库查询
            PersonalStageJumpKeywordDto dto=personalStageJumpKeywordDao.getByKeyword(personalStage.getId(),content, JumpTypeEnum.READER_TRIGGER.key);
            if (dto!=null){
                iskeyword=true;
            }else {
                iskeyword=false;
            }
        }
        //是关键词且有记录
        if (iskeyword){
            LOGGER.info("是关键词且有记录");
            String key = "PERSONAL_STAGE_JUMP_KEYWORD_LOCK".concat(userWxId).concat(robotId);
            try {
                // 加锁，一条一条数据进行处理
                while (!CacheUtils.lock(key)) {
                    Thread.sleep(100);
                }
                //调关键词处理逻辑
                personalStageJumpBiz.handlePersonalStageJump(userWxId,robotId,content,JumpTypeEnum.READER_TRIGGER);
            } catch (Exception e){
                LOGGER.info("缓存加锁失败，userWxId=" + userWxId +" robotId="+robotId);
            } finally {
                CacheUtils.unlock(key);
            }
        }
        //不是关键词且有记录
        if (!iskeyword&&hasRecord){
            //判断是否已经达到熔断标准
            if (personalStage!=null){
                Integer notKeywordSendCount =getSendNKeywordCountFromCache(personalStage.getId(),userWxId,robotId);
                if ((notKeywordSendCount!=null&&notKeywordSendCount>=personalStage.getNotKeywordFusingCount())
                      ||PersonalStageUserStateEnum.FUSING.value.equals(last.getState())){
                    // 走熔断
                    LOGGER.info("走熔断");
                    handleFusingReply(robotId,userWxId,ip,personalStage,last);
                }else {
                    //走正常逻辑非关键词回复并且更新记录
                    if (PersonalStageUserStateEnum.NORMAL.value.equals(last.getState())
                        ||PersonalStageUserStateEnum.WAKEUP.value.equals(last.getState())){
                        LOGGER.info("走正常逻辑非关键词回复并且更新记录");
                        //如果是正常状态或者唤醒状态才回复，并且更新状态为正常状态
                        last.setNotKeywordSendCount(notKeywordSendCount+1);
                        last.setState(PersonalStageUserStateEnum.NORMAL.value);
                        personalStageUserDao.update(last);
                        sendNotKeywordReply(robotId,userWxId,ip,personalStage.getId(), last.getId());
                    }
                }
            }
        }
        //不是关键词且没有记录
        if (!iskeyword&&!hasRecord){
            //走正常非关键词回复并且新增记录
            LOGGER.info("走正常非关键词回复并且新增记录");
            sendNotKeywordReply(robotId,userWxId,ip,personalStage.getId(), last.getId());
            PcloudRobot pcloudRobot = pcloudRobotDao.getByWxId(robotId);
            Long robotClassifyId=pcloudRobot.getRobotType().longValue();
            PersonalStage firstStage = personalStageDao.getFirstStage(robotClassifyId);
            PersonalStageUser personalStageUser=new PersonalStageUser();
            personalStageUser.setRobotId(robotId);
            personalStageUser.setWxId(userWxId);
            personalStageUser.setState(PersonalStageUserStateEnum.NORMAL.value);
            personalStageUser.setNotKeywordSendCount(1);
            personalStageUser.setRobotClassifyId(robotClassifyId);
            personalStageUser.setPersonalStageId(firstStage.getId());
            personalStageUserDao.insert(personalStageUser);
        }
    }


    @ParamLog("从缓存中获取发送非关键词次数")
    private synchronized Integer getSendNKeywordCountFromCache(Long personalStageId, String wxId, String robotId) {
        String key = PersonalStageConstant.NOT_SEND_KEYWORD_COUNT_LOCK + personalStageId;
        Integer count = 0;
        try {
            while (!CacheUtils.lock(key)) {
                Thread.sleep(100);
            }
            String cacheKey = PersonalStageConstant.USER_SEND_KEYWORD_COUNT_LOCK + personalStageId + "_" + wxId;
            long value = JedisClusterUtils.incr(cacheKey);
            if (value != 1) {
                return (int)value - 1;
            }
            count = personalStageUserDao.getNotKeywordSendCount(wxId, robotId, null);
            count = count == null ? 0 : count;
            JedisClusterUtils.set(cacheKey, (count.longValue() + 1) + "", 10 * 60);
            return count;
        } catch (Exception e) {
            LOGGER.warn("从缓存中获取发送非关键词次数失败，personalStageId=" + personalStageId +"wxId="+wxId+"robotId="+robotId);
        } finally {
            CacheUtils.unlock(key);
        }
        return count;
    }

    @Override
    @ParamLog("判断是否是定制化用户")
    public Boolean isPersonalStageUser(String robotId){
        //判断是否是平台端配置的小号
        try {
            PcloudRobot pcloudRobot = pcloudRobotDao.getByWxId(robotId);
            if (pcloudRobot!=null&&pcloudRobot.getRobotType()!=null){
                //查询该分类下的定制化阶段
                Long robotClassifyId=pcloudRobot.getRobotType().longValue();
                PcloudRobotClassify robotClassify = pcloudRobotClassifyDao.getById(robotClassifyId);
                if (robotClassify!=null){
                    //获取该分类下的阶段数量
                    Integer count=personalStageDao.getCountByRobotClassifyId(robotClassifyId);
                    if (count>0){
                        return true;
                    }
                }
            }
        }catch (Exception e) {
            LOGGER.error("判断是否是定制化用户失败" + e.getMessage(), e);
        }
        return false;
    }

    @ParamLog("加好友时插入初始状态")
    @Override
    public void createAddUserStageUser(String robotId, String userWxId, String ip) {
        //查询第一个阶段
        PcloudRobot pcloudRobot = pcloudRobotDao.getByWxId(robotId);
        Long robotClassifyId=pcloudRobot.getRobotType().longValue();
        PersonalStage firstStage = personalStageDao.getFirstStage(robotClassifyId);
        if (firstStage==null){
            return;
        }
        nextStageAddStageUserAndWakeupDelay (robotId, userWxId, ip, firstStage.getId());
    }

    @ParamLog("处理延时唤醒")
    @Override
    public void dealDelayWakeup(DelayQueueDTO dto) {
        String wxId=dto.getKey();
        WakeupDelayDTO wakeupDelayDTO=(WakeupDelayDTO) dto.getMsg();
        //查询该用户最后的阶段
        PersonalStageUser last = personalStageUserDao.getLast(wxId, wakeupDelayDTO.getRobotId(), null);
        Long personalStageId = wakeupDelayDTO.getPersonalStageId();
        //判断用户是否在插入唤醒延时队列的当时阶段且是唤醒状态
        if (last==null||!PersonalStageUserStateEnum.WAKEUP.value.equals(last.getState())
                ||!last.getPersonalStageId().equals(personalStageId)){
            LOGGER.info("用户已跳阶段或者不是唤醒状态"+wakeupDelayDTO.toString());
            return;
        }
        PersonalStage personalStage=personalStageDao.getById(personalStageId);
        if (personalStage==null){
            LOGGER.info("未找到该阶段personalStageId="+personalStageId);
            return;
        }
        Long currentWakeupId=wakeupDelayDTO.getPersonalStageWakeupId();
        //发送唤醒内容
        List<Long> replyIds=personalStageReplyDao.getIdsByRelevance(StageReplyRelevEnum.WAKEUP.value,currentWakeupId);
        if (ListUtils.isEmpty(replyIds)){
            LOGGER.info("没有回复currentWakeupId="+currentWakeupId);
            return;
        }
        List<PersonalStageReplyItem> replyItems = personalStageReplyItemDao.getListByReplyIds(replyIds);
        if (ListUtils.isEmpty(replyItems)){
            LOGGER.info("没有回复项replyIds="+replyIds);
            return;
        }
        String robotId=wakeupDelayDTO.getRobotId();
        String userWxId=wakeupDelayDTO.getWxId();
        String ip=wakeupDelayDTO.getIp();
        sendReplyItems(replyItems,robotId,userWxId,ip,wakeupDelayDTO.getPersonalStageUserId());
        //下一个延时
        List<PersonalStageWakeup> wakeups = personalStageWakeupDao.getListByPersonalStageId(personalStageId);
        int i=0;
        PersonalStageWakeup currentWakeup=null;
        for (PersonalStageWakeup wakeup:wakeups){
            i=i+1;
            if (currentWakeupId.equals(wakeup.getId())){
                currentWakeup=wakeup;
                break;
            }
        }
        if (currentWakeup!=null&&i>0&&i<wakeups.size()){
            PersonalStageWakeup nextWakeup=wakeups.get(i);
            Integer time=(nextWakeup.getToStageStartTime()-currentWakeup.getToStageStartTime())*1000;
            WakeupDelayDTO wakeupDelayDTONew=new WakeupDelayDTO();
            wakeupDelayDTONew.setPersonalStageWakeupId(nextWakeup.getId());
            wakeupDelayDTONew.setRobotId(robotId);
            wakeupDelayDTONew.setWxId(userWxId);
            wakeupDelayDTONew.setIp(ip);
            wakeupDelayDTONew.setPersonalStageId(personalStageId);
            wakeupDelayDTONew.setPersonalStageUserId(wakeupDelayDTO.getPersonalStageUserId());
            DelayQueueDTO delayQueueDTONew = DelayQueueDTO.builder().key(userWxId).type(PersonalStageConstant.PERSONALSTAGE_DELAY_WAKEUP).msg(wakeupDelayDTONew).timeout(time).build();
            delayMessageSender.send(delayQueueDTONew);
            LOGGER.info("下一个延时发送delayQueueDTONew="+delayQueueDTONew.toString());
        }


    }

    @ParamLog("下一阶段新增定制化阶段用户记录和唤醒延时")
    @Override
    public PersonalStageUser nextStageAddStageUserAndWakeupDelay(String robotId, String userWxId, String ip, Long personalStageId) {
        PersonalStageWakeup wakeup = personalStageWakeupDao.getFirstWakeup(personalStageId);
        PersonalStageUser personalStageUser=new PersonalStageUser();
        personalStageUser.setRobotId(robotId);
        personalStageUser.setWxId(userWxId);
        if (wakeup==null){
            //没有唤醒，设置状态为正常状态
            personalStageUser.setState(PersonalStageUserStateEnum.NORMAL.value);
        }else {
            //有唤醒，设置状态为唤醒状态
            personalStageUser.setState(PersonalStageUserStateEnum.WAKEUP.value);
        }
        personalStageUser.setNotKeywordSendCount(0);
        //查询分类
        PcloudRobot pcloudRobot = pcloudRobotDao.getByWxId(robotId);
        Long robotClassifyId=pcloudRobot.getRobotType().longValue();
        personalStageUser.setRobotClassifyId(robotClassifyId);
        personalStageUser.setPersonalStageId(personalStageId);
        personalStageUserDao.insert(personalStageUser);
        //同时插入唤醒队列
        //查询第一个唤醒时长
        if (wakeup==null){
            return personalStageUser;
        }
        Integer toStageStartTime=wakeup.getToStageStartTime();
        WakeupDelayDTO wakeupDelayDTO=new WakeupDelayDTO();
        wakeupDelayDTO.setPersonalStageWakeupId(wakeup.getId());
        wakeupDelayDTO.setRobotId(robotId);
        wakeupDelayDTO.setWxId(userWxId);
        wakeupDelayDTO.setIp(ip);
        wakeupDelayDTO.setPersonalStageId(personalStageId);
        wakeupDelayDTO.setPersonalStageUserId(personalStageUser.getId());
        DelayQueueDTO delayQueueDTO = DelayQueueDTO.builder().key(userWxId).type(PersonalStageConstant.PERSONALSTAGE_DELAY_WAKEUP).msg(wakeupDelayDTO).timeout(toStageStartTime*1000).build();
        delayMessageSender.send(delayQueueDTO);
        return personalStageUser;
    }

    @ParamLog("非关键词回复")
    @Override
    public void sendNotKeywordReply(String robotId, String userWxId, String ip, Long personalStageId, Long personalStageUserId) {
        List<Long> ids = personalStageReplyDao.getIdsByRelevance(StageReplyRelevEnum.STAGE_NOT_KEYWORD.value, personalStageId);
        if (ListUtils.isEmpty(ids)){
            return;
        }
        Random random = new Random();
        Long replyId=ids.get(random.nextInt(ids.size()));
        List<PersonalStageReplyItem> items = personalStageReplyItemDao.getListByReplyIds(Arrays.asList(replyId));
        if (ListUtils.isEmpty(items)){
            return;
        }
        sendReplyItems(items,robotId,userWxId,ip,personalStageUserId);
    }

    @ParamLog("尝试插入单号")
    private void tryInsertNumber(Long personalStageUserId) {
        String requireNumber = DateUtils.formatDate(new Date(), "yyyyMMddHHmmss").concat(String.valueOf((int)((Math.random()*9+1)*100000)));
        personalStageUserDao.updateRequireNumber(personalStageUserId, requireNumber);
    }

    @Override
    @ParamLog("替换进度链接")
    public String replaceProjectProgressUrl(String content, String robotId, String userWxId, Long personalStageUserId) {
        if(content.indexOf(PersonalStageConstant.PERSONAL_STAGE_PROJECT_PROGRESS_TEMPLATE) > -1){
            tryInsertNumber(personalStageUserId);
            String longUrl = wechatLinkPrefix.concat(PersonalStageConstant.PERSONAL_STAGE_PROJECT_PROGRESS).concat("?wxUserId=" + userWxId + "&robotId=" + robotId + "&personalStageUserId=" + personalStageUserId);
            return content.replace(PersonalStageConstant.PERSONAL_STAGE_PROJECT_PROGRESS_TEMPLATE, UrlUtils.getShortUrl4Own(longUrl));
        }
        return content;
    }

    @ParamLog("非关键词熔断回复")
    @Override
    public void sendNotKeywordFusingReply(String robotId, String userWxId, String ip, Long personalStageId, Long personalStageUserId) {
        List<Long> ids = personalStageReplyDao.getIdsByRelevance(StageReplyRelevEnum.FUSING_NOT_KEYWORD.value, personalStageId);
        if (ListUtils.isEmpty(ids)) {
            return;
        }
        Random random = new Random();
        Long replyId = ids.get(random.nextInt(ids.size()));
        List<PersonalStageReplyItem> items = personalStageReplyItemDao.getListByReplyIds(Arrays.asList(replyId));
        if (ListUtils.isEmpty(items)){
            return;
        }
        sendReplyItems(items,robotId,userWxId,ip,personalStageUserId);
    }

    @ParamLog("发送回复")
    private void sendReplyItems(List<PersonalStageReplyItem> replyItems,String robotId, String userWxId, String ip,Long personalStageUserId){
        for (PersonalStageReplyItem item:replyItems){
            Integer type=item.getReplyType();
            if (ReplyTypeEnum.TEXT.value.equals(type)) {
                SendTextMessageVO sendTextMessageVO = new SendTextMessageVO();
                sendTextMessageVO.setContent(replaceProjectProgressUrl(item.getContent(),robotId,userWxId,personalStageUserId));
                sendTextMessageVO.setAltId(robotId);
                sendTextMessageVO.setWxGroupId(userWxId);
                sendTextMessageVO.setIp(ip);
                sendTextMessageVO.setCode(SendMessageTypeEnum.SELF.getCode());
                WxGroupSDK.sendTextMessage(sendTextMessageVO);
            }
            if (ReplyTypeEnum.IMAGE.value.equals(type)) {
                SendPicMessageVO sendPicMessageVO = new SendPicMessageVO();
                sendPicMessageVO.setWxGroupId(userWxId);
                sendPicMessageVO.setAltId(robotId);
                sendPicMessageVO.setPicUrl(item.getPicUrl());
                sendPicMessageVO.setIp(ip);
                sendPicMessageVO.setCode(SendMessageTypeEnum.SELF.getCode());
                WxGroupSDK.sendPicMessage(sendPicMessageVO);
            }
        }
    }

    @ParamLog("处理熔断回复情况")
    private void handleFusingReply(String robotId, String userWxId, String ip, PersonalStage personalStage, PersonalStageUser last) {
        if (!PersonalStageUserStateEnum.FUSING.value.equals(last.getState())) {
            // 置为熔断状态
            PersonalStageUser user = new PersonalStageUser();
            user.setId(last.getId());
            user.setState(PersonalStageUserStateEnum.FUSING.value);
            user.setNotKeywordSendCount(last.getNotKeywordSendCount()+1);
            personalStageUserDao.update(user);
            // 添加延迟队列
            FusingFinishDelayDTO delayDTO = new FusingFinishDelayDTO();
            delayDTO.setPersonalStageUserId(last.getId());
            delayDTO.setPersonalStageId(personalStage.getId());
            delayDTO.setRobotId(robotId);
            delayDTO.setWxId(userWxId);
            delayDTO.setIp(ip);
            DelayQueueDTO delayQueueDTO = DelayQueueDTO.builder().key(userWxId).type(PersonalStageConstant.PERSONALSTAGE_DELAY_FUSING)
                    .msg(delayDTO).timeout(personalStage.getNotKeywordFusingTime().intValue() * 1000).build();
            delayMessageSender.send(delayQueueDTO);
        }
        String key = "BOOK:PERSONAL_STAGE:FUSING:" + last.getId();
        if (!JedisClusterUtils.exists(key)) {
            // 发送熔断回复消息
            JedisClusterUtils.set(key, "true", 60);
            sendNotKeywordFusingReply(robotId, userWxId, ip, personalStage.getId(), last.getId());
        }
    }

    @ParamLog("处理延时熔断结束")
    @Override
    public void dealDelayFusingFinish(DelayQueueDTO dto) {
        String wxId = dto.getKey();
        FusingFinishDelayDTO delayDTO = (FusingFinishDelayDTO) dto.getMsg();
        //查询该用户在该阶段的状态
        Long personalStageId = delayDTO.getPersonalStageId();
        Long personalStageUserId = delayDTO.getPersonalStageUserId();
        PersonalStageUser last = personalStageUserDao.getById(personalStageUserId);
        if (last == null || !PersonalStageUserStateEnum.FUSING.value.equals(last.getState())){
            return;
        }
        PersonalStage personalStage = personalStageDao.getById(personalStageId);
        if (personalStage == null){
            return;
        }
        // 将用户置为正常状态
        PersonalStageUser user = new PersonalStageUser();
        user.setId(personalStageUserId);
        user.setState(PersonalStageUserStateEnum.NORMAL.value);
        user.setNotKeywordSendCount(0);
        personalStageUserDao.update(user);
        //刪除发送非关键词次数緩存
        JedisClusterUtils.del(PersonalStageConstant.USER_SEND_KEYWORD_COUNT_LOCK + personalStageId + "_" + wxId);
    }

    @Override
    public PersonalStageProgressDTO getPersonalProgress(String wxUserId, Long personalStageUserId) {
        PersonalStageProgressDTO personalStageProgressDTO = new PersonalStageProgressDTO();
        PersonalStageUser personalStageUser = personalStageUserDao.getById(personalStageUserId);
        if(personalStageUser == null){
            return personalStageProgressDTO;
        }
        Date startTime = personalStageUser.getCreateTime();
        GroupUserDTO groupUserDTO = wechatGroupConsr.getWxUserInfoByWxUserId(wxUserId);
        personalStageProgressDTO.setWxUserId(wxUserId);
        personalStageProgressDTO.setHeadPic(groupUserDTO.getHeadPic());
        personalStageProgressDTO.setNickName(groupUserDTO.getNickName());
        personalStageProgressDTO.setSex(groupUserDTO.getSex());
        personalStageProgressDTO.setRequireNumber(personalStageUser.getRequireNumber());
        List<PersonalStageProgressMessage> personalStageProgressMessages = personalStageProgressMessageDao.getPersonalProgress(startTime);
        if (!ListUtils.isEmpty(personalStageProgressMessages)){
            setProgressTime(personalStageProgressMessages,startTime);
            personalStageProgressDTO.setPersonalStageProgressMessageList(personalStageProgressMessages);
        }
        return personalStageProgressDTO;
    }

    private void setProgressTime(List<PersonalStageProgressMessage> personalStageProgressMessages, Date startTime) {
        personalStageProgressMessages.stream().forEach(e ->{
            e.setProgressTime(DateUtils.addMinute(startTime,e.getMinutes()));
        });
    }
}
