package com.ard.vtdu.service.impl;

import com.ard.common.constant.HttpStatus;
import com.ard.common.core.page.TableDataInfo;
import com.ard.tianyu.model.receiver.BoxInfoReceiver;
import com.ard.tianyu.model.receiver.BoxStreamUrlReceiver;
import com.ard.tianyu.service.ITyService;
import com.ard.vtdu.client.MediaClient;
import com.ard.vtdu.domain.*;
import com.ard.vtdu.service.IMediaService;
import com.ard.vtdu.util.ArdTool;
import com.dtflys.forest.exceptions.ForestRuntimeException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.OffsetDateTime;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * @Description: 流媒体业务
 * @ClassName: MediaService
 * @Author: 刘苏义
 * @Date: 2023年07月13日9:28
 * @Version: 1.0
 **/
@Service
@Slf4j
public class MediaServiceImpl implements IMediaService {

    @Resource
    private MediaClient mediaClient;
    @Resource
    private ITyService tyService;
    @Value("${mediamtx.host}")
    private String mediamtxHost;


    @PostConstruct
    public void initMediaMTX() {
        //清空mediamtx流媒体数据
        clearAll();
    }


    /**
     * @return
     * @Author 刘苏义
     * @Description 添加流媒体
     * name 相机ID
     * sourceUrl rtsp推流地址
     * isCode 0-不转码 1-转码
     * mode 1-gpu硬解码 0-cpu软解码
     * @Date 2024/7/3 8:45
     * @Param
     */
    @Override
    public ArdVtdu addPath(ArdVtdu ardVtdu) {
        try {
            String rtspUrl = "rtsp://" + mediamtxHost + ":8554/" + ardVtdu.getName();
            String rtmpUrl = "rtmp://" + mediamtxHost + ":1935/" + ardVtdu.getName();
            String webrtcUrl = "http://" + mediamtxHost + ":8889/" + ardVtdu.getName();

            Conf conf = new Conf();
            boolean isCodeEnabled = ardVtdu.getIsCode().equals(1);
            boolean isModeEabled = ardVtdu.getMode().equals(1);
            boolean isRecordEnabled = ardVtdu.getIsRecord().equals(1);
            if (isCodeEnabled) {
                conf.setSource("publisher");
                //默认软解码
                String cmd = "ffmpeg -rtsp_transport tcp -i " + ardVtdu.getRtspSource() + " -vcodec libx264 -preset ultrafast -r 25 -keyint_min 25 -g 60 -sc_threshold 0 -threads 6  -b:v 2048k -acodec opus  -strict -2 -f rtsp rtsp://localhost:$RTSP_PORT/$MTX_PATH";
                if (isModeEabled) {//硬解码
                    cmd = "ffmpeg -hwaccel cuvid -c:v hevc_cuvid  -rtsp_transport tcp  -i " + ardVtdu.getRtspSource() + " -c:v h264_nvenc  -r 25 -g 60 -sc_threshold 0 -threads 6  -b:v 2048k -bf 0 -acodec opus  -strict -2 -f rtsp rtsp://localhost:$RTSP_PORT/$MTX_PATH";
                }
                if (isRecordEnabled) {
                    //录像
                    conf.setRunOnInit(cmd);
                    conf.setRunOnInitRestart(true);
                    conf.setRunOnDemand("");
                    conf.setRecord(true);
                } else {
                    conf.setRunOnDemand(cmd);
                    conf.setRunOnDemandRestart(true);
                    conf.setRunOnDemandCloseAfter("5s");
                    conf.setRecord(false);
                    conf.setRunOnInit("");
                }
            } else {
                conf.setSource(ardVtdu.getRtspSource());
                conf.setRunOnDemandRestart(false);
                conf.setRunOnDemandCloseAfter("5s");
                conf.setSourceOnDemand(!isRecordEnabled);
                conf.setRecord(isRecordEnabled);
            }
            conf.setMaxReaders(100);
            conf.setSourceProtocol("tcp");

            if (!checkNameExist(ardVtdu.getName())) {
                mediaClient.addPath(ardVtdu.getName(), conf);
            } else {
                mediaClient.editPath(ardVtdu.getName(), conf);
            }
            ardVtdu.setCameraId(ardVtdu.getName().split("_")[0]);
            ardVtdu.setRtspUrl(rtspUrl);
            ardVtdu.setRtmpUrl(rtmpUrl);
            ardVtdu.setWebrtcUrl(webrtcUrl);
            return ardVtdu;
        } catch (ForestRuntimeException ex) {
            log.error("添加流媒体异常：" + ex.getMessage());
        }
        return null;
    }

