基于新的云端录像结构实现国标录像

648540858

2023-10-17 5a75381a00a555443925bbbd8e333b14473b3ed1

基于新的云端录像结构实现国标录像

 src/main/java/com/genersoft/iot/vmp/gb28181/session/VideoStreamSessionManager.java

@@ -75,6 +75,33 @@
        return (SsrcTransaction)redisTemplate.opsForValue().get(scanResult.get(0));
    }

    public SsrcTransaction getSsrcTransactionByCallId(String callId){

        if (ObjectUtils.isEmpty(callId)) {
            return null;
        }
        String key = VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetting.getServerId() + "_*_*_" + callId+ "_*";
        List<Object> scanResult = RedisUtil.scan(redisTemplate, key);
        if (!scanResult.isEmpty()) {
            return (SsrcTransaction)redisTemplate.opsForValue().get(scanResult.get(0));
        }else {
            key = VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetting.getServerId() + "_*_*_play_*";
            scanResult = RedisUtil.scan(redisTemplate, key);
            if (scanResult.isEmpty()) {
                return null;
            }
            for (Object keyObj : scanResult) {
                SsrcTransaction ssrcTransaction = (SsrcTransaction)redisTemplate.opsForValue().get(keyObj);
                if (ssrcTransaction.getSipTransactionInfo() != null &&
                        ssrcTransaction.getSipTransactionInfo().getCallId().equals(callId)) {
                    return ssrcTransaction;
                }
            }
            return null;
        }

    }

    public List<SsrcTransaction> getSsrcTransactionForAll(String deviceId, String channelId, String callId, String stream){
        if (ObjectUtils.isEmpty(deviceId)) {
            deviceId ="*";

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java

@@ -31,6 +31,7 @@
import javax.sip.message.Response;
import java.text.ParseException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
@@ -149,7 +150,7 @@
        }else {

            // 可能是设备发送的停止
            SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(null, null, callIdHeader.getCallId(), null);
            SsrcTransaction ssrcTransaction = streamSession.getSsrcTransactionByCallId(callIdHeader.getCallId());
            if (ssrcTransaction == null) {
                return;
            }

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/KeepaliveNotifyMessageHandler.java

@@ -76,7 +76,7 @@

        RemoteAddressInfo remoteAddressInfo = SipUtils.getRemoteAddressFromRequest(request, userSetting.getSipUseSourceIpAsRemoteAddress());
        if (!device.getIp().equalsIgnoreCase(remoteAddressInfo.getIp()) || device.getPort() != remoteAddressInfo.getPort()) {
            logger.info("[心跳] 设备{}地址变化, 远程地址为: {}:{}", device.getDeviceId(), remoteAddressInfo.getIp(), remoteAddressInfo.getPort());
            logger.info("[收到心跳] 设备{}地址变化, 远程地址为: {}:{}", device.getDeviceId(), remoteAddressInfo.getIp(), remoteAddressInfo.getPort());
            device.setPort(remoteAddressInfo.getPort());
            device.setHostAddress(remoteAddressInfo.getIp().concat(":").concat(String.valueOf(remoteAddressInfo.getPort())));
            device.setIp(remoteAddressInfo.getIp());

 src/main/java/com/genersoft/iot/vmp/media/zlm/AssistRESTfulUtils.java

@@ -9,12 +9,16 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import java.io.IOException;
import java.net.ConnectException;
import java.net.SocketTimeoutException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

@Component
public class AssistRESTfulUtils {
@@ -22,21 +26,43 @@
    private final static Logger logger = LoggerFactory.getLogger(AssistRESTfulUtils.class);


    private OkHttpClient client;




    public interface RequestCallback{
        void run(JSONObject response);
    }

    private OkHttpClient getClient(){
        OkHttpClient.Builder httpClientBuilder = new OkHttpClient.Builder();
        if (logger.isDebugEnabled()) {
            HttpLoggingInterceptor logging = new HttpLoggingInterceptor(message -> {
                logger.debug("http请求参数：" + message);
            });
            logging.setLevel(HttpLoggingInterceptor.Level.BASIC);
            // OkHttp進行添加攔截器loggingInterceptor
            httpClientBuilder.addInterceptor(logging);
        return getClient(null);
    }
    
    private OkHttpClient getClient(Integer readTimeOut){
        if (client == null) {
            if (readTimeOut == null) {
                readTimeOut = 10;
            }
            OkHttpClient.Builder httpClientBuilder = new OkHttpClient.Builder();
            // 设置连接超时时间
            httpClientBuilder.connectTimeout(8, TimeUnit.SECONDS);
            // 设置读取超时时间
            httpClientBuilder.readTimeout(readTimeOut,TimeUnit.SECONDS);
            // 设置连接池
            httpClientBuilder.connectionPool(new ConnectionPool(16, 5, TimeUnit.MINUTES));
            if (logger.isDebugEnabled()) {
                HttpLoggingInterceptor logging = new HttpLoggingInterceptor(message -> {
                    logger.debug("http请求参数：" + message);
                });
                logging.setLevel(HttpLoggingInterceptor.Level.BASIC);
                // OkHttp進行添加攔截器loggingInterceptor
                httpClientBuilder.addInterceptor(logging);
            }
            client = httpClientBuilder.build();
        }
        return httpClientBuilder.build();
        return client;

    }


@@ -124,13 +150,91 @@
        return responseJSON;
    }

    public JSONObject sendPost(MediaServerItem mediaServerItem, String api, JSONObject param, ZLMRESTfulUtils.RequestCallback callback, Integer readTimeOut) {
        OkHttpClient client = getClient(readTimeOut);

    public JSONObject fileDuration(MediaServerItem mediaServerItem, String app, String stream, RequestCallback callback){
        Map<String, Object> param = new HashMap<>();
        param.put("app",app);
        param.put("stream",stream);
        param.put("recordIng",true);
        return sendGet(mediaServerItem, "api/record/file/duration",param, callback);
        if (mediaServerItem == null) {
            return null;
        }
        String url = String.format("http://%s:%s/%s",  mediaServerItem.getIp(), mediaServerItem.getRecordAssistPort(), api);
        JSONObject responseJSON = new JSONObject();
        //-2自定义流媒体 调用错误码
        responseJSON.put("code",-2);
        responseJSON.put("msg","ASSIST调用失败");

        RequestBody requestBodyJson = RequestBody.create(MediaType.parse("application/json; charset=utf-8"), param.toString());

        Request request = new Request.Builder()
                .post(requestBodyJson)
                .url(url)
                .addHeader("Content-Type", "application/json")
                .build();
        if (callback == null) {
            try {
                Response response = client.newCall(request).execute();
                if (response.isSuccessful()) {
                    ResponseBody responseBody = response.body();
                    if (responseBody != null) {
                        String responseStr = responseBody.string();
                        responseJSON = JSON.parseObject(responseStr);
                    }
                }else {
                    response.close();
                    Objects.requireNonNull(response.body()).close();
                }
            }catch (IOException e) {
                logger.error(String.format("[ %s ]ASSIST请求失败: %s", url, e.getMessage()));

                if(e instanceof SocketTimeoutException){
                    //读取超时超时异常
                    logger.error(String.format("读取ASSIST数据失败: %s, %s", url, e.getMessage()));
                }
                if(e instanceof ConnectException){
                    //判断连接异常，我这里是报Failed to connect to 10.7.5.144
                    logger.error(String.format("连接ASSIST失败: %s, %s", url, e.getMessage()));
                }

            }catch (Exception e){
                logger.error(String.format("访问ASSIST失败: %s, %s", url, e.getMessage()));
            }
        }else {
            client.newCall(request).enqueue(new Callback(){

                @Override
                public void onResponse(@NotNull Call call, @NotNull Response response){
                    if (response.isSuccessful()) {
                        try {
                            String responseStr = Objects.requireNonNull(response.body()).string();
                            callback.run(JSON.parseObject(responseStr));
                        } catch (IOException e) {
                            logger.error(String.format("[ %s ]请求失败: %s", url, e.getMessage()));
                        }

                    }else {
                        response.close();
                        Objects.requireNonNull(response.body()).close();
                    }
                }

                @Override
                public void onFailure(@NotNull Call call, @NotNull IOException e) {
                    logger.error(String.format("连接ZLM失败: %s, %s", call.request().toString(), e.getMessage()));

                    if(e instanceof SocketTimeoutException){
                        //读取超时超时异常
                        logger.error(String.format("读取ZLM数据失败: %s, %s", call.request().toString(), e.getMessage()));
                    }
                    if(e instanceof ConnectException){
                        //判断连接异常，我这里是报Failed to connect to 10.7.5.144
                        logger.error(String.format("连接ZLM失败: %s, %s", call.request().toString(), e.getMessage()));
                    }
                }
            });
        }



        return responseJSON;
    }

    public JSONObject getInfo(MediaServerItem mediaServerItem, RequestCallback callback){
@@ -138,33 +242,33 @@
        return sendGet(mediaServerItem, "api/record/info",param, callback);
    }

    public JSONObject addStreamCallInfo(MediaServerItem mediaServerItem, String app, String stream, String callId, RequestCallback callback){
        Map<String, Object> param = new HashMap<>();
        param.put("app",app);
        param.put("stream",stream);
        param.put("callId",callId);
        return sendGet(mediaServerItem, "api/record/addStreamCallInfo",param, callback);
    public JSONObject addTask(MediaServerItem mediaServerItem, String app, String stream, String startTime,
                              String endTime, String callId, List<String> filePathList, String remoteHost) {
        
        JSONObject videoTaskInfoJSON = new JSONObject();
        videoTaskInfoJSON.put("app", app);
        videoTaskInfoJSON.put("stream", stream);
        videoTaskInfoJSON.put("startTime", startTime);
        videoTaskInfoJSON.put("endTime", endTime);
        videoTaskInfoJSON.put("callId", callId);
        videoTaskInfoJSON.put("filePathList", filePathList);
        if (!ObjectUtils.isEmpty(remoteHost)) {
            videoTaskInfoJSON.put("remoteHost", remoteHost);
        }

        return sendPost(mediaServerItem, "api/record/file/download/task/add", videoTaskInfoJSON, null, 30);
    }

    public JSONObject getDateList(MediaServerItem mediaServerItem, String app, String stream, int year, int month) {
    public JSONObject queryTaskList(MediaServerItem mediaServerItem, String taskId, Boolean isEnd) {
        Map<String, Object> param = new HashMap<>();
        param.put("app", app);
        param.put("stream", stream);
        param.put("year", year);
        param.put("month", month);
        return sendGet(mediaServerItem, "api/record/date/list", param, null);
    }
        if (!ObjectUtils.isEmpty(taskId)) {
            param.put("taskId", taskId);
        }
        if (!ObjectUtils.isEmpty(isEnd)) {
            param.put("isEnd", isEnd);
        }

    public JSONObject getFileList(MediaServerItem mediaServerItem, int page, int count, String app, String stream,
                                  String startTime, String endTime) {
        Map<String, Object> param = new HashMap<>();
        param.put("app", app);
        param.put("stream", stream);
        param.put("page", page);
        param.put("count", count);
        param.put("startTime", startTime);
        param.put("endTime", endTime);
        return sendGet(mediaServerItem, "api/record/file/listWithDate", param, null);
        return sendGet(mediaServerItem, "api/record/file/download/task/list", param, null);
    }

}

 src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java

@@ -234,12 +234,6 @@
                streamAuthorityInfo.setSign(sign);
                // 鉴权通过
                redisCatchStorage.updateStreamAuthorityInfo(param.getApp(), param.getStream(), streamAuthorityInfo);
                // 通知assist新的callId
                if (mediaInfo != null && mediaInfo.getRecordAssistPort() > 0) {
                    taskExecutor.execute(() -> {
                        assistRESTfulUtils.addStreamCallInfo(mediaInfo, param.getApp(), param.getStream(), callId, null);
                    });
                }
            }
        } else {
            zlmMediaListManager.sendStreamEvent(param.getApp(), param.getStream(), param.getMediaServerId());
@@ -267,15 +261,28 @@
        }
        // 替换流地址
        if ("rtp".equals(param.getApp()) && !mediaInfo.isRtpEnable()) {
            String ssrc = String.format("%010d", Long.parseLong(param.getStream(), 16));;
            InviteInfo inviteInfo = inviteStreamService.getInviteInfoBySSRC(ssrc);
            if (inviteInfo != null) {
                result.setStream_replace(inviteInfo.getStream());
                logger.info("[ZLM HOOK]推流鉴权 stream: {} 替换为 {}", param.getStream(), inviteInfo.getStream());
            if (!mediaInfo.isRtpEnable()) {
                String ssrc = String.format("%010d", Long.parseLong(param.getStream(), 16));;
                InviteInfo inviteInfo = inviteStreamService.getInviteInfoBySSRC(ssrc);
                if (inviteInfo != null) {
                    result.setStream_replace(inviteInfo.getStream());
                    logger.info("[ZLM HOOK]推流鉴权 stream: {} 替换为 {}", param.getStream(), inviteInfo.getStream());
                }
            }

        }

        List<SsrcTransaction> ssrcTransactionForAll = sessionManager.getSsrcTransactionForAll(null, null, null, param.getStream());
        if (ssrcTransactionForAll != null && ssrcTransactionForAll.size() == 1) {

            // 为录制国标模拟一个鉴权信息
            StreamAuthorityInfo streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(param);
            streamAuthorityInfo.setApp(param.getApp());
            streamAuthorityInfo.setStream(ssrcTransactionForAll.get(0).getStream());
            streamAuthorityInfo.setCallId(ssrcTransactionForAll.get(0).getSipTransactionInfo().getCallId());

            redisCatchStorage.updateStreamAuthorityInfo(param.getApp(), ssrcTransactionForAll.get(0).getStream(), streamAuthorityInfo);

            String deviceId = ssrcTransactionForAll.get(0).getDeviceId();
            String channelId = ssrcTransactionForAll.get(0).getChannelId();
            DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
@@ -349,13 +356,11 @@

            List<OnStreamChangedHookParam.MediaTrack> tracks = param.getTracks();
            // TODO 重构此处逻辑
            boolean isPush = false;
            if (param.isRegist()) {
                // 处理流注册的鉴权信息
                // 处理流注册的鉴权信息， 流注销这里不再删除鉴权信息，下次来了新的鉴权信息会对就的进行覆盖
                if (param.getOriginType() == OriginType.RTMP_PUSH.ordinal()
                        || param.getOriginType() == OriginType.RTSP_PUSH.ordinal()
                        || param.getOriginType() == OriginType.RTC_PUSH.ordinal()) {
                    isPush = true;
                    StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(param.getApp(), param.getStream());
                    if (streamAuthorityInfo == null) {
                        streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(param);
@@ -365,8 +370,6 @@
                    }
                    redisCatchStorage.updateStreamAuthorityInfo(param.getApp(), param.getStream(), streamAuthorityInfo);
                }
            } else {
                redisCatchStorage.removeStreamAuthorityInfo(param.getApp(), param.getStream());
            }

            if ("rtsp".equals(param.getSchema())) {

 src/main/java/com/genersoft/iot/vmp/service/ICloudRecordService.java

@@ -1,5 +1,7 @@
package com.genersoft.iot.vmp.service;

import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnRecordMp4HookParam;
import com.genersoft.iot.vmp.service.bean.CloudRecordItem;
@@ -28,4 +30,14 @@
     */
    List<String> getDateList(String app, String stream, int year, int month, List<MediaServerItem> mediaServerItems);

    /**
     * 添加合并任务
     */
    String addTask(String app, String stream, String mediaServerId, String startTime, String endTime, String callId, String remoteHost);


    /**
     * 查询合并任务列表
     */
    JSONArray queryTask(String taskId, String mediaServerId, Boolean isEnd);
}

 src/main/java/com/genersoft/iot/vmp/service/IMediaServerService.java

@@ -87,21 +87,10 @@

    void updateMediaServerKeepalive(String mediaServerId, ServerKeepaliveData data);

    boolean checkRtpServer(MediaServerItem mediaServerItem, String rtp, String stream);

    /**
     * 获取负载信息
     * @return
     */
    MediaServerLoad getLoad(MediaServerItem mediaServerItem);

    /**
     * 按时间查找录像文件
     */
    List<RecordFile> getRecords(String app, String stream, String startTime, String endTime, List<MediaServerItem> mediaServerItems);

    /**
     * 查找存在录像文件的时间
     */
    List<String> getRecordDates(String app, String stream, int year, int month, List<MediaServerItem> mediaServerItems);
}

 src/main/java/com/genersoft/iot/vmp/service/impl/CloudRecordServiceImpl.java

@@ -1,11 +1,17 @@
package com.genersoft.iot.vmp.service.impl;

import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.gb28181.bean.GbStream;
import com.genersoft.iot.vmp.gb28181.bean.SsrcTransaction;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
import com.genersoft.iot.vmp.media.zlm.AssistRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnRecordMp4HookParam;
import com.genersoft.iot.vmp.service.ICloudRecordService;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.bean.CloudRecordItem;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.dao.CloudRecordServiceMapper;
@@ -32,7 +38,16 @@
    private CloudRecordServiceMapper cloudRecordServiceMapper;

    @Autowired
    private IMediaServerService mediaServerService;

    @Autowired
    private IRedisCatchStorage redisCatchStorage;

    @Autowired
    private AssistRESTfulUtils assistRESTfulUtils;

    @Autowired
    private VideoStreamSessionManager streamSession;

    @Override
    public PageInfo<CloudRecordItem> getList(int page, int count, String app, String stream, String startTime, String endTime, List<MediaServerItem> mediaServerItems) {
@@ -54,7 +69,8 @@

        }
        PageHelper.startPage(page, count);
        List<CloudRecordItem> all = cloudRecordServiceMapper.getList(app, stream, startTimeStamp, endTimeStamp, mediaServerItems);
        List<CloudRecordItem> all = cloudRecordServiceMapper.getList(app, stream, startTimeStamp, endTimeStamp,
                null, mediaServerItems);
        return new PageInfo<>(all);
    }

@@ -69,7 +85,8 @@
        }
        long startTimeStamp = startDate.atStartOfDay().toInstant(ZoneOffset.ofHours(8)).getEpochSecond();
        long endTimeStamp = endDate.atStartOfDay().toInstant(ZoneOffset.ofHours(8)).getEpochSecond();
        List<CloudRecordItem> cloudRecordItemList = cloudRecordServiceMapper.getList(app, stream, startTimeStamp, endTimeStamp, mediaServerItems);
        List<CloudRecordItem> cloudRecordItemList = cloudRecordServiceMapper.getList(app, stream, startTimeStamp,
                endTimeStamp, null, mediaServerItems);
        if (cloudRecordItemList.isEmpty()) {
            return new ArrayList<>();
        }
@@ -83,12 +100,71 @@

    @Override
    public void addRecord(OnRecordMp4HookParam param) {
        CloudRecordItem  cloudRecordItem = CloudRecordItem.getInstance(param);
        CloudRecordItem cloudRecordItem = CloudRecordItem.getInstance(param);
        StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(param.getApp(), param.getStream());
        if (streamAuthorityInfo != null) {
            cloudRecordItem.setCallId(streamAuthorityInfo.getCallId());
        }
        logger.info("[添加录像记录] {}/{} 文件大小：{}", param.getApp(), param.getStream(), param.getFile_size());
        logger.info("[添加录像记录] {}/{} 文件大小：{}, 时长： {}秒", param.getApp(), param.getStream(), param.getFile_size(),param.getTime_len());
        cloudRecordServiceMapper.add(cloudRecordItem);
    }

    @Override
    public String addTask(String app, String stream, String mediaServerId, String startTime, String endTime, String callId, String remoteHost) {
        // 参数校验
        assert app != null;
        assert stream != null;
        MediaServerItem mediaServerItem = null;
        if (mediaServerId == null) {
            mediaServerItem = mediaServerService.getDefaultMediaServer();
        }else {
            mediaServerItem = mediaServerService.getOne(mediaServerId);
        }
        if (mediaServerItem == null) {
            throw new ControllerException(ErrorCode.ERROR100.getCode(), "未找到可用的流媒体");
        }else {
            if (remoteHost == null) {
                remoteHost = "http://" + mediaServerItem.getStreamIp() + ":" + mediaServerItem.getRecordAssistPort();
            }
        }
        Long startTimeStamp = null;
        Long endTimeStamp = null;
        if (startTime != null) {
            startTimeStamp = DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(startTime);
        }
        if (endTime != null) {
            endTimeStamp = DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(endTime);
        }

        List<MediaServerItem> mediaServers = new ArrayList<>();
        mediaServers.add(mediaServerItem);
        // 检索相关的录像文件
        List<String> filePathList = cloudRecordServiceMapper.queryRecordFilePathList(app, stream, startTimeStamp, endTimeStamp, callId, mediaServers);
        if (filePathList == null || filePathList.isEmpty()) {
            throw new ControllerException(ErrorCode.ERROR100.getCode(), "未检索到视频文件");
        }
        JSONObject result =  assistRESTfulUtils.addTask(mediaServerItem, app, stream, startTime, endTime, callId, filePathList, remoteHost);
        if (result.getInteger("code") != 0) {
            throw new ControllerException(result.getInteger("code"), result.getString("msg"));
        }
        return result.getString("data");
    }

    @Override
    public JSONArray queryTask(String taskId, String mediaServerId, Boolean isEnd) {
        MediaServerItem mediaServerItem = null;
        if (mediaServerId == null) {
            mediaServerItem = mediaServerService.getDefaultMediaServer();
        }else {
            mediaServerItem = mediaServerService.getOne(mediaServerId);
        }
        if (mediaServerItem == null) {
            throw new ControllerException(ErrorCode.ERROR100.getCode(), "未找到可用的流媒体");
        }
        JSONObject result =  assistRESTfulUtils.queryTaskList(mediaServerItem, taskId, isEnd);
        if (result.getInteger("code") != 0) {
            throw new ControllerException(result.getInteger("code"), result.getString("msg"));
        }
        return result.getJSONArray("data");
    }
}

 src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java

@@ -742,15 +742,6 @@
    }

    @Override
    public boolean checkRtpServer(MediaServerItem mediaServerItem, String app, String stream) {
        JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaServerItem, stream);
        if(rtpInfo.getInteger("code") == 0){
            return rtpInfo.getBoolean("exist");
        }
        return false;
    }

    @Override
    public MediaServerLoad getLoad(MediaServerItem mediaServerItem) {
        MediaServerLoad result = new MediaServerLoad();
        result.setId(mediaServerItem.getId());
@@ -760,91 +751,5 @@
        result.setGbReceive(inviteStreamService.getStreamInfoCount(mediaServerItem.getId()));
        result.setGbSend(redisCatchStorage.getGbSendCount(mediaServerItem.getId()));
        return result;
    }

    @Override
    public List<RecordFile> getRecords(String app, String stream, String startTime, String endTime, List<MediaServerItem> mediaServerItems) {
        Assert.notNull(app, "app不存在");
        Assert.notNull(stream, "stream不存在");
        Assert.notNull(startTime, "startTime不存在");
        Assert.notNull(endTime, "endTime不存在");
        Assert.notEmpty(mediaServerItems, "流媒体列表为空");

        CompletableFuture[] completableFutures = new CompletableFuture[mediaServerItems.size()];
        for (int i = 0; i < mediaServerItems.size(); i++) {
            completableFutures[i] = getRecordFilesForOne(app, stream, startTime, endTime, mediaServerItems.get(i));
        }
        List<RecordFile> result = new ArrayList<>();
        for (int i = 0; i < completableFutures.length; i++) {
            try {
                List<RecordFile> list = (List<RecordFile>) completableFutures[i].get();
                if (!list.isEmpty()) {
                    for (int g = 0; g < list.size(); g++) {
                        list.get(g).setMediaServerId(mediaServerItems.get(i).getId());
                    }
                    result.addAll(list);
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            } catch (ExecutionException e) {
                throw new RuntimeException(e);
            }
        }
        Comparator<RecordFile> comparator = Comparator.comparing(RecordFile::getFileName);
        result.sort(comparator);
        return result;
    }

    @Override
    public List<String> getRecordDates(String app, String stream, int year, int month, List<MediaServerItem> mediaServerItems) {
        Assert.notNull(app, "app不存在");
        Assert.notNull(stream, "stream不存在");
        Assert.notEmpty(mediaServerItems, "流媒体列表为空");
        CompletableFuture[] completableFutures = new CompletableFuture[mediaServerItems.size()];

        for (int i = 0; i < mediaServerItems.size(); i++) {
            completableFutures[i] = getRecordDatesForOne(app, stream, year, month, mediaServerItems.get(i));
        }
        List<String> result = new ArrayList<>();
        CompletableFuture.allOf(completableFutures).join();
        for (CompletableFuture completableFuture : completableFutures) {
            try {
                List<String> list = (List<String>) completableFuture.get();
                result.addAll(list);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            } catch (ExecutionException e) {
                throw new RuntimeException(e);
            }
        }
        Collections.sort(result);
        return result;
    }

    @Async
    public CompletableFuture<List<String>> getRecordDatesForOne(String app, String stream, int year, int month, MediaServerItem mediaServerItem) {
        JSONObject fileListJson = assistRESTfulUtils.getDateList(mediaServerItem, app, stream, year, month);
        if (fileListJson != null && !fileListJson.isEmpty()) {
            if (fileListJson.getString("code") != null && fileListJson.getInteger("code") == 0) {
                JSONArray data = fileListJson.getJSONArray("data");
                return CompletableFuture.completedFuture(data.toJavaList(String.class));
            }
        }
        return CompletableFuture.completedFuture(new ArrayList<>());
    }

    @Async
    public CompletableFuture<List<RecordFile>> getRecordFilesForOne(String app, String stream, String startTime, String endTime, MediaServerItem mediaServerItem) {
        JSONObject fileListJson = assistRESTfulUtils.getFileList(mediaServerItem, 1, 100000000, app, stream, startTime, endTime);
        if (fileListJson != null && !fileListJson.isEmpty()) {
            if (fileListJson.getString("code") != null && fileListJson.getInteger("code") == 0) {
                JSONObject data = fileListJson.getJSONObject("data");
                JSONArray list = data.getJSONArray("list");
                if (list != null) {
                    return CompletableFuture.completedFuture(list.toJavaList(RecordFile.class));
                }
            }
        }
        return CompletableFuture.completedFuture(new ArrayList<>());
    }
}

 src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java

