package com.fs.app.taskService.impl;

import com.alibaba.fastjson.JSON;
import com.fs.app.taskService.QwExternalContactRatingMoreSevenDaysService;
import com.fs.common.core.domain.R;
import com.fs.common.core.redis.RedisCache;
import com.fs.course.mapper.FsCourseWatchLogMapper;
import com.fs.qw.domain.QwExternalContact;
import com.fs.qw.mapper.QwExternalContactMapper;
import com.fs.sop.domain.SopUserLogs;
import com.fs.sop.domain.SopUserLogsInfo;
import com.fs.sop.mapper.SopUserLogsInfoMapper;
import com.fs.sop.mapper.SopUserLogsMapper;
import com.fs.sop.params.QwRatingConfig;
import com.fs.sop.service.IQwSopTempDayService;
import com.fs.sop.service.ISopUserLogsInfoService;
import com.fs.sop.vo.QwRatingVO;
import com.fs.system.service.ISysConfigService;
import com.fs.voice.utils.StringUtil;
import com.google.common.util.concurrent.AtomicDouble;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.stream.Collectors;

@Service
@Slf4j
public class QwExternalContactRatingMoreSevenDaysServiceImpl implements QwExternalContactRatingMoreSevenDaysService {



    @Autowired
    private ISysConfigService configService;

    @Autowired
    private RedisCache redisCache;

    @Autowired
    private SopUserLogsMapper sopUserLogsMapper;

    @Autowired
    private IQwSopTempDayService qwSopTempDayService;

    @Autowired
    private SopUserLogsInfoMapper sopUserLogsInfoMapper;

    @Autowired
    private FsCourseWatchLogMapper fsCourseWatchLogMapper;

    @Autowired
    private QwExternalContactMapper qwExternalContactMapper;

    @Autowired
    private ISopUserLogsInfoService iSopUserLogsInfoService;

    @Autowired
    private ExecutorService sopRatingExecutor;  // 自定义线程池

    // 任务队列
    private final BlockingQueue<SopUserLogs> taskQueue = new LinkedBlockingQueue<>(10000);

    private volatile boolean running = true;
    //批量更新队列
    private final List<CompletableFuture<Void>> updateFutures = Collections.synchronizedList(new ArrayList<>());

    private final Object configLock = new Object();

    // 启动时初始化消费者线程
    @PostConstruct
    public void init() {

        loadCourseConfig();

        int consumerCount = Runtime.getRuntime().availableProcessors(); // 消费者线程数，默认 CPU 核心数
        for (int i = 0; i < consumerCount; i++) {
            sopRatingExecutor.submit(this::consumeTasks); // 提交消费者任务
        }

        log.info("初始化 {} 个消费者线程", consumerCount);
    }

    private  volatile QwRatingConfig qwRatingConfig;

    private void loadCourseConfig() {
        try {
            String json = configService.selectConfigByKey("qwRating:config");
            QwRatingConfig config = JSON.parseObject(json, QwRatingConfig.class);
            if (!StringUtil.strIsNullOrEmpty(json) && config != null) {
                qwRatingConfig = config;
                log.info("Loaded qwRating.config successfully.");
            } else {
                log.error("Failed to load course.config from configService.");
            }
        } catch (Exception e) {
            log.error("Exception while loading qwRating.config: {}", e.getMessage(), e);
        }
    }


