package cn.datax.service.data.quality.schedule.task;

import cn.datax.common.core.DataConstant;
import cn.datax.common.database.DataSourceFactory;
import cn.datax.service.data.metadata.api.feign.MetadataSourceServiceFeign;
import cn.datax.service.data.quality.api.entity.CheckReportEntity;
import cn.datax.service.data.quality.api.entity.CheckRuleEntity;
import cn.datax.service.data.quality.api.entity.ScheduleLogEntity;
import cn.datax.service.data.quality.schedule.exception.ChildThreadException;
import cn.datax.service.data.quality.schedule.thread.MultiThreadHandler;
import cn.datax.service.data.quality.schedule.thread.parallel.ParallelTaskWithThreadPool;
import cn.datax.service.data.quality.service.CheckReportService;
import cn.datax.service.data.quality.service.CheckRuleService;
import cn.datax.service.data.quality.service.ScheduleLogService;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

@Slf4j
@Component("qualityTask")
public class QualityTask {

    @Autowired
    private CheckRuleService checkRuleService;

    @Autowired
    private DataSourceFactory dataSourceFactory;

    @Autowired
    private MetadataSourceServiceFeign metadataSourceServiceFeign;

    @Autowired
    private CheckReportService checkReportService;

    @Autowired
    private ScheduleLogService scheduleLogService;

    public void task(Map<String, Object> map) {
        System.out.println("执行参数：" + map);
        // 结果集
        List<CheckReportEntity> result = new ArrayList<>();
        // 获取可执行的核查规则
        List<CheckRuleEntity> list = checkRuleService.list(Wrappers.<CheckRuleEntity>lambdaQuery().eq(CheckRuleEntity::getStatus, DataConstant.TrueOrFalse.TRUE.getKey()));
        // 定义固定长度的线程池
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>(50),
                new BasicThreadFactory.Builder().namingPattern("executor-schedule-pool-%d").daemon(true).build());
        MultiThreadHandler handler = new ParallelTaskWithThreadPool(threadPoolExecutor);
        // 启动子线程作为要处理的并行任务
        list.stream().forEach(rule -> {
            TaskHander task = new TaskHander(metadataSourceServiceFeign, dataSourceFactory, rule, result);
            handler.addTask(task);
        });
        try {
            handler.run();
        } catch (ChildThreadException e) {
            System.out.println(e.getAllStackTraceMessage());
        }
        // 关闭线程池
        threadPoolExecutor.shutdown();
        // 核查报告
        System.out.println(result);
        result.stream().forEach(s -> {
            // 插入核查结果正常的数据
            String status = StrUtil.isBlank(s.getCheckResult()) ? DataConstant.TrueOrFalse.TRUE.getKey() : DataConstant.TrueOrFalse.FALSE.getKey();
            if (StrUtil.isBlank(s.getCheckResult())) {
                s.setCheckBatch((String) map.get("batch"));
                checkReportService.save(s);
                // 更新最近核查批次号
                LambdaUpdateWrapper<CheckRuleEntity> updateWrapper = new LambdaUpdateWrapper<>();
                updateWrapper.set(CheckRuleEntity::getLastCheckBatch, (String) map.get("batch"));
                updateWrapper.eq(CheckRuleEntity::getId, s.getCheckRuleId());
                checkRuleService.update(updateWrapper);
            }
            // 定时任务日志
            ScheduleLogEntity scheduleLogEntity = new ScheduleLogEntity();
            scheduleLogEntity.setExecuteJobId((String) map.get("id"));
            scheduleLogEntity.setExecuteBatch((String) map.get("batch"));
            scheduleLogEntity.setExecuteDate(s.getCheckDate());
            scheduleLogEntity.setExecuteRuleId(s.getCheckRuleId());
            scheduleLogEntity.setExecuteResult(s.getCheckResult());
            scheduleLogEntity.setStatus(status);
            scheduleLogService.save(scheduleLogEntity);
        });
    }
}
