/**
 * Copyright 2019 DH
 * All right reserved.
 * 项目名称: 大恒泰山系统
 * 创建日期:2023/2/21
 */
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.core.tool.jackson.JsonUtil;
import org.springblade.core.tool.utils.ConcurrentDateFormat;
import org.springblade.core.tool.utils.Func;
import org.springblade.etl.dto.EtlGroundDataDTO;
import org.springblade.etl.dto.EtlRainDataDTO;
import org.springblade.etl.dto.EtlRiverDataDTO;
import org.springblade.etl.dto.EtlRsvrDataDTO;
import org.springblade.etl.service.*;
import org.springblade.modules.baseinfo.rtu.entity.RtuInfoEntity;
import org.springblade.modules.baseinfo.rtu.service.IRtuBaseInfoService;
import org.springblade.modules.business.data.entity.RtuDataGroundEntity;
import org.springblade.modules.business.data.entity.RtuDataRainEntity;
import org.springblade.modules.business.data.entity.RtuDataRiverEntity;
import org.springblade.modules.business.data.entity.RtuDataRsvrEntity;
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.entity.RtuStatusEntity;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;

import java.util.Date;
import java.util.List;
import java.util.concurrent.Callable;

/***
 * Date:2023/2/21
 * Title: ETL数据处理任务模块
 * Description：处理雨情数据、河道水位数据、水库水位数据、墒情数据
 * @author dylan
 * @version 1.0
 * Remark：认为有必要的其他信息
 */
@Slf4j
public class EtlRtuDataTask implements Callable<Integer> {
	private KafkaTemplate<String, String> kafkaTemplate;
	private RedisTemplate redisTemplate;
	private IRtuBaseInfoService rtuBaseInfoService;
	private IEtlRainDataService etlRainDataService;
	private IRtuDataRainService rtuDataRainService;
	private IEtlRiverDataService etlRiverDataService;
	private IRtuDataRiverService rtuDataRiverService;
	private IEtlRsvrDataService etlRsvrDataService;
	private IRtuDataRsvrService rtuDataRsvrService;
	private IEtlGroundDataService etlGroundDataService;
	private IRtuDataGroundService rtuDataGroundService;

	EtlRtuDataTask(
		KafkaTemplate<String, String> kafkaTemplate,
		RedisTemplate redisTemplate,
		IRtuBaseInfoService rtuBaseInfoService,
		IEtlRainDataService etlRainDataService,
		IRtuDataRainService rtuDataRainService,
		IEtlRiverDataService etlRiverDataService,
		IRtuDataRiverService rtuDataRiverService,
		IEtlRsvrDataService etlRsvrDataService,
		IRtuDataRsvrService rtuDataRsvrService,
		IEtlGroundDataService etlGroundDataService,
		IRtuDataGroundService rtuDataGroundService) {
		this.kafkaTemplate = kafkaTemplate;
		this.redisTemplate = redisTemplate;
		this.rtuBaseInfoService = rtuBaseInfoService;
		this.etlRainDataService = etlRainDataService;
		this.rtuDataRainService = rtuDataRainService;
		this.etlRiverDataService = etlRiverDataService;
		this.rtuDataRiverService = rtuDataRiverService;
		this.etlRsvrDataService = etlRsvrDataService;
		this.rtuDataRsvrService = rtuDataRsvrService;
		this.etlGroundDataService = etlGroundDataService;
		this.rtuDataGroundService = rtuDataGroundService;
	}

