package com.bcxin.ferry.service;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.io.FileUtil;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.extension.conditions.query.LambdaQueryChainWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.bcxin.ferry.common.emus.FerryTaskStatusEnum;
import com.bcxin.ferry.common.utils.DateUtil;
import com.bcxin.ferry.common.utils.IdGeneratorSnowflake;
import com.bcxin.ferry.common.utils.ObsUtil;
import com.bcxin.ferry.configs.FerryChannelConfig;
import com.bcxin.ferry.configs.FerrySwitchConfig;
import com.bcxin.ferry.configs.SchedulingConfig;
import com.bcxin.ferry.dao.mapper.FerryTaskMapper;
import com.bcxin.ferry.dtos.MyUserInfo;
import com.bcxin.ferry.dtos.baiduutil.ChangeLogsRequestDto;
import com.bcxin.ferry.dtos.baiduutil.FerryTaskPullResult;
import com.bcxin.ferry.entity.FerryTaskEntity;
import com.jcraft.jsch.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.io.*;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * 摆渡任务表(ferry_task)服务实现类
 * @author : linchunpeng
 * @date : 2024-3-6
 */
@Slf4j
@Service
public class FerryTaskService extends ServiceImpl<FerryTaskMapper, FerryTaskEntity> {

    @Autowired
    private SchedulingConfig schedulingConfig;
    @Autowired
    private RetryService retryService;
    @Autowired
    private IdGeneratorSnowflake snowflake;
    @Autowired
    private RedissonClient redissonClient;
    @Autowired
    private FerryChannelConfig ferryChannelConfig;
    @Autowired
    private FerrySwitchConfig ferrySwitchConfig;

