btxtgxq-system-bigdata/src/main/java/com/cm/bigdata/task/BigDataTask.java

package com.cm.bigdata.task;

import com.cm.bigdata.dao.kpi.IKpiDao;
import com.cm.bigdata.pojo.pos.kpi.CaseAssignPO;
import com.cm.bigdata.pojo.pos.kpi.ReportCasePO;
import com.cm.bigdata.service.kpi.IReportCaseService;
import com.cm.common.utils.DateUtil;
import com.cm.common.utils.HashMapUtil;
import com.cm.common.utils.UUIDUtil;
import org.joda.time.Hours;
import org.joda.time.LocalDateTime;
import org.joda.time.format.DateTimeFormat;
import org.joda.time.format.DateTimeFormatter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Configuration
@EnableScheduling
@Service
public class BigDataTask {

    private static final Logger LOG = LoggerFactory.getLogger(BigDataTask.class);
    @Autowired
    private IKpiDao kpiDao;
    @Autowired
    private IReportCaseService reportCaseService;


    // 每10分钟执行一次
    @Scheduled(cron = "0 0/10 * * * ?")
    public void ddyDayWorkTask() throws Exception {
        DateTimeFormatter dateTimeFormatter = DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss");
        LocalDateTime nowDateTime = LocalDateTime.now();
        LOG.debug("day work task, {}", nowDateTime.toString(dateTimeFormatter));
        // 统计过去24小时内没有转派的案件
        Map<String, Object> params = new HashMap<>();
        params.put("caseStatus", 2);
        // 为下派的案件
        List<ReportCasePO> reportCasePOS = reportCaseService.listPO(params);
        Map<String, ReportCasePO> reportCasePOMap = new HashMap<>();
        reportCasePOS.forEach(reportCasePO -> {
            reportCasePOMap.put(reportCasePO.getReportCaseId(), reportCasePO);
        });
        params.clear();
        params.put("isComplete", 0);
        // 已经保存未完成的调度记录
        List<CaseAssignPO> caseAssignPOS = kpiDao.listCaseAssignPO(params);
        Map<String, CaseAssignPO> caseAssignPOMap = new HashMap<>();
        caseAssignPOS.forEach(caseAssignPO -> {
            caseAssignPOMap.put(caseAssignPO.getCaseId(), caseAssignPO);
        });
        // 原先不存在，需要新增的
        List<CaseAssignPO> saves = new ArrayList<>();
        // 原先存在，现在不存在，判断为已处理且未超时
        List<CaseAssignPO> assigneds = new ArrayList<>();
        // 原先存在，现在依然存在，标记未未处理且已超时
        List<CaseAssignPO> updates = new ArrayList<>();
        // 已经删除的数据
        List<String> deleteCaseIds = new ArrayList<>();
        reportCasePOS.forEach(reportCasePO -> {
            if (reportCasePO.getIsDelete() == 1) {
                deleteCaseIds.add(reportCasePO.getReportCaseId());
            }
            CaseAssignPO caseAssignPO = new CaseAssignPO();
            caseAssignPO.setCaseAssignId(UUIDUtil.getUUID());
            caseAssignPO.setCaseId(reportCasePO.getReportCaseId());
            caseAssignPO.setIsDelete(0);
            // 获得最后一条未完成的数据
            Map<String, Object> queryMap = new HashMap<>();
            queryMap.put("caseId", reportCasePO.getReportCaseId());
            queryMap.put("isComplete", 0);
            CaseAssignPO existCaseAssignPO = kpiDao.getCaseAssignPO(queryMap);
            if (existCaseAssignPO == null) {
                // 需要新增
                caseAssignPO.setGmtCreate(DateUtil.getTime());
                caseAssignPO.setIsTimeout(0);
                caseAssignPO.setIsComplete(0);
                saves.add(caseAssignPO);
                return;
            }
            // 判断是否超过24小时
            LocalDateTime gmtCreateDateTime = LocalDateTime.parse(existCaseAssignPO.getGmtCreate().substring(0, 18), dateTimeFormatter);
            if (Hours.hoursBetween(gmtCreateDateTime, nowDateTime).getHours() >= 24) {
                // 超过24小时标记为已超时
                caseAssignPO.setGmtModified(DateUtil.getTime());
                caseAssignPO.setIsTimeout(1);
                updates.add(caseAssignPO);
            }
        });
        caseAssignPOS.forEach(caseAssignPO -> {
            ReportCasePO reportCasePO = reportCasePOMap.get(caseAssignPO.getCaseId());
            if (reportCasePO == null) {
                caseAssignPO.setGmtModified(DateUtil.getTime());
                // 原先存在，现在不存在，说明已经转派，标记为完成
                caseAssignPO.setGmtComplete(DateUtil.getTime());
                caseAssignPO.setIsComplete(1);
                assigneds.add(caseAssignPO);
            }
        });
        for (CaseAssignPO caseAssignPO : saves) {
            Map<String, Object> map = HashMapUtil.beanToMap(caseAssignPO);
            kpiDao.saveCaseAssign(map);
        }
        for (CaseAssignPO caseAssignPO : updates) {
            Map<String, Object> map = HashMapUtil.beanToMap(caseAssignPO);
            kpiDao.updateCaseAssign(map);
        }
        for (CaseAssignPO caseAssignPO : assigneds) {
            Map<String, Object> map = HashMapUtil.beanToMap(caseAssignPO);
            kpiDao.updateCaseAssign(map);
        }
        if (!deleteCaseIds.isEmpty()) {
            Map<String, Object> deleteMap = new HashMap<>();
            deleteMap.put("caseIds", deleteCaseIds);
            kpiDao.removeCaseAssign(deleteMap);
        }
    }

}