package org.dromara.elevator.listener.message.consumer;

import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import me.chanjar.weixin.mp.api.WxMpService;
import me.chanjar.weixin.mp.bean.template.WxMpTemplateData;
import me.chanjar.weixin.mp.bean.template.WxMpTemplateMessage;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.dromara.common.core.utils.DateUtils;
import org.dromara.common.core.utils.StreamUtils;
import org.dromara.common.core.utils.StringUtils;
import org.dromara.common.json.utils.JsonUtils;
import org.dromara.elevator.constant.ElevatorConstant;
import org.dromara.elevator.domain.EElevatorInfo;
import org.dromara.elevator.domain.EElevatorUser;
import org.dromara.elevator.domain.EElevatorWechatUser;
import org.dromara.elevator.domain.bo.EBehaviorMessageBo;
import org.dromara.elevator.domain.bo.EFaultMessageBo;
import org.dromara.elevator.domain.vo.EElevatorInfoVo;
import org.dromara.elevator.mapper.EElevatorInfoMapper;
import org.dromara.elevator.mapper.EElevatorUserMapper;
import org.dromara.elevator.mapper.EElevatorWechatUserMapper;
import org.dromara.elevator.service.IEBehaviorMessageService;
import org.dromara.elevator.service.IEFaultMessageService;
import org.dromara.system.service.ISysConfigService;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.stream.Collectors;

import static org.dromara.elevator.constant.ElevatorConstant.*;
import static org.dromara.elevator.constant.TopicConstant.ELEVATOR_FAULT_TOPIC;
import static org.dromara.elevator.constant.TopicConstant.ELEVATOR_REAL_TOPIC;

/**
 * 电梯消息消费监听
 *
 * @author ym
 * @version 1.0.0
 * @date 2024-08-23 16:45
 */
@Slf4j
@Component
@ConditionalOnProperty(value = "kafka.enabled", havingValue = "true")
@RequiredArgsConstructor
public class ElevatorMessageListener {

    private final IEFaultMessageService faultMessageService;
    private final IEBehaviorMessageService behaviorMessageService;
    private final EElevatorInfoMapper elevatorInfoMapper;
    private final EElevatorUserMapper elevatorUserMapper;
    private final EElevatorWechatUserMapper elevatorWechatUserMapper;
    private final WxMpService wxMpService;
    private final ISysConfigService sysConfigService;

    @KafkaListener(topics = ELEVATOR_FAULT_TOPIC)
    public void faultMessageListener(ConsumerRecord<String, String> record) {
        String message = record.value();
//        log.info("故障消息:{}", message);
        if (StringUtils.isBlank(message)) {
            return;
        }
        EFaultMessageBo faultMessageBo = JsonUtils.parseObject(message, EFaultMessageBo.class);
        faultMessageService.insertByBo(faultMessageBo);
    }

    @KafkaListener(topics = ELEVATOR_REAL_TOPIC)
    public void realMessageListener(ConsumerRecord<String, String> record) {
        String message = record.value();
//        log.info("实时消息:{}", message);
        if (StringUtils.isBlank(message)) {
            return;
        }
        EBehaviorMessageBo behaviorMessageBo = JsonUtils.parseObject(message, EBehaviorMessageBo.class);
        behaviorMessageService.insertByBo(behaviorMessageBo);
    }

    @KafkaListener(topics = "elevator-user")
    public void elevatorUserListener(ConsumerRecord<String, String> record) {
        String userId = record.value();
        log.info("接收用户id:{}", userId);
        if (StringUtils.isBlank(userId)) {
            return;
        }
        List<EElevatorUser> userList = new ArrayList<>();
        List<EElevatorInfoVo> elevatorList = elevatorInfoMapper.selectVoList();
        for (EElevatorInfoVo elevator : elevatorList) {
            EElevatorUser elevatorUser = new EElevatorUser();
            elevatorUser.setElevatorId(String.valueOf(elevator.getId()));
            elevatorUser.setElevatorNum(elevator.getElevatorNum());
            elevatorUser.setElevatorName(elevator.getElevatorName());
            elevatorUser.setElevatorAddress(elevator.getElevatorAddress());
            elevatorUser.setTenementName(elevator.getTenementName());
            elevatorUser.setMaintenanceName(elevator.getMaintenanceName());
            elevatorUser.setUserId(userId);
            elevatorUser.setIsConcern(0);
            userList.add(elevatorUser);
        }
        elevatorUserMapper.insertBatch(userList);
    }