    @Override
    public R ratingMoreSevenDaysUserLogs() {
        // 分页加载并放入队列
        int pageSize = 1000;
        int offset = 0;
        List<SopUserLogs> sopUserLogs;

        // 获取缓存的配置
        QwRatingConfig config;
        synchronized(configLock) {
            config = qwRatingConfig;
        }

        do {
            sopUserLogs = sopUserLogsMapper.meetsTheRatingByUserInfoWithPaginationStudyDays(offset, pageSize,config.getNotStudyDays());
            if (!sopUserLogs.isEmpty()) {
                sopUserLogs.forEach(item -> {
                    try {
                        taskQueue.put(item); // 将任务放入队列
                    } catch (InterruptedException e) {
                        log.error("任务放入队列失败，sopId: {}", item.getSopId(), e);
                        Thread.currentThread().interrupt();
                    }
                });
                offset += pageSize;
            }
        } while (!sopUserLogs.isEmpty());


        // 等待队列处理完成
        CompletableFuture.runAsync(() -> {
            while (!taskQueue.isEmpty()) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    log.error("等待队列处理时中断", e);
                    Thread.currentThread().interrupt();
                }
            }
        }).join(); // 等待任务完成

        return R.ok();
    }


    private void consumeTasks() {

        if (!running && taskQueue.isEmpty()) {
            log.info("没有评级任务需要处理");
            return; // 如果队列为空且没有正在运行的线程，则直接返回
        }

        while (running) {
            try {
                SopUserLogs item = taskQueue.poll(1, TimeUnit.SECONDS); // 等待 1 秒
                if (item != null) {
                    processSingleTask(item);
                }
            } catch (Exception e) {
                log.error("消费者线程异常", e);
            }
        }
    }

    private void processSingleTask(SopUserLogs item) {

        // 获取缓存的配置
        QwRatingConfig config;
        synchronized(configLock) {
            config = qwRatingConfig;
        }

        List<SopUserLogsInfo> sopUserLogsInfosList = sopUserLogsInfoMapper
                .selectSopUserLogsInfoListBySopId(item.getSopId(), item.getId());

        if (sopUserLogsInfosList == null || sopUserLogsInfosList.isEmpty()) {
            log.error("当前营期没有客户-sopId：{},营期id:{}", item.getSopId(), item.getId());
            return;
        }

        List<QwExternalContact> batchQwExternalContact = sopUserLogsInfosList.stream()
                .map(logsInfo -> processUserLog(logsInfo, config))
                .filter(Objects::nonNull)
                .collect(Collectors.toList());

        if (!batchQwExternalContact.isEmpty()) {
            batchUpdateQwExternalContact(batchQwExternalContact);
        }
    }

    private QwExternalContact processUserLog(SopUserLogsInfo logsInfo, QwRatingConfig config) {
        try {
            Long externalId = logsInfo.getExternalId();
            if (externalId == null) {
                return null;
            }

            // 查询最近 notStudyDays(3天) 的看课记录
            List<QwRatingVO> ratingVOS = fsCourseWatchLogMapper
                    .selectFsCourseWatchLogByExtIdRatingWithTransfer(externalId, config.getNotStudyDays());

            // 如果连续3天没有看课记录，标记为E级
            if (ratingVOS == null || ratingVOS.isEmpty()) {
                return buildELevelContactIfEligible(logsInfo, config,
                        "连续{}天没有看课记录，标记为E级，externalId: {}");
            }

            // 判断连续3天的总观看时长是否大于0
            boolean hasWatchDuration = getScoreMoreStudyLevel(ratingVOS);

            if (!hasWatchDuration) {
                // 有记录但观看时长为0，也认为连续3天未看课，标记为E级
                return buildELevelContactIfEligible(logsInfo, config,
                        "连续{}天看课时长为0，标记为E级，externalId: {}");
            } else {
                // 有看课记录且时长>0，不是E级，恢复为ABCD评级
                log.info("最近{}天有看课记录，不是E级，externalId: {}", config.getNotStudyDays(), externalId);
                QwExternalContact externalContact = new QwExternalContact();
                externalContact.setId(externalId);
                externalContact.setSopUserLogsInfoId(logsInfo.getId());
                // 检查当前等级是否为E级，如果是则需要重新计算
                Integer currentLevel = ratingVOS.get(0).getLevel();
                if (currentLevel != null && currentLevel == 5) {
                    // 如果当前是E级，恢复为D级（默认），后续由ABCD评级任务重新计算
                    externalContact.setLevel(4);
                    externalContact.setLevelType(1);  // 从E级(5)恢复到D级(4)是升级
                    log.info("客户从E级恢复，暂设为D级，externalId: {}", externalId);
                } else {
                    // 保持原有的ABCD评级，不修改
                    if (currentLevel != null) {
                        externalContact.setLevel(currentLevel);
                    }
                    if (ratingVOS.get(0).getLevelType() != null) {
                        externalContact.setLevelType(ratingVOS.get(0).getLevelType());
                    }
                }
                externalContact.setIsDaysNotStudy(0);  // 标记为非连续未看课
                return externalContact;
            }

        } catch (Exception e) {
            log.error("判断E级客户异常，用户：{}", logsInfo, e);
            return null;
        }
    }

    private void batchUpdateQwExternalContact(List<QwExternalContact> notInExternalUseridList) {
        int batchSize = 300;

        for (int i = 0; i < notInExternalUseridList.size(); i += batchSize) {
            int endIndex = Math.min(i + batchSize, notInExternalUseridList.size());
            List<QwExternalContact> batchList = notInExternalUseridList.subList(i, endIndex);

            int finalI = i;
            CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                try {
                    qwExternalContactMapper.batchUpdateQwExternalContactByMoreStudy(batchList);
                    iSopUserLogsInfoService.batchUpdateSopUserLogsInfoByMoreStudy(batchList);
                    log.info("成功更新看课7天数据，起始索引: {}, 数量: {}", finalI, batchList.size());
                } catch (Exception e) {
                    log.error("批量更新异常，批次起始索引: {}", finalI, e);
                }

            }, sopRatingExecutor);

            updateFutures.add(future);
        }
    }

    @PreDestroy
    public void shutdown() {
        running = false;  // 标记消费者停止
        log.info("正在关闭线程池...");

        // **等待任务队列处理完毕**
        while (!taskQueue.isEmpty()) {
            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                log.warn("等待任务队列处理完成时被中断", e);
            }
        }

        // **确保所有 `batchUpdateQwExternalContact` 的任务完成**
        log.info("等待所有批量更新任务完成...");
        CompletableFuture.allOf(updateFutures.toArray(new CompletableFuture[0])).join();

        // 关闭线程池
        sopRatingExecutor.shutdown();
        try {
            if (!sopRatingExecutor.awaitTermination(60, TimeUnit.SECONDS)) {
                List<Runnable> pendingTasks = sopRatingExecutor.shutdownNow();
                log.warn("强制关闭线程池，未完成任务数: {}", pendingTasks.size());
            }
        } catch (InterruptedException e) {
            sopRatingExecutor.shutdownNow();
            Thread.currentThread().interrupt();
        }
        log.info("线程池和消费者已完全关闭");
    }


    /**
     * 每6小时更新一次
     */
    @Scheduled(cron = "0 50 0/6 * * ?")
    public void refreshRatingConfig() {

        synchronized(configLock) {
            try {
                String json = configService.selectConfigByKey("qwRating:config");
                QwRatingConfig config = JSON.parseObject(json, QwRatingConfig.class);
                if (!StringUtil.strIsNullOrEmpty(json) && config != null) {
                    qwRatingConfig = config;
                    log.info("LoadedTime qwRating.config successfully.");
                } else {
                    log.error("Failed to load course.config from configService.");
                }
            } catch (Exception e) {
                log.error("Exception while refreshing course.config: {}", e.getMessage(), e);
            }
        }

    }


    /**
     * 转接客户(is_transfer=1)需 create_time 超过3天才参与E级评定，其他客户不受影响。
     */
    private boolean canRateTransferContactAsELevel(Long externalId) {
        QwExternalContact contact = qwExternalContactMapper.selectQwExternalContactById(externalId);
        if (contact == null) {
            return true;
        }
        if (contact.getIsTransfer() == null || contact.getIsTransfer() != 1) {
            return true;
        }
        Date createTime = contact.getCreateTime();
        if (createTime == null) {
            return true;
        }
        long daysSinceCreate = TimeUnit.MILLISECONDS.toDays(System.currentTimeMillis() - createTime.getTime());
        return daysSinceCreate > 3;
    }

    private QwExternalContact buildELevelContactIfEligible(SopUserLogsInfo logsInfo, QwRatingConfig config, String logPattern) {
        Long externalId = logsInfo.getExternalId();
        if (!canRateTransferContactAsELevel(externalId)) {
            log.info("转接客户创建未满3天，跳过E级评定，externalId: {}", externalId);
            return null;
        }
        log.info(logPattern, config.getNotStudyDays(), externalId);
        QwExternalContact externalContact = new QwExternalContact();
        externalContact.setId(externalId);
        externalContact.setSopUserLogsInfoId(logsInfo.getId());
        externalContact.setLevel(5);
        externalContact.setLevelType(2);
        externalContact.setIsDaysNotStudy(1);
        return externalContact;
    }

    //查 E级
    public boolean getScoreMoreStudyLevel(List<QwRatingVO> qwRatingVOS) {

        AtomicDouble watchCount= new AtomicDouble();

        qwRatingVOS.forEach(vo -> {
            watchCount.addAndGet(vo.getWatchDuration());
        });

        // 判断总 watchDuration 是否 > 0
        return watchCount.get() > 0;
    }

}