    /**
     * description：创建摆渡任务
     * author：linchunpeng
     * date：2024/3/11
     */
    @Transactional
    public void createFerryTask(String regionCode) {
        log.info("开始创建摆渡任务，regionCode：{}", regionCode);
        Date nowTime = new Date();
        //取出最后一次摆渡任务
        LambdaQueryChainWrapper<FerryTaskEntity> lqw = this.lambdaQuery();
        lqw.eq(FerryTaskEntity::getRegionCode, regionCode);
        lqw.orderByDesc(FerryTaskEntity::getCreateTime);
        lqw.last("limit 1");
        List<FerryTaskEntity> list = lqw.list();
        FerryTaskEntity lastTask = CollectionUtil.isNotEmpty(list) ? list.get(0) : null;
        log.info(lastTask == null ? "没有最近一次任务" : "最近一次的任务数据为：" + JSONObject.toJSONString(lastTask));
        //判断上次任务已完成
        if (lastTask != null && lastTask.getTaskStatus() != FerryTaskStatusEnum.FERRY_TASK_COMPLETE.getCode()) {
            //最近一次摆渡任务未完成
            log.info("最近一次摆渡任务未完成");
            //2个小时毫秒数
            long intervalTime = 1000 * 60 * 60 * 2;
            long differenceTime = nowTime.getTime() - lastTask.getCreateTime().getTime();
            if (differenceTime >= intervalTime) {
                //最近一次任务的创建时间>2小时
                log.info("最近一次任务的创建时间>=2小时，进行补偿，复制一个任务出来，重新摆渡");

                String requestId = "";
                boolean isNeedRetryPull = false;
                if (lastTask.getTaskStatus() == FerryTaskStatusEnum.BEGIN_PULL.getCode()) {
                    log.info("最近一次任务的状态是：{}，所以需要重新拉取", FerryTaskStatusEnum.BEGIN_PULL.getDefaultMessage());
                    isNeedRetryPull = true;
                } else {
                    log.info("最近一次任务的状态是：{}，不用需要重新拉取，直接复制摆渡包即可", FerryTaskStatusEnum.BEGIN_PULL.getDefaultMessage());
                    requestId = "" + snowflake.snowflakeId();
                    String oldZip = lastTask.getPackageUrl().concat(File.separator).concat(lastTask.getRequestId()).concat(".zip");
                    String newZip = lastTask.getPackageUrl().concat(File.separator).concat(requestId).concat(".zip");
                    FileUtil.copy(oldZip, newZip, true);
                    log.info("文件复制完成，oldZip：{}, newZip：{}", oldZip, newZip);
                    if (lastTask.getRequestId().contains("zw")) {
                        requestId = requestId.concat("zw");
                    }
                }

                lastTask.setTaskStatus(FerryTaskStatusEnum.FERRY_TASK_COMPLETE.getCode());
                lastTask.setFerryResult("创建时间>=2小时，进行补偿，复制一个任务出来，重新摆渡");
                lastTask.setUpdateTime(new Date());
                this.updateById(lastTask);

                FerryTaskEntity ferryTaskEntity = new FerryTaskEntity();
                ferryTaskEntity.setId(snowflake.snowflakeId());
                ferryTaskEntity.setRegionCode(regionCode);
                ferryTaskEntity.setStartTime(lastTask.getStartTime());
                ferryTaskEntity.setEndTime(lastTask.getEndTime());
                ferryTaskEntity.setTaskStatus(FerryTaskStatusEnum.BEGIN_PULL.getCode());
                ferryTaskEntity.setCreateTime(nowTime);
                ferryTaskEntity.setUpdateTime(nowTime);

                if (!isNeedRetryPull) {
                    //不需要重新拉取，直接改一下状态和requestId
                    ferryTaskEntity.setRequestId(requestId);
                    ferryTaskEntity.setPackageUrl(lastTask.getPackageUrl());
                    ferryTaskEntity.setPullResult("retry_ferry");
                    ferryTaskEntity.setTaskStatus(FerryTaskStatusEnum.PULL_COMPLETE.getCode());
                }

                this.save(ferryTaskEntity);

                if (isNeedRetryPull) {
                    //调用摆渡工具服务，拉取数据
                    updateEndTimeFromPullResult(ferryTaskEntity);
                }
            } else {
                log.info("最近一次任务的创建时间与当前时间差：{} 毫秒，<2小时，无需补偿", differenceTime);
            }
        } else {
            //最近一次摆渡任务已完成
            log.info("最近一次摆渡任务已完成 || 没有最近一次任务");
            Date pullDataBeginTime = DateUtil.getMinuteZeroTime(nowTime);
            //配置拉取间隔
            int after = schedulingConfig.getCreateFerryTask().getTimeInterval();
            //1个小时毫秒数
            long intervalTime = 1000 * 60 * 60;
            if (lastTask != null) {
                pullDataBeginTime = lastTask.getEndTime();
                if (nowTime.getTime() - pullDataBeginTime.getTime() > intervalTime) {
                    //最近一次任务的结束时间>1小时
                    log.info("最近一次任务的结束时间：{}，当前时间：{}，间隔>1小时，需要翻倍拉取", lastTask.getEndTime(), nowTime);
                    log.info("原本拉取：{}分钟", after);
                    after = after * 2;
                    log.info("现在拉取：{}分钟", after);
                }
            }

            Date pullDataEndTime = DateUtil.getAfterNumMinuteTime(pullDataBeginTime, after);
            log.info("新的任务拉取数据开始时间：{}，拉取数据开始时间：{}", DateUtil.formatDateTime(pullDataBeginTime), DateUtil.formatDateTime(pullDataEndTime));
            if (pullDataEndTime.getTime() >= nowTime.getTime()) {
                //新任务的结束时间 >= 当前时间，无法创建任务，等待下次定时任务启动
                log.info("新任务的结束时间 >= 当前时间，无法创建任务，等待下次定时任务启动");
            } else {
                //新任务的结束时间 < 当前时间
                log.info("新任务的结束时间 < 当前时间，可以创建新的任务");
                FerryTaskEntity ferryTaskEntity = new FerryTaskEntity();
                ferryTaskEntity.setId(snowflake.snowflakeId());
                ferryTaskEntity.setRegionCode(regionCode);
//                向前减一分钟，避免数据丢失
                ferryTaskEntity.setStartTime(DateUtil.getBeforeNumMinuteTime(pullDataBeginTime, 1));
                ferryTaskEntity.setEndTime(pullDataEndTime);
                ferryTaskEntity.setTaskStatus(FerryTaskStatusEnum.BEGIN_PULL.getCode());
                ferryTaskEntity.setCreateTime(nowTime);
                ferryTaskEntity.setUpdateTime(nowTime);
                this.save(ferryTaskEntity);
                //调用摆渡工具服务，拉取数据
                updateEndTimeFromPullResult(ferryTaskEntity);
            }
        }
    }

