package org.springblade.mq.kafka.handler;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import org.springblade.core.tool.jackson.JsonUtil;
import org.springblade.core.tool.utils.Func;
import org.springblade.enums.*;
import org.springblade.etl.entity.*;
import org.springblade.etl.service.IEtlTaskInfoService;
import org.springblade.modules.baseinfo.rtu.entity.RtuInfoEntity;
import org.springblade.modules.baseinfo.rtu.service.IRtuBaseInfoService;
import org.springblade.modules.business.check.order.dto.CheckOrderInfoDTO;
import org.springblade.modules.business.check.order.service.IRtuCheckOrderService;
import org.springblade.modules.business.data.service.IRtuDataGroundService;
import org.springblade.modules.business.data.service.IRtuDataRainService;
import org.springblade.modules.business.data.service.IRtuDataRiverService;
import org.springblade.modules.business.data.service.IRtuDataRsvrService;
import org.springblade.modules.business.rtumanage.dto.RtuStatusDTO;
import org.springblade.modules.business.rtumanage.entity.RtuStatusEntity;
import org.springblade.modules.business.rtumanage.service.IRtuManageService;
import org.springblade.modules.business.warning.entity.RtuWarnSettingEntity;
import org.springblade.modules.business.warning.entity.RtuWarningInfoEntity;
import org.springblade.modules.business.warning.service.IRtuWarnSettingService;
import org.springblade.modules.business.warning.service.IRtuWarningService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.ListOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;

/**
 * 消息处理器
 *
 * @author
 */
@Slf4j
@Component
public class MessageHandler {
	@Resource
	private IRtuBaseInfoService rtuBaseInfoService;

	@Resource
	private IRtuCheckOrderService rtuCheckOrderService;

	@Resource
	private IRtuManageService rtuManageService;

	@Resource
	private IRtuWarningService rtuWarningService;

	@Resource
	private IEtlTaskInfoService etlTaskInfoService;

	@Resource
	private IRtuDataRainService rtuDataRainService;

	@Resource
	private IRtuDataRiverService rtuDataRiverService;

	@Resource
	private IRtuDataRsvrService rtuDataRsvrService;

	@Resource
	private IRtuDataGroundService rtuDataGroundService;

	@Resource
	private IRtuWarnSettingService rtuWarnSettingService;

	@Autowired
	private KafkaTemplate<String, String> kafkaTemplate;

	@Resource
	private RedisTemplate redisTemplate;

	@Value("${spring.mq-topic.ywxt-rtu-status}")
	private String topicYwxtRtuStatus;

	/**
	 * ETL任务执行信息
	 *
	 * @param record
	 * @param acknowledgment
	 */
	@KafkaListener(groupId = "${spring.kafka.consumer.group-id}", topics = "${spring.mq-topic.topic-etl-task-run-info}", containerFactory = "ackContainerFactory")
	public void taskRunInfoMessage(ConsumerRecord record, Acknowledgment acknowledgment) {
		try {
			String message = (String) record.value();
			log.info("{} ,{}", "task run info", message);
			JSONObject taskInfo = JSONObject.parseObject(message);
			EtlTaskInfoEntity etlTaskInfoEntity = new EtlTaskInfoEntity();
			etlTaskInfoEntity.setTaskType(taskInfo.getString("taskType"));
			etlTaskInfoEntity.setTaskExecuteTime(new Date(taskInfo.getLong("taskExecuteTime")));
			etlTaskInfoEntity.setTaskExecuteLong(taskInfo.getInteger("taskExecuteLong"));
			etlTaskInfoEntity.setRtuCount(taskInfo.getInteger("rtuCount"));
			etlTaskInfoEntity.setCollectDataCount(taskInfo.getInteger("collectDataCount"));
			etlTaskInfoEntity.setRemark(taskInfo.getString("remark"));
			etlTaskInfoService.save(etlTaskInfoEntity);
		} catch (Exception e) {
			log.error(e.getMessage(), e);
		} finally {
			// 手动提交 offset
			acknowledgment.acknowledge();
		}
	}

