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


import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springblade.core.tool.utils.ConcurrentDateFormat;
import org.springblade.core.tool.utils.Func;
import org.springblade.enums.EtlModeEnum;
import org.springblade.etl.entity.EtlGroundDataEntity;
import org.springblade.etl.service.IEtlGroundDataService;
import org.springblade.etl.service.IEtlMvAttStBaseService;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.kafka.core.KafkaTemplate;

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

/***
 * Date:2022/5/22
 * Title: 墒情同步
 * Description：
 * @author swp
 * @version 1.0
 * Remark：认为有必要的其他信息
 */
@Slf4j
public class EtlGroundDataTask implements Callable<Integer> {

    private IEtlGroundDataService etlGroundDataService;
    private IEtlMvAttStBaseService etlMvAttStBaseService;
    private RedisTemplate redisTemplate;
    private KafkaTemplate<String, String> kafkaTemplate;
    private String topic;
    private int etlMode;
    private String taskId;

    public EtlGroundDataTask(IEtlMvAttStBaseService etlMvAttStBaseService, IEtlGroundDataService etlGroundDataService, RedisTemplate redisTemplate, KafkaTemplate<String, String> kafkaTemplate, String topic, int etlMode, String taskId) {
        this.etlMvAttStBaseService = etlMvAttStBaseService;
        this.etlGroundDataService = etlGroundDataService;
        this.redisTemplate = redisTemplate;
        this.kafkaTemplate = kafkaTemplate;
        this.topic = topic;
        this.etlMode = etlMode;
        this.taskId = taskId;
    }

    @Override
    public Integer call() throws Exception {
        try {
            log.info("墒情数据同步任务,{},{}", taskId, new Date());
            long st = System.currentTimeMillis();
            JSONObject taskInfo = new JSONObject();
            taskInfo.put("taskType", "ground");
            taskInfo.put("remark", "墒情数据同步任务");
            taskInfo.put("taskExecuteTime", st);
            ValueOperations<String, String> ops = redisTemplate.opsForValue();
            int findCound = 0;
            //List<EtlMvAttStBaseEntity> rtuInfoEntityList =  etlMvAttStBaseService.selectList();
            List<EtlGroundDataEntity> rtuInfoEntityList = etlGroundDataService.selectRtus();
            if (null != rtuInfoEntityList && rtuInfoEntityList.size() > 0) {
                for (EtlGroundDataEntity rtuInfoEntity : rtuInfoEntityList) {
                    EtlGroundDataEntity dto = new EtlGroundDataEntity();
                    dto.setRtuCode(rtuInfoEntity.getRtuCode());
                    String key = "etl.ground." + rtuInfoEntity.getRtuCode() + "." + "updatetime";
                    if (etlMode == EtlModeEnum.ETL_ONCE.getCode()) {
                        List<EtlGroundDataEntity> groundDataEntityList;
                        if (redisTemplate.hasKey(key)) {
                            String lastUpdateTime = ops.get(key);
                            Date ut = Func.parse(lastUpdateTime, ConcurrentDateFormat.of("yyyy-MM-dd HH:mm:ss"));
                            dto.setTm(ut);
                            groundDataEntityList = etlGroundDataService.selectNewList(dto);
                        } else {
                            //没有数据，找到一条最近的数据作为基准
                            groundDataEntityList = etlGroundDataService.selectLastList(dto);
                        }
                        if (null != groundDataEntityList && groundDataEntityList.size() > 0) {
                            for (EtlGroundDataEntity etlGroundDataEntity : groundDataEntityList) {
                                //通过KAFKA发送最新一条数据
                                String jsonText = JSONObject.toJSONString(etlGroundDataEntity);
                                kafkaTemplate.send(topic, jsonText);
                            }
                            //更新时间
                            Date lastDataTime = groundDataEntityList.get(groundDataEntityList.size() - 1).getTm();
                            ops.set(key, Func.formatDateTime(lastDataTime));
                            findCound += 1;
                        }
                    } else {
                        List<EtlGroundDataEntity> groundDataEntityList = etlGroundDataService.selectLastList(dto);
                        if (null != groundDataEntityList && groundDataEntityList.size() > 0) {
                            EtlGroundDataEntity lastGroundDataEntity = groundDataEntityList.get(0);
                            //通过KAFKA发送最新一条数据
                            String jsonText = JSONObject.toJSONString(lastGroundDataEntity);
                            kafkaTemplate.send(topic, jsonText);
                            //更新时间
                            ops.set(key, Func.formatDateTime(lastGroundDataEntity.getFromDate()));
                            findCound += 1;
                        }
                    }
                }
            }
            long et = System.currentTimeMillis();
            taskInfo.put("taskExecuteLong", et - st);
            if (null != rtuInfoEntityList && rtuInfoEntityList.size() > 0) {
                taskInfo.put("rtuCount", rtuInfoEntityList.size());
            } else {
                taskInfo.put("rtuCount", 0);
            }
            taskInfo.put("collectDataCount", findCound);
            kafkaTemplate.send("topic-etl-task-run-info", taskInfo.toJSONString());
            return 0;
        } catch (Exception e) {
            log.error("error", e);
            throw e;
        }
    }
}