	@Override
	public Integer call() {
		try {
			log.info("测站定时报数据抽取开始**************");
			long st = System.currentTimeMillis();
			HashOperations<String, String, String> opsHash = redisTemplate.opsForHash();
			LambdaQueryWrapper<RtuInfoEntity> wrapper = Wrappers.<RtuInfoEntity>query().lambda();
			wrapper.eq(RtuInfoEntity::getIsDeleted, 0);
			List<RtuInfoEntity> rtuInfoEntityList = rtuBaseInfoService.list(wrapper);
			for (RtuInfoEntity rtuInfoEntity : rtuInfoEntityList) {
				Date rtuUpTm = null;
				//读取最新一条雨情数据
				EtlRainDataDTO etlRainDataDTO = new EtlRainDataDTO();
				etlRainDataDTO.setRtuCode(rtuInfoEntity.getRtuCode());
				if (opsHash.hasKey("etl_rtu_rain_read_time", rtuInfoEntity.getRtuCode())) {
					String dt = opsHash.get("etl_rtu_rain_read_time", rtuInfoEntity.getRtuCode());
					Date lastDate = Func.parse(dt, ConcurrentDateFormat.of("yyyy-MM-dd HH:mm:ss"));
					etlRainDataDTO.setLastTime(lastDate);
				}
				RtuDataRainEntity rtuDataRainEntity = EtlRainDataTask.dataSync(etlRainDataDTO, etlRainDataService, rtuDataRainService);
				if (null != rtuDataRainEntity) {
					//保存时间
					String dt = Func.formatDateTime(rtuDataRainEntity.getTm());
					opsHash.put("etl_rtu_rain_read_time", rtuInfoEntity.getRtuCode(), dt);
					//最后上报时间
					rtuUpTm = rtuDataRainEntity.getTm();
				}
				//读取最新一条河道数据
				EtlRiverDataDTO etlRiverDataDTO = new EtlRiverDataDTO();
				etlRiverDataDTO.setRtuCode(rtuInfoEntity.getRtuCode());
				if (opsHash.hasKey("etl_rtu_river_read_time", rtuInfoEntity.getRtuCode())) {
					String dt = opsHash.get("etl_rtu_river_read_time", rtuInfoEntity.getRtuCode());
					Date lastDate = Func.parse(dt, ConcurrentDateFormat.of("yyyy-MM-dd HH:mm:ss"));
					etlRiverDataDTO.setLastTime(lastDate);
				}
				RtuDataRiverEntity rtuDataRiverEntity = EtlRiverDataTask.dataSync(etlRiverDataDTO, etlRiverDataService, rtuDataRiverService);
				if (null != rtuDataRiverEntity) {
					//保存时间
					String dt = Func.formatDateTime(rtuDataRiverEntity.getTm());
					opsHash.put("etl_rtu_river_read_time", rtuInfoEntity.getRtuCode(), dt);
					//最后上报时间
					if (rtuUpTm != null) {
						if (rtuDataRiverEntity.getTm().after(rtuUpTm)) {
							rtuUpTm = rtuDataRiverEntity.getTm();
						}
					} else {
						rtuUpTm = rtuDataRiverEntity.getTm();
					}
				}
				//读取最新一条水库数据
				EtlRsvrDataDTO etlRsvrDataDTO = new EtlRsvrDataDTO();
				etlRsvrDataDTO.setRtuCode(rtuInfoEntity.getRtuCode());
				if (opsHash.hasKey("etl_rtu_rsvr_read_time", rtuInfoEntity.getRtuCode())) {
					String dt = opsHash.get("etl_rtu_rsvr_read_time", rtuInfoEntity.getRtuCode());
					Date lastDate = Func.parse(dt, ConcurrentDateFormat.of("yyyy-MM-dd HH:mm:ss"));
					etlRsvrDataDTO.setLastTime(lastDate);
				}
				RtuDataRsvrEntity rsvrDataEntity = EtlRsvrDataTask.dataSync(etlRsvrDataDTO, etlRsvrDataService, rtuDataRsvrService);
				if (null != rsvrDataEntity) {
					//保存时间
					String dt = Func.formatDateTime(rsvrDataEntity.getTm());
					opsHash.put("etl_rtu_rsvr_read_time", rtuInfoEntity.getRtuCode(), dt);
					//最后上报时间
					if (rtuUpTm != null) {
						if (rsvrDataEntity.getTm().after(rtuUpTm)) {
							rtuUpTm = rsvrDataEntity.getTm();
						}
					} else {
						rtuUpTm = rsvrDataEntity.getTm();
					}
				}
				//读取最新一条墒情数据
				EtlGroundDataDTO groundDataDTO = new EtlGroundDataDTO();
				groundDataDTO.setRtuCode(rtuInfoEntity.getRtuCode());
				if (opsHash.hasKey("etl_rtu_ground_read_time", rtuInfoEntity.getRtuCode())) {
					String dt = opsHash.get("etl_rtu_ground_read_time", rtuInfoEntity.getRtuCode());
					Date lastDate = Func.parse(dt, ConcurrentDateFormat.of("yyyy-MM-dd HH:mm:ss"));
					groundDataDTO.setLastTime(lastDate);
				}
				RtuDataGroundEntity groundDataEntity = EtlGroundDataTask.dataSync(groundDataDTO, etlGroundDataService, rtuDataGroundService);
				if (null != groundDataEntity) {
					//保存时间
					String dt = Func.formatDateTime(groundDataEntity.getTm());
					opsHash.put("etl_rtu_ground_read_time", rtuInfoEntity.getRtuCode(), dt);
					//最后上报时间
					if (rtuUpTm != null) {
						if (groundDataEntity.getTm().after(rtuUpTm)) {
							rtuUpTm = groundDataEntity.getTm();
						}
					} else {
						rtuUpTm = groundDataEntity.getTm();
					}
				}
				//更新测站最近上报时间
				if (null != rtuUpTm) {
					RtuStatusEntity statusEntity = new RtuStatusEntity();
					statusEntity.setLastUpTime(rtuUpTm);
					statusEntity.setRtuCode(rtuInfoEntity.getRtuCode());
					kafkaTemplate.send("ywxt.rtu.status", JsonUtil.toJson(statusEntity));
				}
			}
			long et = System.currentTimeMillis();
			log.info("测站定时报数据抽取完成 {}", (et - st) / 1000);
		} catch (Exception e) {
			log.error("实时数据统计任务异常 {}", e.getMessage());
		}
		return 0;
	}
}