	/**
	 * ETL雨情数据
	 *
	 * @param record
	 * @param acknowledgment
	 */
	@KafkaListener(groupId = "${spring.kafka.consumer.group-id}", topics = "${spring.mq-topic.topic-etl-rain}", containerFactory = "ackContainerFactory")
	public void etlRainInfoMessage(ConsumerRecord record, Acknowledgment acknowledgment) {
		try {
			String message = (String) record.value();
			log.info("{} ,{}", "etl rain info", message);
			EtlRainDataEntity rainDataEntity = JSONObject.parseObject(message, EtlRainDataEntity.class);
			//查询是否有效测站，只处理有效测站信息
			LambdaQueryWrapper<RtuInfoEntity> wrapper = Wrappers.<RtuInfoEntity>query().lambda();
			wrapper.eq(RtuInfoEntity::getIsDeleted, 0);
			wrapper.eq(RtuInfoEntity::getRtuCode, rainDataEntity.getRtuCode());
			long count = rtuBaseInfoService.count(wrapper);
			if (count > 0) {
				//更新数据
				rtuDataRainService.updateRainInfo(rainDataEntity);
				//5分钟延时检测
				ValueOperations<String, String> ops = redisTemplate.opsForValue();
				String key = "etl.rain." + rainDataEntity.getRtuCode() + ".up.5min";
				//记录5分钟报文
				if (rainDataEntity.getIntv() == 0.05) {
					if (redisTemplate.hasKey(key)) {
						String value = ops.get(key);
						JSONArray array = JSONArray.parseArray(value);
						JSONObject object = new JSONObject();
						object.put("tm", Func.formatDateTime(rainDataEntity.getTm()));
						object.put("from", Func.formatDateTime(rainDataEntity.getFromDate()));
						if (array.size() > 0) {
							//连续接收到2次5分钟报文时，进行检测
							JSONObject beforeObject = array.getJSONObject(0);
							rtuWarningService.checkRtuRainUpDelayStatus(rainDataEntity.getRtuCode(), beforeObject, object);
							array.clear();
							array.add(object);
						} else {
							array.add(object);
						}
						ops.set(key, array.toJSONString());
					} else {
						JSONArray array = new JSONArray();
						JSONObject object = new JSONObject();
						object.put("tm", Func.formatDateTime(rainDataEntity.getTm()));
						object.put("from", Func.formatDateTime(rainDataEntity.getFromDate()));
						array.add(object);
						ops.set(key, array.toJSONString());
					}
				} else {
					//清空5分钟报文记录
					JSONArray array = new JSONArray();
					ops.set(key, array.toJSONString());
					//记录整点报
					ListOperations<String, String> listOps = redisTemplate.opsForList();
					key = "etl.rain." + rainDataEntity.getRtuCode() + ".hour.cache";
					if (redisTemplate.hasKey(key)) {
						long size = listOps.size(key);
						if (size > 24) {
							listOps.leftPop(key);
						}
					}
					listOps.rightPush(key, JSON.toJSONString(rainDataEntity));
					//漏报检测
					key = "etl.rain." + rainDataEntity.getRtuCode() + ".up.hour";
					if (redisTemplate.hasKey(key)) {
						String value = ops.get(key);
						array = JSONArray.parseArray(value);
					} else {
						array = new JSONArray();
					}
					JSONObject object = new JSONObject();
					object.put("tm", Func.formatDateTime(rainDataEntity.getTm()));
					object.put("from", Func.formatDateTime(rainDataEntity.getFromDate()));
					array.add(object);
					ops.set(key, array.toJSONString());

					//rtuWarningService.checkRtuRainMissOutStatus(rainDataEntity);
					//小时报延时检测
					rtuWarningService.checkRtuRainUpHourDelayStatus(rainDataEntity);
				}
				//测站状态检测
				RtuStatusDTO statusEntity = new RtuStatusDTO();
				statusEntity.setLastUpTime(rainDataEntity.getTm());
				statusEntity.setRtuCode(rainDataEntity.getRtuCode());
				statusEntity.setIsRain(1);
				statusEntity.setFromDate(rainDataEntity.getFromDate());
				kafkaTemplate.send(topicYwxtRtuStatus, JsonUtil.toJson(statusEntity));
			}
		} catch (Exception e) {
			log.error(e.getMessage(), e);
		} finally {
			// 手动提交 offset
			acknowledgment.acknowledge();
		}
	}

