/**
 * 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);
            //更新已抽取的数据
            LambdaQueryWrapper<OriginalWarningInfoEntity> wrapper = Wrappers.<OriginalWarningInfoEntity>query().lambda();
            wrapper.ne(OriginalWarningInfoEntity::getReviewStatus, "1");
            wrapper.eq(OriginalWarningInfoEntity::getType, "4");
            List<OriginalWarningInfoEntity> originalWarningInfoEntities = warnDataCacheService.list(wrapper);
            if (Func.notNull(originalWarningInfoEntities)) {
                for (OriginalWarningInfoEntity entity : originalWarningInfoEntities) {
                    EtlWarningInfoDTO warningInfoDTO = new EtlWarningInfoDTO();
                    warningInfoDTO.setWarnId(entity.getWarnId());
                    EtlWarningInfoEntity etlWarningInfoEntity = etlWarningService.getByWarnId(warningInfoDTO);
                    if (Func.notNull(etlWarningInfoEntity)) {
                        entity.setWarnInfo(etlWarningInfoEntity.getWarnInfo());
                        entity.setWarnStatus(etlWarningInfoEntity.getWarnStatus());
                        entity.setReviewStatus(etlWarningInfoEntity.getReviewStatus());
                        entity.setUpdateTime(etlWarningInfoEntity.getOriginalUpdateTime());
                        warnDataCacheService.updateById(entity);
                    }
                }
            }
            //抽取最新数据
            EtlWarningInfoDTO etlWarningInfoDTO = new EtlWarningInfoDTO();
            ValueOperations<String, String> ops = redisTemplate.opsForValue();
            String key = "etl.warn.updatetime";
            if (redisTemplate.hasKey(key)) {
                String lastUpdateTime = ops.get(key);
                Date ut = Func.parse(lastUpdateTime, ConcurrentDateFormat.of("yyyy-MM-dd HH:mm:ss"));
                etlWarningInfoDTO.setLastTime(ut);
            } else {
                //没有数据，同步最近24小时
                Calendar calendar = Calendar.getInstance();
                calendar.set(Calendar.DAY_OF_MONTH, -1);
                etlWarningInfoDTO.setLastTime(calendar.getTime());
            }
            etlWarningInfoDTO.setReviewStatus("1");
            etlWarningInfoDTO.setType("4");
            List<EtlWarningInfoEntity> updateList = etlWarningService.selectNewList(etlWarningInfoDTO);
            if (Func.notNull(updateList)) {
                for (EtlWarningInfoEntity entity : updateList) {
                    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());
                    warnInfo.setCreateTime(entity.getOriginalCreateTime());
                    warnInfo.setUpdateTime(entity.getOriginalUpdateTime());
                    warnDataCacheService.save(warnInfo);
                }
            }
            //推送到业务系统
            wrapper = Wrappers.<OriginalWarningInfoEntity>query().lambda();
            wrapper.eq(OriginalWarningInfoEntity::getIsDeleted, 0);
            wrapper.ne(OriginalWarningInfoEntity::getReviewStatus, "1");
            wrapper.eq(OriginalWarningInfoEntity::getType, "4");
            originalWarningInfoEntities = warnDataCacheService.list(wrapper);
            if (Func.notNull(originalWarningInfoEntities)) {
                for (OriginalWarningInfoEntity entity : originalWarningInfoEntities) {
                    //通过KAFKA发送最新一条数据
                    String jsonText = JSONObject.toJSONString(entity);
                    log.info("warn info {}", jsonText);
                    // kafkaTemplate.send(topic, jsonText);
                }
            }
            //更新缓存时间
            ops.set(key, Func.formatDateTime(etlWarningInfoDTO.getLastTime()));
            //提交任务运行状态
            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;
        }
    }
}
