开始重构云端录像继续

648540858
1 parent 298a9f42
Showing 10 changed files with 243 additions and 44 deletions
sql/2.6.9更新.sql
sql/初始化.sql
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
src/main/java/com/genersoft/iot/vmp/media/zlm/dto/hook/OnRecordMp4HookParam.java
src/main/java/com/genersoft/iot/vmp/service/ICloudRecordService.java
src/main/java/com/genersoft/iot/vmp/service/bean/CloudRecordItem.java
src/main/java/com/genersoft/iot/vmp/service/impl/CloudRecordServiceImpl.java
src/main/java/com/genersoft/iot/vmp/storager/dao/CloudRecordServiceMapper.java
src/main/java/com/genersoft/iot/vmp/utils/DateUtil.java
src/main/java/com/genersoft/iot/vmp/vmanager/cloudRecord/CloudRecordController.java
@@ -8,17 +8,18 @@ alter table wvp_stream_proxy
     add stream_key varying(255)
 create table wvp_cloud_record (
-                                 id serial primary key,
-                                 app character varying(255),
-                                 stream character varying(255),
-                                 call_id character varying(255),
-                                 start_time integer,
-                                 end_time integer,
-                                 media_server_id character varying(50),
-                                 file_name character varying(50),
-                                 folder character varying(50),
-                                 file_path character varying(255),
-                                 file_size integer,
-                                 time_len integer,
-                                 constraint uk_stream_push_app_stream_path unique (app, stream, file_path)
+      id serial primary key,
+      app character varying(255),
+      stream character varying(255),
+      call_id character varying(255),
+      start_time integer,
+      end_time integer,
+      media_server_id character varying(50),
+      file_name character varying(255),
+      folder character varying(255),
+      file_path character varying(255),
+      collect_type character varying(255),
+      file_size integer,
+      time_len integer,
+      constraint uk_stream_push_app_stream_path unique (app, stream, file_path)
 );