	/**
	 * ETL河道水情数据
	 *
	 * @param record
	 * @param acknowledgment
	 */
	@KafkaListener(groupId = "${spring.kafka.consumer.group-id}", topics = "${spring.mq-topic.topic-etl-river}", containerFactory = "ackContainerFactory")
	public void etlRiverInfoMessage(ConsumerRecord record, Acknowledgment acknowledgment) {
		try {
			String message = (String) record.value();
			log.info("{} ,{}", "etl river info", message);
			EtlRiverDataEntity riverDataEntity = JSONObject.parseObject(message, EtlRiverDataEntity.class);
			//查询是否有效测站，只处理有效测站信息
			LambdaQueryWrapper<RtuInfoEntity> wrapper = Wrappers.<RtuInfoEntity>query().lambda();
			wrapper.eq(RtuInfoEntity::getIsDeleted, 0);
			wrapper.eq(RtuInfoEntity::getRtuCode, riverDataEntity.getRtuCode());
			long count = rtuBaseInfoService.count(wrapper);
			if (count > 0) {
				//更新数据
				rtuDataRiverService.updateRiverInfo(riverDataEntity);
				//延时检测
				ValueOperations<String, String> ops = redisTemplate.opsForValue();
//				String key = "etl.river." + riverDataEntity.getRtuCode() + ".up.5min";
//				//记录5分钟报文
//				if (redisTemplate.hasKey(key)) {
//					String value = ops.get(key);
//					JSONArray array = JSONArray.parseArray(value);
//					if (array.size() > 0) {
//						JSONObject object = new JSONObject();
//						object.put("tm", Func.formatDateTime(riverDataEntity.getTm()));
//						object.put("from", Func.formatDateTime(riverDataEntity.getFromDate()));
//						//连续接收到2次5分钟报文时，进行检测
//						JSONObject beforeObject = array.getJSONObject(array.size() - 1);
//						String upTimeText = beforeObject.getString("tm");
//						Date beforeUpTime = Func.parseDate(upTimeText, "yyyy-MM-dd HH:mm:ss");
//						Calendar beforeDate = Calendar.getInstance();
//						beforeDate.setTime(beforeUpTime);
//						beforeDate.set(Calendar.SECOND, 0);
//						beforeDate.add(Calendar.MINUTE, 5);
//						Calendar nowDate = Calendar.getInstance();
//						nowDate.setTime(riverDataEntity.getTm());
//						nowDate.set(Calendar.SECOND, 0);
//						if (beforeDate.get(Calendar.YEAR) == nowDate.get(Calendar.YEAR) && beforeDate.get(Calendar.MONTH) == nowDate.get(Calendar.MONTH) && beforeDate.get(Calendar.DAY_OF_MONTH) == nowDate.get(Calendar.DAY_OF_MONTH) && beforeDate.get(Calendar.HOUR_OF_DAY) == nowDate.get(Calendar.HOUR_OF_DAY) && beforeDate.get(Calendar.MINUTE) == nowDate.get(Calendar.MINUTE)) {
//							//连续收到5分钟报文
//							rtuWarningService.checkRtuRiverUpDelayStatus(riverDataEntity.getRtuCode(), beforeObject, object);
//						}
//						array.clear();
//						array.add(object);
//					} else {
//						JSONObject object = new JSONObject();
//						object.put("tm", Func.formatDateTime(riverDataEntity.getTm()));
//						object.put("from", Func.formatDateTime(riverDataEntity.getFromDate()));
//						array.add(object);
//					}
//					ops.set(key, array.toJSONString());
//				} else {
//					JSONArray array = new JSONArray();
//					JSONObject object = new JSONObject();
//					object.put("tm", Func.formatDateTime(riverDataEntity.getTm()));
//					object.put("from", Func.formatDateTime(riverDataEntity.getFromDate()));
//					array.add(object);
//					ops.set(key, array.toJSONString());
//				}
				JSONArray array = new JSONArray();
				//漏报检测
				String key = "etl.river." + riverDataEntity.getRtuCode() + ".up.hour";
				if (redisTemplate.hasKey(key)) {
					String value = ops.get(key);
					array = JSONArray.parseArray(value);
				}
				JSONObject object = new JSONObject();
				object.put("tm", Func.formatDateTime(riverDataEntity.getTm()));
				object.put("from", Func.formatDateTime(riverDataEntity.getFromDate()));
				array.add(object);
				ops.set(key, array.toJSONString());

				//rtuWarningService.checkRtuRiverMissOutStatus(riverDataEntity);
				//小时报延时检测
				rtuWarningService.checkRtuRiverUpHourDelayStatus(riverDataEntity);
				//测站状态检测
				RtuStatusDTO statusEntity = new RtuStatusDTO();
				statusEntity.setLastUpTime(riverDataEntity.getTm());
				statusEntity.setRtuCode(riverDataEntity.getRtuCode());
				statusEntity.setIsRiver(1);
				statusEntity.setFromDate(riverDataEntity.getFromDate());
				kafkaTemplate.send(topicYwxtRtuStatus, JsonUtil.toJson(statusEntity));
			}
		} catch (Exception e) {
			log.error(e.getMessage(), e);
		} finally {
			// 手动提交 offset
			acknowledgment.acknowledge();
		}
	}