    @Override
    public ArdVtdu editPath(ArdVtdu ardVtdu) {
        try {
            String rtspUrl = "rtsp://" + mediamtxHost + ":8554/" + ardVtdu.getName();
            String rtmpUrl = "rtmp://" + mediamtxHost + ":1935/" + ardVtdu.getName();
            String webrtcUrl = "http://" + mediamtxHost + ":8889/" + ardVtdu.getName();

            Conf conf = new Conf();
            boolean isCodeEnabled = ardVtdu.getIsCode().equals(1);
            boolean isModeEabled = ardVtdu.getMode().equals(1);
            boolean isRecordEnabled = ardVtdu.getIsRecord().equals(1);
            if (isCodeEnabled) {
                conf.setSource("publisher");
                //默认软解码
                String cmd = "ffmpeg -rtsp_transport tcp -i " + ardVtdu.getRtspSource() + " -vcodec libx264 -preset ultrafast -r 25 -keyint_min 25 -g 60 -sc_threshold 0 -threads 6  -b:v 2048k -acodec opus  -strict -2 -f rtsp rtsp://localhost:$RTSP_PORT/$MTX_PATH";
                if (isModeEabled) {//硬解码
                    cmd = "ffmpeg -hwaccel cuvid -c:v hevc_cuvid  -rtsp_transport tcp  -i " + ardVtdu.getRtspSource() + " -c:v h264_nvenc  -r 25 -g 60 -sc_threshold 0 -threads 6  -b:v 2048k -bf 0 -acodec opus  -strict -2 -f rtsp rtsp://localhost:$RTSP_PORT/$MTX_PATH";
                }
                if (isRecordEnabled) {
                    //录像
                    conf.setRunOnInit(cmd);
                    conf.setRunOnInitRestart(true);
                    conf.setRunOnDemand("");
                    conf.setRecord(true);
                } else {
                    conf.setRunOnDemand(cmd);
                    conf.setRunOnDemandRestart(true);
                    conf.setRunOnDemandCloseAfter("5s");
                    conf.setRecord(false);
                    conf.setRunOnInit("");
                }
            } else {
                conf.setSource(ardVtdu.getRtspSource());
                conf.setRunOnDemand("");
                conf.setRunOnDemandRestart(false);
                conf.setRunOnDemandCloseAfter("5s");
                conf.setSourceOnDemand(!isRecordEnabled);
                conf.setRecord(isRecordEnabled);
            }
            conf.setMaxReaders(100);
            conf.setSourceProtocol("tcp");

            if (checkNameExist(ardVtdu.getName())) {
                mediaClient.editPath(ardVtdu.getName(), conf);
            }
            ardVtdu.setRtspUrl(rtspUrl);
            ardVtdu.setRtmpUrl(rtmpUrl);
            ardVtdu.setWebrtcUrl(webrtcUrl);
            return ardVtdu;
        } catch (ForestRuntimeException ex) {
            log.error("修改流媒体异常：" + ex.getMessage());
        }
        return null;
    }

    @Override
    public void removePath(String[] names) {
        try {
            for (String name : names) {
                if (checkNameExist(name)) {
                    mediaClient.removePath(name);
                }
            }
        } catch (Exception ex) {
            log.error("批量移除流媒体异常：" + ex.getMessage());
        }
    }

    @Override
    public void removePath(String name) {
        try {
            if (checkNameExist(name)) {
                mediaClient.removePath(name);
            }
        } catch (ForestRuntimeException ex) {
            log.error("移除流媒体异常：" + ex.getMessage());
        }
    }