@@ -267,19 +267,20 @@ create table wvp_stream_push (
                                  constraint uk_stream_push_app_stream unique (app, stream)
 );
 create table wvp_cloud_record (
-                                        id serial primary key,
-                                        app character varying(255),
-                                        stream character varying(255),
-                                        call_id character varying(255),
-                                        start_time integer,
-                                        end_time integer,
-                                        mediaServerId character varying(50),
-                                        file_name character varying(50),
-                                        folder character varying(50),
-                                        file_path character varying(255),
-                                        file_size integer,
-                                        time_len integer,
-                                        constraint uk_stream_push_app_stream_path unique (app, stream, file_path)
+                                id serial primary key,
+                                app character varying(255),
+                                stream character varying(255),
+                                call_id character varying(255),
+                                start_time integer,
+                                end_time integer,
+                                media_server_id character varying(50),
+                                file_name character varying(255),
+                                folder character varying(255),
+                                file_path character varying(255),
+                                collect_type character varying(255),
+                                file_size integer,
+                                time_len integer,
+                                constraint uk_stream_push_app_stream_path unique (app, stream, file_path)
 );
 create table wvp_user (
@@ -111,6 +111,9 @@ public class ZLMHttpHookListener {
     private IUserService userService;
     @Autowired
+    private ICloudRecordService cloudRecordService;
+
+    @Autowired
     private VideoStreamSessionManager sessionManager;
     @Autowired
@@ -775,7 +778,7 @@ public class ZLMHttpHookListener {
         logger.info("[ZLM HOOK] 录像完成事件：{}->{}", param.getMediaServerId(), param.getFile_path());
         taskExecutor.execute(() -> {
-
+            cloudRecordService.addRecord(param);
         });
         return HookResult.SUCCESS();
@@ -9,7 +9,7 @@ public class OnRecordMp4HookParam extends HookParam{
     private String stream;
     private String file_name;
     private String file_path;
-    private String file_size;
+    private long file_size;
     private String folder;
     private String url;
     private String vhost;
@@ -48,11 +48,11 @@ public class OnRecordMp4HookParam extends HookParam{
         this.file_path = file_path;
     }
-    public String getFile_size() {
+    public long getFile_size() {
         return file_size;
     }
-    public void setFile_size(String file_size) {
+    public void setFile_size(long file_size) {
         this.file_size = file_size;
     }
 package com.genersoft.iot.vmp.service;
+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;
 import com.github.pagehelper.PageInfo;
@@ -14,14 +16,16 @@ public interface ICloudRecordService {
     /**
      * 分页回去云端录像列表
      */
-    PageInfo<CloudRecordItem> getList(int page, int count, String startTime, String endTime);
+    PageInfo<CloudRecordItem> getList(int page, int count, String app, String stream, String startTime, String endTime, List<MediaServerItem> mediaServerItems);
     /**
-     * 获取所有的日期
+     * 根据hook消息增加一条记录
      */
-    List<String> getDateList(Integer year, Integer month, String app, String stream);
-
-
+    void addRecord(OnRecordMp4HookParam param);
+    /**
+     * 获取所有的日期
+     */
+    List<String> getDateList(String app, String stream, int year, int month, List<MediaServerItem> mediaServerItems);
 }
 package com.genersoft.iot.vmp.service.bean;
+import com.genersoft.iot.vmp.media.zlm.dto.hook.OnRecordMp4HookParam;
+
 /**
  * 云端录像数据
  */
@@ -69,6 +71,21 @@ public class CloudRecordItem {
      */
     private long timeLen;
+    public static CloudRecordItem getInstance(OnRecordMp4HookParam param) {
+        CloudRecordItem cloudRecordItem = new CloudRecordItem();
+        cloudRecordItem.setApp(param.getApp());
+        cloudRecordItem.setStream(param.getStream());
+        cloudRecordItem.setStartTime(param.getStart_time());
+        cloudRecordItem.setFileName(param.getFile_name());
+        cloudRecordItem.setFolder(param.getFolder());
+        cloudRecordItem.setFileSize(param.getFile_size());
+        cloudRecordItem.setFilePath(param.getFile_path());
+        cloudRecordItem.setMediaServerId(param.getMediaServerId());
+        cloudRecordItem.setTimeLen(param.getTime_len());
+        cloudRecordItem.setEndTime(param.getStart_time() + param.getTime_len());
+        return cloudRecordItem;
+    }
+
     public int getId() {
         return id;
     }
+package com.genersoft.iot.vmp.service.impl;
+
+import com.genersoft.iot.vmp.conf.exception.ControllerException;
+import com.genersoft.iot.vmp.gb28181.bean.GbStream;
+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.bean.CloudRecordItem;
+import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
+import com.genersoft.iot.vmp.storager.dao.CloudRecordServiceMapper;
+import com.genersoft.iot.vmp.utils.DateUtil;
+import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
+import com.github.pagehelper.PageHelper;
+import com.github.pagehelper.PageInfo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import org.springframework.util.unit.DataUnit;
+
+import java.time.*;
+import java.time.temporal.TemporalAccessor;
+import java.util.*;
+
+@Service
+public class CloudRecordServiceImpl implements ICloudRecordService {
+
+    private final static Logger logger = LoggerFactory.getLogger(CloudRecordServiceImpl.class);
+
+    @Autowired
+    private CloudRecordServiceMapper cloudRecordServiceMapper;
+
+    @Autowired
+    private IRedisCatchStorage redisCatchStorage;
+
+    @Override
+    public PageInfo<CloudRecordItem> getList(int page, int count, String app, String stream, String startTime, String endTime, List<MediaServerItem> mediaServerItems) {
+        // 开始时间和结束时间在数据库中都是以秒为单位的
+        Long startTimeStamp = null;
+        Long endTimeStamp = null;
+        if (startTime != null ) {
+            if (!DateUtil.verification(startTime, DateUtil.formatter)) {
+                throw new ControllerException(ErrorCode.ERROR100.getCode(), "开始时间格式错误，正确格式为： " + DateUtil.formatter);
+            }
+            startTimeStamp = DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(startTime);
+
+        }
+        if (endTime != null ) {
+            if (!DateUtil.verification(endTime, DateUtil.formatter)) {
+                throw new ControllerException(ErrorCode.ERROR100.getCode(), "结束时间格式错误，正确格式为： " + DateUtil.formatter);
+            }
+            endTimeStamp = DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(endTime);
+
+        }
+        PageHelper.startPage(page, count);
+        List<CloudRecordItem> all = cloudRecordServiceMapper.getList(app, stream, startTimeStamp, endTimeStamp, mediaServerItems);
+        return new PageInfo<>(all);
+    }
+
+    @Override
+    public List<String> getDateList(String app, String stream, int year, int month, List<MediaServerItem> mediaServerItems) {
+        LocalDate startDate = LocalDate.of(year, month, 1);
+        LocalDate endDate;
+        if (month == 12) {
+            endDate = LocalDate.of(year + 1, 1, 1);
+        }else {
+            endDate = LocalDate.of(year, month + 1, 1);
+        }
+        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);
+        if (cloudRecordItemList.isEmpty()) {
+            return new ArrayList<>();
+        }
+        Set<String> resultSet = new HashSet<>();
+        cloudRecordItemList.stream().forEach(cloudRecordItem -> {
+            String date = DateUtil.timestampTo_yyyy_MM_dd(cloudRecordItem.getStartTime());
+            resultSet.add(date);
+        });
+        return new ArrayList<>(resultSet);
+    }
+
+    @Override
+    public void addRecord(OnRecordMp4HookParam 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());
+        cloudRecordServiceMapper.add(cloudRecordItem);
+    }
+}
+package com.genersoft.iot.vmp.storager.dao;
+
+import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
+import com.genersoft.iot.vmp.service.bean.CloudRecordItem;
+import org.apache.ibatis.annotations.Insert;
+import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Select;
+
+import java.util.List;
+
+@Mapper
+public interface CloudRecordServiceMapper {
+
+    @Insert(" <script>" +
+            "INSERT INTO wvp_cloud_record (" +
+            " app," +
+            " stream," +
+            "<if test=\"callId != null\"> call_id,</if>" +
+            " start_time," +
+            " end_time," +
+            " media_server_id," +
+            " file_name," +
+            " folder," +
+            " file_path," +
+            " file_size," +
+            " time_len ) " +
+            "VALUES (" +
+            " #{app}," +
+            " #{stream}," +
+            " <if test=\"callId != null\"> #{callId},</if>" +
+            " #{startTime}," +
+            " #{endTime}," +
+            " #{mediaServerId}," +
+            " #{fileName}," +
+            " #{folder}," +
+            " #{filePath}," +
+            " #{fileSize}," +
+            " #{timeLen})" +
+            " </script>")
+    int add(CloudRecordItem cloudRecordItem);
+
+    @Select(" <script>" +
+            "select * " +
+            "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= '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);
+}
@@ -40,11 +40,17 @@ public class DateUtil {
      */
     public static final String URL_PATTERN = "yyyyMMddHHmmss";
+    /**
+     * 日期格式
+     */
+    public static final String date_PATTERN = "yyyy-MM-dd";
+
     public static final String zoneStr = "Asia/Shanghai";
     public static final DateTimeFormatter formatterCompatibleISO8601 = DateTimeFormatter.ofPattern(ISO8601_COMPATIBLE_PATTERN, Locale.getDefault()).withZone(ZoneId.of(zoneStr));
     public static final DateTimeFormatter formatterISO8601 = DateTimeFormatter.ofPattern(ISO8601_PATTERN, Locale.getDefault()).withZone(ZoneId.of(zoneStr));
     public static final DateTimeFormatter formatter = DateTimeFormatter.ofPattern(PATTERN, Locale.getDefault()).withZone(ZoneId.of(zoneStr));
+    public static final DateTimeFormatter DateFormatter = DateTimeFormatter.ofPattern(date_PATTERN, Locale.getDefault()).withZone(ZoneId.of(zoneStr));
     public static final DateTimeFormatter urlFormatter = DateTimeFormatter.ofPattern(URL_PATTERN, Locale.getDefault()).withZone(ZoneId.of(zoneStr));
 	public static String yyyy_MM_dd_HH_mm_ssToISO8601(String formatTime) {
@@ -72,6 +78,22 @@ public class DateUtil {
 	}
     /**
+     * 时间戳 转 yyyy_MM_dd_HH_mm_ss
+     */
+	public static String timestampTo_yyyy_MM_dd_HH_mm_ss(long timestamp) {
+        Instant instant = Instant.ofEpochSecond(timestamp);
+        return formatter.format(LocalDateTime.ofInstant(instant, ZoneId.of(zoneStr)));
+	}
+
+    /**
+     * 时间戳 转 yyyy_MM_dd
+     */
+    public static String timestampTo_yyyy_MM_dd(long timestamp) {
+        Instant instant = Instant.ofEpochSecond(timestamp);
+        return DateFormatter.format(LocalDateTime.ofInstant(instant, ZoneId.of(zoneStr)));
+    }
+
+    /**
      * 获取当前时间
      * @return
      */
@@ -7,10 +7,12 @@ import com.genersoft.iot.vmp.media.zlm.SendRtpPortManager;
 import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
 import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
 import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
+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.vmanager.bean.ErrorCode;
-import com.genersoft.iot.vmp.vmanager.bean.PageInfo;
 import com.genersoft.iot.vmp.vmanager.bean.RecordFile;
+import com.github.pagehelper.PageInfo;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.Parameter;
 import io.swagger.v3.oas.annotations.tags.Tag;
@@ -41,7 +43,7 @@ public class CloudRecordController {
     private final static Logger logger = LoggerFactory.getLogger(CloudRecordController.class);
     @Autowired
-    private ZlmHttpHookSubscribe hookSubscribe;
+    private ICloudRecordService cloudRecordService;
     @Autowired
     private IMediaServerService mediaServerService;
@@ -95,7 +97,7 @@ public class CloudRecordController {
             return new ArrayList<>();
         }
-        return mediaServerService.getRecordDates(app, stream, year, month, mediaServerItems);
+        return cloudRecordService.getDateList(app, stream, year, month, mediaServerItems);
     }
     @ResponseBody
@@ -108,7 +110,7 @@ public class CloudRecordController {
     @Parameter(name = "startTime", description = "开始时间(yyyy-MM-dd HH:mm:ss)", required = true)
     @Parameter(name = "endTime", description = "结束时间(yyyy-MM-dd HH:mm:ss)", required = true)
     @Parameter(name = "mediaServerId", description = "流媒体ID，置空则查询全部流媒体", required = false)
-    public PageInfo<RecordFile> openRtpServer(
+    public PageInfo<CloudRecordItem> openRtpServer(
             @RequestParam String app,
             @RequestParam String stream,
             @RequestParam int page,
@@ -133,12 +135,9 @@ public class CloudRecordController {
             mediaServerItems = mediaServerService.getAll();
         }
         if (mediaServerItems.isEmpty()) {
-            return new PageInfo<>();
+            throw new ControllerException(ErrorCode.ERROR100.getCode(), "当前无流媒体");
         }
-        List<RecordFile> records = mediaServerService.getRecords(app, stream, startTime, endTime, mediaServerItems);
-        PageInfo<RecordFile> pageInfo = new PageInfo<>(records);
-        pageInfo.startPage(page, count);
-        return pageInfo;
+        return cloudRecordService.getList(page, count, app, stream, startTime, endTime, mediaServerItems);
     }