	/**
	 * ETL水库水情数据
	 *
	 * @param record
	 * @param acknowledgment
	 */
	@KafkaListener(groupId = "${spring.kafka.consumer.group-id}", topics = "${spring.mq-topic.topic-etl-rsvr}", containerFactory = "ackContainerFactory")
	public void etlRsvrInfoMessage(ConsumerRecord record, Acknowledgment acknowledgment) {
		try {
			String message = (String) record.value();
			log.info("{} ,{}", "etl rsvr info", message);
			EtlRsvrDataEntity rsvrDataEntity = JSONObject.parseObject(message, EtlRsvrDataEntity.class);
			//查询是否有效测站，只处理有效测站信息
			LambdaQueryWrapper<RtuInfoEntity> wrapper = Wrappers.<RtuInfoEntity>query().lambda();
			wrapper.eq(RtuInfoEntity::getIsDeleted, 0);
			wrapper.eq(RtuInfoEntity::getRtuCode, rsvrDataEntity.getRtuCode());
			long count = rtuBaseInfoService.count(wrapper);
			if (count > 0) {
				//更新数据
				rtuDataRsvrService.updateRsvrInfo(rsvrDataEntity);
//				//延时检测
//				ValueOperations<String, String> ops = redisTemplate.opsForValue();
//				String key = "etl.river." + rsvrDataEntity.getRtuCode() + ".up.5min";
//				//记录5分钟报文
//				if (redisTemplate.hasKey(key)) {
//					String value = ops.get(key);
//					JSONArray array = JSONArray.parseArray(value);
//					if (array.size() > 0) {
//						JSONObject object = new JSONObject();
//						object.put("tm", Func.formatDateTime(rsvrDataEntity.getTm()));
//						object.put("from", Func.formatDateTime(rsvrDataEntity.getFromDate()));
//						//连续接收到2次5分钟报文时，进行检测
//						JSONObject beforeObject = array.getJSONObject(array.size() - 1);
//						String upTimeText = beforeObject.getString("tm");
//						Date beforeUpTime = Func.parseDate(upTimeText, "yyyy-MM-dd HH:mm:ss");
//						Calendar beforeDate = Calendar.getInstance();
//						beforeDate.setTime(beforeUpTime);
//						beforeDate.set(Calendar.SECOND, 0);
//						beforeDate.add(Calendar.MINUTE, 5);
//						Calendar nowDate = Calendar.getInstance();
//						nowDate.setTime(rsvrDataEntity.getTm());
//						nowDate.set(Calendar.SECOND, 0);
//						if (beforeDate.get(Calendar.YEAR) == nowDate.get(Calendar.YEAR) && beforeDate.get(Calendar.MONTH) == nowDate.get(Calendar.MONTH) && beforeDate.get(Calendar.DAY_OF_MONTH) == nowDate.get(Calendar.DAY_OF_MONTH) && beforeDate.get(Calendar.HOUR_OF_DAY) == nowDate.get(Calendar.HOUR_OF_DAY) && beforeDate.get(Calendar.MINUTE) == nowDate.get(Calendar.MINUTE)) {
//							//连续收到5分钟报文
//							rtuWarningService.checkRtuRsvrUpDelayStatus(rsvrDataEntity.getRtuCode(), beforeObject, object);
//						}
//						array.clear();
//						array.add(object);
//					} else {
//						JSONObject object = new JSONObject();
//						object.put("tm", Func.formatDateTime(rsvrDataEntity.getTm()));
//						object.put("from", Func.formatDateTime(rsvrDataEntity.getFromDate()));
//						array.add(object);
//					}
//					ops.set(key, array.toJSONString());
//				} else {
//					JSONArray array = new JSONArray();
//					JSONObject object = new JSONObject();
//					object.put("tm", Func.formatDateTime(rsvrDataEntity.getTm()));
//					object.put("from", Func.formatDateTime(rsvrDataEntity.getFromDate()));
//					array.add(object);
//					ops.set(key, array.toJSONString());
//				}
				//漏报检测
				String key = "etl.rsvr." + rsvrDataEntity.getRtuCode() + ".up.hour";
				ValueOperations<String, String> ops = redisTemplate.opsForValue();
				JSONArray array = new JSONArray();
				//漏报检测

				if (redisTemplate.hasKey(key)) {
					String value = ops.get(key);
					array = JSONArray.parseArray(value);
				}
				JSONObject object = new JSONObject();
				object.put("tm", Func.formatDateTime(rsvrDataEntity.getTm()));
				object.put("from", Func.formatDateTime(rsvrDataEntity.getFromDate()));
				array.add(object);
				ops.set(key, array.toJSONString());

				//rtuWarningService.checkRtuRsvrMissOutStatus(rsvrDataEntity);
				//小时报延时检测
				rtuWarningService.checkRtuRsvrUpHourDelayStatus(rsvrDataEntity);
				//测站状态检测
				RtuStatusDTO statusEntity = new RtuStatusDTO();
				statusEntity.setLastUpTime(rsvrDataEntity.getTm());
				statusEntity.setRtuCode(rsvrDataEntity.getRtuCode());
				statusEntity.setIsRes(1);
				statusEntity.setFromDate(rsvrDataEntity.getFromDate());
				kafkaTemplate.send(topicYwxtRtuStatus, JsonUtil.toJson(statusEntity));
			}
		} catch (Exception e) {
			log.error(e.getMessage(), e);
		} finally {
			// 手动提交 offset
			acknowledgment.acknowledge();
		}
	}