    @Override
    public List<StreamInfo> paths(Integer pageNum, Integer pageSize) {
        Paths paths = mediaClient.paths(pageNum - 1, pageSize);
        List<Item> items = paths.getItems();
        List<StreamInfo> pathInfoList = new ArrayList<>();
        for (Item item : items) {
            StreamInfo info = new StreamInfo();
            //ID
            String name = item.getName();
            info.setName(name);

            Conf conf = mediaClient.getPathConfig(name);
            String runOn = "";
            if (StringUtils.isNotEmpty(conf.getRunOnDemand())) {
                runOn = conf.getRunOnDemand();
            }
            //RTSP源地址
            Matcher matcher = Pattern.compile("rtsp://[^\\s\"]+").matcher(runOn);
            if (matcher.find()) {
                info.setRtspSource(matcher.group());
                info.setIsCode(1);
            } else {
                info.setRtspSource(conf.getSource());
                info.setIsCode(0);
            }
            //传输协议
            matcher = Pattern.compile("-rtsp_transport\\s+(\\w+)").matcher(runOn);
            if (matcher.find()) {
                info.setProtocol(matcher.group(1));
            }

            pathInfoList.add(info);
        }
        return pathInfoList;
    }

    @Override
    public RtspSession getRtspSessionById(String sessionId) {
        RtspSession rtspSession = mediaClient.getRtspsessionById(sessionId);
        return rtspSession;
    }

    @Override
    public WebrtcSession getWebrtcSessionById(String sessionId) {
        WebrtcSession webrtcSession = mediaClient.getWebrtcsessionById(sessionId);
        return webrtcSession;
    }

    @Override
    public RtmpSession getRtmpSessionById(String sessionId) {
        RtmpSession rtmpSession = mediaClient.getRtmpsessionById(sessionId);
        return rtmpSession;
    }

    /**
     * 获取推流列表
     * 刘苏义
     * 2023/8/29 9:37:05
     */
    @Override
    public TableDataInfo getPushStreamList(Integer pageNum, Integer pageSize) {
        TableDataInfo tableDataInfo = new TableDataInfo();
        List<StreamInfo> PushStreamInfoList = new ArrayList<>();
        Paths paths = mediaClient.paths(pageNum - 1, pageSize);
        int itemCount = paths.getItemCount();
        tableDataInfo.setTotal(itemCount);
        List<Item> items = paths.getItems();
        for (Item item : items) {
            StreamInfo info = new StreamInfo();
            //ID
            String name = item.getName();
            info.setName(name);
            Conf conf = mediaClient.getPathConfig(name);
            //RTMP播放地址
            String rtmpUrl = "rtmp://" + mediamtxHost + ":1935/" + name;
            info.setRtmpUrl(rtmpUrl);
            //RTSP播放地址
            String rtspUrl = "rtsp://" + mediamtxHost + ":8554/" + name;
            info.setRtspUrl(rtspUrl);
            //WEBRTC播放地址
            String webrtcUrl = "http://" + mediamtxHost + ":8889/" + name;
            info.setWebrtcUrl(webrtcUrl);
            Source source = item.getSource();
            if (source == null || source.getId().equals("")) {
                //会话ID
                info.setId("0");
                //下行流量
                long bytesReceived = item.getBytesReceived();
                String formatReceivedSize = ArdTool.formatFileSize(bytesReceived);
                info.setDownTraffic(formatReceivedSize);
                //上行流量
                long bytesSent = item.getBytesSent();
                String formatSentdSize = ArdTool.formatFileSize(bytesSent);
                info.setUpTraffic(formatSentdSize);
                info.setBeginTime(item.getReadyTime());
            } else {
                RtspSession rtspSession = getRtspSessionById(source.getId());
                //会话ID
                info.setId(rtspSession.getId());
                //开始推流时间
                info.setBeginTime(rtspSession.getCreated());
                //上行流量
                long bytesReceived = rtspSession.getBytesReceived();
                String formatReceivedSize = ArdTool.formatFileSize(bytesReceived);
                info.setUpTraffic(formatReceivedSize);
                //下行流量
                long bytesSent = rtspSession.getBytesSent();
                String formatSentSize = ArdTool.formatFileSize(bytesSent);
                info.setDownTraffic(formatSentSize);
                //推流服务器
                info.setRemoteAddr(rtspSession.getRemoteAddr());
            }
            //RTSP源地址
            String runOn = "";
            if (StringUtils.isNotEmpty(conf.getRunOnDemand())) {
                runOn = conf.getRunOnDemand();
            }
            Matcher matcher = Pattern.compile("rtsp://[^\\s\"]+").matcher(runOn);
            if (matcher.find()) {
                info.setRtspSource(matcher.group());
            } else {
                info.setRtspSource(conf.getSource());
            }
            //传输协议
            info.setProtocol(conf.getSourceProtocol());
            //拉流数量
            List<Readers> readers = item.getReaders();
            info.setNum(readers.size());

            PushStreamInfoList.add(info);
        }
        tableDataInfo.setRows(PushStreamInfoList);
        tableDataInfo.setCode(HttpStatus.SUCCESS);
        tableDataInfo.setMsg("查询成功");
        return tableDataInfo;
    }

