/**
 * 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.EtlMvAttStBaseEntity;
import org.springblade.etl.entity.EtlRiverDataEntity;
import org.springblade.etl.service.IEtlMvAttStBaseService;
import org.springblade.etl.service.IEtlRiverDataService;
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 EtlRiverDataTask implements Callable<Integer> {
    private IEtlRiverDataService etlRiverDataService;
    private IEtlMvAttStBaseService etlMvAttStBaseService;
    private RedisTemplate redisTemplate;
    private KafkaTemplate<String, String> kafkaTemplate;
    private String topic;
    private int etlMode;
    private String taskId;

    public EtlRiverDataTask(IEtlMvAttStBaseService etlMvAttStBaseService, IEtlRiverDataService etlRiverDataService, RedisTemplate redisTemplate, KafkaTemplate<String, String> kafkaTemplate, String topic, int etlMode,String taskId) {
        this.etlMvAttStBaseService = etlMvAttStBaseService;
        this.etlRiverDataService = etlRiverDataService;
        this.redisTemplate = redisTemplate;
        this.kafkaTemplate = kafkaTemplate;
        this.topic = topic;
        this.etlMode = etlMode;
        this.taskId=taskId;
    }

    @Override
    public Integer call() throws Exception {
        try {
            long st = System.currentTimeMillis();
            log.info("河道水情数据同步任务,{},{}",taskId,new Date());
            JSONObject taskInfo = new JSONObject();
            taskInfo.put("taskType", "river");
            taskInfo.put("remark", "河道水情数据同步任务");
            taskInfo.put("taskExecuteTime", st);
            ValueOperations<String, String> ops = redisTemplate.opsForValue();
            int findCound = 0;
            List<EtlMvAttStBaseEntity> rtuInfoEntityList = etlMvAttStBaseService.selectList();
            if (null != rtuInfoEntityList && rtuInfoEntityList.size() > 0) {
                for (EtlMvAttStBaseEntity rtuInfoEntity : rtuInfoEntityList) {
                    EtlRiverDataEntity dto = new EtlRiverDataEntity();
                    dto.setRtuCode(rtuInfoEntity.getStCode());
                    String key = "etl.river." + rtuInfoEntity.getStCode() + "." + "updatetime";
                    if (etlMode == EtlModeEnum.ETL_ONCE.getCode()) {
                        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);
                        } else {
                            //没有数据，找到一条最近的数据作为基准
                        }
                        List<EtlRiverDataEntity> riverDataEntityList = etlRiverDataService.selectLastList(dto);
                        if (null != riverDataEntityList && riverDataEntityList.size() > 0) {
                            for (EtlRiverDataEntity etlRiverDataEntity : riverDataEntityList) {
                                //通过KAFKA发送最新一条数据
                                String jsonText = JSONObject.toJSONString(etlRiverDataEntity);
                                kafkaTemplate.send(topic, jsonText);
                            }
                            //更新上报时间
                            Date lastDataTime = riverDataEntityList.get(riverDataEntityList.size() - 1).getTm();
                            ops.set(key, Func.formatDateTime(lastDataTime));
                            findCound += 1;
                        } else {
                            //没有新数据
                            if (dto.getTm() == null) {
                                //没有任何一条数据
                            }
                        }
                    } else {
                        List<EtlRiverDataEntity> riverDataEntityList = etlRiverDataService.selectLastList(dto);
                        if (null != riverDataEntityList && riverDataEntityList.size() > 0) {
                            EtlRiverDataEntity lastRiverDataEntity = riverDataEntityList.get(0);
                            //通过KAFKA发送最新一条数据
                            String jsonText = JSONObject.toJSONString(lastRiverDataEntity);
                            kafkaTemplate.send(topic, jsonText);
                            //更新时间
                            ops.set(key, Func.formatDateTime(lastRiverDataEntity.getFromDate()));
                            findCound += 1;
                        } else {
                            //没有新数据
                            if (dto.getTm() == null) {
                                //没有任何一条数据
                            }
                        }
                    }
                }
            }
            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;
        }
    }
}