	/**
	 * ETL墒情数据
	 *
	 * @param record
	 * @param acknowledgment
	 */
	@KafkaListener(groupId = "${spring.kafka.consumer.group-id}", topics = "${spring.mq-topic.topic-etl-ground}", containerFactory = "ackContainerFactory")
	public void etlGroundInfoMessage(ConsumerRecord record, Acknowledgment acknowledgment) {
		try {
			String message = (String) record.value();
			log.info("{} ,{}", "etl ground info", message);
			EtlGroundDataEntity entity = JSONObject.parseObject(message, EtlGroundDataEntity.class);
			//查询是否有效测站，只处理有效测站信息
			LambdaQueryWrapper<RtuInfoEntity> wrapper = Wrappers.<RtuInfoEntity>query().lambda();
			wrapper.eq(RtuInfoEntity::getIsDeleted, 0);
			wrapper.eq(RtuInfoEntity::getRtuCode, entity.getRtuCode());
			long count = rtuBaseInfoService.count(wrapper);
			if (count > 0) {
				//更新数据
				rtuDataGroundService.updateGroundInfo(entity);
				//测站状态检测
				RtuStatusDTO statusEntity = new RtuStatusDTO();
				statusEntity.setLastUpTime(entity.getTm());
				statusEntity.setRtuCode(entity.getRtuCode());
				statusEntity.setIsGround(1);
				statusEntity.setFromDate(entity.getFromDate());
				kafkaTemplate.send(topicYwxtRtuStatus, JsonUtil.toJson(statusEntity));
			}
		} catch (Exception e) {
			log.error(e.getMessage(), e);
		} finally {
			// 手动提交 offset
			acknowledgment.acknowledge();
		}
	}