    /**
     * 获取webrtc拉流列表
     * 刘苏义
     * 2023/8/29 9:37:05
     */
    @Override
    public TableDataInfo getPullWebrtcStreamList(Integer pageNum, Integer pageSize) {
        TableDataInfo tableDataInfo = new TableDataInfo();
        WebrtcSessions WebrtcSessions = mediaClient.webrtcsessions(pageNum - 1, pageSize);
        List<WebrtcSession> items = WebrtcSessions.getItems();
        items.forEach(
                webrtcSession -> {
                    webrtcSession.setUpStream(ArdTool.formatFileSize(webrtcSession.getBytesReceived()));
                    webrtcSession.setDownStream(ArdTool.formatFileSize(webrtcSession.getBytesSent()));
                    webrtcSession.setBeginTime(webrtcSession.getCreated());
                }
        );
        tableDataInfo.setTotal(WebrtcSessions.getItemCount());
        tableDataInfo.setRows(items);
        tableDataInfo.setCode(HttpStatus.SUCCESS);
        tableDataInfo.setMsg("查询成功");
        return tableDataInfo;
    }

    /**
     * 获取rtmp拉流列表
     * 刘苏义
     * 2023/8/29 9:37:05
     */
    @Override
    public TableDataInfo getPullRtmpStreamList(Integer pageNum, Integer pageSize) {
        TableDataInfo tableDataInfo = new TableDataInfo();
        RtmpSessions rtmpSessions = mediaClient.rtmpsessions(pageNum - 1, pageSize);
        List<RtmpSession> items = rtmpSessions.getItems();
        items.stream().forEach(
                webrtcSession -> {
                    webrtcSession.setUpStream(ArdTool.formatFileSize(webrtcSession.getBytesReceived()));
                    webrtcSession.setDownStream(ArdTool.formatFileSize(webrtcSession.getBytesSent()));
                    webrtcSession.setBeginTime(webrtcSession.getCreated());
                }
        );
        tableDataInfo.setTotal(rtmpSessions.getItemCount());
        tableDataInfo.setRows(items);
        tableDataInfo.setCode(HttpStatus.SUCCESS);
        tableDataInfo.setMsg("查询成功");
        return tableDataInfo;
    }

    /**
     * 获取rtsp拉流列表
     * 刘苏义
     * 2023/8/29 9:37:05
     */
    @Override
    public TableDataInfo getPullRtspStreamList(Integer pageNum, Integer pageSize) {
        TableDataInfo tableDataInfo = new TableDataInfo();
        RtspSessions rtspSessions = mediaClient.rtspsessions(pageNum - 1, pageSize);
        List<RtspSession> items = rtspSessions.getItems();
        items.stream().forEach(
                webrtcSession -> {
                    webrtcSession.setUpStream(ArdTool.formatFileSize(webrtcSession.getBytesReceived()));
                    webrtcSession.setDownStream(ArdTool.formatFileSize(webrtcSession.getBytesSent()));
                    webrtcSession.setBeginTime(webrtcSession.getCreated());
                }
        );
        tableDataInfo.setTotal(rtspSessions.getItemCount());
        tableDataInfo.setRows(items);
        tableDataInfo.setCode(HttpStatus.SUCCESS);
        tableDataInfo.setMsg("查询成功");
        return tableDataInfo;
    }