@@ -27,11 +27,13 @@
import com.genersoft.iot.vmp.media.zlm.dto.hook.HookParam;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
import com.genersoft.iot.vmp.service.*;
import com.genersoft.iot.vmp.service.bean.CloudRecordItem;
import com.genersoft.iot.vmp.service.bean.ErrorCallback;
import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
import com.genersoft.iot.vmp.service.bean.SSRCInfo;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.storager.dao.CloudRecordServiceMapper;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import gov.nist.javax.sip.message.SIPResponse;
@@ -106,6 +108,9 @@

    @Autowired
    private ZlmHttpHookSubscribe subscribe;

    @Autowired
    private CloudRecordServiceMapper cloudRecordServiceMapper;


    @Override
@@ -749,31 +754,43 @@
                logger.warn("查询录像信息时发现节点已离线");
                return null;
            }
            if (mediaServerItem.getRecordAssistPort() > 0) {
                JSONObject jsonObject = assistRESTfulUtils.fileDuration(mediaServerItem, inviteInfo.getStreamInfo().getApp(), inviteInfo.getStreamInfo().getStream(), null);
                if (jsonObject == null) {
                    throw new ControllerException(ErrorCode.ERROR100.getCode(), "连接Assist服务失败");
                }
                if (jsonObject.getInteger("code") == 0) {
                    long duration = jsonObject.getLong("data");

                    if (duration == 0) {
                        inviteInfo.getStreamInfo().setProgress(0);
                    } else {
                        String startTime = inviteInfo.getStreamInfo().getStartTime();
                        String endTime = inviteInfo.getStreamInfo().getEndTime();
                        long start = DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(startTime);
                        long end = DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(endTime);

                        BigDecimal currentCount = new BigDecimal(duration / 1000);
                        BigDecimal totalCount = new BigDecimal(end - start);
                        BigDecimal divide = currentCount.divide(totalCount, 2, RoundingMode.HALF_UP);
                        double process = divide.doubleValue();
                        inviteInfo.getStreamInfo().setProgress(process);
                    }
                    inviteStreamService.updateInviteInfo(inviteInfo);
                }
            if (mediaServerItem.getRecordAssistPort() == 0) {
                throw new ControllerException(ErrorCode.ERROR100.getCode(), "未配置Assist服务，无法完成录像下载");
            }
            SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(deviceId, channelId, null, stream);

            if (ssrcTransaction == null) {
                logger.warn("[获取下载进度]，未找到下载事务信息");
                return null;
            }

            // 为了支持多个数据库，这里不能使用求和函数来直接获取总数了
            List<CloudRecordItem> cloudRecordItemList = cloudRecordServiceMapper.getList("rtp", inviteInfo.getStream(), null, null, ssrcTransaction.getCallId(), null);

            if (cloudRecordItemList.isEmpty()) {
                logger.warn("[获取下载进度]，未找到下载视频信息");
                return null;
            }
            long duration = 0;
            for (CloudRecordItem cloudRecordItem : cloudRecordItemList) {
                duration += cloudRecordItem.getTimeLen();
            }
            if (duration == 0) {
                inviteInfo.getStreamInfo().setProgress(0);
            } else {
                String startTime = inviteInfo.getStreamInfo().getStartTime();
                String endTime = inviteInfo.getStreamInfo().getEndTime();
                long start = DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(startTime);
                long end = DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(endTime);

                BigDecimal currentCount = new BigDecimal(duration);
                BigDecimal totalCount = new BigDecimal(end - start);
                BigDecimal divide = currentCount.divide(totalCount, 2, RoundingMode.HALF_UP);
                double process = divide.doubleValue();
                inviteInfo.getStreamInfo().setProgress(process);
            }
            inviteStreamService.updateInviteInfo(inviteInfo);

            return inviteInfo.getStreamInfo();
        }
        return null;

 src/main/java/com/genersoft/iot/vmp/storager/dao/CloudRecordServiceMapper.java