    /**
     * description：查询摆渡任务状态-拉取完成
     * author：linchunpeng
     * date：2024/3/11
     */
    @Transactional
    public void queryPullIsComplete() {
        //查询最近一条，拉取数据成功的摆渡任务
        LambdaQueryChainWrapper<FerryTaskEntity> lqw = this.lambdaQuery();
        lqw.eq(FerryTaskEntity::getTaskStatus, FerryTaskStatusEnum.BEGIN_PULL.getCode());
        lqw.isNull(FerryTaskEntity::getRequestId);
        lqw.isNull(FerryTaskEntity::getPackageUrl);
        lqw.orderByDesc(FerryTaskEntity::getCreateTime);
        List<FerryTaskEntity> list = lqw.list();
        if (CollectionUtil.isNotEmpty(list)) {
            for (FerryTaskEntity ferryTaskEntity : list) {
                //开始拉取完成的任务不为空，且创建时间小于5个小时，可以开始查询拉取状态
                //调用摆渡工具服务，拉取数据
                try {
                    FerryTaskPullResult result = retryService.postToBaiduutilServerPullResult(ferryTaskEntity);
                    //结果
                    if (result.getTaskStatus() != null) {
                        FerryTaskEntity ferryTask = this.getById(result.getId());
                        //如果是拉取政务网的，需要在requestId前面添加：zw
                        if ("govout-to-govin".equals(ferryTask.getRegionCode())
                                || "govin-to-govout".equals(ferryTask.getRegionCode())) {
                            ferryTask.setRequestId(result.getRequestId().concat("zw"));
                        } else {
                            ferryTask.setRequestId(result.getRequestId());
                        }
                        ferryTask.setPackageUrl(result.getPackageUrl());
                        ferryTask.setPullResult(result.getPullResult());
                        ferryTask.setTaskStatus(result.getTaskStatus());
                        ferryTask.setUpdateTime(new Date());
                        this.updateById(ferryTask);
                    }
                } catch (Exception e) {
                    log.error("调用摆渡工具服务查询拉取状态出错，regionCode：{}, requestId：{}，异常：{}", ferryTaskEntity.getRegionCode(), ferryTaskEntity.getRequestId(), e.getMessage(), e);
                }
            }
        }
    }

