package cn.skcks.docking.gb28181.wvp.service.wvp;

import cn.hutool.cache.CacheUtil;
import cn.hutool.cache.impl.TimedCache;
import cn.hutool.core.date.DateUnit;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.IoUtil;
import cn.hutool.crypto.digest.MD5;
import cn.skcks.docking.gb28181.common.json.JsonException;
import cn.skcks.docking.gb28181.common.json.JsonResponse;
import cn.skcks.docking.gb28181.wvp.config.WvpProxyConfig;
import cn.skcks.docking.gb28181.wvp.dto.common.GeneralTimeReq;
import cn.skcks.docking.gb28181.wvp.dto.device.DeviceChannel;
import cn.skcks.docking.gb28181.wvp.dto.device.GetDeviceChannelsReq;
import cn.skcks.docking.gb28181.wvp.dto.device.GetDeviceChannelsResp;
import cn.skcks.docking.gb28181.wvp.dto.download.DownloadStartReq;
import cn.skcks.docking.gb28181.wvp.dto.login.WvpLoginReq;
import cn.skcks.docking.gb28181.wvp.dto.login.WvpLoginResp;
import cn.skcks.docking.gb28181.wvp.dto.media.proxy.AddDownloadTaskReq;
import cn.skcks.docking.gb28181.wvp.dto.media.proxy.GetDownloadTaskReq;
import cn.skcks.docking.gb28181.wvp.dto.media.proxy.GetDownloadTaskResp;
import cn.skcks.docking.gb28181.wvp.dto.record.QueryRecordReq;
import cn.skcks.docking.gb28181.wvp.dto.record.QueryRecordResp;
import cn.skcks.docking.gb28181.wvp.dto.stream.StreamContent;
import cn.skcks.docking.gb28181.wvp.orm.mybatis.dynamic.model.WvpProxyDevice;
import cn.skcks.docking.gb28181.wvp.orm.mybatis.dynamic.model.WvpProxyDocking;
import cn.skcks.docking.gb28181.wvp.proxy.WvpProxyClient;
import cn.skcks.docking.gb28181.wvp.service.device.DeviceService;
import cn.skcks.docking.gb28181.wvp.service.docking.DockingService;
import cn.skcks.docking.gb28181.wvp.service.download.DownloadService;
import cn.skcks.docking.gb28181.wvp.service.report.ReportService;
import cn.skcks.docking.gb28181.wvp.service.video.VideoService;
import cn.skcks.docking.gb28181.wvp.utils.RetryUtil;
import com.github.rholder.retry.*;
import jakarta.servlet.AsyncContext;
import jakarta.servlet.http.HttpServletRequest;
import jakarta.servlet.http.HttpServletResponse;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.web.context.request.async.DeferredResult;

import java.nio.charset.StandardCharsets;
import java.text.MessageFormat;
import java.util.Date;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
@Service
@RequiredArgsConstructor
public class WvpService {
    private final WvpProxyClient wvpProxyClient;
    private final WvpProxyConfig wvpProxyConfig;
    private final DeviceService deviceService;
    private final DownloadService downloadService;
    private final VideoService videoService;
    private final DockingService dockingService;
    private final TimedCache<String, String> cache = CacheUtil.newTimedCache(TimeUnit.HOURS.toMillis(1));
    private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
    private final ConcurrentMap<String, ScheduledFuture<?>> playing = new ConcurrentHashMap<>();

    private final ReportService reportService;

    public void header(HttpServletResponse response) {
        response.setContentType("video/mp4");
        response.setHeader("Accept-Ranges", "none");
        response.setHeader("Connection", "close");
        // response.setHeader("Content-Disposition","attachment; filename=\"record.mp4\"");
    }

    @SneakyThrows
    private void writeErrorToResponse(HttpServletResponse response, JsonResponse<?> json) {
        response.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
        response.setCharacterEncoding(StandardCharsets.UTF_8.name());
        response.setContentType(MediaType.APPLICATION_JSON_VALUE);
        IoUtil.writeUtf8(response.getOutputStream(), false, json);
    }

