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


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.springblade.core.tool.utils.ConcurrentDateFormat;
import org.springblade.core.tool.utils.Func;
import org.springblade.etl.dto.EtlWarningInfoDTO;
import org.springblade.etl.entity.EtlWarningInfoEntity;
import org.springblade.etl.service.IEtlMvAttStBaseService;
import org.springblade.etl.service.IEtlWarningService;
import org.springblade.modules.data.entity.OriginalWarningInfoEntity;
import org.springblade.modules.data.service.IWarnDataCacheService;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.kafka.core.KafkaTemplate;

import java.util.Calendar;
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 EtlWarnDataTask implements Callable<Integer> {
    private IEtlWarningService etlWarningService;
    private IWarnDataCacheService warnDataCacheService;
    private IEtlMvAttStBaseService etlMvAttStBaseService;
    private RedisTemplate redisTemplate;
    private KafkaTemplate<String, String> kafkaTemplate;
    private String topic;
    private int etlMode;
    private String taskId;

    public EtlWarnDataTask(IEtlMvAttStBaseService etlMvAttStBaseService, IEtlWarningService etlWarningService, IWarnDataCacheService warnDataCacheService, RedisTemplate redisTemplate, KafkaTemplate<String, String> kafkaTemplate, String topic, int etlMode, String taskId) {
        this.etlMvAttStBaseService = etlMvAttStBaseService;
        this.etlWarningService = etlWarningService;
        this.warnDataCacheService = warnDataCacheService;
        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", "rsvr");
            taskInfo.put("remark", "预警数据同步任务");
            taskInfo.put("taskExecuteTime", st);
            //因为预警数据会动态更新，所以需要更新已抽取的数据，只更新未关闭的预警数据
            OriginalWarningInfoEntity dto = new OriginalWarningInfoEntity();
            dto.setType("4");
            dto.setReviewStatus("1");
            List<OriginalWarningInfoEntity> originalWarningInfoEntities = warnDataCacheService.selectWarningList(dto);
            if (Func.notNull(originalWarningInfoEntities) && originalWarningInfoEntities.size() > 0) {
                for (OriginalWarningInfoEntity entity : originalWarningInfoEntities) {
                    //读取源数据库的预警信息
                    EtlWarningInfoDTO warningInfoDTO = new EtlWarningInfoDTO();
                    warningInfoDTO.setWarnId(entity.getWarnId());
                    EtlWarningInfoEntity etlWarningInfoEntity = etlWarningService.getByWarnId(warningInfoDTO);
                    if (Func.notNull(etlWarningInfoEntity)) {
                        entity.setWarnTime(etlWarningInfoEntity.getWarnTime());
                        entity.setWarnInfo(etlWarningInfoEntity.getWarnInfo());
                        entity.setWarnStatus(etlWarningInfoEntity.getWarnStatus());
                        entity.setReviewStatus(etlWarningInfoEntity.getReviewStatus());
                        warnDataCacheService.updateById(entity);
                    }
                }
                log.info("更新已抽取的未关闭预警数据共{}条", originalWarningInfoEntities.size());
            }
            //抽取源数据库最新数据,默认同步所有的数据
            Calendar calendar = Calendar.getInstance();
            calendar.setTimeInMillis(st);
            EtlWarningInfoDTO etlWarningInfoDTO = new EtlWarningInfoDTO();
            Date etlUpdateTime = calendar.getTime();
            ValueOperations<String, String> ops = redisTemplate.opsForValue();
            String key = "etl.warn.etl.updatetime";
            if (redisTemplate.hasKey(key)) {
                String lastUpdateTime = ops.get(key);
                if (Func.notNull(lastUpdateTime)) {
                    etlUpdateTime = Func.parse(lastUpdateTime, ConcurrentDateFormat.of("yyyy-MM-dd HH:mm:ss"));
                    etlWarningInfoDTO.setLastTime(etlUpdateTime);
                }
            }
            etlWarningInfoDTO.setReviewStatus("1");
            etlWarningInfoDTO.setType("4");
            //只抽取新创建的数据，更新的数据不在此部分处理
            List<EtlWarningInfoEntity> updateList = etlWarningService.selectNewList(etlWarningInfoDTO);
            if (Func.notNull(updateList) && updateList.size() > 0) {
                for (EtlWarningInfoEntity entity : updateList) {
                    LambdaQueryWrapper<OriginalWarningInfoEntity> wrapper = Wrappers.<OriginalWarningInfoEntity>query().lambda();
                    wrapper.eq(OriginalWarningInfoEntity::getWarnId, entity.getWarnId());
                    wrapper.last("limit 1");
                    OriginalWarningInfoEntity detail = warnDataCacheService.getOne(wrapper);
                    if (Func.isNull(detail)) {
                        OriginalWarningInfoEntity warnInfo = new OriginalWarningInfoEntity();
                        warnInfo.setWarnId(entity.getWarnId());
                        warnInfo.setStcd(entity.getStcd());
                        warnInfo.setWarnName(entity.getWarnName());
                        warnInfo.setWarnTime(entity.getWarnTime());
                        warnInfo.setWarnLevelCode(entity.getWarnLevelCode());
                        warnInfo.setWarnInfo(entity.getWarnInfo());
                        warnInfo.setWarnStatus(entity.getWarnStatus());
                        warnInfo.setReviewStatus(entity.getReviewStatus());
                        warnInfo.setType(entity.getType());
                        warnInfo.setRainWarnRate(entity.getRainWarnRate());
                        warnInfo.setWarnSort(entity.getWarnSort());
                        warnInfo.setWarnLocation(entity.getWarnLocation());
                        warnInfo.setWarnAdcd(entity.getWarnAdcd());
                        warnInfo.setUserAdcd(entity.getUserAdcd());
                        warnInfo.setWarnLgtd(entity.getWarnLgtd());
                        warnInfo.setWarnLttd(entity.getWarnLttd());
                        warnInfo.setIntv(entity.getIntv());
                        warnInfo.setDrp(entity.getDrp());
                        warnInfo.setZ(entity.getZ());
                        warnInfo.setQ(entity.getQ());
                        warnInfo.setWptn(entity.getWptn());
                        warnInfo.setRz(entity.getRz());
                        warnInfo.setRwptn(entity.getRwptn());
                        warnInfo.setInq(entity.getInq());
                        warnInfo.setRelWarnId(entity.getRelWarnId());
                        warnDataCacheService.save(warnInfo);
                    } else {
                        detail.setWarnTime(entity.getWarnTime());
                        detail.setWarnInfo(entity.getWarnInfo());
                        detail.setWarnStatus(entity.getWarnStatus());
                        if(Func.isNull(entity.getReviewStatus())){
                            detail.setReviewStatus("0");
                        }else {
                            detail.setReviewStatus(entity.getReviewStatus());
                        }
                        warnDataCacheService.updateById(detail);
                    }
                    Date warnTime = entity.getWarnTime();
                    if (warnTime.after(etlUpdateTime)) {
                        etlUpdateTime = warnTime;
                    }
                }
                //更新抽取同步时间
                ops.set(key, Func.formatDateTime(etlUpdateTime));
                log.info("抽取预警数据共{}条", updateList.size());
            }
            //推送到业务系统,任何有修改过的数据都需要进行同步
            dto = new OriginalWarningInfoEntity();
            Date ywxtUpdateTime = calendar.getTime();
            key = "etl.warn.ywxt.updatetime";
            if (redisTemplate.hasKey(key)) {
                //同步最新有变化的数据
                String lastUpdateTime = ops.get(key);
                if (Func.notNull(lastUpdateTime)) {
                    ywxtUpdateTime = Func.parse(lastUpdateTime, ConcurrentDateFormat.of("yyyy-MM-dd HH:mm:ss"));
                    dto.setUpdateTime(ywxtUpdateTime);
                }
            }
            originalWarningInfoEntities = warnDataCacheService.selectWarningList(dto);
            if (Func.notNull(originalWarningInfoEntities) && originalWarningInfoEntities.size() > 0) {
                for (OriginalWarningInfoEntity entity : originalWarningInfoEntities) {
                    //通过KAFKA发送最新一条数据
                    String jsonText = JSONObject.toJSONString(entity);
                    //log.info("warn info {}", jsonText);
                    kafkaTemplate.send(topic, jsonText);
                    Date updateTime = entity.getUpdateTime();
                    if (updateTime.after(ywxtUpdateTime)) {
                        ywxtUpdateTime = updateTime;
                    }
                }
                //更新运维系统同步时间
                ops.set(key, Func.formatDateTime(ywxtUpdateTime));
                log.info("推送预警数据共{}条", originalWarningInfoEntities.size());
            }
            //提交任务运行状态
            long et = System.currentTimeMillis();
            taskInfo.put("taskExecuteLong", et - st);
            kafkaTemplate.send("topic-etl-task-run-info", taskInfo.toJSONString());
            return 0;
        } catch (Exception e) {
            log.error("error", e);
            throw e;
        }
    }
}