    /**
     * 踢出rtsp会话
     * 刘苏义
     * 2023/8/29 9:37:05
     */
    @Override
    public Boolean kickRtspSession(String sessionId) {
        try {
            mediaClient.kickRtspSessions(sessionId);
            return true;
        } catch (Exception ex) {
            return false;
        }
    }

    /**
     * 踢出rtmp会话
     * 刘苏义
     * 2023/8/29 9:37:05
     */
    @Override
    public Boolean kickRtmpSession(String sessionId) {
        try {
            mediaClient.kickRtmpSessions(sessionId);
            return true;
        } catch (Exception ex) {
            return false;
        }
    }

    /**
     * 踢出webrtc会话
     * 刘苏义
     * 2023/8/29 9:37:05
     */
    @Override
    public Boolean kickWebrtcSession(String sessionId) {
        try {
            mediaClient.kickWebrtcSessions(sessionId);
            return true;
        } catch (Exception ex) {
            return false;
        }
    }

    /**
     * 获取流媒体name列表
     * 刘苏义
     * 2023/10/13 14:19:07
     */
    @Override
    public List<String> getNameList() {
        List<String> nameList = new ArrayList<>();
        try {
            Paths paths = mediaClient.paths(0, 1000);
            List<Item> items = paths.getItems();
            for (Item item : items) {
                nameList.add(item.getName());
            }
        } catch (Exception ex) {
            log.error("获取流媒体name列表异常：" + ex.getMessage());
        }
        return nameList;
    }

    /**
     * 检查name是否存在
     * 刘苏义
     * 2023/10/19 15:18:45
     */
    @Override
    public boolean checkNameExist(String name) {
        boolean result = false;
        List<String> nameList = getNameList();
        if (nameList.contains(name)) {
            result = true;
        }
        return result;
    }

    /**
     * 清空所有流媒体信息
     *
     * @author 刘苏义
     * @date 2024/8/22 10:18
     */
    @Override
    public void clearAll() {
        List<String> nameList = getNameList();
        nameList.stream().forEach(name -> {
            removePath(name);
        });
    }

    /**
     * 开启录像
     */
    @Override
    public String setRecord(Conf conf) {
        return mediaClient.editPath(conf.getName(), conf);
    }

    /**
     * 获取全部录像分段列表
     */
    @Override
    public Paths recordingList(Integer pageNum, Integer pageSize) {
        Paths paths = mediaClient.recordingList(pageNum - 1, pageSize);
        paths.getItems().forEach(item -> {
            item.getSegments().forEach(segment -> {
                // 使用 OffsetDateTime 进行解析
                OffsetDateTime offsetDateTime = OffsetDateTime.parse(segment.getStart());
                // 转换为 java.util.Date
                Date date = Date.from(offsetDateTime.toInstant());
                segment.setStartTime(date);
            });
        });
        return paths;
    }

    /**
     * 指定路径名获取录像分段列表
     */
    @Override
    public Item recordingListByName(String name) {
        Item item = mediaClient.recordingListByName(name);
        item.getSegments().forEach(segment -> {
            // 使用 OffsetDateTime 进行解析
            OffsetDateTime offsetDateTime = OffsetDateTime.parse(segment.getStart());
            // 转换为 java.util.Date
            Date date = Date.from(offsetDateTime.toInstant());
            segment.setStartTime(date);
        });
        return item;
    }

    /**
     * 通过路径名和分段录像开始时间删除录像分段
     */
    @Override
    public String deleteSegment(String name, String start) {
        return mediaClient.deleteSegment(name, start);
    }

    //获取通道配置
    @Override
    public Conf getPathConfig(String name) {
        return mediaClient.getPathConfig(name);
    }

    //获取所有通道配置列表
    @Override
    public PathConfig getPathConfigs(Integer pageNum, Integer pageSize) {
        return mediaClient.getPathConfigs(pageNum - 1, pageSize);
    }