    @SneakyThrows
    public void video(HttpServletRequest request, HttpServletResponse response, String deviceCode, Date startTime, Date endTime) {
        Optional<WvpProxyDevice> wvpProxyDeviceList = deviceService.getDeviceByDeviceCode(deviceCode);
        if (wvpProxyDeviceList.isEmpty()) {
            writeErrorToResponse(response, JsonResponse.error("设备不存在"));
            return;
        }
        WvpProxyDevice wvpProxyDevice = wvpProxyDeviceList.get();
        String deviceId = wvpProxyDevice.getGbDeviceId();
        String channelId = wvpProxyDevice.getGbDeviceChannelId();
        log.info("设备编码 (deviceCode=>{}) 查询到的设备信息 国标id(gbDeviceId => {}), 通道(channelId => {})", deviceCode, deviceId, channelId);
        request.startAsync();
        AsyncContext asyncContext = request.getAsyncContext();
        asyncContext.setTimeout(0);
        asyncContext.start(() -> {
            HttpServletResponse asyncResponse = (HttpServletResponse) asyncContext.getResponse();
            try {
                Retryer<JsonResponse<?>> genericRetryer = RetryUtil.getDefaultGenericRetryer("调用 wvp api 查询设备历史");
                genericRetryer.call(() -> video(asyncResponse, deviceCode, deviceId, channelId, startTime, endTime));
            } catch (RetryException e) {
                Attempt<?> failedAttempt = e.getLastFailedAttempt();
                String reason;
                if (failedAttempt.hasException()) {
                    reason = MessageFormat.format("调用 wvp api 查询设备: {0} 历史失败, 已重试 {1} 次, 异常: {2}", deviceCode, e.getNumberOfFailedAttempts(), failedAttempt.getExceptionCause().getMessage());
                } else {
                    reason = MessageFormat.format("调用 wvp api 查询设备: {0} 历史失败, 已重试 {1} 次", deviceCode, e.getNumberOfFailedAttempts());
                }
                log.error(reason);
                writeErrorToResponse(asyncResponse, JsonResponse.error(reason));
            } catch (Exception e) {
                String reason = MessageFormat.format("调用 wvp api 查询设备({0})历史失败, 异常: {1}", deviceCode, e.getMessage());
                writeErrorToResponse(asyncResponse, JsonResponse.error(reason));
            } finally {
                reportService.report(request,wvpProxyDevice, startTime, endTime,-1);
                log.info("asyncContext 结束");
                asyncContext.complete();
            }
        });
    }

