/**
 * Copyright
 * All right reserved.
 * 项目名称:
 * 创建日期:2022/5/22
 */
package org.springblade.etl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import lombok.extern.slf4j.Slf4j;
import org.springblade.etl.entity.EtlWarningInfoEntity;
import org.springblade.etl.service.IEtlWarningService;
import org.springblade.modules.business.warning.entity.RtuWarningInfoEntity;
import org.springblade.modules.business.warning.service.IRtuWarningService;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.messaging.simp.SimpMessagingTemplate;

import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.concurrent.Callable;

/***
 * Date:2022/5/22
 * Title: 环境辐射与气象监测系统(KRS) 模块
 * Description：MODBUS TCP 协议传输 ,读取任务
 * @author swp
 * @version 1.0
 * Remark：认为有必要的其他信息
 */
@Slf4j
public class EtlWarningTask implements Callable<Integer> {

	private RedisTemplate redisTemplate;
	private SimpMessagingTemplate messagingTemplate;
	private KafkaTemplate<String, String> kafkaTemplate;
	private IEtlWarningService etlWarningService;
	private IRtuWarningService rtuWarningService;

	public EtlWarningTask(RedisTemplate redisTemplate, KafkaTemplate<String, String> kafkaTemplate , SimpMessagingTemplate messagingTemplate, IEtlWarningService etlWarningService, IRtuWarningService rtuWarningService){
		this.redisTemplate=redisTemplate;
		this.kafkaTemplate=kafkaTemplate;
		this.messagingTemplate=messagingTemplate;
		this.etlWarningService=etlWarningService;
		this.rtuWarningService=rtuWarningService;
	}

	@Override
	public Integer call()  {
		try {
			HashOperations<String, String, String> opsHash = redisTemplate.opsForHash();
			SimpleDateFormat timeFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
			Calendar calendar = Calendar.getInstance();
			calendar.set(Calendar.HOUR_OF_DAY, 0);
			calendar.set(Calendar.MINUTE, 0);
			calendar.set(Calendar.SECOND, 0);
			Date lastProcessDate = calendar.getTime();
			if (opsHash.hasKey("data_process_time", "warning")) {
				String dt = opsHash.get("data_process_time", "warning");
				lastProcessDate = timeFormat.parse(dt);
			}
			LambdaQueryWrapper<EtlWarningInfoEntity> wrapper = Wrappers.<EtlWarningInfoEntity>query().lambda();
			wrapper.gt(EtlWarningInfoEntity::getOriginalCreateTime,lastProcessDate);
			wrapper.orderByAsc(EtlWarningInfoEntity::getOriginalCreateTime);
			List<EtlWarningInfoEntity> list = etlWarningService.list(wrapper);
			if (null != list && list.size() > 0) {
				for (EtlWarningInfoEntity warningInfo : list) {
					RtuWarningInfoEntity rtuWarningInfoEntity=new RtuWarningInfoEntity();
					rtuWarningService.save(rtuWarningInfoEntity);
				}
				EtlWarningInfoEntity lastwarn= list.get(list.size()-1);
				opsHash.put("data_process_time", "warning", timeFormat.format(lastwarn.getOriginalCreateTime()));
			}

		} catch (Exception e) {
			log.error("数据任务异常 {}", e.getMessage());
		}
		return 0;
	}
}