    /**
     * description：查询摆渡任务状态-拉取完成
     * author：linchunpeng
     * date：2024/3/11
     */
    @Transactional
    public void queryPullComplete() {
        Date nowTime = new Date();
        log.info("当前时间戳：{}", nowTime.getTime());
        //查询最近一条，拉取数据成功的摆渡任务
        LambdaQueryChainWrapper<FerryTaskEntity> lqw = this.lambdaQuery();
        lqw.eq(FerryTaskEntity::getTaskStatus, FerryTaskStatusEnum.PULL_COMPLETE.getCode());
        lqw.isNotNull(FerryTaskEntity::getRequestId);
        lqw.isNotNull(FerryTaskEntity::getPackageUrl);
        lqw.orderByDesc(FerryTaskEntity::getCreateTime);
        List<FerryTaskEntity> list = lqw.list();
        if (CollectionUtil.isNotEmpty(list)) {
            for (FerryTaskEntity ferryTaskEntity : list) {
                //分布式锁key
                String lockKey = "REDISSON_LOCK_QUERY_FERRY_TASK_STATUS_" + ferryTaskEntity.getId().toString();
                //取锁
                log.info("查询摆渡任务状态-拉取完成， lockKey：{}，取锁中.....", lockKey);
                RLock lock = redissonClient.getLock(lockKey);
                //加锁，并设置过期时间 300s
                lock.lock(900, TimeUnit.SECONDS);
                log.info("取到锁");
                try {
                    boolean isDownloadComplete = false;
                    //拉取完成的任务不为空，且拉取完成时间小于2个小时，可以开始下载文件
                    String ferryPackageUrl = ferryTaskEntity.getPackageUrl().concat(File.separator).concat(ferryTaskEntity.getRequestId());
                    String zipPath = ferryPackageUrl.concat(".zip");
                    if (!ferryTaskEntity.getRegionCode().contains("gov") && !"retry_ferry".equals(ferryTaskEntity.getPullResult())) {
                        //不是内网的拉取，也不是政务的拉取
                        log.info("互联网 && 不是重试摆渡，需要下载zip包");
                        boolean downloadFileResult = ObsUtil.downloadFile(zipPath, zipPath.substring(zipPath.indexOf("baidu/")));
                        if (downloadFileResult) {
                            isDownloadComplete = true;
                        } else {
                            log.error("下载摆渡包失败");
                        }
                    } else {
                        log.info("不是互联网 || 是重试摆渡，不需要需要下载zip包");
                        isDownloadComplete = true;
                        //如果是摆渡到政务网的，或者政务网拉取的，摆渡包文件需要增加zw
                        if (zipPath.contains("zw")) {
                            log.info("如果是摆渡到政务网的，或者政务网拉取的，摆渡包文件需要增加zw");
                            File file = new File(zipPath.replace("zw", ""));
                            FileUtil.copy(file, new File(zipPath), true);
                        }
                    }
                    //可以开始复制文件到目标服务器上
                    if (isDownloadComplete) {
                        log.info("开始复制文件到目标服务器上");
                        //目标服务器的地址，sendPath + 日期 + requestId.zip，如果是有摆渡到政务网的，需要在requestId前面添加：zw
                        String targetFilePath = ferryChannelConfig.getSendPath().concat(File.separator)
                                .concat(DateUtil.formatDate(nowTime)).concat(File.separator)
                                .concat(ferryTaskEntity.getRequestId()).concat(".zip");
                        //开始复制文件到目标服务器上
                        this.copyFileToFerryChannel(zipPath, targetFilePath);
                        if (ferrySwitchConfig.getWaitLastTask() != null && ferrySwitchConfig.getWaitLastTask()) {
                            log.info("需要等待上个摆渡任务执行完成再继续摆渡下一个任务，任务状态改为：{}", FerryTaskStatusEnum.PUT_SEND_COMPLETE.getDefaultMessage());
                            ferryTaskEntity.setTaskStatus(FerryTaskStatusEnum.PUT_SEND_COMPLETE.getCode());
                        } else {
                            log.info("不需要等待上个摆渡任务执行完成再继续摆渡下一个任务，任务状态改为：{}", FerryTaskStatusEnum.FERRY_TASK_COMPLETE.getDefaultMessage());
                            ferryTaskEntity.setTaskStatus(FerryTaskStatusEnum.FERRY_TASK_COMPLETE.getCode());
                        }
                        ferryTaskEntity.setUpdateTime(nowTime);
                        this.updateById(ferryTaskEntity);
                        log.info("复制文件结束");
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    log.error("查询摆渡任务状态-拉取完成任务异常，{}", e.getMessage(), e);
                } finally {
                    if (lock.isLocked()) {
                        lock.unlock();
                    }
                    log.info("查询摆渡任务状态-拉取完成任务，lockKey：{}，解锁", lockKey);
                }
            }
        } else {
            log.info("没有拉取完成的任务");
        }
    }

    /**
     * description：查询摆渡任务状态-放入发送文件夹完成
     * author：linchunpeng
     * date：2024/3/11
     */
    @Transactional
    public void queryPutSendComplete() {
        Date nowTime = new Date();
        log.info("当前时间戳：{}", nowTime.getTime());
        //查询最近，放入发送文件夹完成的摆渡任务
        LambdaQueryChainWrapper<FerryTaskEntity> lqw = this.lambdaQuery();
        lqw.eq(FerryTaskEntity::getTaskStatus, FerryTaskStatusEnum.PUT_SEND_COMPLETE.getCode());
        lqw.orderByDesc(FerryTaskEntity::getCreateTime);
        List<FerryTaskEntity> list = lqw.list();
        if (CollectionUtil.isNotEmpty(list)) {
            for (FerryTaskEntity ferryTaskEntity : list) {
                //分布式锁key
                String lockKey = "REDISSON_LOCK_QUERY_FERRY_TASK_STATUS_" + ferryTaskEntity.getId().toString();
                //取锁
                log.info("查询摆渡任务状态-放入发送文件夹完成任务， lockKey：{}，取锁中.....", lockKey);
                RLock lock = redissonClient.getLock(lockKey);
                //加锁，并设置过期时间 300s
                lock.lock(900, TimeUnit.SECONDS);
                log.info("取到锁");
                try {
                    log.info("任务tasId：{}", ferryTaskEntity.getId());
                    FerryTaskEntity task = this.getById(ferryTaskEntity.getId());
                    if (task.getTaskStatus() == FerryTaskStatusEnum.PUT_SEND_COMPLETE.getCode()
                            && nowTime.getTime() - task.getUpdateTime().getTime() > 1000 * 60 * 60) {
                        //放入发送文件夹完成 && 完成时间超过1小时，重新传输文件
                        log.info("开始重新传输文件");
                        String zipPath = task.getPackageUrl().concat(File.separator).concat(task.getRequestId()).concat(".zip");
                        //目标服务器的地址，sendPath + 日期 + requestId.zip，如果是有摆渡到政务网的，需要在requestId前面添加：zw
                        String targetFilePath = ferryChannelConfig.getSendPath().concat(File.separator)
                                .concat(DateUtil.formatDate(task.getUpdateTime())).concat(File.separator)
                                .concat(task.getRequestId()).concat(".zip");
                        //开始复制文件到目标服务器上
                        this.copyFileToFerryChannel(zipPath, targetFilePath);
                        log.info("重新传输文件结束");
                    } else {
                        log.info("任务tasId：{}，已经完成了，不需要重新传输文件", ferryTaskEntity.getId());
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    log.error("查询摆渡任务状态-放入发送文件夹完成任务异常，{}", e.getMessage(), e);
                } finally {
                    if (lock.isLocked()) {
                        lock.unlock();
                    }
                    log.info("查询摆渡任务状态-放入发送文件夹完成任务，lockKey：{}，解锁", lockKey);
                }
            }
        } else {
            log.info("没有放入发送文件夹完成的任务");
        }
    }

    /**
     * description：复制文件到目标服务器上
     * author：linchunpeng
     * date：2024/9/3
     */
    public void copyFileToFerryChannel(String sourceFilePath, String targetFilePath) {
        //scp -P 2222 ~/Documents/test.txt username@192.168.0.1:/home/username
        if (StringUtils.isNotBlank(ferryChannelConfig.getTargetType()) && "local".equals(ferryChannelConfig.getTargetType())) {
            log.info("通道在本机");
            FileUtil.copy(sourceFilePath, targetFilePath, true);
            log.info("复制文件到通道文件夹完成");
        } else {
            log.info("通道在其他服务器");
            StringBuilder result = new StringBuilder("");
            Process process = null;
            InputStream ins = null;
            String command = String.format("scp -P %s %s %s@%s:%s",
                    ferryChannelConfig.getTargetPort(), sourceFilePath, ferryChannelConfig.getTargetUsername(),
                    ferryChannelConfig.getTargetIp(), targetFilePath);
            log.info("通过命令行执行shell脚本，命令：{}", command);
            try {
                process = Runtime.getRuntime().exec(new String[]{"/bin/sh", "-c",command});
                process.waitFor();
                ins = process.getErrorStream();
                byte[] b = new byte[100];
                int num = 0;
                while ((num = ins.read(b)) != -1) {
                    result.append(new String(b, "gb2312"));
                }
                log.info("通过命令行执行shell脚本，命令：{}，执行结果：{}", command, result);
            } catch (Exception e) {
                log.error("命令行执行脚本失败，command：{}，{}",  command, e.getMessage(), e);
                e.printStackTrace();
            } finally {
                try {
                    assert ins != null;
                    ins.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
                process.destroy();
            }
        }
    }


    /**
     * description：修改摆渡任务完成
     * author：linchunpeng
     * date：2024/3/13
     */
    @Transactional
    public void updateFerryTaskComplete(String requestId) {
        log.info("修改摆渡任务完成，requestId：{}", requestId);
        LambdaQueryChainWrapper<FerryTaskEntity> lqw = this.lambdaQuery();
        lqw.eq(FerryTaskEntity::getRequestId, requestId);
        lqw.last("limit 1");
        List<FerryTaskEntity> list = lqw.list();
        FerryTaskEntity ferryTaskEntity = CollectionUtil.isNotEmpty(list) ? list.get(0) : null;
        if (ferryTaskEntity != null) {
            ferryTaskEntity.setTaskStatus(FerryTaskStatusEnum.FERRY_TASK_COMPLETE.getCode());
            ferryTaskEntity.setFerryResult("摆渡完成");
            ferryTaskEntity.setUpdateTime(new Date());
            this.updateById(ferryTaskEntity);
        }
    }

    /**
     * description：根据摆渡任务，更新结束时间
     * author：hongyiwei
     * date：2025/09/25
     */
    private void updateEndTimeFromPullResult(FerryTaskEntity ferryTaskEntity) {
        ChangeLogsRequestDto changeLogsRequestDto = retryService.postToBaiduutilServerPull(ferryTaskEntity);
        if (changeLogsRequestDto != null) {
            ferryTaskEntity.setEndTime(changeLogsRequestDto.getEndTime());
            ferryTaskEntity.setPullResult("根据摆渡任务，更新服务器时间为结束时间");
            this.updateById(ferryTaskEntity);
            log.info("根据摆渡任务，更新结束时间，ferryTaskEntity：{}", ferryTaskEntity);
        }
    }


}