    /**
     * 执行 视频查询
     * <p>返回一个 JsonResponse 类型的执行结果 <p>
     * <p>如果返回值为 null 或 code 不为 0 或 200 则视为执行失败</p>
     * <p>如果排除异常 也视为执行失败</p>
     *
     * @param response      异步响应
     * @param deviceCode    设备编码 21位
     * @param deviceId      国标设备编码 20位
     * @param channelId     通道id
     * @param startDateTime 开始时间
     * @param endDateTime   结束时间
     * @return JsonResponse 类型的执行结果 如果 为 null 或 code 不为 0 或 200 则视为执行失败
     */
    @SneakyThrows
    public JsonResponse<?> video(HttpServletResponse response, String deviceCode, String deviceId, String channelId, Date startDateTime, Date endDateTime) {
        String token = login();

        log.debug("通过 wvp 查询设备 国标id(gbDeviceId => {}) 通道信息", deviceId);
        JsonResponse<GetDeviceChannelsResp> deviceChannels = wvpProxyClient.getDeviceChannels(
                token,
                deviceId,
                GetDeviceChannelsReq.builder().query(channelId).build());
        if (deviceChannels.getData() == null || deviceChannels.getData().getTotal() == 0) {
            writeErrorToResponse(response, JsonResponse.error(MessageFormat.format("未能获取 设备: {0}, 国标id: {1}, 的通道信息", deviceCode, deviceId)));
            return JsonResponse.success(null);
        }
        List<DeviceChannel> list = deviceChannels.getData().getList();
        log.info("通过 wvp 获取到 查询设备 国标id(gbDeviceId => {}), 通道数量 => {}", deviceId, list.size());
        DeviceChannel deviceChannel = list.parallelStream().filter(item -> item.getChannelId().equalsIgnoreCase(channelId)).findFirst().orElse(null);
        if (deviceChannel == null) {
            writeErrorToResponse(response, JsonResponse.error(MessageFormat.format("未查询到 设备: {0}, 国标id: {1}, 通道: {2} 信息", deviceCode, deviceId, channelId)));
            return JsonResponse.success(null);
        }

        String startTime = DateUtil.formatDateTime(startDateTime);
        String endTime = DateUtil.formatDateTime(endDateTime);
        Retryer<JsonResponse<List<QueryRecordResp.RecordListDTO>>> queryRecordRetryer = RetryUtil.getDefaultRetryer("调用 wvp 设备历史查询 api");
        JsonResponse<List<QueryRecordResp.RecordListDTO>> recordListResponse = queryRecordRetryer.call(() -> {
            JsonResponse<QueryRecordResp> queryRecord = wvpProxyClient.queryRecord(token, deviceId, channelId, new QueryRecordReq(startTime, endTime));
            QueryRecordResp queryRecordData = queryRecord.getData();
            if (queryRecordData == null) {
                String reason = MessageFormat.format("通过 wvp 查询历史录像 失败 设备: {0}, 国标id: {1}, 通道: {2}, 错误信息: {3}", deviceCode, deviceId, channelId, queryRecord.getMsg());
                log.error(reason);
                log.error("查询历史录像 返回结果 => {}", queryRecord);
                throw new JsonException(reason);
            }
            List<QueryRecordResp.RecordListDTO> recordList = queryRecordData.getRecordList();
            if (CollectionUtils.isEmpty(recordList)) {
                String reason = MessageFormat.format("通过 wvp 查询历史录像 失败 设备: {0}, 国标id: {1}, 通道: {2}, 查询时间范围 开始时间: {3}, 结束时间: {4}, 录像数量为 0", deviceCode, deviceId, channelId, startTime, endTime);
                throw new JsonException(reason);
            }
            log.info("通过 wvp 查询到 {} 条历史录像 设备: {}, 国标id: {}, 通道: {}, 开始时间: {}, 结束时间: {}", recordList.size(), deviceCode, deviceId, channelId, startTime, endTime);
            return JsonResponse.success(recordList);
        });

        List<QueryRecordResp.RecordListDTO> recordList = recordListResponse.getData();
        recordList.forEach(record -> {
            log.debug("{}", record);
        });

        Boolean useWvpAssist = wvpProxyConfig.getUseWvpAssist();
        log.info("准备下载 deviceCode: {}, deviceId: {}, channelId:{}, ({}~{}) 历史视频, 通过 wvp-assist: {}", deviceCode, deviceId, channelId, startTime, endTime, useWvpAssist);
        if (useWvpAssist) {
            try {
                downloadFromWvpAssist(response, token, deviceCode, deviceId, channelId, startTime, endTime);
            } catch (Exception e) {
                log.warn("尝试通过 wvp-assist 下载视频失败, 尝试通过 视频回放 拉取视频");
                downloadFromPlayback(response, token, deviceId, channelId, startTime, endTime);
            }
        } else {
            downloadFromPlayback(response, token, deviceId, channelId, startTime, endTime);
        }

        return JsonResponse.success(null);
    }

    private void downloadFromWvpAssist(HttpServletResponse response, String token, String deviceCode, String deviceId, String channelId, String startTime, String endTime) {
        JsonResponse<String> videoPathResponse = downloadFromWvpAssist(deviceCode, deviceId, channelId, startTime, endTime, token);
        String videoUrl = videoPathResponse.getData();
        log.info("设备(deviceCode {}) (deviceId {}, channel{}) ({} ~ {}) 视频下载地址 {}", deviceCode, deviceId, channelId, startTime, endTime, videoUrl);
        downloadService.download(response, videoUrl);
    }