    @KafkaListener(topics = ELEVATOR_FAULT_TOPIC, groupId = "elevator_consumer_wechat_push")
    public void faultMsgWechatPushListener(ConsumerRecord<String, String> record) {
        String message = record.value();
        if (StringUtils.isBlank(message)) {
            return;
        }
        EFaultMessageBo faultMessage = JsonUtils.parseObject(message, EFaultMessageBo.class);
        if (Objects.isNull(faultMessage)) {
            return;
        }
        String enabled = sysConfigService.selectConfigByKey(ELE_FAULT_MSG_PUSH_CONFIG_KEY);
        if (StringUtils.isBlank(enabled) || !Objects.equals(Integer.valueOf(enabled), ElevatorConstant.ON)) {
            return;
        }
        // 故障推送类型待确认
        String pushMsgKey = sysConfigService.selectConfigByKey(FAULT_PUSH_MSG_KEY);

        if (StringUtils.isNotBlank(pushMsgKey)) {
            pushMsgKey = pushMsgKey.replaceAll("，", StrUtil.COMMA);
            Set<String> pushMsgFaultCodeSet = Arrays.stream(pushMsgKey.split(StrUtil.COMMA)).collect(Collectors.toSet());
            if (!pushMsgFaultCodeSet.contains(faultMessage.getFaultCode())) {
                // 非指定故障推送类型
                return;
            }
        }
//        log.info("消费故障推送模板消息: {}, {}", faultMessage.getFaultName(), message);
        String elevatorNum = faultMessage.getElevatorNum();

        List<EElevatorUser> elevatorUsers = elevatorUserMapper.selectList(Wrappers.<EElevatorUser>lambdaQuery()
            .eq(EElevatorUser::getElevatorNum, elevatorNum)
            .eq(EElevatorUser::getIsConcern, ON));
        if (CollectionUtils.isEmpty(elevatorUsers)) {
            return;
        }
        List<Long> userIds = StreamUtils.toList(elevatorUsers, EElevatorUser::getUserId)
            .stream().map(Long::valueOf).collect(Collectors.toList());
        List<EElevatorWechatUser> elevatorWechatUsers = elevatorWechatUserMapper.selectList(
            Wrappers.<EElevatorWechatUser>lambdaQuery().in(EElevatorWechatUser::getUserId, userIds));
        elevatorWechatUsers = StreamUtils.filter(elevatorWechatUsers,
            item -> StringUtils.isNotBlank(item.getMpOpenId()) && StringUtils.isNotBlank(item.getMaOpenId()));
        if (CollectionUtils.isEmpty(elevatorWechatUsers)) {
            return;
        }
        String templateId = sysConfigService.selectConfigByKey("ele.fault.msg.templateId");
        if (StringUtils.isBlank(templateId)) {
            log.warn("未设置消息推送模板id");
            return;
        }
        EElevatorInfo elevatorInfo = elevatorInfoMapper.selectOne(Wrappers.<EElevatorInfo>lambdaQuery()
            .eq(EElevatorInfo::getElevatorNum, faultMessage.getElevatorNum()).last(" limit 1 "));
        // 微信模板消息推送
        WxMpTemplateMessage templateMessage = WxMpTemplateMessage.builder()
//            .toUser("OPENID")
            .templateId(templateId)
//            .url("URL")
            .build();
        if (elevatorInfo.getElevatorName().length() > 20) {
            elevatorInfo.setElevatorName(elevatorInfo.getElevatorName().substring(0, 16));
        }

        if (elevatorInfo.getElevatorAddress().length() > 20) {
            elevatorInfo.setElevatorAddress(elevatorInfo.getElevatorAddress().substring(0, 16));
        }

        templateMessage.addData(new WxMpTemplateData("character_string19", elevatorInfo.getElevatorNum(), "#FF0000"));
        templateMessage.addData(new WxMpTemplateData("thing17", elevatorInfo.getElevatorName(), "#FF0000"));
        templateMessage.addData(new WxMpTemplateData("thing18", elevatorInfo.getElevatorAddress(), "#FF0000"));
        templateMessage.addData(new WxMpTemplateData("thing4", faultMessage.getFaultName(), "#FF0000"));
        templateMessage.addData(new WxMpTemplateData("time1", DateUtils.parseDateToStr(DateUtils.YYYY_MM_DD_HH_MM_SS, faultMessage.getHappenTime()), "#FF0000"));
        // 可以添加更多字段
//        WxMpTemplateMessage.MiniProgram miniProgram = new WxMpTemplateMessage.MiniProgram();
//        miniProgram.setAppid("");
//        miniProgram.setPagePath("");
//        miniProgram.setUsePath(Boolean.TRUE);
//        templateMessage.setMiniProgram(miniProgram);

        for (EElevatorWechatUser wechatUser : elevatorWechatUsers) {
            templateMessage.setToUser(wechatUser.getMpOpenId());
            try {
                wxMpService.getTemplateMsgService().sendTemplateMsg(templateMessage);
//                log.info("故障推送模板消息:{}, {}", wechatUser.getUserId(), JsonUtils.toJsonString(templateMessage));
            } catch (Exception e) {
                log.error("模板消息发生失败:{},故障消息:{}", templateMessage.toJson(), JsonUtils.toJsonString(faultMessage), e);
            }
        }
    }

}