	/**
	 * 测站状态检测
	 *
	 * @param record
	 * @param acknowledgment
	 */
	@KafkaListener(groupId = "${spring.kafka.consumer.group-id}", topics = "${spring.mq-topic.ywxt-rtu-status}", containerFactory = "ackContainerFactory")
	public void ywxtRtuStatus(ConsumerRecord record, Acknowledgment acknowledgment) {
		try {
			String message = (String) record.value();
			RtuStatusDTO dto = JsonUtil.parse(message, RtuStatusDTO.class);
			//统计上报间隔
			rtuManageService.rtuUpCount(dto);
			//更新测站最新上报时间状态
			RtuStatusEntity statusEntity = new RtuStatusEntity();
			statusEntity.setRtuCode(dto.getRtuCode());
			statusEntity.setLastUpTime(dto.getLastUpTime());
			rtuManageService.updateLastTime(statusEntity);
		} catch (Exception e) {
			log.error(e.getMessage(), e);
		} finally {
			// 手动提交 offset
			acknowledgment.acknowledge();
		}
	}

	/**
	 * 预警信息
	 *
	 * @param record
	 * @param acknowledgment
	 */
	@KafkaListener(groupId = "${spring.kafka.consumer.group-id}", topics = "${spring.mq-topic.ywxt-warning}", containerFactory = "ackContainerFactory")
	public void handleMessage(ConsumerRecord record, Acknowledgment acknowledgment) {
		try {
			String message = (String) record.value();
			RtuWarningInfoEntity warningInfoEntity = JsonUtil.parse(message, RtuWarningInfoEntity.class);
			int statusCode = warningInfoEntity.getWarningStatus();
			if (statusCode == WarningStatusEnum.STATUS_HAPPEN.getCode()) {
				//更新测站运行状态
				RtuStatusEntity statusEntity = new RtuStatusEntity();
				statusEntity.setRtuCode(warningInfoEntity.getRtuCode());
				statusEntity.setRunStatusId(EquipmentStatusEnum.STATUS_FAILURE.getCode());
				rtuManageService.updateRtuStatus(statusEntity);
				//测站检测参数
				int warnClockAutoTask = 0;
				int warnDelayAutoTask = 0;
				int warnOfflineAutoTask = 0;
				int warnMissOutAutoTask = 0;
				int warnOutlierValueAutoTask = 0;
				int warnHourUpDelayAutoTask = 0;
				LambdaQueryWrapper<RtuWarnSettingEntity> warnSettingWrapper = Wrappers.<RtuWarnSettingEntity>query().lambda();
				warnSettingWrapper.eq(RtuWarnSettingEntity::getIsDeleted, 0);
				List<RtuWarnSettingEntity> warnSettingEntityList = rtuWarnSettingService.list(warnSettingWrapper);
				if (null != warnSettingEntityList && warnSettingEntityList.size() > 0) {
					if (warnSettingEntityList.get(0).getWarnClockAutoTask() != null) {
						warnClockAutoTask = warnSettingEntityList.get(0).getWarnClockAutoTask();
					}
					if (warnSettingEntityList.get(0).getWarnDelayAutoTask() != null) {
						warnDelayAutoTask = warnSettingEntityList.get(0).getWarnDelayAutoTask();
					}
					if (warnSettingEntityList.get(0).getWarnOfflineAutoTask() != null) {
						warnOfflineAutoTask = warnSettingEntityList.get(0).getWarnOfflineAutoTask();
					}
					if (warnSettingEntityList.get(0).getWarnMissOutAutoTask() != null) {
						warnMissOutAutoTask = warnSettingEntityList.get(0).getWarnMissOutAutoTask();
					}
					if (warnSettingEntityList.get(0).getWarnOutlierValueAutoTask() != null) {
						warnOutlierValueAutoTask = warnSettingEntityList.get(0).getWarnOutlierValueAutoTask();
					}
					if (warnSettingEntityList.get(0).getWarnHourUpDelayAutoTask() != null) {
						warnHourUpDelayAutoTask = warnSettingEntityList.get(0).getWarnHourUpDelayAutoTask();
					}
				}
				if (WarnKindEnum.WARN_CLOCK.getCode() == warningInfoEntity.getWarningKind()) {
					if (warnClockAutoTask == WarningActiveEnum.ACTIVE_OPEN.getCode()) {
						//创建或更新维修任务
						rtuCheckOrderService.createOrUpdateOrder(warningInfoEntity);
					}
				} else if (WarnKindEnum.WARN_UP_MIN_DELAY.getCode() == warningInfoEntity.getWarningKind()) {
					if (warnDelayAutoTask == WarningActiveEnum.ACTIVE_OPEN.getCode()) {
						//创建或更新维修任务
						rtuCheckOrderService.createOrUpdateOrder(warningInfoEntity);
					}
				} else if (WarnKindEnum.WARN_OFFLINE.getCode() == warningInfoEntity.getWarningKind()) {
					if (warnOfflineAutoTask == WarningActiveEnum.ACTIVE_OPEN.getCode()) {
						//创建或更新维修任务
						rtuCheckOrderService.createOrUpdateOrder(warningInfoEntity);
					}
				} else if (WarnKindEnum.WARN_RAIN_MISS_OUT.getCode() == warningInfoEntity.getWarningKind()) {
					if (warnMissOutAutoTask == WarningActiveEnum.ACTIVE_OPEN.getCode()) {
						//创建或更新维修任务
						rtuCheckOrderService.createOrUpdateOrder(warningInfoEntity);
					}
				} else if (WarnKindEnum.WARN_WL_MISS_OUT.getCode() == warningInfoEntity.getWarningKind()) {
					if (warnMissOutAutoTask == WarningActiveEnum.ACTIVE_OPEN.getCode()) {
						//创建或更新维修任务
						rtuCheckOrderService.createOrUpdateOrder(warningInfoEntity);
					}
				} else if (WarnKindEnum.WARN_OUTLIER_VALUE.getCode() == warningInfoEntity.getWarningKind()) {
					if (warnOutlierValueAutoTask == WarningActiveEnum.ACTIVE_OPEN.getCode()) {
						//创建或更新维修任务
						rtuCheckOrderService.createOrUpdateOrder(warningInfoEntity);
					}
				} else if (WarnKindEnum.WARN_RAIN_UP_HOUR_DELAY.getCode() == warningInfoEntity.getWarningKind()) {
					if (warnHourUpDelayAutoTask == WarningActiveEnum.ACTIVE_OPEN.getCode()) {
						//创建或更新维修任务
						rtuCheckOrderService.createOrUpdateOrder(warningInfoEntity);
					}
				} else if (WarnKindEnum.WARN_WL_UP_HOUR_DELAY.getCode() == warningInfoEntity.getWarningKind()) {
					if (warnHourUpDelayAutoTask == WarningActiveEnum.ACTIVE_OPEN.getCode()) {
						//创建或更新维修任务
						rtuCheckOrderService.createOrUpdateOrder(warningInfoEntity);
					}
				}
			} else if (statusCode == WarningStatusEnum.STATUS_CLOSE.getCode()) {
				//预警恢复
				LambdaQueryWrapper<RtuWarningInfoEntity> warningWrapper = Wrappers.<RtuWarningInfoEntity>query().lambda();
				warningWrapper.eq(RtuWarningInfoEntity::getIsDeleted, 0);
				warningWrapper.eq(RtuWarningInfoEntity::getRtuCode, warningInfoEntity.getRtuCode());
				warningWrapper.eq(RtuWarningInfoEntity::getWarningStatus, WarningStatusEnum.STATUS_HAPPEN.getCode());
				List<RtuWarningInfoEntity> warningInfoEntityList = rtuWarningService.list(warningWrapper);
				if (null == warningInfoEntityList || warningInfoEntityList.size() == 0) {
					//更新测站运行状态
					RtuStatusEntity statusEntity = new RtuStatusEntity();
					statusEntity.setRtuCode(warningInfoEntity.getRtuCode());
					statusEntity.setRunStatusId(EquipmentStatusEnum.STATUS_RUN.getCode());
					rtuManageService.updateRtuStatus(statusEntity);
					//没有预警，可以关闭维修任务工单
					CheckOrderInfoDTO checkOrderInfoDTO = new CheckOrderInfoDTO();
					checkOrderInfoDTO.setRtuCode(warningInfoEntity.getRtuCode());
					rtuCheckOrderService.closeOrder(checkOrderInfoDTO);
				}
			}
		} catch (Exception e) {
			log.error(e.getMessage(), e);
		} finally {
			// 手动提交 offset
			acknowledgment.acknowledge();
		}
	}
}