    @SneakyThrows
    private void downloadFromPlayback(HttpServletResponse response, String token, String deviceId, String channelId, String startTime, String endTime) {
        Retryer<JsonResponse<StreamContent>> playBackRetryer = RetryUtil
                .<StreamContent>getDefaultRetryerBuilder("通过回放获取实时视频流下载", 100, TimeUnit.MILLISECONDS, 5)
                .build();
        JsonResponse<StreamContent> videoStreamResponse = playBackRetryer.call(() -> {
            JsonResponse<StreamContent> streamContentJsonResponse = wvpProxyClient.playbackStart(token, deviceId, channelId, new GeneralTimeReq(startTime, endTime));
            log.info("实时视频流下载 响应:{} ", streamContentJsonResponse);
            return streamContentJsonResponse;
        });
        StreamContent streamContent = videoStreamResponse.getData();
        String stream = streamContent.getStream();
        String streamUrl = streamContent.getFlv();
        try {
            header(response);
            videoService.record(response, streamUrl, DateUtil.between(DateUtil.parseDateTime(startTime), DateUtil.parseDateTime(endTime), DateUnit.SECOND));
        } finally {
            wvpProxyClient.playbackStop(token, deviceId, channelId, stream);
        }
    }

    @SuppressWarnings("UnstableApiUsage")
    @SneakyThrows
    private JsonResponse<String> downloadFromWvpAssist(String deviceCode, String deviceId, String channelId, String startTime, String endTime, String token) {
        Retryer<JsonResponse<String>> downloadRetryer = RetryUtil.getDefaultRetryer("调用 wvp 设备历史视频下载 api");
        return downloadRetryer.call(() -> {
            JsonResponse<StreamContent> downloadStart = wvpProxyClient.downloadStart(token, deviceId, channelId, DownloadStartReq.builder()
                    .startTime(startTime)
                    .endTime(endTime)
                    .build());
            if (downloadStart.getData() == null) {
                throw new JsonException(downloadStart.getMsg());
            }
            StreamContent downloadStartData = downloadStart.getData();
            String mediaServerId = downloadStartData.getMediaServerId();
            String app = downloadStartData.getApp();
            String stream = downloadStartData.getStream();
            log.info("开始下载 mediaServerId: {}, app: {}, stream: {}", mediaServerId, app, stream);

            AtomicLong keepTime = new AtomicLong();
            Double downloadProgress = 0D;
            Retryer<JsonResponse<StreamContent>> downloadProgressRetryer = RetryUtil.<StreamContent>getDefaultRetryerBuilder(MessageFormat.format("查询设备(deviceCode {0}) (deviceId {1}, channel{2}) ({3} ~ {4}) 历史视频下载进度", deviceCode, deviceId, channelId, startTime, endTime))
                    // 重试间隔
                    .withWaitStrategy(WaitStrategies.fixedWait(1, TimeUnit.SECONDS))
                    .retryIfResult(result -> {
                        keepTime.getAndIncrement();
                        if (keepTime.get() > 60) {
                            return false;
                        }
                        if (result == null || result.getData() == null) {
                            return true;
                        }
                        if (!result.getData().getProgress().equals(downloadProgress)) {
                            keepTime.set(0);
                            return true;
                        }
                        return result.getData().getProgress().equals(downloadProgress) && keepTime.get() <= 60;
                    })
                    .withRetryListener(new RetryListener() {

                        @Override
                        public <V> void onRetry(Attempt<V> attempt) {
                            if (attempt.hasResult()) {
                                log.debug("第 {} 次 获取 设备(deviceCode {}) (deviceId {}, channel{}) ({} ~ {}) 历史视频下载进度 mediaServerId: {}, app: {}, stream: {}", attempt.getAttemptNumber(), deviceCode, deviceId, channelId, startTime, endTime, mediaServerId, app, stream);
                            } else {
                                log.debug("第 {} 次 获取 设备(deviceCode {}) (deviceId {}, channel{}) ({} ~ {}) 历史视频下载进度失败 mediaServerId: {}, app: {}, stream: {}", attempt.getAttemptNumber(), deviceCode, deviceId, channelId, startTime, endTime, mediaServerId, app, stream);
                            }
                        }
                    })
                    .withStopStrategy(StopStrategies.neverStop())
                    .build();
            downloadProgressRetryer.call(() -> {
                JsonResponse<StreamContent> downloadProgressResp = wvpProxyClient.downloadProgress(token, deviceId, channelId, stream);
                StreamContent data = downloadProgressResp.getData();
                log.debug("设备(deviceCode {}) (deviceId {}, channel{}) ({} ~ {}) 历史视频下载进度 {}% mediaServerId: {}, app: {}, stream: {}", deviceCode, deviceId, channelId, startTime, endTime, data.getProgress(), mediaServerId, app, stream);
                return downloadProgressResp;
            });
            wvpProxyClient.downloadStop(token, deviceId, channelId, stream);

            JsonResponse<String> addDownloadTask2MediaServer = wvpProxyClient.addDownloadTask2MediaServer(token, mediaServerId, new AddDownloadTaskReq(app, stream));
            String taskId = addDownloadTask2MediaServer.getData();
            Retryer<JsonResponse<String>> mediaServerRetryer = RetryUtil.<String>getDefaultRetryerBuilder("从 mediaServer 获取视频")
                    // 重试间隔
                    .withWaitStrategy(WaitStrategies.fixedWait(RetryUtil.DEFAULT_RETRY_INTERVAL, TimeUnit.SECONDS))
                    // 重试次数
                    .withStopStrategy(StopStrategies.stopAfterAttempt(RetryUtil.DEFAULT_RETRY_TIME))
                    .retryIfResult(RetryUtil.defaultRetryIf())
                    .withRetryListener(RetryUtil.defaultRetryListener("从 mediaServer 获取视频"))
                    .withStopStrategy(StopStrategies.stopAfterAttempt(10))
                    .build();
            return mediaServerRetryer.call(() -> {
                JsonResponse<List<GetDownloadTaskResp>> downloadTask4MediaServer = wvpProxyClient.getDownloadTask4MediaServer(token, mediaServerId, new GetDownloadTaskReq(app, stream, taskId, true));
                if (downloadTask4MediaServer.getData().size() > 0) {
                    return JsonResponse.success(downloadTask4MediaServer.getData().get(0).getPlayFile());
                } else {
                    return JsonResponse.error(null);
                }
            });
        });
    }

