package org.springblade.mq.kafka.handler;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import org.springblade.common.cache.DictBizCache;
import org.springblade.core.tool.jackson.JsonUtil;
import org.springblade.enums.*;
import org.springblade.modules.business.check.order.entity.CheckOrderInfoEntity;
import org.springblade.modules.business.check.order.service.IRtuCheckOrderService;
import org.springblade.modules.business.check.process.entity.OrderProcessInfoEntity;
import org.springblade.modules.business.check.process.service.IRtuOrderProcessService;
import org.springblade.modules.business.rtu.data.entity.RtuDataGroundEntity;
import org.springblade.modules.business.rtu.data.entity.RtuDataRainEntity;
import org.springblade.modules.business.rtu.data.entity.RtuDataRiverEntity;
import org.springblade.modules.business.rtu.data.entity.RtuDataRsvrEntity;
import org.springblade.modules.business.rtumanage.entity.RtuStatusEntity;
import org.springblade.modules.business.rtumanage.service.IRtuManageService;
import org.springblade.modules.business.warning.entity.RtuWarningInfoEntity;
import org.springblade.modules.business.warning.service.IRtuWarningService;
import org.springblade.modules.system.entity.DictBiz;
import org.springblade.modules.system.service.IDictBizService;
import org.springblade.mq.kafka.constant.KafkaConstant;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;

/**
 * 消息处理器
 *
 * @author
 */
@Slf4j
@Component
public class MessageHandler {

	@Autowired
	private IRtuWarningService rtuWarningService;

	@Autowired
	private IRtuCheckOrderService rtuCheckOrderService;

	@Autowired
	private IRtuOrderProcessService orderProcessService;

	@Autowired
	private IRtuManageService rtuManageService;

	@Autowired
	private IDictBizService dictBizService;



	@KafkaListener(groupId = "ywxt-kafka", topics = "ywxt.rtu.status", containerFactory = "ackContainerFactory")
	public void handleDataRainMessage(ConsumerRecord record, Acknowledgment acknowledgment) {
		try {
			String message = (String) record.value();
			log.info(message);
			RtuStatusEntity statusEntity = JsonUtil.parse(message, RtuStatusEntity.class);

			rtuManageService.updateLastTime(statusEntity);

		} catch (Exception e) {
			log.error(e.getMessage(), e);
		} finally {
			// 手动提交 offset
			acknowledgment.acknowledge();
		}
	}



	@KafkaListener(groupId = "ywxt-kafka", topics = "ywxt.warning", containerFactory = "ackContainerFactory")
	public void handleMessage(ConsumerRecord record, Acknowledgment acknowledgment) {
		try {
			String message = (String) record.value();
			log.info(message);
			RtuWarningInfoEntity warningInfoEntity = JsonUtil.parse(message, RtuWarningInfoEntity.class);
			CheckOrderInfoEntity checkOrderInfoEntity = new CheckOrderInfoEntity();
			checkOrderInfoEntity.setOrderDesc(warningInfoEntity.getWarningDesc());
			List<DictBiz> statusdicts = dictBizService.getList("rtu_check_order_status");
			for (DictBiz biz : statusdicts) {
				if (biz.getDictKey().equals("" + OrderStatusEnum.STATUS_CREATE.getCode())) {
					checkOrderInfoEntity.setOrderStatusKey(biz.getId());
					break;
				}
			}
			List<DictBiz> fromdicts = dictBizService.getList("rtu_check_order_from");
			for (DictBiz biz : fromdicts) {
				if (biz.getDictKey().equals(OrderFromEnum.FROM_WARNING.getCode())) {
					checkOrderInfoEntity.setOrderFromKey(biz.getId());
					break;
				}
			}
			checkOrderInfoEntity.setWarnId(warningInfoEntity.getId());
			checkOrderInfoEntity.setRtuCode(warningInfoEntity.getRtuCode());
			checkOrderInfoEntity.setOrderClose(OrderCloseEnum.ACTIVE_OPEN.getCode());
			checkOrderInfoEntity.setOrderConfirm(OrderConfirmEnum.ACTIVE_CREATE.getCode());
			checkOrderInfoEntity.setUpdateUser(KafkaConstant.SYS_ADMIN_ID);
			checkOrderInfoEntity.setCreateUser(KafkaConstant.SYS_ADMIN_ID);
			checkOrderInfoEntity.setUpdateTime(new Date());
            checkOrderInfoEntity.setCreateTime(new Date());
			rtuCheckOrderService.save(checkOrderInfoEntity);

			OrderProcessInfoEntity processInfoEntity = new OrderProcessInfoEntity();
			processInfoEntity.setOrderId(checkOrderInfoEntity.getId());
			processInfoEntity.setOrderStatusKey(checkOrderInfoEntity.getOrderStatusKey());
			processInfoEntity.setProcessDesc(checkOrderInfoEntity.getOrderDesc());
			processInfoEntity.setUpdateUser(KafkaConstant.SYS_ADMIN_ID);
			processInfoEntity.setCreateUser(KafkaConstant.SYS_ADMIN_ID);
			processInfoEntity.setUpdateTime(new Date());
			processInfoEntity.setCreateTime(new Date());
			orderProcessService.save(processInfoEntity);


			RtuStatusEntity statusEntity=new RtuStatusEntity();
			statusEntity.setRtuCode(warningInfoEntity.getRtuCode());
			statusEntity.setRunStatusId(EquipmentStatusEnum.STATUS_FAILURE.getCode());
			rtuManageService.updateRtuStatus(statusEntity);

		} catch (Exception e) {
			log.error(e.getMessage(), e);
		} finally {
			// 手动提交 offset
			acknowledgment.acknowledge();
		}
	}
}