@@ -42,17 +42,37 @@

    @Select(" <script>" +
            "select * " +
            "from wvp_cloud_record " +
            "where 0 = 0" +
            " from wvp_cloud_record " +
            " where 0 = 0" +
            " <if test= 'app != null '> and app=#{app}</if>" +
            " <if test= 'stream != null '> and stream=#{stream}</if>" +
            " <if test= 'startTimeStamp != null '> and start_time &gt;= #{startTimeStamp}</if>" +
            " <if test= 'endTimeStamp != null '> and end_time &lt;= #{endTimeStamp}</if>" +
            " <if test= 'startTimeStamp != null '> and end_time &gt;= #{startTimeStamp}</if>" +
            " <if test= 'endTimeStamp != null '> and start_time &lt;= #{endTimeStamp}</if>" +
            " <if test= 'callId != null '> and call_id = #{callId}</if>" +
            " <if test= 'mediaServerItemList != null  ' > and media_server_id in " +
            " <foreach collection='mediaServerItemList'  item='item'  open='(' separator=',' close=')' > #{item.id}</foreach>" +
            " </if>" +
            " </script>")
    List<CloudRecordItem> getList(@Param("app") String app, @Param("stream") String stream,
                                  @Param("startTimeStamp")Long startTimeStamp, @Param("endTimeStamp")Long endTimeStamp,
                                  List<MediaServerItem> mediaServerItemList);
                                  @Param("callId")String callId, List<MediaServerItem> mediaServerItemList);


    @Select(" <script>" +
            "select file_path" +
            " from wvp_cloud_record " +
            " where 0 = 0" +
            " <if test= 'app != null '> and app=#{app}</if>" +
            " <if test= 'stream != null '> and stream=#{stream}</if>" +
            " <if test= 'startTimeStamp != null '> and end_time &gt;= #{startTimeStamp}</if>" +
            " <if test= 'endTimeStamp != null '> and start_time &lt;= #{endTimeStamp}</if>" +
            " <if test= 'callId != null '> and call_id = #{callId}</if>" +
            " <if test= 'mediaServerItemList != null  ' > and media_server_id in " +
            " <foreach collection='mediaServerItemList'  item='item'  open='(' separator=',' close=')' > #{item.id}</foreach>" +
            " </if>" +
            " </script>")
    List<String> queryRecordFilePathList(@Param("app") String app, @Param("stream") String stream,
                                  @Param("startTimeStamp")Long startTimeStamp, @Param("endTimeStamp")Long endTimeStamp,
                                  @Param("callId")String callId, List<MediaServerItem> mediaServerItemList);

}

 src/main/java/com/genersoft/iot/vmp/vmanager/cloudRecord/CloudRecordController.java