    @SneakyThrows
    public Optional<DeviceChannel> getDeviceChannelByDeviceCode(String token, String deviceCode) {
        WvpProxyDevice device = deviceService.getDeviceByDeviceCode(deviceCode).orElse(null);
        if (device == null) {
            throw new JsonException("设备不存在");
        }
        String channelId = device.getGbDeviceChannelId();

        WvpProxyDocking docking = dockingService.getDeviceByDeviceCode(device.getGbDeviceId()).orElse(null);
        if (docking == null) {
            throw new JsonException("设备(通道)不存在");
        }

        Optional<DeviceChannel> deviceChannel = Optional.empty();
        for (String deviceId : wvpProxyConfig.getParents()) {
            log.debug("通过 wvp 查询设备 国标id(gbDeviceId => {}) 通道信息", deviceId);
            JsonResponse<GetDeviceChannelsResp> deviceChannels = wvpProxyClient.getDeviceChannels(
                    token,
                    deviceId,
                    GetDeviceChannelsReq.builder()
                            .query(channelId)
                            .count(Integer.MAX_VALUE)
                            .build());
            if (deviceChannels.getData() != null && deviceChannels.getData().getTotal() > 0) {
                deviceChannel = deviceChannels.getData()
                        .getList()
                        .parallelStream()
                        .filter(item -> item.getChannelId().equalsIgnoreCase(channelId))
                        .findFirst();
                if (deviceChannel.isPresent()) {
                    break;
                }
            }
        }
        return deviceChannel;
    }