    //同步无人机推流到流媒体
    @Scheduled(cron = "0/20 * * * * ?")
    public void syncVtdu() {
        log.info("定时同步MediaMTX任务开始");
        List<String> nameList = getNameList();
        List<BoxInfoReceiver> boxList = tyService.getBoxList();
        if (boxList == null || boxList.isEmpty()) {
            return;
        }
        List<ArdVtdu> ardVtduList = boxList.stream()
                .filter(boxReceiver -> boxReceiver.getOnLine() != 0)  // 过滤掉不在线的
                .map(boxReceiver -> {
                    BoxStreamUrlReceiver boxLiveUrl = tyService.getBoxLiveUrl(boxReceiver.getBoxSn());
                    return ArdVtdu.builder().name(boxReceiver.getBoxSn()).rtspSource(boxLiveUrl.getRtmp()).build();
                })
                .collect(Collectors.toList());
        asyncMediaMTX(ardVtduList, nameList);
    }

    /**
     * @return
     * @Author 刘苏义
     * @Description 同步本地vtdu库和流媒体中的数据
     * @Date 2024/7/10 15:26
     * @Param vtdus vtdu库的集合
     * @Param names 流媒体的name集合
     */
    public void asyncMediaMTX(List<ArdVtdu> vtdus, List<String> names) {
        //需要更新的数据,参数顺序注意
        sameList(vtdus, names).forEach(vtdu -> {
            Conf conf = getPathConfig(vtdu.getName());
            if (conf == null || conf.getSource() == null) {
                return;
            }
            String currentRtspSource = vtdu.getRtspSource();  // 提前获取需要比对的RTSP源
            String confRtspSource = null;

            // 未转码
            if (vtdu.getIsCode().equals(0)) {
                //获取source中的rtsp地址
                confRtspSource = conf.getSource();

            } else {
                // 转码
                if (vtdu.getIsRecord().equals(1)) {
                    //开启录像获取RunOnInit中的rtsp地址
                    confRtspSource = conf.getRunOnInit() != null ? matchRtsp(conf.getRunOnInit()) : null;
                } else {
                    //关闭录像获取RunOnDemand中的rtsp地址
                    confRtspSource = conf.getRunOnDemand() != null ? matchRtsp(conf.getRunOnDemand()) : null;
                }
            }
            // 比较 录像状态 是否一致
            boolean isRecordChanged = !Objects.equals(conf.getRecord() != null && conf.getRecord() ? 1 : 0, vtdu.getIsRecord());
            // 比较 RTSP 源是否一致
            boolean isRtspSourceChanged = !Objects.equals(confRtspSource, currentRtspSource);
            // 如果 RTSP 源或录制状态发生变化，则更新路径
            if (isRecordChanged || isRtspSourceChanged) {
                editPath(vtdu);
            }
        });
        //需要删除的数据
        diffListToDel(vtdus, names).forEach(name -> {
            removePath(name);
        });
        //需要新增的数据
        diffListToAdd(vtdus, names).forEach(vtdu -> {
            addPath(vtdu);
        });
    }

    //需要更新的（流媒体和vtdu相同）
    public List<ArdVtdu> sameList(List<ArdVtdu> vtduses, List<String> names) {
        return vtduses.stream().filter(vtdu -> names.contains(vtdu.getName())).collect(Collectors.toList());
    }

    //需要删除的（流媒体中多的）
    public List<String> diffListToDel(List<ArdVtdu> vtduses, List<String> names) {
        if (vtduses.size() >= names.size()) {
            List<String> nameList = vtduses.stream().map(ArdVtdu::getName)
                    .filter(item -> !names.contains(item))
                    .collect(Collectors.toList());
            return nameList;
        } else {
            List<String> nameList = names.stream().filter(item -> !vtduses.stream().map(ArdVtdu::getName).collect(Collectors.toList()).contains(item))
                    .collect(Collectors.toList());
            return nameList;
        }
    }

    //需要新增的（流媒体中少的）
    public List<ArdVtdu> diffListToAdd(List<ArdVtdu> vtduses, List<String> names) {
        return vtduses.stream().filter(vtdu -> !names.contains(vtdu.getName())).collect(Collectors.toList());
    }

    //匹配rtsp地址
    private String matchRtsp(String value) {
        Matcher matcher = Pattern.compile("rtsp://[^\\s\"]+").matcher(value);
        if (matcher.find()) {
            return matcher.group();
        } else {
            return "";
        }
    }
}