@@ -1,5 +1,7 @@
package com.genersoft.iot.vmp.vmanager.cloudRecord;

import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
@@ -145,19 +147,53 @@
    @Operation(summary = "添加合并任务")
    @Parameter(name = "app", description = "应用名", required = true)
    @Parameter(name = "stream", description = "流ID", required = true)
    @Parameter(name = "mediaServerId", description = "流媒体ID", required = false)
    @Parameter(name = "startTime", description = "鉴权ID", required = false)
    @Parameter(name = "endTime", description = "鉴权ID", required = false)
    @Parameter(name = "callId", description = "鉴权ID", required = false)
    @Parameter(name = "remoteHost", description = "返回地址时的远程地址", required = false)
    public String addTask(
            @RequestParam String app,
            @RequestParam String stream,
            @RequestParam String startTime,
            @RequestParam String endTime,
            @RequestParam String callId,
            @RequestParam String remoteHost
            @RequestParam(required = true) String app,
            @RequestParam(required = true) String stream,
            @RequestParam(required = false) String mediaServerId,
            @RequestParam(required = false) String startTime,
            @RequestParam(required = false) String endTime,
            @RequestParam(required = false) String callId,
            @RequestParam(required = false) String remoteHost
    ){
        return cloudRecordService.addTask(app, stream, startTime, endTime, callId, remoteHost);
        return cloudRecordService.addTask(app, stream, mediaServerId, startTime, endTime, callId, remoteHost);
    }

    @ResponseBody
    @GetMapping("/task/list")
    @Operation(summary = "查询合并任务")
    @Parameter(name = "taskId", description = "任务Id", required = false)
    @Parameter(name = "mediaServerId", description = "流媒体ID", required = false)
    @Parameter(name = "isEnd", description = "是否结束", required = false)
    public JSONArray queryTaskList(
            @RequestParam(required = false) String taskId,
            @RequestParam(required = false) String mediaServerId,
            @RequestParam(required = false) Boolean isEnd
    ){
        return cloudRecordService.queryTask(taskId, mediaServerId, isEnd);
    }

    @ResponseBody
    @GetMapping("/collect/add")
    @Operation(summary = "添加收藏")
    @Parameter(name = "app", description = "应用名", required = true)
    @Parameter(name = "stream", description = "流ID", required = true)
    @Parameter(name = "mediaServerId", description = "流媒体ID", required = false)
    @Parameter(name = "startTime", description = "鉴权ID", required = false)
    @Parameter(name = "endTime", description = "鉴权ID", required = false)
    @Parameter(name = "callId", description = "鉴权ID", required = false)
    @Parameter(name = "collectType", description = "收藏类型", required = false)
    public JSONArray addCollect(
            @RequestParam(required = false) String taskId,
            @RequestParam(required = false) String mediaServerId,
            @RequestParam(required = false) Boolean isEnd
    ){
        return cloudRecordService.queryTask(taskId, mediaServerId, isEnd);
    }



 web_src/src/components/CloudRecordDetail.vue