    @SneakyThrows
    @SuppressWarnings("UnstableApiUsage")
    public DeferredResult<JsonResponse<String>> realtimeVideoUrl(String deviceCode) {
        DeferredResult<JsonResponse<String>> result = new DeferredResult<>(TimeUnit.SECONDS.toMillis(60));
        result.onTimeout(() -> {
            log.error("timeout");
            result.setResult(JsonResponse.error("请求超时"));
        });

        ScheduledFuture<?> schedule = playing.get(deviceCode);
        if (schedule != null) {
            schedule.cancel(true);
        }

        final String[] token = {login()};
        Retryer<Optional<DeviceChannel>> defaultGenericRetryer = RetryerBuilder.<Optional<DeviceChannel>>newBuilder()
                // 异常就重试
                .retryIfException()
                .retryIfRuntimeException()
                // 重试间隔
                .withWaitStrategy(WaitStrategies.fixedWait(3, TimeUnit.SECONDS))
                // 重试次数
                .withStopStrategy(StopStrategies.stopAfterAttempt(3))
                .withRetryListener(new RetryListener() {
                    @Override
                    public <V> void onRetry(Attempt<V> attempt) {
                        if (attempt.hasException()) {
                            log.info("异常 {}", attempt.getExceptionCause().getMessage());
                            cache.remove("token");
                            token[0] = login();
                        }
                    }
                }).build();
        Optional<DeviceChannel> deviceChannel = defaultGenericRetryer.call(() -> getDeviceChannelByDeviceCode(token[0], deviceCode));

        if (deviceChannel.isEmpty()) {
            result.setResult(JsonResponse.error(MessageFormat.format("未能获取 设备: {0} 的通道信息", deviceCode)));
            return result;
        }

        DeviceChannel dc = deviceChannel.get();
        log.info("设备通道信息 => {}", dc);

        try {
            StreamContent data = wvpProxyClient.playStart(token[0], dc.getDeviceId(), dc.getChannelId(), false).getData();
            log.info("实时流信息 {}", data);

            String url = StringUtils.joinWith("/",
                    wvpProxyConfig.getProxyMediaUrl(), data.getApp(),
                    StringUtils.joinWith(".", data.getStream(), "live", "flv"));


            // 定时任务
            schedule = scheduledExecutorService.schedule(() -> {
                log.info("[定时任务] 关闭设备(deviceCode => {}) 视频", deviceCode);
                playing.remove(deviceCode);
                closeRealtimeVideo(deviceCode);
            }, wvpProxyConfig.getRealtimeVideoDuration().getSeconds(), TimeUnit.SECONDS);
            playing.put(deviceCode, schedule);

            result.setResult(JsonResponse.success(url));
        } catch (Exception e) {
            log.error("点播失败", e);
            result.setResult(JsonResponse.error("点播失败"));
            return result;
        }

        return result;
    }

    public void closeRealtimeVideo(String deviceCode) {
        String token = login();
        Optional<DeviceChannel> deviceChannel = getDeviceChannelByDeviceCode(token, deviceCode);
        if (deviceChannel.isEmpty()) {
            return;
        }

        DeviceChannel dc = deviceChannel.get();
        log.info("设备通道信息 => {}", dc);

        wvpProxyClient.playStop(token, dc.getDeviceId(), dc.getChannelId(), false);
    }

    private String login() {
        return cache.get("token", false, () -> {
            String passwdMd5 = MD5.create().digestHex(wvpProxyConfig.getPasswd());
            WvpLoginReq loginReq = WvpLoginReq.builder()
                    .username(wvpProxyConfig.getUser())
                    .password(passwdMd5)
                    .build();
            JsonResponse<WvpLoginResp> login = wvpProxyClient.login(loginReq);
            String token = login.getData().getAccessToken();
            log.info("wvp 登录成功 token => {}", token);
            return token;
        });
    }
}