@@ -480,12 +480,13 @@
        let that = this;
        this.$axios({
          method: 'get',
          url:`/record_proxy/${that.mediaServerId}/api/record/file/download/task/add`,
          url:`/api/cloud/record/task/add`,
          params: {
            app: that.app,
            stream: that.stream,
            startTime: moment(this.taskTimeRange[0]).format('YYYY-MM-DD HH:mm:ss'),
            endTime: moment(this.taskTimeRange[1]).format('YYYY-MM-DD HH:mm:ss'),
              app: this.app,
              stream: this.stream,
              mediaServerId: this.mediaServerId,
              startTime: moment(this.taskTimeRange[0]).format('YYYY-MM-DD HH:mm:ss'),
              endTime: moment(this.taskTimeRange[1]).format('YYYY-MM-DD HH:mm:ss'),
          }
        }).then(function (res) {
          if (res.data.code === 0 ) {
@@ -505,8 +506,9 @@
        let that = this;
        this.$axios({
          method: 'get',
          url:`/record_proxy/${that.mediaServerId}/api/record/file/download/task/list`,
          url:`/api/cloud/record/task/list`,
          params: {
            mediaServerId: this.mediaServerId,
            isEnd: isEnd,
          }
        }).then(function (res) {

 web_src/src/components/dialog/recordDownload.vue

@@ -137,10 +137,11 @@
        getFileDownload: function (){
          this.$axios({
            method: 'get',
            url:`/record_proxy/${this.mediaServerId}/api/record/file/download/task/add`,
            url:`/api/cloud/record/task/add`,
            params: {
              app: this.app,
              stream: this.stream,
              mediaServerId: this.mediaServerId,
              startTime: null,
              endTime: null,
            }
@@ -169,10 +170,9 @@
        getProgressForFile: function (callback){
          this.$axios({
            method: 'get',
            url:`/record_proxy/${this.mediaServerId}/api/record/file/download/task/list`,
            url:`/api/cloud/record/task/list`,
            params: {
              app: this.app,
              stream: this.stream,
              mediaServerId: this.mediaServerId,
              taskId: this.taskId,
              isEnd: true,
            }